Project repo_url: worker can run tasks from any git repo

Projects gain repo_url (validated: https:// or git@ only, no
userinfo/password, no file:// or local paths; old DBs get the
column via idempotent ALTER in both SQLite and Postgres init).
Worker resolves a project's repo_url to a local checkout under
--repos-dir (default ./repos) before dispatch: clone if absent,
verify remote origin matches and fetch if present, never force
reset/clean, never push (push needs write credentials the
control plane does not store). Resolve failure fails the task
with a clear error, never completed. Docker workers need
repos_dir under a same-path extra mount (e.g.
/var/lib/codex-app-server/repos).
Also: driver-missing test is now mocked instead of depending
on the host really having no psycopg driver installed.
Deploy package: Caddyfile template exposing only /webhooks/*
and /oauth/* (tasks/projects/admin stay 403) plus a
webhook-secret generator script that never prints the secret.
Tests: host 27 pass, cloud 286 pass (8 skipped are
environment-dependent real-Postgres tests).

cloud/README.md

@@ -533,9 +533,40 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   绝不抛异常），`Worker.cleanup_stale_containers()` 在 run_loop 启动
   与 CLI 启动时各调一次；取消/超时/失败路径上已有的容器清理保留
   （docker run `--rm`、超时 kill+rm、取消按名 kill）。
-- 仍未做（不要宣称已做）：webhook 建的任务仍需 worker 认领执行，
-  worker 的 repo 仍是配置里的 repo，不是任意 GitHub 仓库的自动克隆；
-  `CloudExecutor` 仍是占位符；公网 https 入口仍需部署侧反向代理 +
-  TLS；Docker 超时杀容器与真实 `pg_dump`、真实 GitHub webhook 送达
-  仍需在真实环境验证（本轮只用 fake runner / 本地 HTTP 单测）。
+## 项目仓库（任意仓库任务）
+
+- 项目级 `repo_url`：`POST /projects` 接受可选 `repo_url`，
+  `GET /projects` 列表与 `POST /projects` 的 201 返回里都含它，
+  不传时为 `None`（老行为不变）。校验与 worker 共用同一套
+  `cloud.repos.validate_repo_url`：只接受 `https://` 开头（且不含
+  userinfo，即不含用户名/密码）或 `git@` 开头（`git@host:path` 形）；
+  `file://`、`ssh://`、本地路径、带密码的 URL 一律 400 且不建项目。
+- 存储与迁移：`projects` 表新增 `repo_url TEXT` 可空列；SQLite 的
+  `init_db` 建表时就含该列，老库用 `ALTER TABLE` 幂等补列（仿照
+  `migrate_tasks_table` 的写法）；Postgres 的 `POSTGRES_SCHEMA`
+  同样加该列，`init_postgres_schema` 也对老库补列。
+- worker 派发：`run_one` 派发前查任务所属项目的 `repo_url`，非空时
+  经 `cloud.repos.resolve_repo` 在 `--repos-dir`（默认 `./repos`）下
+  解析：目录不存在时 `git clone`；已存在时校验它是 git 仓库且
+  remote origin 与 `repo_url` 一致（不一致时报错，绝不强制改
+  remote），一致时 `git fetch origin`（失败明确报错，不静默用旧代码；
+  fetch 成功后不 checkout/pull/reset/clean，工作树由 host.py 的
+  worktree 逻辑处理）。解析出的路径设为 `task["repo"]` 再交给执行器，
+  Local 与 Docker 的 `dispatch_task` 都已读 `task["repo"]`，同一条路径；
+  查不到项目或 `repo_url` 为空时走原有默认 repo，不报错；解析失败时
+  任务置 `failed`，错误文本写清是仓库解析失败，绝不标 completed。
+- Docker 下的路径前提：容器内 `task["repo"]` 路径必须与宿主机一致，
+  所以 `repos_dir` 必须放在 `--docker-extra-mounts` 已挂载的目录下
+  （例如把 `/var/lib/codex-app-server/repos` 以同路径挂载进容器），
+  否则容器内 repo 路径不存在——不要假装任意目录都行。生产部署示例：
+  `python3 -m cloud.worker --db <库> --executor docker
+  --repos-dir /var/lib/codex-app-server/repos
+  --docker-extra-mounts /var/lib/codex-app-server:/var/lib/codex-app-server,…`。
+- 仍未做（不要宣称已做）：推送到远端明确不在本轮范围——任务在本地
+  检出上执行、本地 worktree 合并仍按 host.py 现有流程，`git push`
+  到远端需要仓库写凭证，控制面不存、不推，所以任意仓库的自动 PR
+  仍未实现；私有仓库 clone 会失败，错误文本写明“需要仓库访问凭证，
+  控制面不存仓库凭证”，绝不尝试用任何 key 去 clone；webhook 建的
+  任务仍需 worker 认领执行；`CloudExecutor` 仍是占位符；公网 https
+  入口仍需部署侧反向代理 + TLS。
 

cloud/db.py

@@ -780,7 +780,8 @@ POSTGRES_SCHEMA = (
     "id TEXT PRIMARY KEY, "
     "name TEXT, "
     "owner_id TEXT, "
-    "created_at DOUBLE PRECISION)",
+    "created_at DOUBLE PRECISION, "
+    "repo_url TEXT)",
     # tasks 含 server.migrate_tasks_table 补的那 6 列，建表时一次建齐；
     # 老库迁移仍走 migrate_tasks_table（PRAGMA 翻译后在 PG 上是 no-op）。
     "CREATE TABLE IF NOT EXISTS tasks("
@@ -861,9 +862,19 @@ def init_postgres_schema(conn):
     conn 须是 PgConnection（或具同名 execute/commit 的适配连接）；
     SQLite 连接传进来会在首条 DDL 的 IDENTITY 语法上报错——
     调用方须按后端选对连接，不要把 SQLite 连接混进来。
+
+    建表后还给老库 projects 表补 repo_url 列（新库建表时已含）：
+    PRAGMA table_info 经适配层翻译成 information_schema 查询，
+    调用方取 row[1] 即列名；缺列时 ALTER TABLE 补列，已迁移的
+    库是 no-op。server.init_db 的 Postgres 分支还会再跑一次
+    migrate_projects_table，两次补列互不冲突（第二次是 no-op）。
     """
     for statement in POSTGRES_SCHEMA:
         conn.execute(statement)
+    cols = {row[1] for row in conn.execute(
+        "PRAGMA table_info(projects)").fetchall()}
+    if "repo_url" not in cols:
+        conn.execute("ALTER TABLE projects ADD COLUMN repo_url TEXT")
     conn.commit()
 
 

cloud/repos.py

@@ -0,0 +1,150 @@
+"""项目级仓库解析：把项目的 repo_url 变成 worker 可派发的本地检出。
+
+零第三方依赖，只用 Python 标准库；不改 host.py 的 worktree 逻辑，
+不存任何仓库凭证、不推远端。
+
+- validate_repo_url：API（POST /projects）与 worker 共用的同一套
+  校验，绝不另写第二份校验逻辑让两者漂移。只接受：
+  https:// 开头（且不含 userinfo，即不含用户名/密码），或
+  git@ 开头（git@host:path 形）。file://、ssh://、本地路径、
+  带密码的 URL 一律拒绝。
+- checkout_dir：按 project_id 的安全字符命名，防路径穿越。
+- resolve_repo：目录不存在时 git clone；已存在时校验它是 git
+  仓库且 remote origin 与 repo_url 一致（不一致时报错，绝不强制
+  改 remote），一致时 git fetch origin（失败明确报错，不静默用
+  旧代码）；fetch 成功后不自动 checkout/pull、不 git reset/clean，
+  工作树脏不脏由 host.py 的 worktree 逻辑处理。
+- 私有仓库 clone 失败时错误文本写明：需要仓库访问凭证，控制面
+  不存仓库凭证——绝不尝试用任何 key 去 clone。
+"""
+
+import os
+import re
+import subprocess
+from urllib.parse import urlparse
+
+__all__ = [
+    "RepoError",
+    "validate_repo_url",
+    "checkout_dir",
+    "resolve_repo",
+]
+
+
+class RepoError(ValueError):
+    """仓库解析失败（校验、clone、remote 不一致、fetch 失败）。"""
+
+
+_GIT_SCP_RE = re.compile(r"^git@[^/:@\s]+:[^@\s]+$")
+_SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9_.-]")
+
+
+def validate_repo_url(repo_url):
+    """校验 repo_url，合法时返回原值，否则抛 ValueError。
+
+    拒绝：非字符串、空串、含空白/控制字符、非 https:// 与 git@ 开头、
+    https:// 带 userinfo（username 或 password 任一存在即拒绝，
+    含 https://user@host/ 这种只有用户名无密码的也拒绝，因为
+    userinfo 本身就是凭证载体）、https:// 无 host、git@ 形不合规。
+    """
+    if not isinstance(repo_url, str):
+        raise ValueError("repo_url 必须是非空字符串")
+    # 含空白/控制字符的 URL 直接拒绝，不静默 strip：strip 会把
+    # “https://…\n”这种畸形输入洗成合法 URL 存进库，git 调用时
+    # 再出难以定位的问题；调用方应传干净的 URL。
+    if not repo_url or any(ch.isspace() for ch in repo_url):
+        raise ValueError("repo_url 必须是非空字符串且不得含空白字符")
+    url = repo_url
+    if url.startswith("https://"):
+        parsed = urlparse(url)
+        if parsed.username is not None or parsed.password is not None:
+            raise ValueError(
+                "repo_url 不得含 userinfo（用户名/密码）: %s" % url)
+        if not parsed.hostname:
+            raise ValueError("repo_url 缺少 host: %s" % url)
+        return url
+    if url.startswith("git@"):
+        if not _GIT_SCP_RE.match(url):
+            raise ValueError("git@ 形 repo_url 不合法: %s" % url)
+        return url
+    raise ValueError(
+        "repo_url 只接受 https:// 或 git@ 开头，"
+        "不接受 file://、ssh://、本地路径: %s" % url)
+
+
+def checkout_dir(repos_dir, project_id):
+    """按 project_id 的安全字符命名检出目录，防路径穿越。
+
+    project_id 本身是 hex，仍做安全字符过滤；过滤后为空时报错。
+    返回的绝对路径一定在 repos_dir 之下，否则报错（防御性双检）。
+    """
+    raw = str(project_id)
+    safe = _SAFE_NAME_RE.sub("-", raw).strip(".")
+    if not safe or safe in (".", ".."):
+        raise RepoError("project_id 无法映射为安全的检出目录名: %r" % raw)
+    base = os.path.abspath(repos_dir)
+    path = os.path.join(base, safe)
+    if os.path.commonpath([base, path]) != base:
+        raise RepoError("检出目录越出 repos_dir，拒绝: %s" % path)
+    return path
+
+
+def _default_runner(cmd):
+    return subprocess.run(cmd, capture_output=True, text=True)
+
+
+def _run(runner, cmd):
+    """调 runner：优先只传 cmd，假 runner 只接受 cmd 时即走此路；
+    接受 (cmd, timeout) 的 runner 也兼容（TypeError 时回退）。"""
+    try:
+        return runner(cmd)
+    except TypeError:
+        return runner(cmd, timeout=None)
+
+
+def resolve_repo(repo_url, repos_dir, project_id, runner=None):
+    """把 repo_url 解析为本地检出路径，返回该路径。
+
+    目录不存在：git clone repo_url 到该目录；clone 失败抛 RepoError，
+    文本含“需要仓库访问凭证，控制面不存仓库凭证”的提示（私有仓
+    库、仓库不存在、无权访问都走同一文案，不猜测、不尝试凭证）。
+    目录已存在：须是 git 仓库（.git 目录或文件存在）且 remote
+    origin 的 URL 与 repo_url 一致；不是 git 仓库、取不到 origin、
+    origin 不一致时都抛 RepoError，绝不强制改 remote、不覆盖。
+    一致时 git fetch origin：失败抛 RepoError，不静默用旧代码；
+    成功后不 checkout/pull/reset/clean，工作树由 host.py 处理。
+    """
+    url = validate_repo_url(repo_url)
+    run = runner if runner is not None else _default_runner
+    target = checkout_dir(repos_dir, project_id)
+    os.makedirs(repos_dir, exist_ok=True)
+    if not os.path.exists(target):
+        result = _run(run, ["git", "clone", url, target])
+        if getattr(result, "returncode", 1) != 0:
+            err = (getattr(result, "stderr", "") or "").strip()
+            raise RepoError(
+                "git clone 失败: %s；若为私有仓库，需要仓库访问凭证，"
+                "控制面不存仓库凭证，请先在可访问该仓库的机器上完成"
+                "检出或改用公开仓库" % (err or url))
+        return target
+    git_path = os.path.join(target, ".git")
+    if not (os.path.isdir(git_path) or os.path.isfile(git_path)):
+        raise RepoError(
+            "检出目录已存在但不是 git 仓库，拒绝覆盖: %s" % target)
+    result = _run(run, ["git", "-C", target, "remote", "get-url", "origin"])
+    if getattr(result, "returncode", 1) != 0:
+        raise RepoError(
+            "检出目录 %s 取不到 remote origin，拒绝派发" % target)
+    origin = (getattr(result, "stdout", "") or "").strip()
+    if origin != url:
+        raise RepoError(
+            "检出目录 %s 的 remote origin（%s）与项目 repo_url（%s）"
+            "不一致，拒绝派发（绝不强制改 remote）"
+            % (target, origin, url))
+    fetched = _run(run, ["git", "-C", target, "fetch", "origin"])
+    if getattr(fetched, "returncode", 1) != 0:
+        err = (getattr(fetched, "stderr", "") or "").strip()
+        raise RepoError(
+            "git fetch origin 失败，拒绝用旧代码派发: %s"
+            % (err or target))
+    return target

cloud/server.py

@@ -67,6 +67,7 @@ from cloud.oauth import (
     map_external_identity_to_local_user_with_key,
     start_oauth_login,
 )
+from cloud.repos import validate_repo_url
 from cloud.push import (
     create_subscription as _push_create,
     delete_subscription as _push_delete,
@@ -192,6 +193,20 @@ def migrate_tasks_table(conn):
                 "ALTER TABLE tasks ADD COLUMN %s %s" % (name, coltype))
 
 
+def migrate_projects_table(conn):
+    """幂等迁移：给老库 projects 表补 repo_url 列，不丢数据。
+
+    与 migrate_tasks_table 同款写法：PRAGMA table_info 查现有列
+    （Postgres 适配层把它翻译成 information_schema 查询，调用方
+    取 row[1] 即列名），缺 repo_url 时 ALTER TABLE 补列，已迁移
+    的库再跑一次是 no-op。SQLite 与 Postgres 两支都走它。
+    """
+    cols = {row[1] for row in conn.execute(
+        "PRAGMA table_info(projects)").fetchall()}
+    if "repo_url" not in cols:
+        conn.execute("ALTER TABLE projects ADD COLUMN repo_url TEXT")
+
+
 def _is_postgres_backend():
     backend = os.environ.get("CLOUD_DB_BACKEND") or "sqlite"
     return str(backend).strip().lower() in ("postgres", "postgresql")
@@ -210,6 +225,7 @@ def init_db(db_path):
         if _is_postgres_backend():
             _init_postgres_schema(conn)
             migrate_tasks_table(conn)
+            migrate_projects_table(conn)
             conn.commit()
             return
         conn.execute(
@@ -217,7 +233,8 @@ def init_db(db_path):
             "id TEXT PRIMARY KEY, "
             "name TEXT, "
             "owner_id TEXT, "
-            "created_at REAL)"
+            "created_at REAL, "
+            "repo_url TEXT)"
         )
         conn.execute(
             "CREATE TABLE IF NOT EXISTS tasks("
@@ -235,6 +252,7 @@ def init_db(db_path):
             "UNIQUE(project_id, idempotency_key))"
         )
         migrate_tasks_table(conn)
+        migrate_projects_table(conn)
         conn.commit()
     finally:
         conn.close()
@@ -296,6 +314,9 @@ def _project_to_dict(row):
         "name": row["name"],
         "owner_id": row["owner_id"],
         "created_at": row["created_at"],
+        # 老库行缺 repo_url 列时 _row_get 给 None，不炸（列本身由
+        # migrate_projects_table 在 init_db 时补，老行读到也是 None）。
+        "repo_url": _row_get(row, "repo_url"),
     }
 
 
@@ -547,6 +568,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
         if not isinstance(project_id, str):
             self._send_json(400, {"error": "id must be a string"})
             return
+        # repo_url 可选：不传时为 None（老行为不变）；传了则与 worker
+        # 共用 cloud.repos.validate_repo_url 的同一套校验，不合法时
+        # 400 且不建项目，绝不把带凭证的 URL 存进库。
+        repo_url = body.get("repo_url")
+        if repo_url is not None:
+            try:
+                repo_url = validate_repo_url(repo_url)
+            except ValueError as exc:
+                self._send_json(400, {"error": str(exc)})
+                return
         now = time.time()
         conn = self._db()
         try:
@@ -557,9 +588,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 self._send_json(409, {"error": "project exists"})
                 return
             conn.execute(
-                "INSERT INTO projects(id, name, owner_id, created_at)"
-                " VALUES(?, ?, ?, ?)",
-                (project_id, name, user_id, now),
+                "INSERT INTO projects(id, name, owner_id, created_at,"
+                " repo_url) VALUES(?, ?, ?, ?, ?)",
+                (project_id, name, user_id, now, repo_url),
             )
             conn.commit()
             project = {
@@ -567,6 +598,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 "name": name,
                 "owner_id": user_id,
                 "created_at": now,
+                "repo_url": repo_url,
             }
         except (sqlite3.IntegrityError, _PgIntegrityError):
             self._send_json(409, {"error": "project exists"})
@@ -591,8 +623,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 ).fetchall()
             }
             rows = conn.execute(
-                "SELECT id, name, owner_id, created_at FROM projects"
-                " ORDER BY created_at, id"
+                "SELECT id, name, owner_id, created_at, repo_url"
+                " FROM projects ORDER BY created_at, id"
             ).fetchall()
         finally:
             conn.close()

cloud/tests/test_postgres_backend.py

@@ -288,7 +288,13 @@ class SchemaStatementTestCase(unittest.TestCase):
         conn = db.connect()
         init_postgres_schema(conn)
         executed = [sql for sql, _ in fake.conn.cursor_obj.executed]
-        self.assertEqual(len(executed), len(POSTGRES_SCHEMA))
+        # 改动理由：init_postgres_schema 现在还要给老库 projects 表补
+        # repo_url 列（PRAGMA 查列 + ALTER TABLE），所以执行语句数
+        # 不再精确等于 schema 语句数；断言改为“schema 语句全部执行
+        # 且补列语句也执行”，不放宽 schema 本身的覆盖要求。
+        for stmt in POSTGRES_SCHEMA:
+            self.assertIn(stmt, executed)
+        self.assertTrue(any("repo_url" in sql for sql in executed))
         self.assertTrue(fake.conn.committed)
 
 

cloud/tests/test_prod_gaps.py

@@ -16,6 +16,7 @@ import tempfile
 import threading
 import time
 import unittest
+from unittest import mock
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -309,9 +310,14 @@ class MetricsAndBackupTestCase(unittest.TestCase):
             conn.close()
 
     def test_postgres_backend_without_driver_fails_before_file(self):
+        # 环境无关：用 mock 模拟驱动缺失，而不是依赖本机真的没装驱动
+        # （另一任务已在本 VM 装了 psycopg2 + Postgres，原写法在该环境下
+        # 前提不存在而失败，并非代码回归；CI runner 无驱动时原逻辑同样成立）。
         other_db = os.path.join(self.tmp.name, "pg.db")
-        with self.assertRaises(PostgresDriverMissingError):
-            create_server(other_db, "127.0.0.1", 0, backend="postgres")
+        with mock.patch("cloud.db._load_postgres_driver",
+                         side_effect=PostgresDriverMissingError("no driver")):
+            with self.assertRaises(PostgresDriverMissingError):
+                create_server(other_db, "127.0.0.1", 0, backend="postgres")
         self.assertFalse(os.path.exists(other_db))
 
 

cloud/tests/test_project_repo.py

@@ -0,0 +1,397 @@
+"""项目级 repo_url 单测：校验、API、老库迁移、resolve、worker 派发。
+
+只用标准库与假 runner/假执行器，不跑真实模型回合、不联网、
+不用真实 git clone（clone 逻辑用 mock runner 模拟 git 的副作用：
+建出目标目录 + .git；真实公开仓库 clone 列入真实环境验证清单）。
+测试里出现的 URL 与密钥均为占位值，绝无真实凭证。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.repos import (
+        RepoError,
+        checkout_dir,
+        resolve_repo,
+        validate_repo_url,
+    )
+    from cloud.server import create_server, 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.auth import AuthStore
+    from cloud.repos import (
+        RepoError,
+        checkout_dir,
+        resolve_repo,
+        validate_repo_url,
+    )
+    from cloud.server import create_server, init_db
+    from cloud.worker import Worker
+
+
+def _git_runner_factory(behavior, calls, side_effect=None):
+    """behavior: dict，键为命令首段特征，值为 (returncode, stdout, stderr)。
+
+    clone 命令默认成功且模拟 git 副作用（建出目标目录 + .git）；
+    调用方可覆写任意命令的行为。
+    """
+    def runner(cmd):
+        calls.append(list(cmd))
+        if cmd[:2] == ["git", "clone"]:
+            key = "clone"
+            rc, out, err = behavior.get(
+                key, (0, "", ""))
+            if rc == 0 and side_effect is None:
+                target = cmd[-1]
+                os.makedirs(os.path.join(target, ".git"), exist_ok=True)
+            return SimpleNamespace(returncode=rc, stdout=out, stderr=err)
+        if cmd[:3] == ["git", "-C", cmd[2]] and "remote" in cmd:
+            rc, out, err = behavior.get(
+                "get-url", (0, behavior.get("origin", ""), ""))
+            return SimpleNamespace(returncode=rc, stdout=out, stderr=err)
+        if "fetch" in cmd:
+            rc, out, err = behavior.get("fetch", (0, "", ""))
+            return SimpleNamespace(returncode=rc, stdout=out, stderr=err)
+        return SimpleNamespace(returncode=0, stdout="", stderr="")
+    return runner
+
+
+class ValidateRepoUrlTestCase(unittest.TestCase):
+    def test_valid_https_and_git_scp(self):
+        self.assertEqual(
+            validate_repo_url("https://github.com/openqj/codex-app-server.git"),
+            "https://github.com/openqj/codex-app-server.git")
+        self.assertEqual(
+            validate_repo_url("git@github.com:openqj/codex-app-server.git"),
+            "git@github.com:openqj/codex-app-server.git")
+
+    def test_rejects_userinfo_file_ssh_local_empty(self):
+        bad = [
+            "https://user:pass@github.com/o/r.git",
+            "https://user@github.com/o/r.git",
+            "file:///tmp/repo",
+            "ssh://git@github.com/o/r.git",
+            "ssh://git:pass@github.com/o/r.git",
+            "/tmp/local-repo",
+            "./relative",
+            "",
+            "   ",
+            "https://",
+            "git@github.com",
+            "https://github.com/o/r.git\n",
+        ]
+        for url in bad:
+            with self.subTest(url=url):
+                with self.assertRaises(ValueError):
+                    validate_repo_url(url)
+        with self.assertRaises(ValueError):
+            validate_repo_url(None)
+        with self.assertRaises(ValueError):
+            validate_repo_url(123)
+
+
+class CheckoutDirTestCase(unittest.TestCase):
+    def test_sanitizes_and_stays_inside(self):
+        tmp = tempfile.TemporaryDirectory()
+        base = os.path.join(tmp.name, "repos")
+        path = checkout_dir(base, "abc123")
+        self.assertEqual(path, os.path.join(os.path.abspath(base), "abc123"))
+        # 穿越尝试被过滤，仍在 repos_dir 之下。
+        path2 = checkout_dir(base, "../../etc")
+        self.assertTrue(os.path.commonpath(
+            [os.path.abspath(base), path2]) == os.path.abspath(base))
+        tmp.cleanup()
+
+
+class ResolveRepoTestCase(unittest.TestCase):
+    def test_clone_when_missing(self):
+        tmp = tempfile.TemporaryDirectory()
+        calls = []
+        runner = _git_runner_factory({}, calls)
+        url = "https://github.com/openqj/codex-app-server.git"
+        path = resolve_repo(url, os.path.join(tmp.name, "repos"),
+                            "proj1", runner=runner)
+        self.assertTrue(os.path.isdir(os.path.join(path, ".git")))
+        self.assertEqual(calls[0][:3], ["git", "clone", url])
+        tmp.cleanup()
+
+    def test_clone_failure_mentions_no_stored_credentials(self):
+        tmp = tempfile.TemporaryDirectory()
+        calls = []
+        runner = _git_runner_factory(
+            {"clone": (128, "", "fatal: Authentication failed")}, calls)
+        with self.assertRaises(RepoError) as ctx:
+            resolve_repo("https://github.com/o/private.git",
+                         os.path.join(tmp.name, "repos"),
+                         "proj1", runner=runner)
+        self.assertIn("控制面不存仓库凭证", str(ctx.exception))
+        tmp.cleanup()
+
+    def test_existing_matching_origin_fetches(self):
+        tmp = tempfile.TemporaryDirectory()
+        repos = os.path.join(tmp.name, "repos")
+        target = checkout_dir(repos, "proj1")
+        os.makedirs(os.path.join(target, ".git"))
+        url = "https://github.com/openqj/codex-app-server.git"
+        calls = []
+        runner = _git_runner_factory({"origin": url}, calls)
+        path = resolve_repo(url, repos, "proj1", runner=runner)
+        self.assertEqual(path, target)
+        self.assertTrue(any("fetch" in c for c in calls))
+        tmp.cleanup()
+
+    def test_existing_mismatched_origin_refuses(self):
+        tmp = tempfile.TemporaryDirectory()
+        repos = os.path.join(tmp.name, "repos")
+        target = checkout_dir(repos, "proj1")
+        os.makedirs(os.path.join(target, ".git"))
+        calls = []
+        runner = _git_runner_factory(
+            {"origin": "https://github.com/other/repo.git"}, calls)
+        with self.assertRaises(RepoError) as ctx:
+            resolve_repo("https://github.com/openqj/codex-app-server.git",
+                         repos, "proj1", runner=runner)
+        self.assertIn("不一致", str(ctx.exception))
+        tmp.cleanup()
+
+    def test_existing_not_git_refuses(self):
+        tmp = tempfile.TemporaryDirectory()
+        repos = os.path.join(tmp.name, "repos")
+        os.makedirs(checkout_dir(repos, "proj1"))
+        calls = []
+        runner = _git_runner_factory({}, calls)
+        with self.assertRaises(RepoError):
+            resolve_repo("https://github.com/o/r.git",
+                         repos, "proj1", runner=runner)
+        self.assertEqual(calls, [])
+        tmp.cleanup()
+
+    def test_fetch_failure_refuses(self):
+        tmp = tempfile.TemporaryDirectory()
+        repos = os.path.join(tmp.name, "repos")
+        target = checkout_dir(repos, "proj1")
+        os.makedirs(os.path.join(target, ".git"))
+        url = "https://github.com/o/r.git"
+        calls = []
+        runner = _git_runner_factory(
+            {"origin": url, "fetch": (128, "", "fatal: unable to access")},
+            calls)
+        with self.assertRaises(RepoError) as ctx:
+            resolve_repo(url, repos, "proj1", runner=runner)
+        self.assertIn("fetch", str(ctx.exception))
+        tmp.cleanup()
+
+
+class _FakeExecutor:
+    def __init__(self):
+        self.seen = []
+
+    def _host_name(self, task):
+        return task["id"]
+
+    def dispatch_task(self, task):
+        self.seen.append(dict(task))
+        return {"executor": "local", "task_id": task["id"],
+                "status": "completed", "stdout": "ok", "returncode": 0,
+                "host_status": None}
+
+
+def _make_worker_db(tmp, project_repo_url="__unset__"):
+    """建库 + 项目行 + queued 任务行，返回 (db_path, project_id)。"""
+    db_path = os.path.join(tmp, "control.db")
+    init_db(db_path)
+    project_id = "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"
+    conn = sqlite3.connect(db_path)
+    try:
+        if project_repo_url == "__unset__":
+            conn.execute(
+                "INSERT INTO projects(id, name, owner_id, created_at)"
+                " VALUES(?, 'p', 'owner', 0)", (project_id,))
+        else:
+            conn.execute(
+                "INSERT INTO projects(id, name, owner_id, created_at,"
+                " repo_url) VALUES(?, 'p', 'owner', 0, ?)",
+                (project_id, project_repo_url))
+        conn.execute(
+            "INSERT INTO tasks(id, project_id, prompt, status,"
+            " depends_on, priority, created_by, created_at, updated_at,"
+            " last_heartbeat) VALUES('t1', ?, 'do it', 'queued',"
+            " '[]', 0, 'system', 0, 0, NULL)", (project_id,))
+        conn.commit()
+    finally:
+        conn.close()
+    return db_path, project_id
+
+
+class WorkerProjectRepoTestCase(unittest.TestCase):
+    def test_dispatch_uses_project_repo(self):
+        tmp = tempfile.TemporaryDirectory()
+        url = "https://github.com/openqj/codex-app-server.git"
+        db_path, _ = _make_worker_db(tmp.name, url)
+        repos = os.path.join(tmp.name, "repos")
+        calls = []
+        executor = _FakeExecutor()
+        worker = Worker(db_path, "host.py", repo=None,
+                        env={"XCM_API_KEY": "test-placeholder"},
+                        env_key=None, executor=executor,
+                        repos_dir=repos,
+                        repos_runner=_git_runner_factory({}, calls),
+                        retry_backoff_seconds=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(len(executor.seen), 1)
+        self.assertEqual(executor.seen[0].get("repo"),
+                         checkout_dir(repos,
+                                      "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"))
+        conn = sqlite3.connect(db_path)
+        try:
+            status = conn.execute(
+                "SELECT status FROM tasks WHERE id='t1'").fetchone()[0]
+        finally:
+            conn.close()
+        self.assertEqual(status, "completed")
+        tmp.cleanup()
+
+    def test_no_repo_url_uses_default(self):
+        tmp = tempfile.TemporaryDirectory()
+        db_path, _ = _make_worker_db(tmp.name, "__unset__")
+        executor = _FakeExecutor()
+        worker = Worker(db_path, "host.py", repo="/default/repo",
+                        env={"XCM_API_KEY": "test-placeholder"},
+                        env_key=None, executor=executor,
+                        repos_dir=os.path.join(tmp.name, "repos"),
+                        retry_backoff_seconds=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        # task 本身无 repo 字段：执行器走自身默认 repo，不被改写。
+        self.assertNotIn("repo", executor.seen[0])
+        tmp.cleanup()
+
+    def test_repo_resolve_failure_is_failed_not_completed(self):
+        tmp = tempfile.TemporaryDirectory()
+        url = "https://github.com/o/private.git"
+        db_path, _ = _make_worker_db(tmp.name, url)
+        executor = _FakeExecutor()
+        worker = Worker(db_path, "host.py", repo=None,
+                        env={"XCM_API_KEY": "test-placeholder"},
+                        env_key=None, executor=executor,
+                        repos_dir=os.path.join(tmp.name, "repos"),
+                        repos_runner=_git_runner_factory(
+                            {"clone": (128, "",
+                                       "fatal: Authentication failed")},
+                            []),
+                        retry_backoff_seconds=0)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("仓库解析失败", result["error"])
+        self.assertEqual(executor.seen, [])
+        conn = sqlite3.connect(db_path)
+        try:
+            status, error = conn.execute(
+                "SELECT status, error_text FROM tasks WHERE id='t1'"
+            ).fetchone()
+        finally:
+            conn.close()
+        self.assertEqual(status, "failed")
+        self.assertIn("仓库解析失败", error)
+        tmp.cleanup()
+
+
+class ProjectRepoApiTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        # 老库：projects 表不含 repo_url 列，init_db 必须幂等补列。
+        conn = sqlite3.connect(self.db_path)
+        conn.execute(
+            "CREATE TABLE projects(id TEXT PRIMARY KEY, name TEXT,"
+            " owner_id TEXT, created_at REAL)")
+        conn.execute(
+            "CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT,"
+            " prompt TEXT, status TEXT, idempotency_key TEXT,"
+            " depends_on TEXT, priority INTEGER, created_by TEXT,"
+            " created_at REAL, updated_at REAL, last_heartbeat REAL,"
+            " UNIQUE(project_id, idempotency_key))")
+        conn.commit()
+        conn.close()
+        auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = auth.create_user("owner")
+        auth.conn.close()
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None):
+        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        payload = json.dumps(body).encode("utf-8") if body is not None else None
+        headers = {"Content-Type": "application/json"}
+        if key is not None:
+            headers["X-API-Key"] = key
+        try:
+            conn.request(method, path, body=payload, headers=headers)
+            resp = conn.getresponse()
+            return resp.status, json.loads(resp.read() or b"{}")
+        finally:
+            conn.close()
+
+    def test_old_db_migrated_and_create_with_repo_url(self):
+        conn = sqlite3.connect(self.db_path)
+        cols = {r[1] for r in conn.execute(
+            "PRAGMA table_info(projects)").fetchall()}
+        conn.close()
+        self.assertIn("repo_url", cols)
+        status, project = self._req(
+            "POST", "/projects",
+            {"name": "p1",
+             "repo_url": "https://github.com/openqj/codex-app-server.git"},
+            self.owner_key)
+        self.assertEqual(status, 201, project)
+        self.assertEqual(
+            project["repo_url"],
+            "https://github.com/openqj/codex-app-server.git")
+        status, listed = self._req("GET", "/projects", key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(len(listed["projects"]), 1)
+        self.assertEqual(
+            listed["projects"][0]["repo_url"],
+            "https://github.com/openqj/codex-app-server.git")
+
+    def test_create_without_repo_url_is_none(self):
+        status, project = self._req(
+            "POST", "/projects", {"name": "p2"}, self.owner_key)
+        self.assertEqual(status, 201, project)
+        self.assertIsNone(project["repo_url"])
+
+    def test_create_with_bad_repo_url_is_400_and_not_created(self):
+        for bad in ("https://user:pass@github.com/o/r.git",
+                    "file:///tmp/r", "ssh://git@github.com/o/r.git",
+                    "/tmp/local"):
+            status, body = self._req(
+                "POST", "/projects",
+                {"name": "bad", "repo_url": bad}, self.owner_key)
+            self.assertEqual(status, 400, (bad, body))
+        status, listed = self._req("GET", "/projects", key=self.owner_key)
+        self.assertEqual(listed["projects"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/worker.py

@@ -82,6 +82,18 @@ interrupted，不用执行结果覆盖它。
 按任务派生的容器名 docker kill，不只改数据库状态；本机无
 docker 时 DockerExecutor 在调 runner 前就明确报错，任务标
 failed 且 error_text 含 docker 缺失，绝不静默降级为本地执行。
+
+项目仓库：run_one 派发前查任务所属项目的 repo_url（projects
+表），非空时经 cloud.repos.resolve_repo 在 repos_dir（CLI
+--repos-dir，默认 ./repos）下 clone/fetch 该项目仓库，设
+task["repo"] 再交给执行器；Local 与 Docker 执行器都已读
+task["repo"]，同一条路径。repo_url 为空时走原有默认 repo；
+解析失败时任务直接 failed，错误文本写清是仓库解析失败。
+Docker 下 repos_dir 必须放在 extra-mounts 已挂载的目录下，
+否则容器内 repo 路径不存在（见 cloud/README.md）。
+推送到远端不在本轮范围：任务在本地检出上执行、本地 worktree
+合并仍按 host.py 现有流程，git push 需要仓库写凭证，控制面
+不存、不推。
 """
 
 import argparse
@@ -100,6 +112,7 @@ from cloud.db import connect as _db_connect
 from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, is_fallbackable, record_fallback
 from cloud.queue import budget_exceeded, ready_tasks
+from cloud.repos import RepoError, resolve_repo
 from cloud.sandbox import DockerExecutor, LocalExecutor
 from cloud.sandbox import _default_runner as _sandbox_default_runner
 
@@ -194,7 +207,8 @@ class Worker:
                  max_turns=None, max_cost_usd=None, executor=None,
                  cost_store=None, process_factory=None,
                  task_timeout=1800, max_retries=1,
-                 retry_backoff_seconds=5.0):
+                 retry_backoff_seconds=5.0, repos_dir="./repos",
+                 repos_runner=None):
         self.db_path = str(db_path)
         self.host_py_path = host_py_path
         self.config_path = config_path
@@ -232,6 +246,12 @@ class Worker:
         self.task_timeout = task_timeout
         self.max_retries = max_retries
         self.retry_backoff_seconds = retry_backoff_seconds
+        # 项目仓库检出根目录：CLI --repos-dir 可调（默认 ./repos）；
+        # Docker 执行器下它必须放在 extra-mounts 已挂载的目录下，
+        # 否则容器内 task["repo"] 路径不存在——见 cloud/README.md。
+        # repos_runner 可注入（单测用假 git runner），缺省走真实 git。
+        self.repos_dir = repos_dir
+        self.repos_runner = repos_runner
 
     # -- 成本存储（惰性建表，避免只认领不执行时也建表）--------------
     @property
@@ -615,6 +635,53 @@ class Worker:
                 "stdout": out or "", "stderr": err or "",
                 "returncode": proc.returncode, "host_status": None}
 
+    # -- 项目仓库 ------------------------------------------------------
+    def _project_repo_url(self, project_id):
+        """查任务所属项目的 repo_url；查不到项目、列缺失、表缺失时
+        一律返回 None（走原有默认 repo，不报错、不炸掉派发）。
+        repo_url 本身只在 projects 表里，绝不从任务 prompt 里猜。"""
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                row = conn.execute(
+                    "SELECT repo_url FROM projects WHERE id = ?",
+                    (project_id,)).fetchone()
+            finally:
+                conn.close()
+        except Exception:
+            return None
+        if row is None:
+            return None
+        try:
+            url = row["repo_url"]
+        except (KeyError, IndexError, TypeError):
+            return None
+        return url or None
+
+    def _apply_project_repo(self, task):
+        """派发前按项目 repo_url 解析仓库，返回错误文本或 None。
+
+        repo_url 为空：不动 task，返回 None（走原有默认 repo）。
+        repo_url 非空：调 cloud.repos.resolve_repo 得到本地检出，
+        设 task["repo"] 再交给执行器——LocalExecutor 与
+        DockerExecutor 的 dispatch_task 都已读 task["repo"]，同一条
+        路径，不另起分支。resolve 失败返回错误文本，调用方据此把
+        任务置 failed（仓库解析失败绝不标 completed，也不在此重试，
+        避免对不可用的仓库反复 clone 烧时间）。
+        """
+        repo_url = self._project_repo_url(task.get("project_id"))
+        if not repo_url:
+            return None
+        try:
+            task["repo"] = resolve_repo(
+                repo_url, self.repos_dir, task.get("project_id"),
+                runner=self.repos_runner)
+        except (RepoError, ValueError) as exc:
+            return "仓库解析失败: %s" % exc
+        except Exception as exc:
+            return "仓库解析失败: %s" % exc
+        return None
+
     # -- 执行 ----------------------------------------------------------
     def _run_attempt(self, task):
         """单次执行尝试（不含认领、key 门、重试、成本与最终审计）。
@@ -735,6 +802,25 @@ class Worker:
                     "stdout": "", "error": error, "events": [],
                     "cost_record_id": None}
 
+        # 项目仓库解析：认领 + 缺 key 门之后、真正执行之前。
+        # repo_url 为空时不动（默认 repo）；解析失败时任务直接 failed，
+        # 错误文本写清是仓库解析失败，绝不标 completed，也不在此重试。
+        repo_error = self._apply_project_repo(task)
+        if repo_error is not None:
+            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())
+            self._audit(task["project_id"], "worker_repo_failed", repo_error)
+            log_event(self.logger, "worker_repo_failed", task_id=task_id,
+                      project_id=task["project_id"], status="failed")
+            return {"dispatched": True, "task_id": task_id,
+                    "host_name": self.executor._host_name(task)
+                    if hasattr(self.executor, "_host_name") else task_id,
+                    "status": "failed", "executor": self._executor_name(),
+                    "stdout": "", "error": repo_error, "events": [],
+                    "cost_record_id": None}
+
         events = []
         fallback_count = 0
         result = None
@@ -939,6 +1025,12 @@ def main(argv=None):
                         help="host.py 路径，缺省为 cloud/ 上一级的 host.py")
     parser.add_argument("--config", default=None)
     parser.add_argument("--repo", default=None)
+    parser.add_argument("--repos-dir", default="./repos",
+                        help="项目仓库检出根目录（默认 ./repos）；项目有"
+                             " repo_url 时 worker 在此 clone/fetch 各项目"
+                             "仓库；Docker 执行器下它必须放在"
+                             " --docker-extra-mounts 已挂载的目录下，"
+                             "否则容器内 task repo 路径不存在")
     parser.add_argument("--env-key", default=None)
     parser.add_argument("--codex-config", default=None,
                         help="含 env_key 行的 codex 配置文件（只读名字）")
@@ -988,7 +1080,8 @@ def main(argv=None):
                     executor=_build_cli_executor(args, host_py),
                     task_timeout=args.task_timeout,
                     max_retries=args.max_retries,
-                    retry_backoff_seconds=args.retry_backoff_seconds)
+                    retry_backoff_seconds=args.retry_backoff_seconds,
+                    repos_dir=args.repos_dir)
     # CLI 启动清理兜底（常驻与单次都做）：清掉上一轮残留容器。
     worker.cleanup_stale_containers()
     # CLI 起 worker 前先建表/迁移：全新库直接起循环不应因缺表而炸。