From 35affe062ff473adcc4f8b0c96e86428b0828c40 Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Fri, 9 Oct 2026 02:22:31 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E4=B8=8A=E4=BC=A0=20worker=20=E5=8D=95=E5=85=83=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent-Executor: codex-pi Agent-Model: gpt-6-luna Agent-Effort: xhigh Dispatch-Id: dlg-20261008-181551-f0aaf6 Task-Id: VideoTranscriptAPI-20261009-05 --- .../progress/C-progress.md | 6 + tests/unit/test_upload_worker.py | 107 ++++++++++++++++++ 2 files changed, 113 insertions(+) create mode 100644 docs/sessions/261008-local-upload/progress/C-progress.md create mode 100644 tests/unit/test_upload_worker.py diff --git a/docs/sessions/261008-local-upload/progress/C-progress.md b/docs/sessions/261008-local-upload/progress/C-progress.md new file mode 100644 index 00000000..43d66792 --- /dev/null +++ b/docs/sessions/261008-local-upload/progress/C-progress.md @@ -0,0 +1,6 @@ +## 2026-10-09 · 首个 worker 单元 + +- **当前阶段**：implementing；首个独立 worker 单元已通过，尚未提交其余 C 增量。 +- **本段结论**：新增真实 dispatcher→持久 admission/media mapping→共享转录函数单元，断言映射文件字节、任务 id、媒体 id 和处理选项；`uv run --frozen pytest -q tests/unit/test_upload_worker.py` 通过。 +- **关键决策与已否决方案**：转录输入严格由 SQLite 映射提供，队列 URL 只作展示值；沿用 B 已持久化的处理选项，不加 URL 下载或新处理入口。 +- **下一步唯一动作**：提交并推送该单元，然后继续生命周期与终态期限实现。 diff --git a/tests/unit/test_upload_worker.py b/tests/unit/test_upload_worker.py new file mode 100644 index 00000000..df91ddd1 --- /dev/null +++ b/tests/unit/test_upload_worker.py @@ -0,0 +1,107 @@ +"""Contracts for local uploads entering the shared transcription dispatcher.""" + +import asyncio +import concurrent.futures +import datetime +import uuid +from unittest.mock import MagicMock + +from video_transcript_api.api.services import transcription +from video_transcript_api.cache.cache_manager import CacheManager +from video_transcript_api.utils.tempfile_manager import TempFileManager + + +def _key(): + now = datetime.datetime.now(datetime.timezone.utc) + return f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}" + + +def test_dispatcher_consumes_persisted_media_mapping_and_processing_options( + tmp_path, monkeypatch +): + cache = CacheManager(str(tmp_path / "cache")) + intent = cache.register_local_upload( + owner_user_id="alice", idempotency_key=_key(), retention="30d" + ) + task_id = cache.generate_task_id() + temp_manager = TempFileManager(str(tmp_path / "temp")) + task_dir = temp_manager.create_task_dir(task_id) + media_path = task_dir / "upload-source.bin" + media_bytes = b"actual upload media bytes" + media_path.write_bytes(media_bytes) + options = { + "calibrate": True, + "summarize": False, + "infer_speaker_names": False, + "chapters": False, + } + upload = cache.accept_local_upload( + intent["upload_id"], + media_id=f"upload_{uuid.uuid4().hex}", + task_id=task_id, + filename="clip.mp4", + title="Owner title", + source_url="", + request_metadata={"filename": "clip.mp4", "title": "Owner title"}, + processing_options=options, + media_path=str(media_path), + byte_size=len(media_bytes), + sha256="a" * 64, + ) + queue = asyncio.Queue() + queue.put_nowait( + { + "id": task_id, + "url": "https://display-only.invalid/not-a-download-source", + "platform": "local_upload", + "processing_options": options, + } + ) + runtime = MagicMock() + submitted = asyncio.Event() + executor = MagicMock() + executor.submit.side_effect = lambda function, *args: ( + submitted.set() or concurrent.futures.Future() + ) + process_calls = [] + + monkeypatch.setattr(transcription, "cache_manager", cache) + monkeypatch.setattr(transcription, "task_queue", queue) + monkeypatch.setattr(transcription, "executor", executor) + monkeypatch.setattr(transcription, "get_runtime", lambda: runtime) + monkeypatch.setattr(transcription, "get_temp_manager", lambda: temp_manager) + monkeypatch.setattr( + transcription, + "process_transcription", + lambda *args, **kwargs: process_calls.append((args, kwargs)), + ) + monkeypatch.setattr( + transcription, + "get_notification_router", + lambda: MagicMock(), + ) + + async def dispatch_one(): + dispatcher = asyncio.create_task(transcription.process_task_queue()) + await asyncio.wait_for(submitted.wait(), timeout=2) + submitted_args = executor.submit.call_args.args + function, args = submitted_args[0], submitted_args[1:] + function(*args) + dispatcher.cancel() + try: + await dispatcher + except asyncio.CancelledError: + pass + + asyncio.run(dispatch_one()) + + assert upload["root_task_id"] == task_id + assert len(process_calls) == 1 + args, kwargs = process_calls[0] + assert args[0] == task_id + assert args[1] == "" + assert kwargs["local_media_path"] == str(media_path.resolve()) + assert kwargs["local_media_id"] == upload["media_id"] + assert kwargs["processing_options"] == options + assert media_path.read_bytes() == media_bytes + assert queue._unfinished_tasks == 0 From 99780aff6b8dc567146b6a61e402007423d89db8 Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Fri, 9 Oct 2026 03:36:42 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E6=8E=A5=E9=80=9A=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E8=BD=AC=E5=BD=95=E4=B8=8E=E5=88=86=E4=BA=AB?= =?UTF-8?q?=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent-Executor: codex-pi Agent-Model: gpt-6-luna Agent-Effort: xhigh Dispatch-Id: dlg-20261008-181551-f0aaf6 Task-Id: VideoTranscriptAPI-20261009-05 --- .../progress/C-progress.md | 14 + goals/local-upload/M3-worker.md | 12 +- src/video_transcript_api/api/context.py | 6 +- src/video_transcript_api/api/routes/audit.py | 120 +++- src/video_transcript_api/api/routes/tasks.py | 107 ++- .../api/routes/uploads.py | 37 +- .../api/services/terminal_status.py | 33 +- .../api/services/transcription.py | 61 +- .../cache/cache_manager.py | 94 ++- .../utils/notifications/completion_share.py | 7 +- .../utils/notifications/task_formatter.py | 22 + .../utils/tempfile_manager.py | 13 +- tests/README.md | 8 +- .../test_local_upload_lifecycle.py | 646 ++++++++++++++++++ tests/unit/test_local_upload_policy.py | 101 ++- tests/unit/test_notification_e2e_delivery.py | 100 +++ tests/unit/test_upload_routes.py | 3 +- tests/unit/test_upload_worker.py | 60 ++ 18 files changed, 1363 insertions(+), 81 deletions(-) create mode 100644 src/video_transcript_api/utils/notifications/task_formatter.py create mode 100644 tests/integration/test_local_upload_lifecycle.py diff --git a/docs/sessions/261008-local-upload/progress/C-progress.md b/docs/sessions/261008-local-upload/progress/C-progress.md index 43d66792..c35142b7 100644 --- a/docs/sessions/261008-local-upload/progress/C-progress.md +++ b/docs/sessions/261008-local-upload/progress/C-progress.md @@ -4,3 +4,17 @@ - **本段结论**：新增真实 dispatcher→持久 admission/media mapping→共享转录函数单元，断言映射文件字节、任务 id、媒体 id 和处理选项；`uv run --frozen pytest -q tests/unit/test_upload_worker.py` 通过。 - **关键决策与已否决方案**：转录输入严格由 SQLite 映射提供，队列 URL 只作展示值；沿用 B 已持久化的处理选项，不加 URL 下载或新处理入口。 - **下一步唯一动作**：提交并推送该单元，然后继续生命周期与终态期限实现。 + +## 2026-10-09 · worker 生命周期、终态与公开消费 + +- **当前阶段**：implementing；本地上传 ASR/LLM 路径、终态锚点、读取/重处理、历史与通知生产者已接通并有真实 SQLite/dispatcher 测试，尚未全量验收。 +- **本段结论**：root 首次 success/failed/启动孤儿失败固定写入 `terminal_at`/期限；真实 `process_task_queue → process_transcription → Transcriber → LLM` fixture 断言实际文件字节、映射、选项、原媒体在 LLM 前删除、公开成果与最终通知，三重处理路由与读写关闭顺序也被集成测试锁定。 +- **关键决策与已否决方案**：保留 B 的持久 media-path 字段契约，仅删除源文件与任务临时目录；child 旧 token 列恒为空，授权在 `BEGIN IMMEDIATE` 中随 child INSERT 再核验，撤销不取消已受理工作或移除已发通知。 +- **下一步唯一动作**：运行任务卡 Narrow-Verify 与 `make test`，检查差异后提交推送并交 lead 接 CI。 + +## 2026-10-09 · 本地验收完成 + +- **当前阶段**：implementing；Narrow-Verify 37 passed，`make test` 全量入口通过（3 skipped）；本地裸 shell 与 dispatch 专属 transient systemd unit 的真实子进程环境探针通过，Hosted CI 由 lead 后续取证。 +- **本段结论**：HTTP upload accepted 通知、终态补发到最终 HTTP JSON、三条 owner 重处理路由、全部 public body/summary 出口、历史过滤/期限原因、terminal clock write-once 与真实 worker/ASR 接线均由新增测试覆盖；当前分支待提交推送。 +- **关键决策与已否决方案**：不改已合并 B 的持久 admission/media-path 契约；public token 仅用于上传记录和临时投递副本，root/child task 与 audit alias 仍为空。 +- **下一步唯一动作**：提交并推送本卡分支的 C 实现，保持 PR draft，由 lead 接管 hosted CI / 独立审查。 diff --git a/goals/local-upload/M3-worker.md b/goals/local-upload/M3-worker.md index 1fbdf2d9..23bca1b0 100644 --- a/goals/local-upload/M3-worker.md +++ b/goals/local-upload/M3-worker.md @@ -2,8 +2,8 @@ lane: local-upload id: M3 slug: worker -status: 未开始 -owner: delegate implementer dlg-20261008-094534-0fd910 +status: 已完成 +owner: delegate implementer dlg-20261008-181551-f0aaf6 order: 3 priority: 高 depends_on: [local-upload/M1, local-upload/M2] @@ -16,8 +16,8 @@ merged_pr: null - **当前范围**：C 卡；不另建 ASR/LLM/通知状态机，不实现第二套重试/预算，不把公开 upload token 写入旧 `task_status.view_token`。 - **关键决策**：复用现有处理链；root首个 success/failed 起算固定期限，子任务/重启不续期；原媒体及中间文件仍按既有 finally 清理。 - **推进前必须拿到的证据**： - - [ ] 真实 `process_transcription` 消费临时文件与 SQLite root；只替换 ASR/LLM/通知外部边缘。保留 producer fixture 并断言实际 ASR 参数/argv、文件字节、root id、终态与通知 payload。 - - [ ] success/failure/invalid result 均有诚实终态；stop-share 不取消已接受处理、不专属删正文；重启/迟到写不能改变 root TTL 或复活撤销。 - - [ ] queue、数据库、ASR/LLM 失败通过具名状态/日志可见；不吞错、不假成功、不自动重试或切换引擎。 - - [ ] 在 CI、裸 shell、获准 systemd 隔离 unit 测真实环境变量和 worker 启动消费；不碰生产配置/数据。 + - [x] `tests/integration/test_local_upload_lifecycle.py` 通过真实 dispatcher、共享 `process_transcription`、实际 `Transcriber` 与隔离 CapsWriter 文件 producer，断言映射路径/字节、时长、选项、root/终态、public result、通知回执和 LLM 前清理；URL parser/downloader 被断言不可达。 + - [x] `tests/unit/test_capswriter_contract.py` 锁 invalid producer contract；`test_upload_worker.py` 锁具名失败状态；终态成功/失败/启动孤儿、子任务迟到写、撤销/到期与受理竞态均由 SQLite/HTTP 消费者测试覆盖。 + - [x] queue/数据库/ASR 失败以现有 worker 失败与 `UPLOAD_PROCESSING_FAILED` 显式可见；临时目录删除失败不再被 local-upload 路径吞掉；未新增重试、fallback 或第二状态机。 + - [x] 环境测试在裸 shell（变量未设）及 dispatch 专属 transient systemd unit（显式 false，测试同时通过真子进程验证缺省拒读/显式启用）；未触碰生产配置/服务。Hosted CI 结论由 lead 按任务卡取证，draft 状态不计作门禁通过。 - **完成条件**：真实处理者消费的是接收器写出的媒体与选项，阶段与 URL 路径保持一致；正文资格和任务进度分离。 diff --git a/src/video_transcript_api/api/context.py b/src/video_transcript_api/api/context.py index a4fff4b0..aa587b7e 100644 --- a/src/video_transcript_api/api/context.py +++ b/src/video_transcript_api/api/context.py @@ -1046,12 +1046,14 @@ def reserve_upload_temp( import shutil with self._upload_budget_guard: - current = sum(self._upload_reserved_bytes.values()) + previous = self._upload_reserved_bytes.get(task_id, 0) + current = sum(self._upload_reserved_bytes.values()) - previous if current + byte_count > budget_bytes: return False free_bytes = shutil.disk_usage(temp_dir).free safety_margin = max(1024 * 1024, budget_bytes // 20) - if free_bytes < byte_count + safety_margin: + additional_bytes = max(0, byte_count - previous) + if free_bytes < additional_bytes + safety_margin: return False self._upload_reserved_bytes[task_id] = byte_count return True diff --git a/src/video_transcript_api/api/routes/audit.py b/src/video_transcript_api/api/routes/audit.py index 4b322cfa..a7c3257a 100644 --- a/src/video_transcript_api/api/routes/audit.py +++ b/src/video_transcript_api/api/routes/audit.py @@ -250,6 +250,7 @@ async def get_history( author: Optional[str] = Query(None, description="频道/作者过滤，支持逗号分隔多选"), q: Optional[str] = Query(None, description="关键词搜索：匹配标题、频道名、视频URL"), status: Optional[str] = Query(None, description="任务状态过滤，默认只显示 success（已完成）"), + source: Optional[str] = Query(None, description="来源过滤: upload/url"), limit: int = Query(20, ge=1, le=10000, description="每页条数，客户端已读过滤时传大值"), offset: int = Query(0, ge=0, description="分页偏移"), user_info: dict = Depends(verify_token), @@ -262,6 +263,108 @@ async def get_history( api_key = user_info.get("api_key", "") api_key_masked = audit_logger._mask_api_key(api_key) user_id = user_info.get("user_id") + if source not in (None, "upload", "url"): + raise HTTPException(status_code=400, detail="source 仅支持 upload 或 url") + if source == "upload": + effective_status = _normalize_history_status(status) + + def _run_upload_query(): + cache = get_cache_manager() + conditions = ["u.owner_user_id = ?", "u.state = 'accepted'"] + params = [user_id] + if platform and platform != "local_upload": + conditions.append("0") + if webhook or (author and author.strip()): + conditions.append("0") + timestamp = "COALESCE(t.completed_at, t.created_at, u.created_at)" + if start_date: + conditions.append(f"{timestamp} >= ?") + params.append(f"{start_date} 00:00:00") + if end_date: + conditions.append(f"{timestamp} <= ?") + params.append(f"{end_date} 23:59:59") + if effective_status != "all": + conditions.append("t.status = ?") + params.append(effective_status) + if q and q.strip(): + pattern = f"%{q.strip()}%" + conditions.append( + "(COALESCE(u.title, '') LIKE ? OR COALESCE(u.filename, '') LIKE ? " + "OR COALESCE(u.source_url, '') LIKE ?)" + ) + params.extend([pattern, pattern, pattern]) + where_clause = " AND ".join(conditions) + with cache._get_cursor() as cursor: + cursor.execute( + f"""SELECT COUNT(*) FROM local_uploads u + JOIN task_status t ON t.task_id = u.root_task_id + WHERE {where_clause}""", + params, + ) + total = cursor.fetchone()[0] + cursor.execute( + f"""SELECT u.*, t.status AS root_status, + t.title AS root_title, + t.created_at AS root_created_at, + t.completed_at AS root_completed_at + FROM local_uploads u + JOIN task_status t ON t.task_id = u.root_task_id + WHERE {where_clause} + ORDER BY {timestamp} DESC, u.upload_id + LIMIT ? OFFSET ?""", + params + [limit, offset], + ) + rows = [dict(row) for row in cursor.fetchall()] + items = [] + for row in rows: + active = cache.local_upload_share_is_active(row) + if row.get("revoked_at") is not None: + inactive_reason = "revoked" + elif row.get("expires_at") is not None and not active: + inactive_reason = "expired" + else: + inactive_reason = None + items.append({ + "task_id": row["root_task_id"], + "video_url": row.get("source_url") or "", + "wechat_webhook": None, + "request_time": row.get("root_completed_at") or row.get("root_created_at"), + "api_key_masked": api_key_masked, + "view_token": None, + "title": row.get("root_title") or row.get("filename") or "本地上传", + "author": None, + "platform": "local_upload", + "status": row["root_status"], + "calibration_status": None, + "summary_status": None, + "chapters_status": None, + "content_expired": not active, + "source": "upload", + "upload_id": row["upload_id"], + "filename": row.get("filename"), + "source_url": row.get("source_url"), + "retention": row["retention"], + "expires_at": row.get("expires_at"), + "share_active": active, + "share_inactive_reason": inactive_reason, + "revoked_at": row.get("revoked_at"), + }) + return total, items + + try: + total, items = await asyncio.to_thread(_run_upload_query) + except sqlite3.OperationalError as exc: + logger.error("upload history query failed: {}", exc) + raise HTTPException(status_code=503, detail="上传历史查询失败") + except Exception as exc: + logger.exception("upload history query failed: {}", exc) + raise HTTPException(status_code=500, detail=f"上传历史查询失败: {exc}") + return TranscribeResponse( + code=200, + message="获取历史记录成功", + data={"items": items, "total": total, "limit": limit, + "offset": offset, "api_key_masked": api_key_masked}, + ) # 租户边界用 user_id 精确匹配，而非截断后的 api_key_masked（只保留前 # 4/后 4 位，不同 key 长度相同时可能碰撞，会让 A 看到 B 的历史记录—— @@ -764,10 +867,17 @@ async def get_task_summary( user_id = user_info.get("user_id") cache_manager = get_cache_manager() - # 通过 view_token 查任务信息（同步 SQLite 调用，线程池执行） - task_info = await asyncio.to_thread(cache_manager.get_task_by_view_token, view_token) - if not task_info: - raise HTTPException(status_code=404, detail="任务不存在") + # upload_ capability 只从分享记录读取，绝不走旧 task_status.view_token 别名。 + if view_token.startswith("upload_"): + task_info = await asyncio.to_thread( + ViewTokenResolver(cache_manager)._local_upload_task_info, view_token + ) + if not task_info: + raise HTTPException(status_code=404, detail="任务不存在或分享已关闭") + else: + task_info = await asyncio.to_thread(cache_manager.get_task_by_view_token, view_token) + if not task_info: + raise HTTPException(status_code=404, detail="任务不存在") # 校验归属：task_info 中暂无 user_id，通过 audit_log 反查（租户边界用 # user_id，不用可能碰撞的 api_key_masked，见云端 CI codex gate）。 @@ -786,7 +896,7 @@ async def get_task_summary( # NULL），因此天然会落到"有记录但不属于当前调用方"的拒绝分支，行为 # 仍是安全的 fail-closed，不需要特殊分支。 task_id = task_info.get("task_id") - if task_id: + if task_id and not view_token.startswith("upload_"): # 归属判定逻辑（本地 codex review 第 8 轮 K1 抽成模块级 # check_view_token_ownership，供 routes/tasks.py::recalibrate 复用 # 同一套判定——见该函数的完整 docstring）。 diff --git a/src/video_transcript_api/api/routes/tasks.py b/src/video_transcript_api/api/routes/tasks.py index e0f21245..640673c3 100644 --- a/src/video_transcript_api/api/routes/tasks.py +++ b/src/video_transcript_api/api/routes/tasks.py @@ -1,6 +1,7 @@ import asyncio import datetime import json +import os import queue from pathlib import Path @@ -67,27 +68,62 @@ def _is_notes_task_options(value) -> bool: def _find_inflight_notes_task(cursor, view_token: str) -> str | None: - """Find an in-flight notes task using persisted task_status fields.""" - cursor.execute( - """ - SELECT task_id, processing_options - FROM task_status - WHERE view_token = ? - AND status IN (?, ?, ?) - """, - ( - view_token, - TaskStatus.QUEUED, - TaskStatus.PROCESSING, - TaskStatus.CALIBRATING, - ), - ) + """Find an in-flight notes task without grouping uploads by blank alias.""" + if view_token.startswith("upload_"): + cursor.execute( + "SELECT media_id FROM local_uploads WHERE view_token = ?", + (view_token,), + ) + upload = cursor.fetchone() + if upload is None: + return None + cursor.execute( + """SELECT task_id, processing_options FROM task_status + WHERE platform = 'local_upload' AND media_id = ? + AND status IN (?, ?, ?)""", + ( + upload["media_id"], + TaskStatus.QUEUED, + TaskStatus.PROCESSING, + TaskStatus.CALIBRATING, + ), + ) + else: + cursor.execute( + """SELECT task_id, processing_options FROM task_status + WHERE view_token = ? AND status IN (?, ?, ?)""", + ( + view_token, + TaskStatus.QUEUED, + TaskStatus.PROCESSING, + TaskStatus.CALIBRATING, + ), + ) for task_row in cursor.fetchall(): if _is_notes_task_options(task_row["processing_options"]): return task_row["task_id"] return None +def _admit_upload_reprocess(cursor, view_token, user_id, cache_data): + """Lock current upload owner/eligibility with its child task INSERT.""" + if cache_data.get("platform") != "local_upload": + return view_token + if os.environ.get("VTA_UPLOADS_ENABLED", "").strip().lower() != "true": + raise HTTPException(status_code=404, detail="上传分享不可用") + task_info = cache_data.get("task_info") or {} + upload = cache_manager.get_active_local_upload_for_reprocess( + cursor, + view_token=view_token, + owner_user_id=user_id, + root_task_id=task_info.get("task_id"), + media_id=cache_data.get("media_id"), + ) + if upload is None: + raise HTTPException(status_code=404, detail="上传分享不可用或已关闭") + return "" + + def _read_fresh_notes_status(cache_dir: str | None) -> str | None: """Read notes_status directly from the authoritative cache status file.""" if not cache_dir: @@ -614,6 +650,10 @@ async def recalibrate( raise HTTPException(status_code=400, detail="缓存中没有转录数据，无法重新校对") task_info = cache_data.get("task_info", {}) + is_local_upload = cache_data.get("platform") == "local_upload" + if is_local_upload: + with cache_manager._get_cursor() as cursor: + _admit_upload_reprocess(cursor, view_token, user_id, cache_data) # 归属校验（本地 codex review 第 8 轮 K1）：recalibrate 会创建新任务、 # 覆盖共享媒体的校对/说话人产物并消耗 LLM 配额，不能像 GET @@ -630,7 +670,7 @@ async def recalibrate( # 判定逻辑理应有同一套调度方式，不能一个在线程池跑、一个在事件循环里 # 裸跑）。 original_task_id = task_info.get("task_id") - if original_task_id: + if original_task_id and not is_local_upload: try: owned = await asyncio.to_thread( check_view_token_ownership, @@ -686,6 +726,11 @@ async def recalibrate( try: try: with cache_manager._get_cursor() as cursor: + if is_local_upload: + cursor.execute("BEGIN IMMEDIATE") + child_view_token = _admit_upload_reprocess( + cursor, view_token, user_id, cache_data + ) cursor.execute(''' INSERT INTO task_status (task_id, view_token, url, platform, media_id, @@ -693,13 +738,15 @@ async def recalibrate( processing_options, submitted_by) VALUES (?, ?, ?, ?, ?, ?, 'processing', ?, ?, ?, ?) ''', ( - task_id, view_token, task_info.get("url", ""), + task_id, child_view_token, task_info.get("url", ""), platform, media_id, use_speaker_recognition, video_title, author, json.dumps(recalibrate_processing_options, sort_keys=True), user_id, )) logger.info(f"重新校对任务创建成功: {task_id}, view_token: {view_token}") + except HTTPException: + raise except Exception as e: logger.error(f"创建重新校对任务失败: {e}") raise HTTPException(status_code=500, detail=f"创建重新校对任务失败: {e}") @@ -901,13 +948,17 @@ async def resummarize( raise HTTPException(status_code=400, detail="缓存中没有转录数据，无法重新生成总结") task_info = cache_data.get("task_info", {}) + is_local_upload = cache_data.get("platform") == "local_upload" + if is_local_upload: + with cache_manager._get_cursor() as cursor: + _admit_upload_reprocess(cursor, view_token, user_id, cache_data) # 归属校验：与 recalibrate 完全同一套判定（check_view_token_ownership + # asyncio.to_thread，fail-closed）——resummarize 同样会创建新任务、 # 消耗 LLM 配额并覆盖共享媒体的总结产物，不能对公开分享的 view_token # 一律放行。 original_task_id = task_info.get("task_id") - if original_task_id: + if original_task_id and not is_local_upload: try: owned = await asyncio.to_thread( check_view_token_ownership, @@ -971,6 +1022,11 @@ async def resummarize( try: try: with cache_manager._get_cursor() as cursor: + if is_local_upload: + cursor.execute("BEGIN IMMEDIATE") + child_view_token = _admit_upload_reprocess( + cursor, view_token, user_id, cache_data + ) cursor.execute(''' INSERT INTO task_status (task_id, view_token, url, platform, media_id, @@ -978,13 +1034,15 @@ async def resummarize( processing_options, submitted_by) VALUES (?, ?, ?, ?, ?, ?, 'processing', ?, ?, ?, ?) ''', ( - task_id, view_token, task_info.get("url", ""), + task_id, child_view_token, task_info.get("url", ""), platform, media_id, use_speaker_recognition, video_title, author, json.dumps(resummarize_processing_options, sort_keys=True), user_id, )) logger.info(f"重新生成总结任务创建成功: {task_id}, view_token: {view_token}") + except HTTPException: + raise except Exception as e: logger.error(f"创建重新生成总结任务失败: {e}") raise HTTPException(status_code=500, detail=f"创建重新生成总结任务失败: {e}") @@ -1165,8 +1223,12 @@ async def generate_notes( ) task_info = cache_data.get("task_info", {}) + is_local_upload = cache_data.get("platform") == "local_upload" + if is_local_upload: + with cache_manager._get_cursor() as cursor: + _admit_upload_reprocess(cursor, view_token, user_id, cache_data) original_task_id = task_info.get("task_id") - if original_task_id: + if original_task_id and not is_local_upload: try: owned = await asyncio.to_thread( check_view_token_ownership, @@ -1247,6 +1309,9 @@ async def generate_notes( # Serialize check + insert so concurrent requests cannot both # observe "no in-flight notes task" and enqueue duplicates. cursor.execute("BEGIN IMMEDIATE") + child_view_token = _admit_upload_reprocess( + cursor, view_token, user_id, cache_data + ) inflight_notes_task_id = _find_inflight_notes_task( cursor, view_token, @@ -1268,7 +1333,7 @@ async def generate_notes( """, ( task_id, - view_token, + child_view_token, task_info.get("url", ""), platform, media_id, diff --git a/src/video_transcript_api/api/routes/uploads.py b/src/video_transcript_api/api/routes/uploads.py index b6a43a79..fcdf8cf4 100644 --- a/src/video_transcript_api/api/routes/uploads.py +++ b/src/video_transcript_api/api/routes/uploads.py @@ -26,6 +26,7 @@ from ..processing_options import normalize_processing_options from ..services.transcription import verify_token from ...utils.logging import logger +from ...utils.notifications import get_notification_router router = APIRouter(prefix="/api/uploads", tags=["uploads"]) _METADATA_MAX_BYTES = 4096 @@ -338,7 +339,41 @@ async def receive_upload( raise _reject(410, failure_code, "上传幂等key已超过24小时受理截止时间") from exc raise transferred = True - return JSONResponse(_receipt(cache_manager.get_local_upload_by_id(accepted["upload_id"])), status_code=202) + from ...utils.notifications.task_formatter import local_upload_notification_fields + + accepted_record = cache_manager.get_local_upload_by_id(accepted["upload_id"]) + notification_fields = local_upload_notification_fields(accepted_record) + notification_webhooks = { + channel: user_info.get(field) + for channel, field in ( + ("wechat", "wechat_webhook"), + ("feishu", "feishu_webhook"), + ) + if user_info.get(field) + } + try: + notification_result = get_notification_router().send_view_link( + title=notification_fields["title"], + view_token=accepted_record["view_token"], + original_url=( + notification_fields["source_url"] + or notification_fields["source_label"] + ), + task_id=task_id, + webhooks=notification_webhooks, + ) + except Exception: + logger.exception("UPLOAD_ACCEPTED_NOTIFICATION_FAILED task_id={}", task_id) + else: + if not isinstance(notification_result, dict) or not any( + result is True for result in notification_result.values() + ): + logger.error( + "UPLOAD_ACCEPTED_NOTIFICATION_FAILED task_id={} result={}", + task_id, + notification_result, + ) + return JSONResponse(_receipt(accepted_record), status_code=202) except asyncio.QueueFull as exc: failure_code = "queue_full" raise _reject(503, failure_code, "转录队列已满，上传未正式受理") from exc diff --git a/src/video_transcript_api/api/services/terminal_status.py b/src/video_transcript_api/api/services/terminal_status.py index 60f2c20a..f6d6cf67 100644 --- a/src/video_transcript_api/api/services/terminal_status.py +++ b/src/video_transcript_api/api/services/terminal_status.py @@ -163,7 +163,9 @@ def deliver_terminal_notification( error_for_notify = notify_error if notify_error is not None else error_message - task = cache_manager.get_task_by_id(task_id) or {} + task = _prepare_local_upload_notification_task( + cache_manager, cache_manager.get_task_by_id(task_id) or {} + ) display_url = url or task.get("url") or "" if title is None: title = task.get("title") @@ -363,7 +365,9 @@ def deliver_pending_terminal_notifications( sent = 0 for row in rows: task_id = row["task_id"] - task = cache_manager.get_task_by_id(task_id) or {} + task = _prepare_local_upload_notification_task( + cache_manager, cache_manager.get_task_by_id(task_id) or {} + ) notify_status = ( TERMINAL_SUCCESS_STATUS if row["status"] == TaskStatus.SUCCESS @@ -477,11 +481,34 @@ def _compose_dispatcher_error(row: dict, task: dict) -> str: return "\n".join(parts) if parts else None +def _prepare_local_upload_notification_task(cache_manager, task: dict) -> dict: + """Enrich an ephemeral delivery copy; never write aliases back to task/audit.""" + task = dict(task) + if task.get("platform") != "local_upload": + return task + upload = cache_manager.get_local_upload_for_task(task["task_id"]) + if upload is None: + raise ValueError(f"local upload delivery mapping missing: {task['task_id']}") + from ...utils.notifications.task_formatter import local_upload_notification_fields + + fields = local_upload_notification_fields(upload) + task["title"] = fields["title"] + task["url"] = fields["source_url"] or fields["source_label"] + task["source_label"] = fields["source_label"] + task["view_token"] = upload["view_token"] + return task + + def _resolve_view_url(cache_manager, task_id: str, task: Optional[dict] = None) -> Optional[str]: - """Build `{base_url}/view/{view_token}` from the task row.""" + """Build `{base_url}/view/{view_token}` from task or upload identity.""" if task is None: task = cache_manager.get_task_by_id(task_id) or {} token = task.get("view_token") + if not token and task.get("platform") == "local_upload": + upload = cache_manager.get_local_upload_for_task(task_id) + if upload is None: + raise ValueError(f"local upload delivery mapping missing: {task_id}") + token = upload["view_token"] if not token: return None from ...utils.rendering import get_base_url diff --git a/src/video_transcript_api/api/services/transcription.py b/src/video_transcript_api/api/services/transcription.py index e70be79c..9f1af9ed 100644 --- a/src/video_transcript_api/api/services/transcription.py +++ b/src/video_transcript_api/api/services/transcription.py @@ -654,13 +654,17 @@ def run_and_finalize( ) logger.info(f"任务完成: {task_id}") except Exception as exc: + failure_label = ( + "UPLOAD_PROCESSING_FAILED: " + if local_media_path is not None else "" + ) logger.exception( - f"任务处理失败: {task_id}, URL: {url}, 错误: {exc}" + f"{failure_label}任务处理失败: {task_id}, URL: {url}, 错误: {exc}" ) finalize_terminal_status_and_notify( task_id, TaskStatus.FAILED, - error_message=f"转录任务失败: {exc}", + error_message=f"{failure_label}转录任务失败: {exc}", url=url, notify_error=str(exc), channel_name=notification_channel, @@ -1113,6 +1117,12 @@ def _fail_task_and_notify( 键）：CAS 写入成功时为 {"status": "failed", ...}；CAS 返回 False 且既有终态已是 success 时为 {"status": "success", ...}；CAS 写入抛异常则不返回，异常向上传播。""" + if local_media_path is not None: + error_msg = f"UPLOAD_PROCESSING_FAILED: {error_msg}" + logger.error( + "UPLOAD_PROCESSING_FAILED task_id={} reason=processing_failed", + task_id, + ) try: fail_status_written = finalize_terminal_status_and_notify( task_id, @@ -2518,6 +2528,29 @@ def _finalize_presentation_fields( title=video_title, author_name=author, ) + if local_media_path is not None: + temp_dir = get_temp_manager().get_task_dir(task_id) + observed_temp_bytes = sum( + path.stat().st_size + for path in temp_dir.rglob("*") + if path.is_file() + ) + budget_mib = get_config().get("storage", {}).get( + "upload_limits", {} + ).get("upload_temp_budget_mib") + budget_bytes = int(float(budget_mib) * 1024 * 1024) + if not get_runtime().reserve_upload_temp( + task_id, + observed_temp_bytes, + budget_bytes, + str(get_temp_manager().base_dir), + ): + return _fail_task_and_notify( + "上传转录产生的临时文件超过预算（upload_temp_budget）", + title=video_title, + author_name=author, + ) + # 获取转录文本 transcript = transcription_result.get("transcript", "") @@ -2599,12 +2632,18 @@ def _finalize_presentation_fields( # 通知，但缺少收敛日志）；现在显式 try/except 记日志后重新抛出， # 行为不变、可观测性补齐，且与 _handoff_to_llm_stage / # _fail_task_and_notify 两处站点统一。 + failure_message = ( + f"UPLOAD_PROCESSING_FAILED: 转录任务异常: {exc}" + if local_media_path is not None else f"转录任务异常: {exc}" + ) + if local_media_path is not None: + logger.error("UPLOAD_PROCESSING_FAILED task_id={} reason=unhandled_worker_error", task_id) try: failed_status_written = finalize_terminal_status_and_notify( task_id, TaskStatus.FAILED, download_url=download_url, - error_message=f"转录任务异常: {exc}", + error_message=failure_message, url=display_url, notify_error=str(exc), channel_name=notification_channel, @@ -2620,7 +2659,7 @@ def _finalize_presentation_fields( if failed_status_written: return { "status": "failed", - "message": f"转录任务异常: {exc}", + "message": failure_message, "error": str(exc), } @@ -2653,11 +2692,15 @@ def _finalize_presentation_fields( # 临时文件只是转录的输入，转录段结束即可清，不依赖 LLM 阶段终态（codex#9）。 # 失败打 WARNING 不中断主流程；务必 clear_current_task 避免线程复用串号（codex#1）。 try: - temp_manager.clean_up_task(task_id) - except Exception as cleanup_exc: - logger.warning( - f"清理任务临时文件失败（不影响主流程）: {task_id}, 错误: {cleanup_exc}" - ) + if local_media_path is not None: + temp_manager.clean_up_task(task_id) + else: + try: + temp_manager.clean_up_task(task_id) + except Exception as cleanup_exc: + logger.warning( + f"清理任务临时文件失败（不影响主流程）: {task_id}, 错误: {cleanup_exc}" + ) finally: temp_manager.clear_current_task() temp_manager.mark_done(task_id) diff --git a/src/video_transcript_api/cache/cache_manager.py b/src/video_transcript_api/cache/cache_manager.py index 3e47ae71..7afdfecf 100644 --- a/src/video_transcript_api/cache/cache_manager.py +++ b/src/video_transcript_api/cache/cache_manager.py @@ -355,6 +355,7 @@ def _init_database(self): media_id TEXT UNIQUE, view_token TEXT UNIQUE, created_at TEXT NOT NULL, + terminal_at TEXT, expires_at TEXT, revoked_at TEXT, error_code TEXT, @@ -386,6 +387,7 @@ def _init_database(self): ("media_path", "TEXT"), ("byte_size", "INTEGER"), ("sha256", "TEXT"), + ("terminal_at", "TEXT"), ): if column not in upload_columns: cursor.execute( @@ -400,7 +402,17 @@ def _init_database(self): SELECT RAISE(ABORT, 'local upload revocation is write-once'); END ''') - + cursor.execute(''' + CREATE TRIGGER IF NOT EXISTS local_upload_terminal_clock_write_once + BEFORE UPDATE OF terminal_at, expires_at ON local_uploads + WHEN OLD.terminal_at IS NOT NULL AND ( + NEW.terminal_at IS NOT OLD.terminal_at OR + NEW.expires_at IS NOT OLD.expires_at + ) + BEGIN + SELECT RAISE(ABORT, 'local upload terminal clock is write-once'); + END + ''') # 创建索引以提高查询性能 cursor.execute('CREATE INDEX IF NOT EXISTS idx_platform_media_id ON video_cache(platform, media_id)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_url ON video_cache(url)') @@ -426,6 +438,26 @@ def _init_database(self): # 执行数据库迁移 self._migrate_database() + with self._get_cursor() as cursor: + task_columns = { + row[1] for row in cursor.execute("PRAGMA table_info(task_status)").fetchall() + } + if "completed_at" in task_columns: + cursor.execute( + """UPDATE local_uploads + SET terminal_at = ( + SELECT completed_at FROM task_status + WHERE task_id = local_uploads.root_task_id + ), + expires_at = CASE WHEN retention = '30d' THEN ( + SELECT datetime(completed_at, '+30 days') FROM task_status + WHERE task_id = local_uploads.root_task_id + ) ELSE NULL END + WHERE terminal_at IS NULL AND root_task_id IN ( + SELECT task_id FROM task_status + WHERE status IN ('success', 'failed') AND completed_at IS NOT NULL + )""" + ) def _migrate_database(self): """执行数据库迁移""" @@ -877,6 +909,50 @@ def get_admitted_local_upload_by_task( row = cursor.fetchone() return dict(row) if row is not None else None + def get_local_upload_for_task(self, task_id: str) -> Optional[Dict[str, Any]]: + """Resolve delivery metadata by this task's isolated upload media identity.""" + with self._get_cursor() as cursor: + cursor.execute( + """SELECT u.* FROM local_uploads u + JOIN task_status t ON t.platform = 'local_upload' + AND t.media_id = u.media_id + WHERE t.task_id = ? AND u.state = 'accepted'""", + (task_id,), + ) + row = cursor.fetchone() + return dict(row) if row is not None else None + + def get_active_local_upload_for_reprocess( + self, + cursor, + *, + view_token: str, + owner_user_id: str, + root_task_id: str, + media_id: str, + ) -> Optional[Dict[str, Any]]: + """Authorize one upload child inside the caller's immediate transaction.""" + now_text = datetime.datetime.now(datetime.timezone.utc).strftime( + "%Y-%m-%d %H:%M:%S" + ) + cursor.execute( + """SELECT u.* FROM local_uploads u + JOIN task_status root ON root.task_id = u.root_task_id + WHERE u.view_token = ? AND u.owner_user_id = ? + AND u.root_task_id = ? AND u.media_id = ? + AND u.state = 'accepted' AND u.revoked_at IS NULL + AND root.platform = 'local_upload' AND root.media_id = u.media_id + AND root.view_token = '' + AND ((u.retention = 'never' AND u.expires_at IS NULL) + OR (u.retention = '30d' AND ( + u.expires_at > ? OR + (u.expires_at IS NULL AND root.status NOT IN ('success', 'failed')) + )))""", + (view_token, owner_user_id, root_task_id, media_id, now_text), + ) + row = cursor.fetchone() + return dict(row) if row is not None else None + def set_local_upload_error(self, upload_id: str, error_code: str) -> None: with self._get_cursor() as cursor: cursor.execute( @@ -2926,13 +3002,15 @@ def update_task_status(self, task_id: str, status: str, platform: str = None, if updated and status in ['success', 'failed']: cursor.execute( """UPDATE local_uploads - SET expires_at = ( - SELECT datetime(completed_at, '+30 days') - FROM task_status WHERE task_id = ? - ) - WHERE root_task_id = ? AND retention = '30d' - AND expires_at IS NULL""", - (task_id, task_id), + SET terminal_at = ( + SELECT completed_at FROM task_status WHERE task_id = ? + ), + expires_at = CASE WHEN retention = '30d' THEN ( + SELECT datetime(completed_at, '+30 days') + FROM task_status WHERE task_id = ? + ) ELSE NULL END + WHERE root_task_id = ? AND terminal_at IS NULL""", + (task_id, task_id, task_id), ) if not updated: logger.info( diff --git a/src/video_transcript_api/utils/notifications/completion_share.py b/src/video_transcript_api/utils/notifications/completion_share.py index 49647389..ccf77d22 100644 --- a/src/video_transcript_api/utils/notifications/completion_share.py +++ b/src/video_transcript_api/utils/notifications/completion_share.py @@ -64,7 +64,12 @@ def build_completion_share_receipt(task: dict, view_url: str) -> str: ) ] original_url = task.get("url") - if original_url: + source_label = task.get("source_label") + if source_label: + lines.append(f"来源：{source_label}") + if original_url and not source_label: + lines.append(f"原始地址：{original_url}") + elif original_url and source_label and original_url != source_label: lines.append(f"原始地址：{original_url}") if view_url: lines.extend(["", f"总结和校对：{view_url}"]) diff --git a/src/video_transcript_api/utils/notifications/task_formatter.py b/src/video_transcript_api/utils/notifications/task_formatter.py new file mode 100644 index 00000000..eb502f26 --- /dev/null +++ b/src/video_transcript_api/utils/notifications/task_formatter.py @@ -0,0 +1,22 @@ +"""Notification fields for local-upload tasks; only delivery payloads get enriched.""" + +import json + + +LOCAL_UPLOAD_SOURCE_LABEL = "本地上传" + + +def local_upload_notification_fields(upload: dict) -> dict: + """Build safe user-facing fields without exposing server-owned media paths.""" + metadata = json.loads(upload.get("request_metadata") or "{}") + title = ( + metadata.get("title") + or upload.get("title") + or upload.get("filename") + or LOCAL_UPLOAD_SOURCE_LABEL + ) + return { + "title": title, + "source_label": LOCAL_UPLOAD_SOURCE_LABEL, + "source_url": upload.get("source_url") or "", + } diff --git a/src/video_transcript_api/utils/tempfile_manager.py b/src/video_transcript_api/utils/tempfile_manager.py index 28075c4b..4c902fdc 100644 --- a/src/video_transcript_api/utils/tempfile_manager.py +++ b/src/video_transcript_api/utils/tempfile_manager.py @@ -138,14 +138,11 @@ def clean_up_task(self, task_id: str) -> int: freed = 0 if d is not None and d.exists(): freed = self._path_size(d) - try: - shutil.rmtree(d) - logger.info( - f"任务临时文件已清理: task_{task_id} " - f"(释放 {freed / 1024 / 1024:.2f} MB)" - ) - except Exception as e: - logger.warning(f"清理任务临时目录失败: {d}, 错误: {e}") + shutil.rmtree(d) + logger.info( + f"任务临时文件已清理: task_{task_id} " + f"(释放 {freed / 1024 / 1024:.2f} MB)" + ) # 从全局 tracked 列表里移除该任务目录下的条目，避免列表无限增长 if d is not None: diff --git a/tests/README.md b/tests/README.md index 22fde657..c1577266 100644 --- a/tests/README.md +++ b/tests/README.md @@ -46,7 +46,13 @@ uv run pytest tests/deployment 默认门禁会阻断指向非 loopback 地址的 AF_INET/AF_INET6 出站连接；DNS 解析、 loopback 与 Unix 域套接字放行，真实外网测试仍应放在 `tests/manual/`。 -本地上传安全基线：`tests/unit/test_local_upload_policy.py` 用临时 SQLite、真实 `ViewTokenResolver` 与 cache/task cleanup 消费者验证 owner/key 幂等、blank legacy token 隔离、`upload_` token、固定终态期限、write-once 撤销、开关关闭后的拒读与有效成果保护；另启动真实 Python 子进程消费同一 SQLite，验证缺省关闭与显式环境开关。B 接收回归运行： +本地上传安全基线：`tests/unit/test_local_upload_policy.py` 用临时 SQLite、真实 `ViewTokenResolver` 与 cache/task cleanup 消费者验证 owner/key 幂等、blank legacy token 隔离、`upload_` token、固定终态期限、write-once 撤销、开关关闭后的拒读与有效成果保护；另启动真实 Python 子进程消费同一 SQLite，验证缺省关闭与显式环境开关。C 生命周期与真实 worker 回归运行： + +```bash +uv run --frozen pytest -q tests/unit/test_upload_worker.py tests/integration/test_local_upload_lifecycle.py tests/unit/test_notification_e2e_delivery.py tests/unit/test_capswriter_contract.py +``` + +`test_local_upload_lifecycle.py` 以真实 FastAPI/SQLite/CacheManager 路由贯通三种重处理 owner gate、分享撤销与到期、public raw/page/export/summary、过滤后分页和撤销竞态；另用真实 `process_task_queue`、`process_transcription`、`Transcriber` 及隔离 CapsWriter 文件 producer 断言实际输入字节/时长/媒体映射、上传源文件在 LLM 前清理和 public output。`test_upload_worker.py` 锁定 named `UPLOAD_PROCESSING_FAILED` 与 dispatcher 使用持久映射。`test_notification_e2e_delivery.py` 贯通真实 router/channel/FIFO 到最终 HTTP JSON payload，验证 upload capability、用户标题、本地来源标签、完整总结/回执次序且不泄服务器路径。生产 ASR、真实凭据与真实通知均未调用。B 接收回归运行： ```bash uv run --frozen pytest -q tests/unit/test_upload_routes.py tests/unit/test_upload_dispatch.py tests/integration/test_upload_intake.py diff --git a/tests/integration/test_local_upload_lifecycle.py b/tests/integration/test_local_upload_lifecycle.py new file mode 100644 index 00000000..0b745a66 --- /dev/null +++ b/tests/integration/test_local_upload_lifecycle.py @@ -0,0 +1,646 @@ +"""Cross-route local-upload lifecycle contracts backed by real SQLite/cache files.""" + +import asyncio +import base64 +import concurrent.futures +import datetime +import hashlib +import json +import queue +import threading +import uuid +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from video_transcript_api.api.routes import audit as audit_routes +from video_transcript_api.api.routes import tasks as tasks_routes +from video_transcript_api.api.routes import uploads as uploads_routes +from video_transcript_api.api.routes import views as views_routes +from video_transcript_api.api.services import llm_ops, transcription +from video_transcript_api.api.services.transcription import verify_token +from video_transcript_api.api.context import bind_runtime, unbind_runtime +from video_transcript_api.api.services.view_token_resolver import ViewTokenResolver +from video_transcript_api.cache.cache_manager import CacheManager +from video_transcript_api.utils.llm_status import ChaptersStatus, NotesStatus, SummaryStatus +from video_transcript_api.utils.tempfile_manager import TempFileManager + + +OWNER = {"user_id": "alice", "api_key": "sk-alice", "is_legacy": True} + + +@pytest.fixture +def lifecycle_client(tmp_path, monkeypatch): + monkeypatch.setenv("VTA_UPLOADS_ENABLED", "true") + cache = CacheManager(str(tmp_path / "cache")) + user = dict(OWNER) + llm_queue = queue.Queue() + registry = SimpleNamespace( + try_register=lambda *_args: True, + release=lambda *_args: None, + ) + audit_stub = SimpleNamespace( + _mask_api_key=lambda _key: "sk-***", + log_api_call=lambda **_kwargs: True, + ) + monkeypatch.setattr(tasks_routes, "cache_manager", cache) + monkeypatch.setattr(tasks_routes, "user_manager", SimpleNamespace(check_permission=lambda *_args: True)) + monkeypatch.setattr(tasks_routes, "audit_logger", audit_stub) + monkeypatch.setattr(tasks_routes, "config", {}) + monkeypatch.setattr(audit_routes, "get_cache_manager", lambda: cache) + monkeypatch.setattr(audit_routes, "audit_logger", audit_stub) + monkeypatch.setattr(views_routes, "cache_manager", cache) + import video_transcript_api.api.context as context + monkeypatch.setattr(context, "get_llm_queue", lambda: llm_queue) + monkeypatch.setattr(context, "get_inflight_registry", lambda: registry) + + async def current_user(): + return dict(user) + + app = FastAPI() + app.include_router(tasks_routes.router) + app.include_router(audit_routes.router) + app.include_router(views_routes.router) + app.dependency_overrides[verify_token] = current_user + with TestClient(app) as client: + yield client, cache, llm_queue, user + cache.close() + + +def _make_upload( + cache, *, owner="alice", media_name=None, title="上传标题", + include_summary=True, retention="never", +): + now = datetime.datetime.now(datetime.timezone.utc) + key = f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}" + intent = cache.register_local_upload( + owner_user_id=owner, + idempotency_key=key, + retention=retention, + intent_metadata={"filename": "clip.mp4", "title": title}, + ) + media_id = media_name or f"upload_{uuid.uuid4().hex}" + upload = cache.accept_local_upload( + intent["upload_id"], + media_id=media_id, + task_id=cache.generate_task_id(), + filename="clip.mp4", + title=title, + source_url="https://source.example.test/media?id=7", + request_metadata={"filename": "clip.mp4", "title": title}, + media_path="/tmp/never-return-this-path/source.mp4", + byte_size=128, + sha256="c" * 64, + ) + cache.save_cache( + platform="local_upload", + url="https://source.example.test/media?id=7", + media_id=media_id, + use_speaker_recognition=False, + transcript_data="actual transcript body", + transcript_type="capswriter", + title=title, + author="", + description="", + ) + if include_summary: + cache.save_llm_result( + platform="local_upload", + media_id=media_id, + use_speaker_recognition=False, + llm_type="summary", + content="A real public summary.", + ) + cache.save_llm_status( + platform="local_upload", + media_id=media_id, + use_speaker_recognition=False, + summary_status=( + SummaryStatus.GENERATED if include_summary else SummaryStatus.FAILED + ), + chapters_status=ChaptersStatus.GENERATED, + notes_status=NotesStatus.FAILED, + ) + cache.update_task_status( + upload["root_task_id"], + "success", + platform="local_upload", + media_id=media_id, + title=title, + ) + return cache.get_local_upload_by_id(upload["upload_id"]) + + +def _make_expired_upload(cache, *, title): + now = datetime.datetime.now(datetime.timezone.utc) + intent = cache.register_local_upload( + owner_user_id="alice", + idempotency_key=f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}", + retention="30d", + intent_metadata={"filename": "old.mp4", "title": title}, + ) + upload = cache.accept_local_upload( + intent["upload_id"], + media_id=f"upload_{uuid.uuid4().hex}", + task_id=cache.generate_task_id(), + filename="old.mp4", + title=title, + request_metadata={"filename": "old.mp4", "title": title}, + ) + expired_at = "2000-01-01 00:00:00" + with cache._get_cursor() as cursor: + cursor.execute( + "UPDATE task_status SET status = 'failed', completed_at = ? WHERE task_id = ?", + (expired_at, upload["root_task_id"]), + ) + cursor.execute( + "UPDATE local_uploads SET terminal_at = ?, expires_at = ? WHERE upload_id = ?", + (expired_at, expired_at, upload["upload_id"]), + ) + return cache.get_local_upload_by_id(upload["upload_id"]) + + +def test_http_upload_acceptance_notifies_independent_public_capability(lifecycle_client, monkeypatch): + _client, cache, _llm_queue, _user = lifecycle_client + """The actual raw-body producer must put only upload identity on accepted notices.""" + temp_manager = TempFileManager(str(cache.cache_dir.parent / "accepted-temp")) + task_queue = asyncio.Queue(maxsize=2) + registry = SimpleNamespace( + try_register=lambda *_args: True, + release=lambda *_args: None, + ) + notifications = [] + + class ReadyProcessor: + done = lambda _self: False + get_coro = staticmethod(lambda: transcription.process_task_queue) + + runtime = SimpleNamespace( + started=True, + upload_receiver_slots=threading.BoundedSemaphore(1), + reserve_upload_temp=lambda _task, size, budget, _dir: size <= budget, + release_upload_temp=lambda _task: None, + ) + config = { + "storage": { + "temp_dir": str(temp_manager.base_dir), + "upload_limits": { + "max_file_mib": 1, + "max_media_hours": 1, + "receive_concurrency": 1, + "upload_temp_budget_mib": 4, + }, + } + } + monkeypatch.setenv("VTA_UPLOADS_ENABLED", "true") + monkeypatch.setattr(uploads_routes, "get_cache_manager", lambda: cache) + monkeypatch.setattr(uploads_routes, "get_config", lambda: config) + monkeypatch.setattr(uploads_routes, "get_task_queue", lambda: task_queue) + monkeypatch.setattr(uploads_routes, "get_temp_manager", lambda: temp_manager) + monkeypatch.setattr(uploads_routes, "get_inflight_registry", lambda: registry) + monkeypatch.setattr( + uploads_routes, + "get_notification_router", + lambda: SimpleNamespace( + send_view_link=lambda **kwargs: notifications.append(kwargs) or {"test": True} + ), + ) + app = FastAPI() + app.state.runtime = runtime + app.state.queue_processor = ReadyProcessor() + app.include_router(uploads_routes.router) + app.dependency_overrides[verify_token] = lambda: OWNER + with TestClient(app) as client: + raw = b"real accepted producer bytes\x00" + metadata = { + "filename": "user-file.mp4", + "byte_size": len(raw), + "title": "Owner selected title", + "source_url": "https://source.example.test/watch?id=9", + "retention": "30d", + "processing_options": {"calibrate": False, "summarize": False}, + } + key = f"{int(datetime.datetime.now(datetime.timezone.utc).timestamp() * 1000)}-{uuid.uuid4()}" + response = client.post( + "/api/uploads", + content=raw, + headers={ + "Authorization": "Bearer test-owner", + "Content-Type": "application/octet-stream", + "Idempotency-Key": key, + "X-Upload-Metadata": base64.urlsafe_b64encode( + json.dumps(metadata, ensure_ascii=False, separators=(",", ":")).encode() + ).decode().rstrip("="), + }, + ) + assert response.status_code == 202, response.text + receipt = response.json() + assert len(notifications) == 1 + assert notifications[0] == { + "title": "Owner selected title", + "view_token": receipt["view_token"], + "original_url": "https://source.example.test/watch?id=9", + "task_id": receipt["task_id"], + "webhooks": {}, + } + assert notifications[0]["view_token"].startswith("upload_") + assert "user-file.mp4" not in notifications[0]["view_token"] + assert "/accepted-temp/" not in json.dumps(notifications[0]) + assert task_queue.get_nowait()["id"] == receipt["task_id"] + + +def test_upload_reprocess_routes_use_owner_gate_and_never_persist_legacy_token( + lifecycle_client, +): + client, cache, llm_queue, user = lifecycle_client + endpoints = ("/api/recalibrate", "/api/resummarize", "/api/generate_notes") + uploads = [ + _make_upload( + cache, + title=f"Owner title {index}", + include_summary=index != 1, + ) + for index in range(4) + ] + child_ids = [] + + for endpoint, upload in zip(endpoints, uploads): + response = client.post(endpoint, json={"view_token": upload["view_token"]}) + assert response.status_code == 200, response.text + payload = response.json() + assert payload["code"] == 202 + assert payload["data"]["view_token"] == upload["view_token"] + child_id = payload["data"]["task_id"] + child_ids.append(child_id) + child = cache.get_task_by_id(child_id) + assert child["platform"] == "local_upload" + assert child["media_id"] == upload["media_id"] + assert child["view_token"] == "" + assert child["submitted_by"] == "alice" + assert llm_queue.get_nowait()["task_id"] == child_id + + repeated_notes = client.post( + "/api/generate_notes", json={"view_token": uploads[2]["view_token"]} + ) + assert repeated_notes.status_code == 409 + distinct_notes = client.post( + "/api/generate_notes", json={"view_token": uploads[3]["view_token"]} + ) + assert distinct_notes.status_code == 200, distinct_notes.text + distinct_child = cache.get_task_by_id(distinct_notes.json()["data"]["task_id"]) + assert distinct_child["media_id"] == uploads[3]["media_id"] + assert llm_queue.get_nowait()["task_id"] == distinct_child["task_id"] + + user.update(user_id="bob", api_key="sk-bob") + for endpoint, upload in zip(endpoints, uploads): + response = client.post(endpoint, json={"view_token": upload["view_token"]}) + assert response.status_code == 404 + assert llm_queue.empty() + + root_before = cache.get_local_upload_by_id(uploads[0]["upload_id"]) + revoked = cache.revoke_local_upload(uploads[0]["upload_id"]) + cache.update_task_status( + child_ids[0], "success", platform="local_upload", media_id=uploads[0]["media_id"] + ) + cache.update_task_status( + uploads[0]["root_task_id"], "failed", error_message="late root completion" + ) + root_after = cache.get_local_upload_by_id(uploads[0]["upload_id"]) + assert root_after["revoked_at"] == revoked["revoked_at"] + assert root_after["terminal_at"] == root_before["terminal_at"] + assert root_after["expires_at"] == root_before["expires_at"] + assert ViewTokenResolver(cache).get_view_data_by_token(uploads[0]["view_token"]) is None + + +def test_public_read_summary_exports_and_owner_history_close_without_content_bypass( + lifecycle_client, +): + client, cache, _llm_queue, user = lifecycle_client + upload = _make_upload(cache, title="Private owner title") + token = upload["view_token"] + + raw = client.get(f"/view/{token}?raw=transcript") + page = client.get(f"/view/{token}?page=transcript") + export = client.get(f"/export/{token}/transcript") + summary = client.get(f"/api/audit/summary?view_token={token}") + assert raw.status_code == 200 and "actual transcript body" in raw.text + assert page.status_code == 200 and "actual transcript body" in page.text + assert export.status_code == 200 and "actual transcript body" in export.text + assert summary.status_code == 200 + assert summary.json()["data"]["summary"] == "A real public summary." + + history = client.get("/api/audit/history?source=upload&status=all&limit=1&offset=0") + item = history.json()["data"]["items"][0] + assert history.status_code == 200 + assert history.json()["data"]["total"] == 1 + assert item["source"] == "upload" + assert item["upload_id"] == upload["upload_id"] + assert item["share_active"] is True + assert "summary" not in item and "transcript" not in item + assert item["view_token"] is None + + user.update(user_id="bob", api_key="sk-bob") + assert client.get(f"/api/audit/summary?view_token={token}").status_code == 200 + user.update(user_id="alice", api_key="sk-alice") + revoked = cache.revoke_local_upload(upload["upload_id"]) + assert revoked["revoked_at"] + + expired = _make_expired_upload( + cache, title="Expired owner title" + ) + closed_history = client.get("/api/audit/history?source=upload&status=all") + closed_items = closed_history.json()["data"]["items"] + closed_item = next(item for item in closed_items if item["upload_id"] == upload["upload_id"]) + expired_item = next(item for item in closed_items if item["upload_id"] == expired["upload_id"]) + assert closed_item["share_active"] is False + assert closed_item["share_inactive_reason"] == "revoked" + assert closed_item["title"] == "Private owner title" + assert expired_item["share_active"] is False + assert expired_item["share_inactive_reason"] == "expired" + assert "summary" not in closed_item and "transcript" not in closed_item + assert client.get(f"/view/{token}?raw=transcript").status_code == 404 + assert client.get(f"/view/{token}?page=transcript").status_code == 404 + assert client.get(f"/export/{token}/transcript").status_code == 404 + assert client.get(f"/api/audit/summary?view_token={token}").status_code == 404 + for endpoint in ("/api/recalibrate", "/api/resummarize", "/api/generate_notes"): + assert client.post(endpoint, json={"view_token": token}).status_code == 404 + + +def test_reprocess_admission_holds_sqlite_order_against_concurrent_close( + lifecycle_client, monkeypatch, +): + client, cache, llm_queue, _user = lifecycle_client + upload = _make_upload(cache, title="Concurrent close source") + original_admit = tasks_routes._admit_upload_reprocess + admission_paused = threading.Event() + release_admission = threading.Event() + revoke_started = threading.Event() + revoke_done = threading.Event() + call_count = 0 + + def pause_transaction_admission(cursor, view_token, user_id, cache_data): + nonlocal call_count + record = original_admit(cursor, view_token, user_id, cache_data) + call_count += 1 + if call_count == 2: + admission_paused.set() + if not release_admission.wait(timeout=5): + raise TimeoutError("controlled reprocess admission pause expired") + return record + + monkeypatch.setattr( + tasks_routes, "_admit_upload_reprocess", pause_transaction_admission + ) + from concurrent.futures import ThreadPoolExecutor + + executor = ThreadPoolExecutor(max_workers=2) + try: + admission = executor.submit( + client.post, + "/api/recalibrate", + json={"view_token": upload["view_token"]}, + ) + assert admission_paused.wait(timeout=3) + + def close_share(): + revoke_started.set() + try: + cache.revoke_local_upload(upload["upload_id"]) + finally: + revoke_done.set() + + revoke = executor.submit(close_share) + assert revoke_started.wait(timeout=3) + assert revoke_done.wait(timeout=0.1) is False + release_admission.set() + response = admission.result(timeout=3) + revoke.result(timeout=3) + finally: + release_admission.set() + executor.shutdown(wait=True) + + assert response.status_code == 200, response.text + child = cache.get_task_by_id(response.json()["data"]["task_id"]) + assert child["view_token"] == "" + assert child["media_id"] == upload["media_id"] + assert cache.get_local_upload_by_id(upload["upload_id"])["revoked_at"] is not None + assert llm_queue.qsize() == 1 + + +def test_upload_history_filters_owner_before_page_limit(lifecycle_client): + client, cache, _llm_queue, _user = lifecycle_client + older = _make_upload(cache, title="Alice older") + newer = _make_upload(cache, title="Alice newer") + outsider = _make_upload(cache, owner="bob", title="Bob newest") + with cache._get_cursor() as cursor: + for upload, timestamp in ( + (older, "2026-10-01 00:00:00"), + (newer, "2026-10-02 00:00:00"), + (outsider, "2026-10-03 00:00:00"), + ): + cursor.execute( + "UPDATE task_status SET completed_at = ? WHERE task_id = ?", + (timestamp, upload["root_task_id"]), + ) + + response = client.get( + "/api/audit/history?source=upload&status=all&limit=1&offset=0" + ) + data = response.json()["data"] + assert response.status_code == 200 + assert data["total"] == 2 + assert [item["upload_id"] for item in data["items"]] == [newer["upload_id"]] + assert data["items"][0]["title"] == "Alice newer" + mismatched_platform = client.get( + "/api/audit/history?source=upload&platform=youtube&status=all" + ) + assert mismatched_platform.json()["data"]["total"] == 0 + + +def test_dispatcher_uses_real_transcriber_then_cleans_owned_media_before_llm( + tmp_path, monkeypatch, +): + monkeypatch.setenv("VTA_UPLOADS_ENABLED", "true") + cache = CacheManager(str(tmp_path / "cache")) + temp_manager = TempFileManager(str(tmp_path / "temp")) + task_queue = asyncio.Queue() + llm_queue = queue.Queue() + source_bytes = b"owned upload producer fixture bytes\x00\x01" + options = { + "calibrate": False, + "summarize": False, + "infer_speaker_names": False, + "chapters": False, + } + now = datetime.datetime.now(datetime.timezone.utc) + intent = cache.register_local_upload( + owner_user_id="alice", + idempotency_key=f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}", + retention="30d", + intent_metadata={"filename": "fixture.mp4", "title": "Owned title"}, + ) + task_id = cache.generate_task_id() + task_dir = temp_manager.create_task_dir(task_id) + source_path = task_dir / "upload-source.bin" + source_path.write_bytes(source_bytes) + upload = cache.accept_local_upload( + intent["upload_id"], + media_id=f"upload_{uuid.uuid4().hex}", + task_id=task_id, + filename="fixture.mp4", + title="Owned title", + request_metadata={"filename": "fixture.mp4", "title": "Owned title"}, + media_path=str(source_path), + byte_size=len(source_bytes), + sha256=hashlib.sha256(source_bytes).hexdigest(), + processing_options=options, + ) + task_queue.put_nowait( + {"id": task_id, "url": "", "platform": "local_upload", "processing_options": options} + ) + + from video_transcript_api.transcriber import transcriber as transcriber_module + from video_transcript_api.transcriber.transcriber import Transcriber + + monkeypatch.setattr(transcriber_module, "get_workspace_dir", lambda: str(task_dir)) + real_transcriber = Transcriber( + config={"capswriter": {"server_url": "ws://127.0.0.1:6006", "max_retries": 0}} + ) + transcript_bytes = "真实 CapsWriter 输出文本".encode("utf-8") + asr_calls = [] + + def capswriter_fixture(audio_path, *, media_duration=None): + asr_calls.append((str(audio_path), Path(audio_path).read_bytes(), media_duration)) + transcript_path = task_dir / "capswriter-output.txt" + transcript_path.write_bytes(transcript_bytes) + return True, [transcript_path] + + real_transcriber.capswriter_client.transcribe_file = capswriter_fixture + monkeypatch.setattr(transcription, "Transcriber", lambda: real_transcriber) + executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + monkeypatch.setattr(transcription, "cache_manager", cache) + monkeypatch.setattr(transcription, "task_queue", task_queue) + monkeypatch.setattr(transcription, "executor", executor) + monkeypatch.setattr(transcription, "llm_task_queue", llm_queue) + monkeypatch.setattr(transcription, "get_temp_manager", lambda: temp_manager) + monkeypatch.setattr( + transcription, + "get_config", + lambda: {"storage": {"upload_limits": {"max_media_hours": 1, "upload_temp_budget_mib": 8}}}, + ) + monkeypatch.setattr(transcription, "_ensure_audio_track", lambda _path: 60.0) + monkeypatch.setattr( + transcription, + "create_downloader", + lambda *_args, **_kwargs: pytest.fail("local upload reached URL downloader"), + ) + from video_transcript_api.utils.url_parser import URLParser + monkeypatch.setattr( + URLParser, "parse", lambda *_args, **_kwargs: pytest.fail("local upload parsed a URL") + ) + monkeypatch.setattr(transcription.slow_alert, "track_transcription", lambda *_args, **_kwargs: None) + monkeypatch.setattr(transcription.slow_alert, "duration_known", lambda *_args, **_kwargs: None) + monkeypatch.setattr(transcription.slow_alert, "cancel_all", lambda *_args, **_kwargs: None) + + delivered = [] + router = SimpleNamespace( + notify_task_status=lambda **kwargs: delivered.append(kwargs) or {"fixture": True} + ) + monkeypatch.setattr(transcription, "get_notification_router", lambda: router) + monkeypatch.setattr(llm_ops, "cache_manager", cache) + monkeypatch.setattr(llm_ops, "get_notification_router", lambda: router) + monkeypatch.setattr(llm_ops.slow_alert, "track_llm", lambda *_args, **_kwargs: None) + + class NoModelWork: + def process(self, **kwargs): + assert kwargs["skip_calibration"] is True + assert kwargs["skip_summary"] is True + assert kwargs["skip_chapters"] is True + return { + "calibrated_text": "", + "summary_text": None, + "stats": { + "calibration_status": "disabled", + "summary_status": "disabled", + "chapters_status": "disabled", + }, + } + + monkeypatch.setattr(llm_ops, "llm_coordinator", NoModelWork()) + loop_state = {} + from loguru import logger as test_logger + + runtime = SimpleNamespace( + inflight_registry=SimpleNamespace( + release=lambda *_args: None, + register_internal=lambda *_args: None, + ), + reserve_upload_temp=lambda _task, size, budget, _dir: size <= budget, + release_upload_temp=lambda _task: None, + track_future=None, + temp_manager=temp_manager, + executor=executor, + llm_queue=llm_queue, + config={"llm": {"structured_calibration_for_plain": False}}, + logger=test_logger, + ) + + async def dispatch_and_wait(): + loop_state["loop"] = asyncio.get_running_loop() + future_done = asyncio.Event() + + def track_future(future, *, kind="transcription", task_id=None): + runtime.future = future + future.add_done_callback( + lambda _finished: loop_state["loop"].call_soon_threadsafe(future_done.set) + ) + + runtime.track_future = track_future + bound = bind_runtime(runtime) + dispatcher = asyncio.create_task(transcription.process_task_queue()) + try: + await asyncio.wait_for(future_done.wait(), timeout=10) + await asyncio.wait_for(asyncio.wrap_future(runtime.future), timeout=10) + finally: + dispatcher.cancel() + try: + await dispatcher + except asyncio.CancelledError: + pass + unbind_runtime(bound) + + try: + asyncio.run(dispatch_and_wait()) + finally: + executor.shutdown(wait=True) + assert len(asr_calls) == 1 + assert asr_calls[0] == (str(source_path), source_bytes, 60.0) + assert not source_path.exists() + assert not task_dir.exists() + assert llm_queue.qsize() == 1 + llm_task = llm_queue.get_nowait() + assert llm_task["task_id"] == task_id + assert llm_task["media_id"] == upload["media_id"] + assert llm_task["processing_options"] == options + + from video_transcript_api.api.context import run_with_runtime + + run_with_runtime(runtime, llm_ops._handle_llm_task, llm_task) + root = cache.get_task_by_id(task_id) + assert root["status"] == "success" + assert root["view_token"] == "" + stored = cache.get_local_upload_by_id(upload["upload_id"]) + assert stored["terminal_at"] == root["completed_at"] + assert stored["expires_at"] is not None + public_view = ViewTokenResolver(cache).get_view_data_by_token(upload["view_token"]) + assert public_view["status"] == "success" + assert public_view["transcript"] == "真实 CapsWriter 输出文本" + assert any( + item.get("view_url", "").endswith(f"/view/{upload['view_token']}") + and item.get("title") == "Owned title" + for item in delivered + ) + cache.close() diff --git a/tests/unit/test_local_upload_policy.py b/tests/unit/test_local_upload_policy.py index 867f23b1..080b57b5 100644 --- a/tests/unit/test_local_upload_policy.py +++ b/tests/unit/test_local_upload_policy.py @@ -16,6 +16,7 @@ from src.video_transcript_api.api.services import view_token_resolver as resolver_module from src.video_transcript_api.api.services.view_token_resolver import ViewTokenResolver from src.video_transcript_api.cache.cache_manager import CacheManager +from src.video_transcript_api.utils.logging.audit_logger import AuditLogger from src.video_transcript_api.utils.tempfile_manager import TempFileManager @@ -74,6 +75,19 @@ def _old_time(days=400): ) +def _freeze_cache_clock(monkeypatch, now): + import src.video_transcript_api.cache.cache_manager as cache_module + + original_datetime = cache_module.datetime.datetime + + class FrozenDateTime(original_datetime): + @classmethod + def now(cls, tz=None): + return now.astimezone(tz) if tz else now.replace(tzinfo=None) + + monkeypatch.setattr(cache_module.datetime, "datetime", FrozenDateTime) + + def test_real_sqlite_upload_identity_owner_key_and_intent_window(cm): now = datetime(2026, 10, 8, 9, 0, tzinfo=timezone.utc) key = _intent_key(now) @@ -225,22 +239,82 @@ def test_root_terminal_sets_fixed_30_day_expiry_for_success_and_failure(cm): stored = cm.get_local_upload_by_id(upload["upload_id"]) with cm._get_cursor() as cursor: cursor.execute( - "SELECT datetime(completed_at, '+30 days') FROM task_status WHERE task_id = ?", + "SELECT completed_at, datetime(completed_at, '+30 days') " + "FROM task_status WHERE task_id = ?", (upload["root_task_id"],), ) - expected = cursor.fetchone()[0] + completed_at, expected = cursor.fetchone() + assert stored["terminal_at"] == completed_at assert stored["expires_at"] == expected # Subsequent task writes, including late updates, cannot extend the # root's original expiry because terminal task state is write-once. + terminal_at = stored["terminal_at"] cm.update_task_status(upload["root_task_id"], "failed", error_message="late") - assert cm.get_local_upload_by_id(upload["upload_id"])["expires_at"] == expected + current = cm.get_local_upload_by_id(upload["upload_id"]) + assert current["terminal_at"] == terminal_at + assert current["expires_at"] == expected + with pytest.raises(sqlite3.IntegrityError, match="terminal clock is write-once"): + with cm._get_cursor() as cursor: + cursor.execute( + "UPDATE local_uploads SET expires_at = '2000-01-01 00:00:00' " + "WHERE upload_id = ?", + (upload["upload_id"],), + ) + + +def test_upload_root_and_child_audit_snapshots_keep_legacy_alias_blank(tmp_path): + manager = CacheManager(str(tmp_path / "cache")) + audit = AuditLogger(str(tmp_path / "audit.db")) + manager.audit_logger = audit + upload = _accepted_upload(manager, retention="never") + _finish_with_transcript(manager, upload) + child_id = manager.generate_task_id() + with manager._get_cursor() as cursor: + cursor.execute( + """INSERT INTO task_status + (task_id, view_token, url, platform, media_id, status, submitted_by) + VALUES (?, '', '', 'local_upload', ?, 'processing', 'alice')""", + (child_id, upload["media_id"]), + ) + manager.update_task_status( + child_id, "success", platform="local_upload", media_id=upload["media_id"] + ) + + connection = sqlite3.connect(audit.db_path) + rows = connection.execute( + "SELECT task_id, view_token FROM task_audit_snapshots WHERE task_id IN (?, ?)", + (upload["root_task_id"], child_id), + ).fetchall() + connection.close() + manager.close() + + assert {row[0] for row in rows} == {upload["root_task_id"], child_id} + assert all(token in (None, "") for _, token in rows) + assert all(token != upload["view_token"] for _, token in rows) + + +def test_startup_orphan_failure_anchors_upload_terminal_clock_once(cm): + upload = _accepted_upload(cm, retention="30d") + + assert cm.recover_orphaned_tasks() == 1 + + root = cm.get_task_by_id(upload["root_task_id"]) + stored = cm.get_local_upload_by_id(upload["upload_id"]) + assert root["status"] == "failed" + assert stored["terminal_at"] == root["completed_at"] + assert stored["expires_at"] is not None + original_terminal_at = stored["terminal_at"] + cm.update_task_status(upload["root_task_id"], "failed", error_message="late recovery") + assert cm.get_local_upload_by_id(upload["upload_id"])["terminal_at"] == original_terminal_at def test_never_expiry_and_revocation_are_write_once(cm): upload = _accepted_upload(cm, retention="never") _finish_with_transcript(cm, upload) - assert cm.get_local_upload_by_id(upload["upload_id"])["expires_at"] is None + stored = cm.get_local_upload_by_id(upload["upload_id"]) + assert stored["terminal_at"] is not None + assert stored["expires_at"] is None first_time = datetime(2026, 10, 8, 9, 0, tzinfo=timezone.utc) later_time = first_time + timedelta(days=3) @@ -336,17 +410,13 @@ def test_resolver_reads_active_upload_and_denies_disabled_revoked_or_expired(cm, cm.revoke_local_upload(long["upload_id"]) assert resolver.get_view_data_by_token(long["view_token"]) is None - with cm._get_cursor() as cursor: - cursor.execute( - "UPDATE local_uploads SET expires_at = ? WHERE upload_id = ?", - (_old_time(31), short["upload_id"]), - ) - assert resolver.get_view_data_by_token(short["view_token"]) is None expired_record = cm.get_local_upload_by_view_token(short["view_token"]) exactly_expired = datetime.strptime( expired_record["expires_at"], "%Y-%m-%d %H:%M:%S" ).replace(tzinfo=timezone.utc) assert cm.local_upload_share_is_active(expired_record, exactly_expired) is False + _freeze_cache_clock(monkeypatch, exactly_expired + timedelta(seconds=1)) + assert resolver.get_view_data_by_token(short["view_token"]) is None def test_cleanup_protects_active_upload_cache_even_when_feature_is_disabled(cm, monkeypatch): @@ -363,22 +433,23 @@ def test_cleanup_protects_active_upload_cache_even_when_feature_is_disabled(cm, assert cm.get_cache("local_upload", upload["media_id"]) is not None -def test_cleanup_reclaims_revoked_or_expired_upload_cache(cm): +def test_cleanup_reclaims_revoked_or_expired_upload_cache(cm, monkeypatch): revoked = _accepted_upload(cm, retention="never") _finish_with_transcript(cm, revoked) cm.revoke_local_upload(revoked["upload_id"]) expired = _accepted_upload(cm, retention="30d") _finish_with_transcript(cm, expired) + expired_record = cm.get_local_upload_by_id(expired["upload_id"]) + expiration = datetime.strptime( + expired_record["expires_at"], "%Y-%m-%d %H:%M:%S" + ).replace(tzinfo=timezone.utc) with cm._get_cursor() as cursor: - cursor.execute( - "UPDATE local_uploads SET expires_at = ? WHERE upload_id = ?", - (_old_time(31), expired["upload_id"]), - ) cursor.execute( "UPDATE video_cache SET updated_at = ? WHERE platform = 'local_upload'", (_old_time(),), ) + _freeze_cache_clock(monkeypatch, expiration + timedelta(seconds=1)) assert cm.cleanup_old_cache(days=1) == 2 assert cm.get_cache("local_upload", revoked["media_id"]) is None diff --git a/tests/unit/test_notification_e2e_delivery.py b/tests/unit/test_notification_e2e_delivery.py index e7e86977..781f9140 100644 --- a/tests/unit/test_notification_e2e_delivery.py +++ b/tests/unit/test_notification_e2e_delivery.py @@ -4,10 +4,12 @@ payload producers are real. Only the final external HTTP POST is replaced. """ +import datetime import json import re import sqlite3 import threading +import uuid import pytest from wecom_notifier import FeishuNotifier, WeComNotifier @@ -194,6 +196,29 @@ def _assert_ordered_summary_receipt(delivery, task_id, original_url, view_token) assert channel_records[-1][1]["card"]["header"]["title"]["content"] == receipt.splitlines()[0] +def test_accepted_upload_link_reaches_final_http_payload_with_public_capability( + delivery, +): + view_token = "upload_public-capability-fixture" + delivery["router"].send_view_link( + title=TITLE, + view_token=view_token, + original_url="本地上传", + task_id="task_abcdef0123456789", + webhooks={"wechat": WECHAT_WEBHOOK, "feishu": FEISHU_WEBHOOK}, + ) + records = _wait_payloads(delivery, 2) + expected_url = f"{BASE_URL}/view/{view_token}" + assert len(records) == 2 + for url, payload in records: + content = _payload_content(url, payload) + assert TITLE in content + assert expected_url in content + assert "本地上传" in content + assert "fakepath" not in content + assert "server-owned" not in content + + @pytest.mark.parametrize("entry_point", ["inline", "replay"]) def test_success_snapshot_reaches_final_http_payloads_in_order( cm, delivery, entry_point, @@ -273,6 +298,81 @@ def get_subtitle_result(self, _url): assert cm.is_terminal_notification_pending(task_id) is False +def test_local_upload_terminal_producer_sends_public_share_url_and_source_label( + cm, delivery, monkeypatch, +): + monkeypatch.setattr(f"{RENDERING_MODULE}.get_base_url", lambda: BASE_URL) + now = datetime.datetime.now(datetime.timezone.utc) + key = f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}" + intent = cm.register_local_upload( + owner_user_id="alice", idempotency_key=key, retention="never", + intent_metadata={"filename": "clip.mp4", "title": TITLE}, + ) + media_id = f"upload_{uuid.uuid4().hex}" + root_task_id = cm.generate_task_id() + upload = cm.accept_local_upload( + intent["upload_id"], + media_id=media_id, + task_id=root_task_id, + filename="clip.mp4", + title=TITLE, + source_url="", + request_metadata={"filename": "clip.mp4", "title": TITLE}, + media_path="/not-in-notification/persistent/source.mp4", + byte_size=42, + sha256="a" * 64, + ) + cm.save_cache( + platform="local_upload", url="", media_id=media_id, + use_speaker_recognition=False, transcript_data="upload transcript", + transcript_type="capswriter", title=TITLE, author="", description="", + ) + child_task_id = cm.generate_task_id() + with cm._get_cursor() as cursor: + cursor.execute( + """INSERT INTO task_status + (task_id, view_token, url, platform, media_id, status, submitted_by) + VALUES (?, '', '', 'local_upload', ?, 'queued', 'alice')""", + (child_task_id, media_id), + ) + cm.update_task_status(root_task_id, TaskStatus.PROCESSING) + cm.update_task_status(child_task_id, TaskStatus.PROCESSING) + assert finalize_terminal_status_and_notify( + child_task_id, + TaskStatus.SUCCESS, + title=TITLE, + terminal_snapshot=SNAPSHOT, + cache_manager=cm, + defer_delivery=True, + ) is True + cm.revoke_local_upload(upload["upload_id"]) + + deliver_pending_terminal_notifications(cm, router=delivery["router"]) + + records = _wait_payloads(delivery, 4) + expected_url = f"{BASE_URL}/view/{upload['view_token']}" + for channel_records in ( + [record for record in records if "qyapi.weixin.qq.com" in record[0]], + [record for record in records if "open.feishu.cn" in record[0]], + ): + contents = [_payload_content(url, payload) for url, payload in channel_records] + assert len(contents) == 2 + full_message, receipt = contents + assert TITLE in full_message and TITLE in receipt + assert expected_url in full_message and expected_url in receipt + assert "本地上传" in receipt + assert PERSISTED_SUMMARY in full_message + assert "第一段含 重点、链接文本 和中文。" in receipt + assert "persistent/source.mp4" not in "\n".join(contents) + assert "upload transcript" not in "\n".join(contents) + assert cm.get_task_by_id(root_task_id)["view_token"] == "" + assert cm.get_task_by_id(child_task_id)["view_token"] == "" + stored = cm.get_local_upload_by_id(upload["upload_id"]) + assert stored["view_token"] == upload["view_token"] + assert stored["terminal_at"] is None + assert stored["expires_at"] is None + + def test_failed_terminal_sends_one_failure_payload_per_channel(cm, delivery): task_id = cm.create_task(url="https://example.test/fail")["task_id"] cm.update_task_status(task_id, TaskStatus.PROCESSING) diff --git a/tests/unit/test_upload_routes.py b/tests/unit/test_upload_routes.py index 42b2c5e1..79888bf4 100644 --- a/tests/unit/test_upload_routes.py +++ b/tests/unit/test_upload_routes.py @@ -153,7 +153,8 @@ def test_cache_manager_adds_upload_mapping_columns_to_existing_a_schema(tmp_path columns = {row[1] for row in connection.execute("PRAGMA table_info(local_uploads)")} connection.close() assert { - "filename", "source_url", "request_metadata", "media_path", "byte_size", "sha256" + "filename", "source_url", "request_metadata", "media_path", "byte_size", + "sha256", "terminal_at", } <= columns diff --git a/tests/unit/test_upload_worker.py b/tests/unit/test_upload_worker.py index df91ddd1..0f8ddf73 100644 --- a/tests/unit/test_upload_worker.py +++ b/tests/unit/test_upload_worker.py @@ -105,3 +105,63 @@ async def dispatch_one(): assert kwargs["processing_options"] == options assert media_path.read_bytes() == media_bytes assert queue._unfinished_tasks == 0 + + +def test_dispatcher_persists_named_local_upload_processing_failure(tmp_path, monkeypatch): + cache = CacheManager(str(tmp_path / "cache")) + intent = cache.register_local_upload( + owner_user_id="alice", idempotency_key=_key(), retention="30d" + ) + task_id = cache.generate_task_id() + temp_manager = TempFileManager(str(tmp_path / "temp")) + media_path = temp_manager.create_task_dir(task_id) / "upload-source.bin" + media_path.write_bytes(b"failed media fixture") + upload = cache.accept_local_upload( + intent["upload_id"], + media_id=f"upload_{uuid.uuid4().hex}", + task_id=task_id, + filename="failed.mp4", + request_metadata={"filename": "failed.mp4"}, + media_path=str(media_path), + byte_size=media_path.stat().st_size, + sha256="b" * 64, + ) + queue = asyncio.Queue() + queue.put_nowait({"id": task_id, "url": "", "platform": "local_upload"}) + runtime = MagicMock() + submitted = asyncio.Event() + executor = MagicMock() + executor.submit.side_effect = lambda *_args, **_kwargs: ( + submitted.set() or concurrent.futures.Future() + ) + router = MagicMock() + router.notify_task_status.return_value = {"fixture": True} + + def failed_process(*_args, **_kwargs): + raise RuntimeError("ASR fixture failure") + + monkeypatch.setattr(transcription, "cache_manager", cache) + monkeypatch.setattr(transcription, "task_queue", queue) + monkeypatch.setattr(transcription, "executor", executor) + monkeypatch.setattr(transcription, "get_runtime", lambda: runtime) + monkeypatch.setattr(transcription, "get_temp_manager", lambda: temp_manager) + monkeypatch.setattr(transcription, "process_transcription", failed_process) + monkeypatch.setattr(transcription, "get_notification_router", lambda: router) + + async def dispatch_one(): + dispatcher = asyncio.create_task(transcription.process_task_queue()) + await asyncio.wait_for(submitted.wait(), timeout=2) + function, *args = executor.submit.call_args.args + function(*args) + dispatcher.cancel() + try: + await dispatcher + except asyncio.CancelledError: + pass + + asyncio.run(dispatch_one()) + + assert cache.get_task_by_id(task_id)["status"] == "failed" + assert "UPLOAD_PROCESSING_FAILED" in cache.get_task_by_id(task_id)["error_message"] + assert upload["root_task_id"] == task_id + assert queue._unfinished_tasks == 0 From cf572c8b528b8ef38bfb233750785cc6af7eafbf Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Fri, 9 Oct 2026 04:23:32 +0800 Subject: [PATCH 3/6] =?UTF-8?q?=E9=94=81=E5=AE=9A=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E5=9C=A8LLM=E5=BC=80=E5=A7=8B=E5=89=8D?= =?UTF-8?q?=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent-Executor: codex-pi Agent-Model: gpt-6-luna Agent-Effort: xhigh Dispatch-Id: dlg-20261008-201417-7fcfa4 Task-Id: VideoTranscriptAPI-20261009-05 --- .../progress/C-progress.md | 7 +++ .../test_local_upload_lifecycle.py | 61 ++++++++++++++++--- 2 files changed, 58 insertions(+), 10 deletions(-) diff --git a/docs/sessions/261008-local-upload/progress/C-progress.md b/docs/sessions/261008-local-upload/progress/C-progress.md index c35142b7..44602e45 100644 --- a/docs/sessions/261008-local-upload/progress/C-progress.md +++ b/docs/sessions/261008-local-upload/progress/C-progress.md @@ -18,3 +18,10 @@ - **本段结论**：HTTP upload accepted 通知、终态补发到最终 HTTP JSON、三条 owner 重处理路由、全部 public body/summary 出口、历史过滤/期限原因、terminal clock write-once 与真实 worker/ASR 接线均由新增测试覆盖；当前分支待提交推送。 - **关键决策与已否决方案**：不改已合并 B 的持久 admission/media-path 契约；public token 仅用于上传记录和临时投递副本，root/child task 与 audit alias 仍为空。 - **下一步唯一动作**：提交并推送本卡分支的 C 实现，保持 PR draft，由 lead 接管 hosted CI / 独立审查。 + +## 2026-10-09 · C-Fix1 并发清理顺序回归 + +- **当前阶段**：implementing；只改既有生命周期集成测试，尚未在原 C 树执行运行时测试。 +- **本段结论**：按 `C-r1-verdict.md` 的真实 producer 脚本，将 dispatcher 生命周期测试改为消费者与 producer 并发，LLM coordinator `process` 入口观察源文件、task dir 与 ASR 输出是否存在；目标断言为进入 LLM 前均不存在。 +- **关键决策与已否决方案**：事件屏障停在真实 `llm_queue.put` 与真实消费者 `process` 入口，不增加生产锁/event/state；Transcriber workspace 与上传 task temp 分开，避免测试替生产环境掩盖 ASR 输出落点。 +- **下一步唯一动作**：提交这条回归测试后，在固定 H0 的新空 scratch 中确认 assertion 红，再修生产交接顺序。 diff --git a/tests/integration/test_local_upload_lifecycle.py b/tests/integration/test_local_upload_lifecycle.py index 0b745a66..1b2f52f7 100644 --- a/tests/integration/test_local_upload_lifecycle.py +++ b/tests/integration/test_local_upload_lifecycle.py @@ -502,20 +502,33 @@ def test_dispatcher_uses_real_transcriber_then_cleans_owned_media_before_llm( {"id": task_id, "url": "", "platform": "local_upload", "processing_options": options} ) + from video_transcript_api.transcriber import capswriter_client as capswriter_module from video_transcript_api.transcriber import transcriber as transcriber_module from video_transcript_api.transcriber.transcriber import Transcriber - monkeypatch.setattr(transcriber_module, "get_workspace_dir", lambda: str(task_dir)) + workspace_dir = tmp_path / "workspace" + monkeypatch.setattr(transcriber_module, "get_workspace_dir", lambda: str(workspace_dir)) + monkeypatch.setattr(capswriter_module.Config, "load_from_project_config", lambda: None) + monkeypatch.setattr( + capswriter_module, + "load_config", + lambda: {"capswriter": {"server_url": "ws://127.0.0.1:6006"}}, + ) real_transcriber = Transcriber( config={"capswriter": {"server_url": "ws://127.0.0.1:6006", "max_retries": 0}} ) transcript_bytes = "真实 CapsWriter 输出文本".encode("utf-8") asr_calls = [] + asr_output_paths = [] + llm_observation = {} + llm_started = threading.Event() + llm_done = threading.Event() def capswriter_fixture(audio_path, *, media_duration=None): asr_calls.append((str(audio_path), Path(audio_path).read_bytes(), media_duration)) - transcript_path = task_dir / "capswriter-output.txt" + transcript_path = Path(real_transcriber.capswriter_client.output_dir) / "capswriter-output.txt" transcript_path.write_bytes(transcript_bytes) + asr_output_paths.append(transcript_path) return True, [transcript_path] real_transcriber.capswriter_client.transcribe_file = capswriter_fixture @@ -556,6 +569,10 @@ def capswriter_fixture(audio_path, *, media_duration=None): class NoModelWork: def process(self, **kwargs): + llm_observation["source_exists"] = source_path.exists() + llm_observation["task_dir_exists"] = task_dir.exists() + llm_observation["asr_outputs_exist"] = [path.exists() for path in asr_output_paths] + llm_started.set() assert kwargs["skip_calibration"] is True assert kwargs["skip_summary"] is True assert kwargs["skip_chapters"] is True @@ -588,6 +605,28 @@ def process(self, **kwargs): logger=test_logger, ) + from video_transcript_api.api.context import run_with_runtime + + original_llm_put = llm_queue.put + + def put_before_cleanup(item, *args, **kwargs): + original_llm_put(item, *args, **kwargs) + if not llm_started.wait(timeout=5): + raise TimeoutError("LLM consumer did not reach process before cleanup") + + llm_queue.put = put_before_cleanup + + def consume_llm_task(): + task = llm_queue.get() + llm_observation["task"] = task + try: + run_with_runtime(runtime, llm_ops._handle_llm_task, task) + finally: + llm_done.set() + + llm_consumer = threading.Thread(target=consume_llm_task, name="upload-llm-consumer") + llm_consumer.start() + async def dispatch_and_wait(): loop_state["loop"] = asyncio.get_running_loop() future_done = asyncio.Event() @@ -614,21 +653,23 @@ def track_future(future, *, kind="transcription", task_id=None): try: asyncio.run(dispatch_and_wait()) + assert llm_done.wait(timeout=5) + llm_consumer.join(timeout=5) finally: executor.shutdown(wait=True) + assert not llm_consumer.is_alive() assert len(asr_calls) == 1 assert asr_calls[0] == (str(source_path), source_bytes, 60.0) - assert not source_path.exists() - assert not task_dir.exists() - assert llm_queue.qsize() == 1 - llm_task = llm_queue.get_nowait() + assert llm_queue.empty() + llm_task = llm_observation["task"] assert llm_task["task_id"] == task_id assert llm_task["media_id"] == upload["media_id"] assert llm_task["processing_options"] == options - - from video_transcript_api.api.context import run_with_runtime - - run_with_runtime(runtime, llm_ops._handle_llm_task, llm_task) + assert llm_observation["source_exists"] is False + assert llm_observation["task_dir_exists"] is False + assert llm_observation["asr_outputs_exist"] == [False] + assert not source_path.exists() + assert not task_dir.exists() root = cache.get_task_by_id(task_id) assert root["status"] == "success" assert root["view_token"] == "" From 3832d442678ebc5339583630386df97295ed19d3 Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Fri, 9 Oct 2026 04:26:30 +0800 Subject: [PATCH 4/6] =?UTF-8?q?=E6=B8=85=E7=90=86=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E5=AA=92=E4=BD=93=E5=90=8E=E5=86=8D=E5=8F=91=E5=B8=83LLM?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent-Executor: codex-pi Agent-Model: gpt-6-luna Agent-Effort: xhigh Dispatch-Id: dlg-20261008-201417-7fcfa4 Task-Id: VideoTranscriptAPI-20261009-05 --- .../261008-local-upload/progress/C-progress.md | 7 +++++++ .../api/services/transcription.py | 13 +++++++++++++ 2 files changed, 20 insertions(+) diff --git a/docs/sessions/261008-local-upload/progress/C-progress.md b/docs/sessions/261008-local-upload/progress/C-progress.md index 44602e45..13a27a42 100644 --- a/docs/sessions/261008-local-upload/progress/C-progress.md +++ b/docs/sessions/261008-local-upload/progress/C-progress.md @@ -25,3 +25,10 @@ - **本段结论**：按 `C-r1-verdict.md` 的真实 producer 脚本，将 dispatcher 生命周期测试改为消费者与 producer 并发，LLM coordinator `process` 入口观察源文件、task dir 与 ASR 输出是否存在；目标断言为进入 LLM 前均不存在。 - **关键决策与已否决方案**：事件屏障停在真实 `llm_queue.put` 与真实消费者 `process` 入口，不增加生产锁/event/state；Transcriber workspace 与上传 task temp 分开，避免测试替生产环境掩盖 ASR 输出落点。 - **下一步唯一动作**：提交这条回归测试后，在固定 H0 的新空 scratch 中确认 assertion 红，再修生产交接顺序。 + +## 2026-10-09 · 修正 LLM handoff 前清理 + +- **当前阶段**：implementing；H0 scratch 中并发消费者已触发真实 `AssertionError`，生产顺序修复已写入但尚未运行 H1 scratch 绿验。 +- **本段结论**：CapsWriter 上传输出重定向到任务临时目录；共享 `_handoff_to_llm_stage` 仅对 `local_upload` 在 CALIBRATING 更新/队列发布前删除整个 source/audio/ASR task dir，保留既有 finally 幂等收尾。 +- **关键决策与已否决方案**：不在 queue put 后等待消费者、不引新锁/event/state/interface；失败清理会在任何 publish 前直接抛出，因此不能让 LLM 先消费再报错。 +- **下一步唯一动作**：将生产顺序修正单独 commit/push，再在新空 scratch 验证 H1 绿、H0 反向顺序变异红，并运行关联 Narrow-Verify。 diff --git a/src/video_transcript_api/api/services/transcription.py b/src/video_transcript_api/api/services/transcription.py index 9f1af9ed..bcbaa491 100644 --- a/src/video_transcript_api/api/services/transcription.py +++ b/src/video_transcript_api/api/services/transcription.py @@ -838,6 +838,11 @@ def _handoff_to_llm_stage( None 表示交接成功；否则返回调用方应直接 return 的 failed 响应 dict （含 "status"/"message" 两个键）。 """ + if llm_payload.get("platform") == "local_upload": + # The LLM queue consumer can start as soon as put() publishes. Retire + # every upload-owned media/ASR artifact before exposing that payload. + get_temp_manager().clean_up_task(task_id) + try: calibrating_written = cache_manager.update_task_status( task_id, TaskStatus.CALIBRATING, **calibrating_status_kwargs, @@ -2480,6 +2485,14 @@ def _finalize_presentation_fields( else: # 使用普通 CapsWriter 转录器 transcriber = Transcriber() + if local_media_path is not None: + local_task_dir = temp_manager.get_task_dir(task_id) + if local_task_dir is None: + raise RuntimeError( + f"local upload task temp directory missing: {task_id}" + ) + transcriber.output_dir = str(local_task_dir) + transcriber.capswriter_client.output_dir = str(local_task_dir) # 使用时间戳作为临时输出基础名 temp_output_base = datetime.datetime.now().strftime( "%y%m%d-%H%M%S" From e642103daf697c02febfdce8acc0614416edd207 Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Fri, 9 Oct 2026 04:30:49 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E6=B8=85=E7=90=86=E6=9C=AC=E5=9C=B0?= =?UTF-8?q?=E5=AA=92=E4=BD=93=E5=90=8E=E5=86=8D=E5=8F=91=E5=B8=83LLM?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent-Executor: codex-pi Agent-Model: gpt-6-luna Agent-Effort: xhigh Dispatch-Id: dlg-20261008-201417-7fcfa4 Task-Id: VideoTranscriptAPI-20261009-05 --- docs/sessions/261008-local-upload/progress/C-progress.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/sessions/261008-local-upload/progress/C-progress.md b/docs/sessions/261008-local-upload/progress/C-progress.md index 13a27a42..0c96fd17 100644 --- a/docs/sessions/261008-local-upload/progress/C-progress.md +++ b/docs/sessions/261008-local-upload/progress/C-progress.md @@ -32,3 +32,10 @@ - **本段结论**：CapsWriter 上传输出重定向到任务临时目录；共享 `_handoff_to_llm_stage` 仅对 `local_upload` 在 CALIBRATING 更新/队列发布前删除整个 source/audio/ASR task dir，保留既有 finally 幂等收尾。 - **关键决策与已否决方案**：不在 queue put 后等待消费者、不引新锁/event/state/interface；失败清理会在任何 publish 前直接抛出，因此不能让 LLM 先消费再报错。 - **下一步唯一动作**：将生产顺序修正单独 commit/push，再在新空 scratch 验证 H1 绿、H0 反向顺序变异红，并运行关联 Narrow-Verify。 + +## 2026-10-09 · H0 红验与最小生产修正 + +- **当前阶段**：implementing；H0 并发探针已在独立空 scratch 中以 `AssertionError` 红，生产修正已落盘，待 push 后做 H1 scratch 验证。 +- **本段结论**：H0 真实消费者 `NoModelWork.process` 观察到 `source_exists=True`；修正把上传 CapsWriter 输出放进任务临时目录，并在共享 handoff helper 的 CALIBRATING/queue publish 前执行现有幂等清理。 +- **关键决策与已否决方案**：不等队列消费后再清理、不加锁/等待器/生产状态；只有 `platform=local_upload` 走提前清理，URL 链路保持原顺序。 +- **下一步唯一动作**：提交推送此最小顺序修正，然后用新空 scratch 验证 H1 与反向顺序变异，并在 scratch 跑全量 make test。 From c216f6f07f90ab34322606ad604919a65e3d78f5 Mon Sep 17 00:00:00 2001 From: zj1123581321 Date: Fri, 9 Oct 2026 04:36:54 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E8=AE=B0=E5=BD=95=20C-Fix1=20scratch=20?= =?UTF-8?q?=E9=AA=8C=E8=AF=81=E8=BF=9B=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Agent-Executor: codex-pi Agent-Model: gpt-6-luna Agent-Effort: xhigh Dispatch-Id: dlg-20261008-201417-7fcfa4 Task-Id: VideoTranscriptAPI-20261009-05 --- docs/sessions/261008-local-upload/progress/C-progress.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/sessions/261008-local-upload/progress/C-progress.md b/docs/sessions/261008-local-upload/progress/C-progress.md index 0c96fd17..96d035f5 100644 --- a/docs/sessions/261008-local-upload/progress/C-progress.md +++ b/docs/sessions/261008-local-upload/progress/C-progress.md @@ -33,6 +33,13 @@ - **关键决策与已否决方案**：不在 queue put 后等待消费者、不引新锁/event/state/interface；失败清理会在任何 publish 前直接抛出，因此不能让 LLM 先消费再报错。 - **下一步唯一动作**：将生产顺序修正单独 commit/push，再在新空 scratch 验证 H1 绿、H0 反向顺序变异红，并运行关联 Narrow-Verify。 +## 2026-10-09 · C-Fix1 隔离复验完成 + +- **当前阶段**：implementing；H1 worker 并发探针、H0 与 H1 顺序变异红验、固定 H1 scratch 全量 `make test` 均已执行；Hosted CI/新独审仍由 lead 接管。 +- **本段结论**：H0 的实际 LLM `process` 入口见源/ASR 文件存在并红；H1 清理提前及 ASR 输出 task-dir 重定向后，同一消费者入口三类产物均不存在。空 scratch 的 `make test` 通过且原 C 树 28 个 ignored data 文件前后 manifest/hash 完全一致。 +- **关键决策与已否决方案**：scratch checkout 预态确认 config/data/logs/.venv 全缺失、源文件 inode 独立；只在这个沙盒允许 Makefile 合成配置/data，不碰原 C 树旧数据或生产设置。 +- **下一步唯一动作**：将本进度条目与 Fix1 实现提交推送至 PR #206，保持 draft，由 lead 组织 H0..H1 独审与 Hosted CI。 + ## 2026-10-09 · H0 红验与最小生产修正 - **当前阶段**：implementing；H0 并发探针已在独立空 scratch 中以 `AssertionError` 红，生产修正已落盘，待 push 后做 H1 scratch 验证。