worker: release running slot after coordinator completes

Coordinator tasks bypass _run_attempt, so their _claim placeholder in _running was never cleared: each completed coordinator permanently reduced the worker process effective capacity until it could claim nothing, while heartbeat stayed fresh and the capacity gate returned silently. This caused the 2026-10-28 40-person stall (autoscale-1 had one leaked slot, then claimed coordinator A and filled up) and the old 18-person C-6 never-claim (autoscale-1 leaked both A and B). Fix: wrap the run_one coordinator branch in try/finally _running_clear, so normal and exception paths both release. Also log capacity-gate refusals (worker_claim_refused_capacity) throttled to once per 30s; throttling applies to the log only, refusal always returns False. Tests: coordinator completion releases slot, next claim succeeds, capacity-log throttle; both leak tests fail on pre-fix code.

cloud/tests/test_coordinator.py

@@ -270,6 +270,60 @@ class WorkerCoordinatorTestCase(unittest.TestCase):
         self.assertIn("后端好了", result["stdout"])
         self.assertEqual(_row(self.db_path, "coord")["status"], "completed")
 
+    def test_coordinator_completion_releases_running_slot(self):
+        # 回归：协调员不走 _run_attempt 的 finally，旧版完成
+        # 后 _running 占位永不释放，同进程 worker 的有效容量
+        # 逐个协调员递减直至再也认领不了（2026-10-28 40 人
+        # 与 18 人旧轮的第 4 槽/C-6 未认领即此漏）。
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="后端好了")
+        _insert_task(self.db_path, "c2", status="completed",
+                     parent_task_id="coord", coordinator_part="前端",
+                     result_text="前端好了")
+        worker = self._worker()
+        result = worker.run_one(worker.get_task("coord"))
+        self.assertEqual(result["status"], "completed")
+        self.assertNotIn("coord", worker._running_snapshot())
+        self.assertEqual(worker._running_count(), 0)
+
+    def test_coordinator_then_next_claim_succeeds(self):
+        # 容量门视角的同一回归：协调员完成后，下一个任务
+        # 的 _claim 必须成功，而不是被漏掉的占位静默挡住。
+        _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
+        _insert_task(self.db_path, "c1", status="completed",
+                     parent_task_id="coord", coordinator_part="后端",
+                     result_text="后端好了")
+        _insert_task(self.db_path, "c2", status="completed",
+                     parent_task_id="coord", coordinator_part="前端",
+                     result_text="前端好了")
+        _insert_task(self.db_path, "next1")
+        worker = self._worker()
+        worker.run_one(worker.get_task("coord"))
+        self.assertTrue(
+            worker._claim("next1", executor=worker._executor_name()))
+        self.assertIn("next1", worker._running_snapshot())
+        worker._running_clear("next1")
+
+    def test_capacity_refused_log_throttled(self):
+        # 容量门被拦以前是静默返回；修法后 30 秒内只记一次，
+        # 避免池线程自旋把日志打爆（这里断言时间戳节流，
+        # 不捕获日志输出）。
+        _insert_task(self.db_path, "t1")
+        _insert_task(self.db_path, "t2")
+        _insert_task(self.db_path, "t3")
+        worker = self._worker()
+        self.assertTrue(
+            worker._claim("t1", executor=worker._executor_name()))
+        self.assertFalse(
+            worker._claim("t2", executor=worker._executor_name()))
+        first = worker._capacity_refused_logged_at
+        self.assertGreater(first, 0.0)
+        self.assertFalse(
+            worker._claim("t3", executor=worker._executor_name()))
+        self.assertEqual(worker._capacity_refused_logged_at, first)
+
     def test_run_timeout_cancels_pending(self):
         _insert_task(self.db_path, "coord", coordinator_parts=self._parts())
         conn = sqlite3.connect(self.db_path)