实现本地上传转录、期限、重处理与通知

docs/sessions/261008-local-upload/progress/C-progress.md

@@ -0,0 +1,48 @@
+## 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 下载或新处理入口。
+- **下一步唯一动作**：提交并推送该单元，然后继续生命周期与终态期限实现。
+
+## 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 / 独立审查。
+
+## 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 红，再修生产交接顺序。
+
+## 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。
+
+## 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 验证。
+- **本段结论**：H0 真实消费者 `NoModelWork.process` 观察到 `source_exists=True`；修正把上传 CapsWriter 输出放进任务临时目录，并在共享 handoff helper 的 CALIBRATING/queue publish 前执行现有幂等清理。
+- **关键决策与已否决方案**：不等队列消费后再清理、不加锁/等待器/生产状态；只有 `platform=local_upload` 走提前清理，URL 链路保持原顺序。
+- **下一步唯一动作**：提交推送此最小顺序修正，然后用新空 scratch 验证 H1 与反向顺序变异，并在 scratch 跑全量 make test。

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 路径保持一致；正文资格和任务进度分离。

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

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）。