P0 security batch: true global admin, quota gates on all spending paths, default quota for quota-less users, fix-review author trust + docker/owner attribution, GitHub App setup issuer binding, real cancel kill, wider shell env exclusion

- CODEX_GLOBAL_ADMINS replaces maintainer-derived global admin;
  admin endpoints (quotas, sweep, backup, push dispatch) fail-closed
  403 when unset, with audit on allow and deny.
- continue/followups/messages/retry check both caller and original
  creator quotas (USD + unpriced token cap), 402 on either overrun.
- Quota-less users now get the #86 default quota instead of unlimited;
  init_db backfills and OAuth create/bind/existing-login branches
  INSERT OR IGNORE a default row; explicit NULL row stays unlimited.
  user_quotas added to migration tables.
- /fix-review: untrusted authors rejected before dedup claim, no task;
  trusted tasks force executor=docker, created_by=project owner, owner
  quota gate, prompt is a fixed template plus a 2000-char excerpt.
- GitHub App setup state only issued by global admins, state rows bind
  issued_by (ALTER-compatible for old DBs), callback rejects issuers
  no longer in the admin list without consuming the state.
- Cancel: popen poll checks DB status every second, Docker path runs a
  cancel watchdog that kills by container name, write-back is CAS-guarded
  so late results never overwrite a new status.
- SHELL_ENV_EXCLUDE adds *_TOKEN, *_AUTH*, *_CREDENTIAL*, *API_KEY*;
  comment states the residual risk: same-UID /proc environ remains
  readable, true isolation needs UID separation or hidepid.
- Tests: new cloud/tests/test_p0_security.py (17 cases); old tests
  updated to the new semantics (admin env, default quota, trusted
  fix-review author) without weakening any gate.

cloud/worker.py

@@ -700,45 +700,57 @@ class Worker:
                                 error_text=None, actual_model=None,
                                 started_at=None, finished_at=None,
                                 child_pid=None, turn_id=None,
-                                root_turn_id=None):
+                                root_turn_id=None, only_if_status=None):
         """把执行结果写进 tasks 新列；老库缺列时先迁移再写。
 
         turn_id/root_turn_id 只在新值非空时覆盖：某轮没产出
         turn（如提前失败门、解析不到 JSON）时保留上一轮的值，
         列表里始终显示最近一个真实回合的编号，而不是被空值
         洗掉。
+        P0 安全批后：only_if_status 非空时写回带状态守卫
+        （UPDATE ... WHERE id = ? AND status = ?），状态已
+        变（例如迟到的旧进程在任务已被取消/新一轮认领后才
+        写回）时不覆盖新结果，返回 False；不传时保持旧行为
+        并返回 True。调用方在终局路径必须传守卫，不能只靠
+        写回前的 _set_status_if_running（两者之间仍有竞态窗）。
         """
         conn = _db_connect(self.db_path)
         try:
             self._ensure_result_columns(conn)
+            status_clause = ""
+            extra = ()
+            if only_if_status is not None:
+                status_clause = " AND status = ?"
+                extra = (only_if_status,)
             if child_pid is None:
-                conn.execute(
+                cur = conn.execute(
                     "UPDATE tasks SET result_text = ?, error_text = ?,"
                     " actual_model = ?, started_at = ?, finished_at = ?,"
                     " turn_id = COALESCE(?, turn_id),"
                     " root_turn_id = COALESCE(?, root_turn_id),"
-                    " updated_at = ? WHERE id = ?",
+                    " updated_at = ? WHERE id = ?" + status_clause,
                     ((result_text or "")[:4000] if result_text else None,
                      (error_text or "")[:4000] if error_text else None,
                      actual_model, started_at, finished_at,
                      turn_id, root_turn_id,
-                     time.time(), task_id),
+                     time.time(), task_id) + extra,
                 )
             else:
-                conn.execute(
+                cur = conn.execute(
                     "UPDATE tasks SET result_text = ?, error_text = ?,"
                     " actual_model = ?, started_at = ?, finished_at = ?,"
                     " child_pid = ?,"
                     " turn_id = COALESCE(?, turn_id),"
                     " root_turn_id = COALESCE(?, root_turn_id),"
-                    " updated_at = ? WHERE id = ?",
+                    " updated_at = ? WHERE id = ?" + status_clause,
                     ((result_text or "")[:4000] if result_text else None,
                      (error_text or "")[:4000] if error_text else None,
                      actual_model, started_at, finished_at, child_pid,
                      turn_id, root_turn_id,
-                     time.time(), task_id),
+                     time.time(), task_id) + extra,
                 )
             conn.commit()
+            return cur.rowcount == 1
         finally:
             conn.close()
 
@@ -1079,6 +1091,38 @@ class Worker:
                       project_id=task["project_id"], status="interrupted")
         return claimed
 
+    def _docker_cancel_watchdog(self, task_id, container_name,
+                                stop_event):
+        """Docker 同步路径的取消看门狗（_run_attempt 起的线程）。
+
+        同步路径阻塞在 docker run 里，worker 的 cancel() 在
+        另一线程只能改库、清 _running 项，阻塞中的 run 本身
+        不会醒——P0 安全批前这就是“取消杀不掉容器”的根因。
+        本线程每 1 秒查一次库：任务变 interrupted 或本进程
+        收到取消标志时，按容器名真杀容器（kill+rm，尽力而
+        为），然后退出；调用方在 _run_attempt 结束时置
+        stop_event 并 join。查库失败时当没取消，不误杀。
+        """
+        while not stop_event.wait(1.0):
+            try:
+                if task_id in self._cancel_requested:
+                    cancelled = True
+                else:
+                    db_task = self.get_task(task_id)
+                    cancelled = (db_task is not None and
+                                 db_task.get("status") == "interrupted")
+            except Exception:
+                cancelled = False
+            if not cancelled:
+                continue
+            try:
+                kill = getattr(self.executor, "kill_container", None)
+                if kill is not None and container_name:
+                    kill(container_name)
+            except Exception:
+                pass
+            return
+
     def cleanup_stale_containers(self):
         """启动清理兜底：清掉上一轮残留的 codex- 容器。
 
@@ -1298,12 +1342,29 @@ class Worker:
         quota_hit = False
         started = time.time()
         last_quota_check = 0.0
+        last_db_cancel_check = 0.0
         try:
             while proc.poll() is None:
                 if task_id in self._cancel_requested or self._loop_stop:
                     cancelled = True
                     self._kill_proc_group(proc)
                     break
+                # P0 安全批后：popen 轮询也要轮询库状态，不只
+                # 看本进程标志——控制台取消 / sweep 是另一进程
+                # 改库，旧代码只看本进程标志时杀不掉本进程的
+                # 子进程。节流到每 1 秒查一次库，避免 0.05 秒
+                # 轮询把库打爆；查库失败时当没取消，不误杀。
+                if time.time() - last_db_cancel_check >= 1.0:
+                    last_db_cancel_check = time.time()
+                    try:
+                        db_task = self.get_task(task_id)
+                    except Exception:
+                        db_task = None
+                    if db_task is not None and \
+                            db_task.get("status") == "interrupted":
+                        cancelled = True
+                        self._kill_proc_group(proc)
+                        break
                 if deadline_seconds is not None and \
                         time.time() - started >= deadline_seconds:
                     timed_out = True
@@ -2003,6 +2064,23 @@ class Worker:
                         float(self.executor.timeout))
                 except (TypeError, ValueError):
                     effective_docker_timeout = self.task_timeout
+        # P0 安全批后：Docker 同步路径起取消看门狗线程
+        # （阻塞在 docker run 里时，库状态变 interrupted 就
+        # 按容器名真杀；非 Docker 路径不启动，避免空线程）。
+        _watchdog_stop = None
+        _watchdog_thread = None
+        if is_docker and hasattr(self.executor, "kill_container"):
+            try:
+                _container_name = self.executor.container_name(task)
+            except Exception:
+                _container_name = None
+            if _container_name:
+                _watchdog_stop = threading.Event()
+                _watchdog_thread = threading.Thread(
+                    target=self._docker_cancel_watchdog,
+                    args=(task_id, _container_name, _watchdog_stop),
+                    daemon=True)
+                _watchdog_thread.start()
         try:
             gateway_provider, gateway_env = self._ensure_gateway(task)
             if self._use_popen():
@@ -2138,6 +2216,10 @@ class Worker:
                       "host_status": None, "stderr": ""}
             error = str(exc)
         finally:
+            if _watchdog_stop is not None:
+                _watchdog_stop.set()
+                if _watchdog_thread is not None:
+                    _watchdog_thread.join(timeout=2)
             # 同步路径执行结束：清掉取消用的容器标记（取消路径里
             # 若已清过，这里再清一次无害）。进程模式的 proc 由
             # _execute_via_process 自己的 finally 清，这里只清
@@ -2417,10 +2499,12 @@ class Worker:
         if not final_written:
             final = "interrupted"
         finished_at = time.time()
-        self._write_execution_result(
-            task_id, result_text=report, error_text=error,
-            actual_model=None, started_at=started_at,
-            finished_at=finished_at)
+        # P0 安全批后：CAS 没改到行时不写回（见终局路径注释）。
+        if final_written:
+            self._write_execution_result(
+                task_id, result_text=report, error_text=error,
+                actual_model=None, started_at=started_at,
+                finished_at=finished_at, only_if_status=final)
         self._cleanup_task_seed(task)
         if final_written:
             self._audit(project_id,
@@ -2489,7 +2573,8 @@ class Worker:
             self._set_status(task_id, "failed")
             self._write_execution_result(
                 task_id, error_text=error, actual_model=self.model,
-                started_at=started_at, finished_at=time.time())
+                started_at=started_at, finished_at=time.time(),
+                only_if_status="failed")
             # 缺 key 门提前返回：同样要清暂存，否则与仓库解析
             # 失败一路同款残留（见 _cleanup_task_seed）。
             self._cleanup_task_seed(task)
@@ -2516,7 +2601,8 @@ class Worker:
             self._set_status(task_id, "failed")
             self._write_execution_result(
                 task_id, error_text=repo_error, actual_model=self.model,
-                started_at=started_at, finished_at=time.time())
+                started_at=started_at, finished_at=time.time(),
+                only_if_status="failed")
             # 仓库解析失败提前返回：够不到正常终局后的清理，
             # 这里必须先清暂存，否则 bundle 目录永久残留。
             self._cleanup_task_seed(task)
@@ -2544,7 +2630,8 @@ class Worker:
             self._set_status(task_id, "interrupted")
             self._write_execution_result(
                 task_id, error_text=quota_error, actual_model=self.model,
-                started_at=started_at, finished_at=time.time())
+                started_at=started_at, finished_at=time.time(),
+                only_if_status="interrupted")
             self._audit(task["project_id"], "worker_quota_stop",
                         "%s: %s" % (task_id, quota_error))
             log_event(self.logger, "worker_quota_stop", task_id=task_id,
@@ -2594,7 +2681,8 @@ class Worker:
             self._set_status(task_id, "failed")
             self._write_execution_result(
                 task_id, error_text=error, actual_model=self.model,
-                started_at=started_at, finished_at=time.time())
+                started_at=started_at, finished_at=time.time(),
+                only_if_status="failed")
             self._audit(task["project_id"], "worker_gateway_failed",
                         "%s: %s" % (task_id, error))
             log_event(self.logger, "worker_gateway_failed",
@@ -2810,7 +2898,8 @@ class Worker:
                 error_text="cancelled", actual_model=actual_model,
                 started_at=started_at, finished_at=time.time(),
                 turn_id=(result or {}).get("turn_id"),
-                root_turn_id=(result or {}).get("root_turn_id"))
+                root_turn_id=(result or {}).get("root_turn_id"),
+                only_if_status="interrupted")
             log_event(self.logger, "worker_cancelled", task_id=task_id,
                       project_id=task["project_id"], status="interrupted")
             return {"dispatched": True, "task_id": task_id,
@@ -2874,12 +2963,17 @@ class Worker:
                                     "%s -> %s" % (self.model, nxt))
         final_written = self._set_status_if_running(task_id, final)
         finished_at = time.time()
-        self._write_execution_result(
-            task_id, result_text=stdout, error_text=error,
-            actual_model=actual_model, started_at=started_at,
-            finished_at=finished_at,
-            turn_id=(result or {}).get("turn_id"),
-            root_turn_id=(result or {}).get("root_turn_id"))
+        # P0 安全批后：CAS 没改到行（任务已被取消/回收）时
+        # 绝不写回结果，否则迟到的旧进程会覆盖新一轮的结果
+        # 文本；CAS 成功时写回也要带状态守卫。
+        if final_written:
+            self._write_execution_result(
+                task_id, result_text=stdout, error_text=error,
+                actual_model=actual_model, started_at=started_at,
+                finished_at=finished_at,
+                turn_id=(result or {}).get("turn_id"),
+                root_turn_id=(result or {}).get("root_turn_id"),
+                only_if_status=final)
         # 附件暂存清理：任务终局后删 bundle 目录（重试走已存在
         # 续跑，文件已在 worktree 里，不再需要暂存）。
         self._cleanup_task_seed(task)