612 lines
28 KiB
Markdown
612 lines
28 KiB
Markdown
# WP4 韧性与持久化 Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** 修复 LLM 治理栈与持久化层的 10 处韧性缺陷，使训练信号不被缓存重放/截断响应/断连/INFRA 故障污染，且 workspace 元数据崩溃可恢复。
|
||
|
||
**Architecture:** 缓存加 salt 维度让跨 epoch 评估真实重采样（同 epoch 续跑仍命中）；SSE 未收 `[DONE]` 视为截断进重试且不写缓存；扩展瞬时错误清单覆盖断连族；熔断半开只放一个探针；gate 基线臂 INFRA 故障不污染 BaselineCache（算法 #6 保真区，逐行比对）；manifest 补原子写；只读查询不改基线元数据。
|
||
|
||
**Tech Stack:** Python 3.11、httpx、Redis、asyncio、SQLite、pytest、pydantic-settings。
|
||
|
||
**设计源**：`research-wiki/designs/2026-07-16-preflight-fixes-design.md §8`
|
||
|
||
---
|
||
|
||
## 关键锚点（实现前必读）
|
||
|
||
| 用途 | 位置 |
|
||
|------|------|
|
||
| chat 缓存/重试 | `adapters/llm.py:271` chat（get L299 / set L373 / 重试 L336）；`_call_streaming:506`；`_consume_stream:533`（不校验 done）；`_iter_sse_deltas:94`（done 标志 L116）；`_is_transient_error:173-186`；`_SseAnomaly:40` |
|
||
| 缓存键 | `adapters/redis_cache.py:32` `_build_key`（无 salt）；`get:50`；`set:73`（ttl 分支 L89-92） |
|
||
| Protocol | `core/protocols.py:22-28` `LLMProvider.chat` |
|
||
| TTL | `main.py:41` `redis_cache_ttl=86400`；`:93` `ttl_s = ... if >0 else None` |
|
||
| 熔断 | `adapters/breaker.py:24` `is_open`（L36-37 到期即放行，无探针锁）；`record_failure:39`；`record_success:63` |
|
||
| gate INFRA | `app/harness/validate.py:257` `_resolve_baseline_block`（put L303-304）；`_check_infra_guard:384`（累计检查 L593）；`BaselineCache` in `gate_ladder.py:344`（put L384 原子 L401-403） |
|
||
| 持久化 | `app/harness/workspace.py` 4 处非原子 `manifest.json` write_text（WP2 后行号）：`_scaffold:111`/`update_manifest:289`/`record_run:318`/`update_best:372`；原子范式 `checkpoint.py:257-260`。**以 `grep -n 'manifest.json").write_text' app/harness/workspace.py` 现场定位为准** |
|
||
| 基线元数据 | `app/harness/log.py:57` `HarnessLog.__init__`（upsert `_runs` L73-82）；`RunLogImpl._read_table:247`（只读） |
|
||
| dual_metric | `app/harness/observation.py:119` `write_dual_metric`（version_kind final L138）；runner 调用 `runner.py:1444-1448` |
|
||
| 配置 | `.env:65` `REDIS_CACHE_TTL=0`；`.env.example:49` `=86400` |
|
||
|
||
## 核心算法保真校验
|
||
|
||
触及算法 #6（块顺序验证：基线缓存/INFRA 护栏/配对翻转）——Task 6（gate 基线臂 INFRA 隔离）改 `_resolve_baseline_block` 与护栏时序，**必须逐行比对** TRM4 `/home/iomgaa/Projects/Video-Tree-TRM4/core/harness/validate.py` 的 INFRA 护栏语义，确保只改"INFRA unit 不写缓存/不计 W/L + 护栏前置"，不动配对翻转与基线快照复用逻辑。其余 task 不触碰核心算法。
|
||
|
||
---
|
||
|
||
## Task 1: 缓存 cache_salt 贯穿
|
||
|
||
**Files:**
|
||
- Modify: `core/protocols.py:22-28`
|
||
- Modify: `adapters/llm.py:271-373`
|
||
- Modify: `adapters/redis_cache.py:32-92`
|
||
- Modify: VLM 适配转发（`adapters/` 内实现 `LLMProvider`/转发 chat 的类，grep 定位）
|
||
- Test: `tests/unit/test_redis_cache.py`、`tests/unit/test_governed_llm.py`
|
||
|
||
- [ ] **Step 1: 写失败测试（salt 入键）**
|
||
|
||
在 `tests/unit/test_redis_cache.py` 追加：
|
||
```python
|
||
@pytest.mark.asyncio
|
||
async def test_salt_changes_key(fake_redis):
|
||
from adapters.redis_cache import RedisResponseCache
|
||
|
||
cache = RedisResponseCache(redis=fake_redis, ttl_s=None)
|
||
k_none = cache._build_key("m", [{"role": "user", "content": "x"}], None)
|
||
k_e1 = cache._build_key("m", [{"role": "user", "content": "x"}], "run:e1")
|
||
k_e2 = cache._build_key("m", [{"role": "user", "content": "x"}], "run:e2")
|
||
assert k_none != k_e1 != k_e2 and k_e1 != k_e2
|
||
```
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_redis_cache.py::test_salt_changes_key -v`
|
||
Expected: FAIL（`_build_key() takes 3 positional arguments but 4 were given`）。
|
||
|
||
- [ ] **Step 3: redis_cache 加 salt 维度**
|
||
|
||
`adapters/redis_cache.py`：
|
||
- `_build_key(self, model, messages, cache_salt: str | None = None)`：payload 字典加 `"salt": cache_salt`：
|
||
```python
|
||
payload = json.dumps(
|
||
{"model": model, "messages": messages, "salt": cache_salt},
|
||
sort_keys=True,
|
||
ensure_ascii=False,
|
||
)
|
||
```
|
||
- `get(self, model, messages, cache_salt: str | None = None)`：`key = self._build_key(model, messages, cache_salt)`
|
||
- `set(self, model, messages, response, cache_salt: str | None = None)`：同样传 `cache_salt`
|
||
|
||
- [ ] **Step 4: Protocol + chat 透传 salt**
|
||
|
||
`core/protocols.py` `LLMProvider.chat` 签名加 `cache_salt: str | None = None`（keyword-only，放 `parent_call_id` 后）。
|
||
`adapters/llm.py` `chat` 签名加同参；L299 改 `await self._cache.get(self._model, messages, cache_salt)`；L373 改 `await self._cache.set(self._model, messages, response, cache_salt)`。
|
||
grep `adapters/` 找到转发 `chat` 的 VLM/包装类（如有），同步加 `cache_salt` 透传。
|
||
|
||
- [ ] **Step 5: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_redis_cache.py tests/unit/test_governed_llm.py -q`
|
||
Expected: 全 PASS（默认 `cache_salt=None` 保持旧键，既有缓存测试不受影响）。
|
||
|
||
- [ ] **Step 6: 提交**
|
||
|
||
```bash
|
||
git add core/protocols.py adapters/llm.py adapters/redis_cache.py tests/unit/test_redis_cache.py
|
||
git commit -m "feat: add cache_salt dimension to LLM response cache"
|
||
```
|
||
|
||
> 注：训练链路注入 epoch 盐（`f"{run_id}:e{epoch}"`）在 WP3 的推理调用点完成（本 WP 只提供 salt 能力）。
|
||
|
||
---
|
||
|
||
## Task 2: TTL 语义修正（≤0 报错）
|
||
|
||
**Files:**
|
||
- Modify: `main.py:93`
|
||
- Modify: `.env:65`、`.env.example:49`
|
||
- Test: `tests/unit/test_governed_llm.py` 或新增 `tests/unit/test_infra_settings.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
新增 `tests/unit/test_infra_settings.py`：
|
||
```python
|
||
import pytest
|
||
|
||
|
||
def test_redis_cache_ttl_zero_rejected(monkeypatch):
|
||
"""REDIS_CACHE_TTL<=0 必须启动即报错，消灭'0=永不过期'隐式语义。"""
|
||
from main import _resolve_cache_ttl # Step 3 抽出的纯函数
|
||
|
||
with pytest.raises(ValueError, match="REDIS_CACHE_TTL"):
|
||
_resolve_cache_ttl(0)
|
||
with pytest.raises(ValueError, match="REDIS_CACHE_TTL"):
|
||
_resolve_cache_ttl(-1)
|
||
assert _resolve_cache_ttl(86400) == 86400
|
||
```
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_infra_settings.py -v`
|
||
Expected: FAIL（`cannot import name '_resolve_cache_ttl'`）。
|
||
|
||
- [ ] **Step 3: 抽出并实现 fail-loud**
|
||
|
||
`main.py`：新增纯函数并替换 L93 逻辑：
|
||
```python
|
||
def _resolve_cache_ttl(ttl: int) -> int:
|
||
"""校验 Redis 缓存 TTL：必须为正整数（消灭 0=永不过期 的隐式语义）。"""
|
||
if ttl <= 0:
|
||
raise ValueError(
|
||
f"REDIS_CACHE_TTL 必须为正整数秒，实际 {ttl}。"
|
||
"训练场景建议 >= 单次训练时长（如 86400）。"
|
||
)
|
||
return ttl
|
||
```
|
||
L93 改为 `ttl_s = _resolve_cache_ttl(settings.redis_cache_ttl)`（删除 `if >0 else None` 分支；`RedisResponseCache` 的 `ttl_s` 从此恒为正）。
|
||
|
||
- [ ] **Step 4: 改 .env / .env.example**
|
||
|
||
`.env:65` `REDIS_CACHE_TTL=0` → `REDIS_CACHE_TTL=86400`。
|
||
`.env.example:49` 确认为正整数（当前 86400，OK）；补注释 `# 正整数秒，禁止 0（0 会被拒绝启动）`。
|
||
|
||
- [ ] **Step 5: 运行测试确认通过**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_infra_settings.py -v`
|
||
Expected: PASS。
|
||
|
||
- [ ] **Step 6: 提交**
|
||
|
||
```bash
|
||
git add main.py .env.example tests/unit/test_infra_settings.py
|
||
git commit -m "fix: reject REDIS_CACHE_TTL<=0 (kill implicit never-expire)"
|
||
```
|
||
> 注：`.env` 不提交（gitignore）；改动需手动同步到运行环境，runbook 会提示。
|
||
|
||
---
|
||
|
||
## Task 3: SSE 截断检测（未收 [DONE] 即重试，不写缓存）
|
||
|
||
**Files:**
|
||
- Modify: `adapters/llm.py:533-578`（_consume_stream）
|
||
- Test: `tests/unit/test_governed_llm.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_governed_llm.py` 追加：
|
||
```python
|
||
@pytest.mark.asyncio
|
||
async def test_truncated_stream_without_done_raises(_build_client):
|
||
"""SSE 流耗尽但未收 [DONE] → _SseAnomaly（进重试，不当成功）。"""
|
||
from adapters.llm import _SseAnomaly, GovernedLLMClient
|
||
|
||
async def _lines():
|
||
yield 'data: {"choices":[{"delta":{"content":"半"}}]}'
|
||
# 无 data: [DONE] —— 模拟服务端截断
|
||
|
||
client = _build_client() # 依现有 fixture
|
||
with pytest.raises(_SseAnomaly):
|
||
await client._consume_stream(_lines())
|
||
```
|
||
> 依 `test_governed_llm.py` 现有 `_build_client`（L50）/`_FakeRedisCache` 构造；实现前读该文件对齐 client 构造签名。
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_governed_llm.py -k truncated_stream -v`
|
||
Expected: FAIL（当前耗尽即正常返回，不抛）。
|
||
|
||
- [ ] **Step 3: _consume_stream 校验 done**
|
||
|
||
`adapters/llm.py` `_consume_stream` 在 `return content, thinking, ttft_ms, max_inter_token_ms, usage`（L578）之前加：
|
||
```python
|
||
if not usage_sink.get("done"):
|
||
raise _SseAnomaly("truncated_no_done")
|
||
```
|
||
（`_iter_sse_deltas` 收到 `[DONE]` 时置 `usage_sink["done"]=True`，L116；未置说明流被截断。`_SseAnomaly` 已在 `_is_transient_error` L186 归为可重试，故自动进重试梯且不走成功路径、不写缓存。）
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_governed_llm.py tests/unit/test_streaming.py -q`
|
||
Expected: 全 PASS（正常流带 `[DONE]` 的既有测试仍通过；若既有 streaming 测试的假流未含 `[DONE]`，同步补 `data: [DONE]` 终帧）。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add adapters/llm.py tests/unit/test_governed_llm.py
|
||
git commit -m "fix: treat SSE stream without [DONE] as truncated (retry, no cache)"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 4: 瞬时错误清单扩展（断连族）
|
||
|
||
**Files:**
|
||
- Modify: `adapters/llm.py:182-186`
|
||
- Test: `tests/unit/test_governed_llm.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_governed_llm.py` 追加：
|
||
```python
|
||
def test_transient_covers_disconnect_family():
|
||
import httpx
|
||
from adapters.llm import _is_transient_error
|
||
|
||
assert _is_transient_error(httpx.RemoteProtocolError("peer reset"))
|
||
assert _is_transient_error(httpx.ReadError("read"))
|
||
assert _is_transient_error(httpx.ConnectTimeout("ct"))
|
||
assert _is_transient_error(httpx.PoolTimeout("pt"))
|
||
```
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_governed_llm.py -k disconnect_family -v`
|
||
Expected: FAIL（RemoteProtocolError/PoolTimeout 不在当前清单）。
|
||
|
||
- [ ] **Step 3: 扩展 _is_transient_error**
|
||
|
||
`adapters/llm.py` L182-183 的：
|
||
```python
|
||
if isinstance(exc, (httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout)):
|
||
return True
|
||
```
|
||
改为（两族基类覆盖 RemoteProtocolError/ReadError/ConnectTimeout/PoolTimeout 等）：
|
||
```python
|
||
if isinstance(exc, (httpx.TimeoutException, httpx.TransportError)):
|
||
return True
|
||
```
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_governed_llm.py -q`
|
||
Expected: 全 PASS（`httpx.HTTPStatusError` 非 TransportError 子类，401/403 致命分支不受影响）。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add adapters/llm.py tests/unit/test_governed_llm.py
|
||
git commit -m "fix: cover httpx disconnect family in transient error set"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 5: 熔断半开单探针锁
|
||
|
||
**Files:**
|
||
- Modify: `adapters/breaker.py:18-70`
|
||
- Test: `tests/unit/test_breaker.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_breaker.py::TestCircuitBreaker` 追加：
|
||
```python
|
||
def test_half_open_admits_single_probe(self):
|
||
from adapters.breaker import CircuitBreaker
|
||
|
||
b = CircuitBreaker(fail_threshold=2, cooldown_s=10.0)
|
||
b.record_failure("p", now=0.0)
|
||
b.record_failure("p", now=0.0) # 开路至 t=10
|
||
assert b.is_open("p", now=5.0) is True # 冷却中
|
||
# 冷却到期：只放行第一个探针
|
||
assert b.is_open("p", now=11.0) is False # 探针 1 放行
|
||
assert b.is_open("p", now=11.0) is True # 探针 2 被挡（探针在途）
|
||
b.record_success("p") # 探针成功 → 闭合
|
||
assert b.is_open("p", now=12.0) is False
|
||
```
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_breaker.py -k half_open_admits_single -v`
|
||
Expected: FAIL（当前到期后所有调用都返回 False）。
|
||
|
||
- [ ] **Step 3: 实现半开单探针锁**
|
||
|
||
`adapters/breaker.py`：
|
||
- `__init__` 加 `self._half_open_inflight: dict[str, bool] = {}`
|
||
- `is_open` 改为：
|
||
```python
|
||
def is_open(self, source_name: str, now: float) -> bool:
|
||
until = self._open_until.get(source_name)
|
||
if until is None:
|
||
return False
|
||
if now < until:
|
||
return True # 冷却中，全挡
|
||
# 冷却到期：half-open，只放行一个探针
|
||
if self._half_open_inflight.get(source_name):
|
||
return True # 已有探针在途，继续挡
|
||
self._half_open_inflight[source_name] = True
|
||
return False
|
||
```
|
||
- `record_success` 加 `self._half_open_inflight.pop(source_name, None)`（探针成功 → 清在途 + 已有的清 fails/open_until）
|
||
- `record_failure`：探针失败会累计并可能重开路；末尾加 `self._half_open_inflight.pop(source_name, None)`（让下一轮 cooldown 后可再探）
|
||
- `force_open`：加 `self._half_open_inflight.pop(source_name, None)`
|
||
|
||
（is_open 在 asyncio 单线程内同步执行，"检查+标记探针"原子，无竞态。）
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_breaker.py tests/unit/test_governed_llm.py -q`
|
||
Expected: 全 PASS（既有 circuit_open 测试若假设"到期即多次放行"需同步更新为单探针语义）。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add adapters/breaker.py tests/unit/test_breaker.py
|
||
git commit -m "fix: half-open circuit admits single probe (no thundering herd)"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 6: gate 基线臂 INFRA 隔离（算法 #6 保真区）
|
||
|
||
**Files:**
|
||
- Modify: `app/harness/validate.py:257-311`（_resolve_baseline_block）+ 护栏时序
|
||
- Test: `tests/unit/test_harness_validate.py`
|
||
|
||
> **保真前置**：实现前完整读 `app/harness/validate.py:257-311` `_resolve_baseline_block`、`_candidate_correctness_from_db`、块循环（575-600），并逐行比对 TRM4 `core/harness/validate.py` 的 INFRA 护栏语义。目标仅为：① 基线臂 INFRA 故障（stop_reason∈{error,parse_error}）的 unit **不写 BaselineCache**、**不计入 W/L 翻转**；② 护栏检查移到"写缓存之前"。不得改动配对翻转、基线快照复用、unit 折叠逻辑。
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_harness_validate.py` 追加（用 fake run_inference 让某基线 unit 返回 stop_reason=error）：
|
||
```python
|
||
@pytest.mark.asyncio
|
||
async def test_baseline_infra_error_not_cached(tmp_path):
|
||
"""基线臂 INFRA error 的 unit 不写入 BaselineCache（不永久污染）。"""
|
||
# 依现有 test_harness_validate.py 的 fake run_inference / BaselineCache fixture 构造；
|
||
# 让 miss unit u1 的推理返回 prediction=None, stop_reason='error'；
|
||
# 调 _resolve_baseline_block 后断言 baseline_cache.get(..., u1) is None
|
||
...
|
||
```
|
||
> 依 `test_harness_validate.py` 现有 fixture（fake `run_inference`、`BaselineCache`、`HarnessLog`）；实现前读该文件对齐构造，勿臆造签名。
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_validate.py -k baseline_infra_error_not_cached -v`
|
||
Expected: FAIL（当前 INFRA unit 的对错被 put 进缓存）。
|
||
|
||
- [ ] **Step 3: 实现 INFRA 隔离**
|
||
|
||
`_resolve_baseline_block`：miss 跑完后，识别 INFRA 题（从 db 读 stop_reason∈{error,parse_error} 或经 run_inference 结果），其所属 unit：不 `baseline_cache.put`、不计入返回的 `b_units`（从块 unit 集合排除，使配对与 W/L 不含它）。护栏所需的 `errors_inc` 在 put 之前累计并检查（护栏前置）。具体实现按 Step 前"保真前置"要求对齐现场与 TRM4。
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 保真回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_validate.py tests/unit/test_gate_block_unit.py tests/unit/test_gates.py -q`
|
||
Expected: 全 PASS（配对翻转/e-process 既有测试不受影响 = 保真达成）。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add app/harness/validate.py tests/unit/test_harness_validate.py
|
||
git commit -m "fix: isolate gate baseline-arm INFRA errors from BaselineCache (algo #6)"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 7: manifest 原子写
|
||
|
||
**Files:**
|
||
- Modify: `app/harness/workspace.py`（`_scaffold:111`/`update_manifest:282`/`record_run:311`/`update_best:365`）
|
||
- Test: `tests/unit/test_harness_workspace.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_harness_workspace.py` 追加：
|
||
```python
|
||
def test_update_manifest_is_atomic(workspace_dir, monkeypatch):
|
||
"""写 manifest 途中崩溃不产生半截 JSON（原子写：tmp 存在即失败也不损原文件）。"""
|
||
import json
|
||
from app.harness import workspace as ws
|
||
|
||
# 先建合法 manifest
|
||
... # 依现有 fixture 初始化 workspace
|
||
original = (workspace_dir / "manifest.json").read_text()
|
||
|
||
# monkeypatch os.replace 抛异常，模拟替换阶段崩溃
|
||
def _boom(src, dst):
|
||
raise OSError("crash during replace")
|
||
monkeypatch.setattr(ws.os, "replace", _boom)
|
||
with pytest.raises(OSError):
|
||
ws.update_manifest(workspace_dir, skills="skills/v2")
|
||
# 原 manifest 未被破坏
|
||
assert (workspace_dir / "manifest.json").read_text() == original
|
||
assert json.loads((workspace_dir / "manifest.json").read_text())
|
||
```
|
||
> 依 `test_harness_workspace.py` 现有 workspace 初始化 fixture；实现前读对齐。
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_workspace.py -k update_manifest_is_atomic -v`
|
||
Expected: FAIL（当前 write_text 非原子，崩溃留半截）。
|
||
|
||
- [ ] **Step 3: 加原子写 helper 并替换 4 处**
|
||
|
||
`app/harness/workspace.py` 顶部确认 `import os`。新增模块级 helper：
|
||
```python
|
||
def _atomic_write_json(path: Path, data: dict) -> None:
|
||
"""原子写 JSON：tmp + os.replace（对齐 checkpoint.py 范式，防半截损坏）。"""
|
||
tmp = path.with_name(path.name + ".tmp")
|
||
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
```
|
||
把 `_scaffold_workspace`（~L111）、`update_manifest`（~L289）、`record_run`（~L318）、`update_best`（~L372）四处的 `(workspace_dir / "manifest.json").write_text(json.dumps(...))` 替换为 `_atomic_write_json(workspace_dir / "manifest.json", <data>)`（用 `grep -n 'manifest.json").write_text' app/harness/workspace.py` 定位全部 4 处，行号随 WP2 改动漂移）。
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_workspace.py -q`
|
||
Expected: 全 PASS。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add app/harness/workspace.py tests/unit/test_harness_workspace.py
|
||
git commit -m "fix: atomic writes for manifest/record_run/update_best (tmp+replace)"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 8: 只读查询不改基线元数据
|
||
|
||
**Files:**
|
||
- Modify: `app/harness/log.py:57-83`（HarnessLog.__init__ 加 register_run 开关）
|
||
- Modify: 只读查询基线的调用点（`app/harness/runner.py:923` `_init_gate`、`app/harness/pools.py:765,818`）
|
||
- Test: `tests/unit/test_harness_log.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_harness_log.py::TestHarnessLogUpsert` 追加：
|
||
```python
|
||
def test_register_run_false_skips_upsert(self, tmp_path):
|
||
"""register_run=False 时只读打开不改写已有 _runs 行（started_at/status 不变）。"""
|
||
db = str(tmp_path / "h.db")
|
||
from app.harness.log import HarnessLog
|
||
|
||
with HarnessLog(db, "r1") as log:
|
||
log # 初次注册
|
||
row0 = _read_run_row(db, "r1") # 依现有 helper 读 started_at/status
|
||
|
||
with HarnessLog(db, "r1", register_run=False) as log:
|
||
log.query("SELECT 1") # 只读
|
||
row1 = _read_run_row(db, "r1")
|
||
assert row1["started_at"] == row0["started_at"]
|
||
assert row1["status"] == row0["status"]
|
||
```
|
||
> 依 `test_harness_log.py` 现有读行 helper；实现前读 `TestHarnessLogUpsert`（L196）对齐。
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_log.py -k register_run_false -v`
|
||
Expected: FAIL（`unexpected keyword argument 'register_run'`）。
|
||
|
||
- [ ] **Step 3: HarnessLog 加 register_run 开关**
|
||
|
||
`app/harness/log.py` `__init__` 签名加 `register_run: bool = True`（keyword）。把 L73-82 的 `_runs` upsert 包进 `if register_run:`；`register_run=False` 时跳过 upsert（仅 `_init_fixed_tables` 建表 + 连接，供只读查询）。`__exit__`/`close` 的 status 更新同样在 `register_run` 为 True 时才执行（避免只读关闭把 status 改 completed）。
|
||
|
||
- [ ] **Step 4: 只读调用点传 register_run=False**
|
||
|
||
`app/harness/runner.py:923`（_init_gate 用 baseline_run_id 只读查 predictions）、`app/harness/pools.py:765,818`（build_or_load_pools 只读查 predictions）三处 `HarnessLog(...)` 调用加 `register_run=False`。
|
||
|
||
- [ ] **Step 5: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_log.py tests/unit/test_harness_pools.py -q`
|
||
Expected: 全 PASS。
|
||
|
||
- [ ] **Step 6: 提交**
|
||
|
||
```bash
|
||
git add app/harness/log.py app/harness/runner.py app/harness/pools.py tests/unit/test_harness_log.py
|
||
git commit -m "fix: read-only baseline queries skip _runs upsert (register_run flag)"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 9: dual_metric version_kind 口径修正
|
||
|
||
**Files:**
|
||
- Modify: `app/harness/runner.py:1444-1448`（第二次 write_dual_metric）
|
||
- Test: `tests/unit/test_harness_observation.py` 或 `test_harness_runner.py`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
在 `tests/unit/test_harness_observation.py` 追加（验证 slow_candidate 与 final 可区分）：
|
||
```python
|
||
def test_slow_candidate_kind_distinct_from_final(tmp_path):
|
||
from app.harness.observation import write_dual_metric, read_dual_metric
|
||
|
||
db = str(tmp_path / "h.db")
|
||
write_dual_metric(db, run_id="r", epoch=1, version_kind="final",
|
||
skills_version="v1", prompts_version="v1", pool="val",
|
||
hard_acc=0.7, soft_score=None, mixed_score=None)
|
||
write_dual_metric(db, run_id="r", epoch=1, version_kind="slow_candidate",
|
||
skills_version="v2", prompts_version="v2", pool="val",
|
||
hard_acc=0.6, soft_score=None, mixed_score=None)
|
||
rows = read_dual_metric(db, run_id="r", epoch=1)
|
||
kinds = {r["version_kind"] for r in rows}
|
||
assert "final" in kinds and "slow_candidate" in kinds
|
||
```
|
||
> 依 `test_harness_observation.py` 现有 read_dual_metric 签名；实现前读对齐。
|
||
|
||
- [ ] **Step 2: 运行确认失败/通过基线**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_observation.py -k slow_candidate -v`
|
||
Expected: 若 read_dual_metric 已支持则 PASS 基线；重点是 Step 3 改 runner 调用点。
|
||
|
||
- [ ] **Step 3: 改 runner 慢更新第二次写为 slow_candidate**
|
||
|
||
`app/harness/runner.py` L1444-1448 的 `write_dual_metric(..., version_kind="final", ...)`（`_slow_update_cycle` Phase 8 的 R2 行）改为 `version_kind="slow_candidate"`，使被 revert 的慢更新候选不再占用 `final` 语义。确认 Phase 2 的第一次（L1399-1403）仍为 `final`（epoch 终值唯一）。
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_harness_observation.py tests/unit/test_harness_runner.py -q`
|
||
Expected: 全 PASS。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add app/harness/runner.py tests/unit/test_harness_observation.py
|
||
git commit -m "fix: slow-update R2 dual_metric uses slow_candidate kind (not final)"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 10: 离线 --retry-uncertain + docstring 修正
|
||
|
||
**Files:**
|
||
- Modify: `app/harness/video_split_cli.py`（build_arg_parser + run_pipeline）
|
||
- Modify: `app/harness/baseline_diagnosis.py`（done 集排除 uncertain + docstring）
|
||
- Test: `tests/unit/test_video_split_cli.py` 或 `test_baseline_diagnosis`
|
||
|
||
- [ ] **Step 1: 写失败测试**
|
||
|
||
定位 `baseline_diagnosis` 的 done_question_ids 计算函数（`app/harness/baseline_diagnosis.py`，grep `done_question_ids`），追加测试：`retry_uncertain=True` 时 tier=uncertain 的已落库题不计入 done（会被重诊）。
|
||
```python
|
||
def test_retry_uncertain_excludes_uncertain_from_done(...):
|
||
# 构造 baseline_diagnosis 表含 tier=T2 与 tier=uncertain 行；
|
||
# done_ids(retry_uncertain=False) 含 uncertain 题；
|
||
# done_ids(retry_uncertain=True) 不含 uncertain 题
|
||
...
|
||
```
|
||
> 实现前读 `baseline_diagnosis.py` 的 done 集查询函数签名对齐。
|
||
|
||
- [ ] **Step 2: 运行确认失败**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest -k retry_uncertain -v`
|
||
Expected: FAIL（无该参数）。
|
||
|
||
- [ ] **Step 3: 实现 retry_uncertain**
|
||
|
||
`app/harness/baseline_diagnosis.py`：done 集查询函数加 `retry_uncertain: bool = False`，True 时 `WHERE ... AND tier != 'uncertain'`（uncertain 题不算完成，会被重诊）。修正模块 docstring 把"崩溃最多丢正在写的一行/逐行落库"表述改为实际的"run 末批量落库（Phase 3），崩溃丢本次 run 全部未落库结果，靠 Redis 缓存缓解重烧"。
|
||
`app/harness/video_split_cli.py`：`build_arg_parser` 加 `--retry-uncertain`（store_true）；`run_pipeline` 透传到 `run_baseline_diagnosis` 的 done 集计算。
|
||
|
||
- [ ] **Step 4: 运行测试确认通过 + 回归**
|
||
|
||
Run: `conda run -n Video-Tree-TRM python -m pytest tests/unit/test_video_split_cli.py -q`
|
||
Expected: 全 PASS。
|
||
|
||
- [ ] **Step 5: 提交**
|
||
|
||
```bash
|
||
git add app/harness/video_split_cli.py app/harness/baseline_diagnosis.py tests/unit/
|
||
git commit -m "feat: --retry-uncertain re-diagnoses uncertain rows; fix docstring"
|
||
```
|
||
|
||
---
|
||
|
||
## Self-Review（作者自查，执行者复核）
|
||
|
||
- [ ] cache_salt 四处（protocols/llm/redis_cache/VLM 转发）贯通，默认 None 保持旧键。
|
||
- [ ] TTL≤0 fail-loud + .env/.env.example 同步（.env 手动，runbook 提示）。
|
||
- [ ] SSE 未收 [DONE] → _SseAnomaly（已在瞬时清单 → 重试 + 不写缓存）。
|
||
- [ ] 瞬时清单用 TimeoutException+TransportError 两族基类覆盖断连族，不误纳 HTTPStatusError。
|
||
- [ ] 熔断半开单探针：is_open 检查+标记原子（asyncio 单线程）。
|
||
- [ ] Task 6 触及算法 #6，已设保真前置（逐行比对 TRM4，只改 INFRA 不写缓存 + 护栏前置）。
|
||
- [ ] manifest 4 处原子写；只读查询 register_run=False 不改基线元数据。
|
||
|
||
## 核心算法保真校验结论
|
||
|
||
Task 6 触及算法 #6（块顺序验证），已在该 Task 设"保真前置"要求逐行比对 TRM4 `validate.py`，仅改 INFRA 隔离与护栏时序，不动配对翻转/基线快照/unit 折叠。其余 Task 均为治理栈/持久化层，不涉及核心算法。
|
||
|
||
## 验收标准
|
||
|
||
1. `pytest tests/unit/test_redis_cache.py tests/unit/test_governed_llm.py tests/unit/test_breaker.py tests/unit/test_streaming.py tests/unit/test_harness_validate.py tests/unit/test_harness_workspace.py tests/unit/test_harness_log.py tests/unit/test_harness_observation.py tests/unit/test_infra_settings.py` 全绿。
|
||
2. 缓存加 salt 后跨 epoch 键不同、同 epoch 键相同。
|
||
3. 熔断半开只放一个探针；SSE 截断进重试不写缓存。
|
||
4. manifest 原子写；基线 _runs 元数据只读查询不被改写。
|