test: G14/open 旧口径测试按 P1 口径更新

- test_system_quota_exempt_server → test_system_quota_not_exempt_server：system 按普通属主算，quota 0 时照常拦（402）。
- test_system_task_exempt_from_quota → test_system_task_not_exempt_from_quota：system quota 0 时认领门拦，标 interrupted。
- test_poll_quota_lookup_failure_open → test_poll_quota_lookup_failure_open_returns_check_failed：open 异常时回 quota_check_failed（边界拒绝、轮询经 _quota_hit 过滤只警告），closed 回 quota_check_unavailable。

cloud/tests/test_task_quotas.py

@@ -375,7 +375,10 @@ class ServerQuotasTestCase(unittest.TestCase):
         self.assertEqual(n, 0)
 
     # -- G13：system 豁免（server 侧） -----------------------------------
-    def test_system_quota_exempt_server(self):
+    def test_system_quota_not_exempt_server(self):
+        # P1 口径：system 按普通属主算，不豁免（G14 的 system 豁免
+        # 与 P1 冲突，rebase 时未合入）。quota 0、used 0 时
+        # used >= quota，照常拦。
         conn = sqlite3.connect(self.db_path)
         try:
             conn.execute(
@@ -386,8 +389,9 @@ class ServerQuotasTestCase(unittest.TestCase):
             conn.close()
         handler = object.__new__(CloudControlPlaneHandler)
         handler.server = SimpleNamespace(db_path=self.db_path)
-        self.assertIsNone(handler._check_quota("system"))
-        self.assertFalse(handler._reject_if_over_quota("system"))
+        handler._send_json = lambda *args, **kwargs: None
+        self.assertIsNotNone(handler._check_quota("system"))
+        self.assertTrue(handler._reject_if_over_quota("system"))
 
     # -- G13/G14：/pr 按请求人过配额门、队列记归属人 ----------------------
     def _merged_remote_task(self):

cloud/tests/test_worker_quota.py

@@ -360,7 +360,12 @@ class WorkerQuotaTestCase(unittest.TestCase):
             conn.close()
         self.assertEqual(rows, [])
 
-    def test_poll_quota_lookup_failure_open(self):
+    def test_poll_quota_lookup_failure_open_returns_check_failed(self):
+        # 合并后口径（G7 + P1）：open 不再回 None 放行，而是回
+        # quota_check_failed——新回合边界见此串即拒绝；执行中
+        # 轮询经 _quota_hit 闭包过滤此串、只警告不杀轮（轮询
+        # 不杀的语义由闭包保证，不是由这里回 None 保证）。
+        # closed（认领门）回 quota_check_unavailable。
         _insert_task(self.db_path, "T1")
         worker = self._worker(_usage_runner(
             {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
@@ -369,21 +374,26 @@ class WorkerQuotaTestCase(unittest.TestCase):
         with mock.patch.object(
                 Worker, "_quota_limit_and_used",
                 side_effect=RuntimeError("db down")):
-            # 默认 on_error="open"：异常不冒充超限，放行。
-            self.assertIsNone(worker._quota_exceeded(task, None))
-            self.assertIsNone(worker._quota_exceeded(task, None,
-                                                     on_error="open"))
+            err = worker._quota_exceeded(task, None)
+            self.assertTrue(err.startswith("quota_check_failed:"))
+            err = worker._quota_exceeded(task, None, on_error="open")
+            self.assertTrue(err.startswith("quota_check_failed:"))
+            err = worker._quota_exceeded(task, None, on_error="closed")
+            self.assertTrue(err.startswith("quota_check_unavailable:"))
 
-    # -- G13：system 豁免（worker 侧） ----------------------------------
-    def test_system_task_exempt_from_quota(self):
+    # -- G13：system 不豁免（worker 侧，P1 口径） ------------------------
+    def test_system_task_not_exempt_from_quota(self):
+        # P1 口径：system 按普通属主算，不豁免（G14 的 system 豁免
+        # 与 P1 冲突，rebase 时未合入）。quota 0 时认领门即拦，
+        # 标 interrupted（可续跑），不是 completed。
         _insert_quota(self.db_path, "system", 0)
         _insert_task(self.db_path, "T1", created_by="system")
         worker = self._worker(_usage_runner(
             {"input_tokens": 100, "output_tokens": 10, "total_tokens": 110},
             self.calls))
         result = worker.run_one()
-        self.assertEqual(result["status"], "completed")
-        self.assertEqual(_status(self.db_path, "T1"), "completed")
+        self.assertEqual(result["status"], "interrupted")
+        self.assertEqual(_status(self.db_path, "T1"), "interrupted")
 
     # -- G8 Docker 轮询掐断 ---------------------------------------------
     def test_docker_quota_poll_kills_container(self):