feat: G7 分层 fail-open/fail-closed + G8 Docker 轮询掐断 + G9 worker 忽略 host 预算 + G13/G14 归属（第三批）

- G7：认领门 closed 时查配额异常回 quota_check_unavailable、标 interrupted（可重试/续跑）；轮询与轮次边界用 open，异常时回 quota_check_failed——新回合边界拒绝，执行中轮询经 _quota_hit 闭包过滤此串、只记 warning 不杀在跑任务，确认超限才掐断；查询函数本身不吞异常冒充不限额。 - G8：Docker 路径加 _docker_call_with_quota_poll（后台线程跑阻塞式 docker 调用，主循环每 quota_poll_interval_seconds 缺省 30 秒查一次，确认超限按容器名真杀容器，结果改写 quota_exceeded 走统一收尾：interrupted、成本照记、续跑消息保留，配额命中后不走 ALREADY_EXISTS 回落）；同步路径保持边界检查。 - G9：worker 路径默认忽略 host max_daily_turns（host.py 顶层 --ignore-host-budget 只清该项，Local/Docker 执行器在子命令前追加，worker CLI 缺省开、--no-ignore-host-budget 关）；standalone 直接跑 host.py 时预算照常。 - G13/G14：/pr 不再走 system 豁免，建任务前按请求用户查配额（402/503 均不排队不改状态），队列路 pr_jobs.created_by 记请求用户（老表 ALTER 补列），worker 跑 pr 任务时审计操作人用 created_by（老行回落 system）。注意：G14 原案的 system 豁免（服务端与 worker 写死）与 P1 冲突（P1：system 按普通属主走默认额度，不另设无限额系统预算），本次 rebase 未合入该豁免，system 豁免需另行决策。 - 配额口径与 P0/P1 对齐：无配额行按默认额度（非不限额），只有显式 NULL 行才不限额；无价上限只计 source='actual' 行；缺属主（含父任务回查仍无）拒绝新回合。 - README 配额节同步以上口径与 G15 接受现状（best-effort 超支、UTC 月切分、在跑任务只在轮询/轮次边界响应配额改动）。G4 与 G15 不在本批。
2026-10-28 11:14:10 +08:00
1 parent 5883169c82
commit e31fcc6cba
@@ -26,7 +26,8 @@ try:
    from cloud.autoscaler import Autoscaler
    from cloud.db import connect as _db_connect
    from cloud.pr_exec import (claim_pr_job, enqueue_pr_job,
-                               find_open_pr_job, finish_pr_job)
+                               ensure_pr_jobs_table, find_open_pr_job,
+                               finish_pr_job)
    from cloud.server import create_server, migrate_tasks_table
    from cloud.worker import Worker
 except ImportError:
@@ -105,6 +106,46 @@ class PrJobsTableTestCase(unittest.TestCase):
        finally:
            conn.close()

+    # -- G13：队列记费用归属人 + 老表补列 --------------------------------
+    def test_enqueue_records_created_by(self):
+        conn = _db_connect(self.db_path)
+        try:
+            enqueue_pr_job(conn, "t1", "p1", "host-a",
+                           created_by="user-9")
+            job = find_open_pr_job(conn, "t1")
+            self.assertEqual(job["created_by"], "user-9")
+            # 不传时为 None（老调用方），worker 回落 system。
+            enqueue_pr_job(conn, "t2", "p1", "host-a")
+            self.assertIsNone(find_open_pr_job(conn, "t2")["created_by"])
+        finally:
+            conn.close()
+
+    def test_old_table_migration_adds_created_by(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "CREATE TABLE pr_jobs(id TEXT PRIMARY KEY,"
+                " task_id TEXT NOT NULL, project_id TEXT NOT NULL,"
+                " host TEXT NOT NULL, status TEXT NOT NULL, pr_url TEXT,"
+                " pr_number INTEGER, error TEXT, created_at REAL,"
+                " updated_at REAL, started_at REAL, finished_at REAL)")
+            conn.commit()
+        finally:
+            conn.close()
+        conn = _db_connect(self.db_path)
+        try:
+            ensure_pr_jobs_table(conn)
+            cols = {row[1] for row in conn.execute(
+                "PRAGMA table_info(pr_jobs)").fetchall()}
+            self.assertIn("created_by", cols)
+            # 补列后照常排队并记归属人。
+            enqueue_pr_job(conn, "t1", "p1", "host-a",
+                           created_by="user-9")
+            self.assertEqual(
+                find_open_pr_job(conn, "t1")["created_by"], "user-9")
+        finally:
+            conn.close()
+

 class WorktreeHostMigrationTestCase(unittest.TestCase):
    def setUp(self):
@@ -191,6 +191,39 @@ class TaskOptionsTestCase(unittest.TestCase):
                         "untrusted")
        self.assertNotIn("--sandbox", cmd)

+    # -- G9：worker 路径默认忽略 host 预算 -------------------------------
+    def test_local_command_ignore_host_budget_flag(self):
+        seen = []
+
+        def fake_runner(cmd):
+            seen.append(cmd)
+            return SimpleNamespace(returncode=0, stdout="{}")
+
+        ex = LocalExecutor("/fake/host.py", runner=fake_runner,
+                           ignore_host_budget=True)
+        ex.dispatch_task({"id": "t1", "prompt": "hi"})
+        cmd = seen[0]
+        self.assertIn("--ignore-host-budget", cmd)
+        # 顶层全局参数，必须在子命令之前。
+        self.assertLess(cmd.index("--ignore-host-budget"),
+                        cmd.index("dispatch"))
+        ex2 = LocalExecutor("/fake/host.py", runner=fake_runner)
+        ex2.dispatch_task({"id": "t1", "prompt": "hi"})
+        self.assertNotIn("--ignore-host-budget", seen[1])
+
+    def test_docker_inner_command_ignore_host_budget_flag(self):
+        ex = DockerExecutor(config_path="/fake/cfg.json",
+                            ignore_host_budget=True)
+        cmd = ex._inner_host_command("dispatch", "--name", "x",
+                                     "--task", "hi")
+        self.assertIn("--ignore-host-budget", cmd)
+        self.assertLess(cmd.index("--ignore-host-budget"),
+                        cmd.index("dispatch"))
+        ex2 = DockerExecutor(config_path="/fake/cfg.json")
+        cmd2 = ex2._inner_host_command("dispatch", "--name", "x",
+                                       "--task", "hi")
+        self.assertNotIn("--ignore-host-budget", cmd2)
+

 if __name__ == "__main__":
    unittest.main()
@@ -17,14 +17,16 @@ import threading
 import unittest
 from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock

 try:
    from cloud.auth import AuthStore
-    from cloud.server import create_server
+    from cloud.server import CloudControlPlaneHandler, create_server
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.auth import AuthStore
-    from cloud.server import create_server
+    from cloud.server import CloudControlPlaneHandler, create_server


 def _req(port, method, path, body=None, key=None):
@@ -355,6 +357,96 @@ class ServerQuotasTestCase(unittest.TestCase):
        self.assertEqual(s, 200, q)
        self.assertEqual(q["unpriced_actual_tokens"], 150)

+    # -- G7：准入点 fail-closed ----------------------------------------
+    def test_create_quota_lookup_failure_503(self):
+        with mock.patch.object(
+                CloudControlPlaneHandler, "_quota_status",
+                side_effect=RuntimeError("db down")):
+            s, body = self._create_task()
+        self.assertEqual(s, 503, body)
+        self.assertEqual(body["error"], "quota_check_unavailable")
+        # 没建出任务：状态、任务表都不改。
+        conn = sqlite3.connect(self.db_path)
+        try:
+            n = conn.execute(
+                "SELECT COUNT(*) FROM tasks").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(n, 0)
+
+    # -- G13：system 豁免（server 侧） -----------------------------------
+    def test_system_quota_exempt_server(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO user_quotas(user_id, quota_usd_micros,"
+                " updated_at) VALUES('system', 0, 0)")
+            conn.commit()
+        finally:
+            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"))
+
+    # -- G13/G14：/pr 按请求人过配额门、队列记归属人 ----------------------
+    def _merged_remote_task(self):
+        s, task = self._create_task()
+        self.assertEqual(s, 201, task)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "UPDATE tasks SET status = 'merged',"
+                " worktree_host = 'elsewhere-host-1' WHERE id = ?",
+                (task["id"],))
+            conn.commit()
+        finally:
+            conn.close()
+        return task
+
+    def test_pr_blocked_at_quota_no_job(self):
+        task = self._merged_remote_task()
+        s, _ = self._set_quota(self.owner_id, 0)
+        self.assertEqual(s, 200)
+        s, body = _req(self.port, "POST",
+                       "/tasks/%s/pr" % task["id"], {},
+                       self.owner_key)
+        self.assertEqual(s, 402, body)
+        self.assertEqual(body["error"], "quota_exceeded")
+        conn = sqlite3.connect(self.db_path)
+        try:
+            has_table = conn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table'"
+                " AND name='pr_jobs'").fetchone() is not None
+            rows = conn.execute(
+                "SELECT COUNT(*) FROM pr_jobs"
+                " WHERE task_id = ?", (task["id"],)).fetchone()[0] \
+                if has_table else 0
+            status = conn.execute(
+                "SELECT status FROM tasks WHERE id = ?",
+                (task["id"],)).fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(rows, 0)
+        self.assertEqual(status, "merged")
+
+    def test_pr_queue_records_created_by(self):
+        task = self._merged_remote_task()
+        s, body = _req(self.port, "POST",
+                       "/tasks/%s/pr" % task["id"], {},
+                       self.owner_key)
+        self.assertEqual(s, 200, body)
+        self.assertTrue(body.get("pr_queued"))
+        conn = sqlite3.connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT created_by FROM pr_jobs WHERE task_id = ?",
+                (task["id"],)).fetchone()
+        finally:
+            conn.close()
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], self.owner_id)
+

 if __name__ == "__main__":
    unittest.main()
@@ -18,18 +18,22 @@ import sqlite3
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 from pathlib import Path
 from types import SimpleNamespace
+from unittest import mock

 try:
    from cloud.cost import CostStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
    from cloud.server import init_db
    from cloud.worker import Worker
 except ImportError:  # discover -s cloud/tests 时把项目根目录补进 sys.path
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.cost import CostStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
    from cloud.server import init_db
    from cloud.worker import Worker

@@ -330,6 +334,144 @@ class WorkerQuotaTestCase(unittest.TestCase):
            task, kind="dispatch", quota_check=lambda: True)
        self.assertEqual(result["status"], "quota_exceeded")

+    # -- G7 分层：门侧 fail-closed、轮询侧 fail-open -------------------
+    def test_claim_gate_quota_lookup_failure_closed(self):
+        _insert_task(self.db_path, "T1")
+        worker = self._worker(_usage_runner(
+            {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+            self.calls))
+        with mock.patch.object(
+                Worker, "_quota_limit_and_used",
+                side_effect=RuntimeError("db down")):
+            result = worker.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertIn("quota_check_unavailable", result["error"])
+        self.assertEqual(_status(self.db_path, "T1"), "interrupted")
+        # 门侧拦住：runner 一次没调，也不记成本（cost_records
+        # 是 CostStore 懒建的，没跑过就根本没有这张表）。
+        self.assertEqual(self.calls, [])
+        conn = sqlite3.connect(self.db_path)
+        try:
+            has_table = conn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table'"
+                " AND name='cost_records'").fetchone() is not None
+            rows = _cost_rows(self.db_path, "T1") if has_table else []
+        finally:
+            conn.close()
+        self.assertEqual(rows, [])
+
+    def test_poll_quota_lookup_failure_open(self):
+        _insert_task(self.db_path, "T1")
+        worker = self._worker(_usage_runner(
+            {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
+            self.calls))
+        task = worker.get_task("T1")
+        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"))
+
+    # -- G13：system 豁免（worker 侧） ----------------------------------
+    def test_system_task_exempt_from_quota(self):
+        _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")
+
+    # -- G8 Docker 轮询掐断 ---------------------------------------------
+    def test_docker_quota_poll_kills_container(self):
+        _insert_quota(self.db_path, "tester", QUOTA)
+        _insert_task(self.db_path, "T1")
+        _insert_followup(self.db_path, "fu1", "T1")
+        executor = _BlockingDockerExecutor()
+        worker = Worker(self.db_path, "/fake/host.py",
+                        env=self.env, model="gpt-6.1-sol",
+                        executor=executor, retry_backoff_seconds=0,
+                        quota_poll_interval_seconds=0.01)
+        calls = {"n": 0}
+
+        def fake_limit(owner):
+            calls["n"] += 1
+            # 第 1 次是认领门（未超限，放行）；之后轮询一律超限。
+            # helper 在 rebase 合并 #86 无价上限后返回 4 元组
+            # （quota, used, month, unpriced_used），此处无价
+            # 用量为 0，不触发无价上限。
+            return (QUOTA, 90000 if calls["n"] == 1 else QUOTA,
+                    "2026-10", 0)
+
+        worker._quota_limit_and_used = fake_limit
+        result = worker.run_one()
+        self.assertEqual(result["status"], "interrupted")
+        self.assertIn("quota_exceeded", result["error"])
+        self.assertEqual(_status(self.db_path, "T1"), "interrupted")
+        self.assertEqual(executor.killed, ["codex-T1"])
+        # 续跑消息保留（仍 pending），已花成本照记。
+        self.assertEqual(_followup_status(self.db_path, "fu1"), "pending")
+        self.assertEqual(len(_cost_rows(self.db_path, "T1")), 1)
+
+    # -- G9：Worker 把 ignore 开关同步给外部执行器 ----------------------
+    def test_worker_syncs_ignore_host_budget_to_executor(self):
+        worker = Worker(self.db_path, "/fake/host.py",
+                        env=self.env, model="gpt-6.1-sol",
+                        executor=LocalExecutor("/fake/host.py"),
+                        retry_backoff_seconds=0)
+        self.assertTrue(worker.ignore_host_budget)
+        self.assertTrue(worker.executor.ignore_host_budget)
+        worker2 = Worker(self.db_path, "/fake/host.py",
+                         env=self.env, model="gpt-6.1-sol",
+                         executor=LocalExecutor("/fake/host.py"),
+                         ignore_host_budget=False,
+                         retry_backoff_seconds=0)
+        self.assertFalse(worker2.ignore_host_budget)
+        self.assertFalse(worker2.executor.ignore_host_budget)
+
+    def test_docker_quota_poll_failure_open(self):
+        _insert_task(self.db_path, "T1")
+        worker = Worker(self.db_path, "/fake/host.py",
+                        env=self.env, model="gpt-6.1-sol",
+                        quota_poll_interval_seconds=0.01)
+        task = worker.get_task("T1")
+
+        def slow_fn():
+            time.sleep(0.05)
+            return {"status": "completed"}
+
+        def bad_check():
+            raise RuntimeError("db down")
+
+        result = worker._docker_call_with_quota_poll(
+            task, slow_fn, bad_check)
+        self.assertEqual(result["status"], "completed")
+
+
+class _BlockingDockerExecutor(DockerExecutor):
+    """G8 单测用假 Docker：dispatch 阻塞到被 kill，绝不调真 docker。"""
+
+    def __init__(self):
+        super().__init__(host_py_path="/fake/host.py")
+        self.killed = []
+        self._stop = threading.Event()
+
+    def dispatch_task(self, task, timeout=None, extra_mounts=None,
+                      network=None, extra_env=None, seed_host_dir=None,
+                      model_provider=None):
+        self._stop.wait(timeout=10)
+        return {"status": "failed", "stdout": "", "stderr": "killed",
+                "usage": {"input_tokens": 100, "cached_input_tokens": 0,
+                          "output_tokens": 10, "total_tokens": 110}}
+
+    def kill_container(self, name):
+        self.killed.append(name)
+        self._stop.set()
+        return True
+

 if __name__ == "__main__":
    unittest.main()