fix: gate slot cancel-safety + post-inference freeze discard (algo #6)

Codex 质量审 4 项:推理后二次冻结检查(τ 后 in-flight 结果整体丢弃)、
acquire 取消回滚(半持有 permit 自动归还)、BoundedSemaphore 防静默扩容、
补取消恢复与冻结丢弃两个回归测试。

app/harness/validate.py

@@ -976,7 +976,8 @@ class _QuestionSlots:
         """
         assert width > 0, f"并发宽度必须为正: {width}"
         self._width = width
-        self._sem = asyncio.Semaphore(width)
+        # BoundedSemaphore:多还立即 ValueError 而非静默扩容(Codex 质量审 3)
+        self._sem = asyncio.BoundedSemaphore(width)
         self._acquire_lock = asyncio.Lock()
 
     async def acquire(self, n: int) -> None:
@@ -984,6 +985,8 @@ class _QuestionSlots:
 
         fail-fast:n > 宽度时任务持锁等待永不满足的槽位 → 自死锁
         (AR pair 单元 2 题 + width=1 的病态配置,Codex plan 审 C2),直接报错。
+        取消安全:半持有自动回滚——逐槽获取途中被取消(或任何 BaseException)
+        时,已拿到的 permit 全部归还再重抛,容量不泄漏(Codex 质量审 2)。
 
         参数:
             n: 申请的题槽数(单元内题目数,single=1 / AR pair=2)。
@@ -997,8 +1000,15 @@ class _QuestionSlots:
         if n > self._width:
             raise ValueError(f"单次申请题槽 {n} 超过并发宽度 {self._width},将自死锁")
         async with self._acquire_lock:
-            for _ in range(n):
-                await self._sem.acquire()
+            got = 0
+            try:
+                for _ in range(n):
+                    await self._sem.acquire()
+                    got += 1
+            except BaseException:
+                for _ in range(got):
+                    self._sem.release()
+                raise
 
     def release(self, n: int) -> None:
         """归还 n 个题槽。
@@ -1029,7 +1039,8 @@ async def _run_unit_arm(
 ) -> None:
     """执行一个 (单元, 臂) 任务:缓存/推理 → 到达登记 → 前缀消费推进。
 
-    冻结检查两次:启动时(排队任务撤销点)与获得题槽后(获槽期间被冻结)。
+    冻结检查三次:启动时(排队任务撤销点)、获得题槽后(获槽期间被冻结)、
+    推理返回后(τ 之后的 in-flight 结果不计入,整体丢弃)。
     base 臂缓存命中不占题槽(零推理);INFRA 单元不写缓存(不永久污染基线快照)。
     护栏在每次臂完成时检查(等价迁移自跨块累计,设计 v3 §2.3),超阈值 raise
     中止整轮(与现行行为一致)。
@@ -1073,6 +1084,12 @@ async def _run_unit_arm(
         run_id = f"{spec.gate_run_prefix}_{arm}"
         skills_dir = base_skills_dir if arm == "base" else cand_dir
         r = await run_inference(questions, run_id=run_id, skills_dir=skills_dir)
+        # 推理 await 期间该题型可能已被其他任务判定冻结:设计语义是
+        # "τ(冻结时刻)之后的 in-flight 结果不计入"——整体丢弃,不写
+        # slot/infra_denom/errors,滞后 INFRA 也不得触发护栏 raise 掀翻
+        # 整轮 gather(Codex 质量审 1)。
+        if run.frozen:
+            return
         _register_arm_arrival(
             run=run,
             slot=slot,

tests/unit/test_gate_unit_arm.py

@@ -196,3 +196,62 @@ async def test_question_slots_rejects_oversized_request() -> None:
     slots = _QuestionSlots(1)
     with pytest.raises(ValueError, match="自死锁"):
         await slots.acquire(2)
+
+
+@pytest.mark.asyncio
+async def test_acquire_cancellation_restores_capacity() -> None:
+    """acquire 半持有时被取消:已拿 permit 自动回滚,容量完全恢复(Codex 质量审 2)。"""
+    slots = _QuestionSlots(2)
+    await slots.acquire(1)  # 预占 1 槽,使 acquire(2) 卡在第二槽
+    task = asyncio.ensure_future(slots.acquire(2))
+    for _ in range(5):
+        await asyncio.sleep(0)  # 让 task 拿到第 1 个 permit 并阻塞在第 2 个
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    slots.release(1)  # 归还预占
+    # 半持有的 permit 若泄漏,此处 acquire(2) 将永久阻塞 → wait_for 超时暴露泄漏
+    await asyncio.wait_for(slots.acquire(2), timeout=1.0)
+    slots.release(2)
+
+
+@pytest.mark.asyncio
+async def test_frozen_during_inference_discards_result(tmp_path) -> None:
+    """推理 await 期间被冻结:in-flight 结果整体丢弃(不写 slot/计数器/缓存)。
+
+    设计语义:τ(冻结时刻)之后到达的结果不计入,滞后 INFRA 也不得触发护栏。
+    """
+    run, cache = _mk_gate_run(1, tmp_path)
+    log = _FakeLog()
+    gate_open = asyncio.Event()
+
+    async def _slow_run(questions, *, run_id: str, skills_dir: Path):
+        await gate_open.wait()
+        return await _fake_run_inference(log, correct=True)(
+            questions, run_id=run_id, skills_dir=skills_dir
+        )
+
+    task = asyncio.ensure_future(
+        _run_unit_arm(
+            run,
+            0,
+            "base",
+            _QuestionSlots(4),
+            _slow_run,
+            log,
+            cache,
+            "v1",
+            tmp_path,
+            tmp_path,
+            _PARAMS,
+            0.10,
+        )
+    )
+    for _ in range(5):
+        await asyncio.sleep(0)  # 让 task 进入推理等待
+    run.frozen = True
+    gate_open.set()
+    await task
+    assert run.slots[0].base is None
+    assert run.infra_denom == 0 and run.errors == 0
+    assert cache.get("Action Reasoning", run.s_hash, "v1", "q0") is None