feat: 任务附件——server/worker/console 链路打通（+ popen goal 新语义补漏）

server：POST /uploads（multipart，413 预检、路径/数量/大小校验、 24h 过期包、used 防一包多用）+ 建任务绑定（同事务写附件行 + 包置 used，cloud 执行器 400）+ 列表/详情带 attachments + 取消删暂存 + 分支带走附件清单；db.py PG 建表 + init_db 补表。 worker：_task_seed_dir（无附件/目录已删静默 None）+ dispatch/ goal 三路（同步本地、popen、Docker 只读挂载 /tmp/codex-seed） + cloud 执行器兜底失败 + 终局删暂存。 console：文件/文件夹/图片三路输入，先上传拿 upload_id 再建任务，列表 [附件×n]、详情列附件路径。附带：popen 的 goal 命令此前强制要求 verify_command，goal 新语义任务到本地 worker（走 popen）会直接报错，改成与同步 goal_task 一致（无验收不传 --verify，轮次 0 原样传）。测试：新增 test_task_attachments.py 15 个。
2026-10-27 04:30:43 +08:00
1 parent 9537e6f0d2
commit 7c444a688b
@@ -43,6 +43,7 @@ import json
 import logging
 import os
 import re
+import shutil
 import socket
 import sqlite3
 import threading
@@ -64,6 +65,15 @@ from cloud.db import PostgresOperationalError as _PgOperationalError
 from cloud.kb import KBStore
 from cloud.memory import MemoryStore
 from cloud.pr import PrSkipped, PrStepError
+from cloud.uploads import (
+    BUNDLE_TTL_SECONDS,
+    MAX_RAW_BYTES,
+    parse_multipart,
+    parse_paths_field,
+    save_bundle,
+    upload_root as _upload_root,
+    validate_bundle,
+)
 from cloud.pr_exec import (enqueue_pr_job, execute_task_pr,
                           find_open_pr_job)
 from cloud.workers import (
@@ -479,6 +489,61 @@ def migrate_projects_table(conn):
        conn.execute("ALTER TABLE projects ADD COLUMN auto_pr INTEGER")


+def _upload_dir_for(server):
+    """附件暂存根目录：server 属性（单测注入）> 环境变量 > 默认路径
+    （与 cloud.uploads.upload_root 同款解析，worker 侧也用它）。"""
+    d = getattr(server, "upload_dir", None)
+    if d:
+        return str(d)
+    return _upload_root()
+
+
+def _ensure_uploads_tables(conn):
+    """幂等建附件包与任务附件表：server init_db 调它，老库第一次
+    跑到这里时自动补表，不丢数据。uploads 记包的归属与状态
+    （pending/used，任务绑定时置 used，防一包多用）；
+    task_attachments 记任务绑定的每个文件（相对路径 + 大小），
+    worker 派发时按 bundle_id 找到暂存目录。"""
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS uploads("
+        "id TEXT PRIMARY KEY, "
+        "created_by TEXT, "
+        "created_at DOUBLE PRECISION, "
+        "status TEXT)"
+    )
+    conn.execute(
+        "CREATE TABLE IF NOT EXISTS task_attachments("
+        "task_id TEXT, "
+        "bundle_id TEXT, "
+        "rel_path TEXT, "
+        "size INTEGER, "
+        "PRIMARY KEY(task_id, rel_path))"
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_attachments_task"
+        " ON task_attachments(task_id)")
+
+
+def _attachments_map(conn, task_ids):
+    """一批任务的附件相对路径（按路径排序）：{task_id: [rel_path]}。
+    任务列表接口用一次查询取全量，不按任务逐条查。"""
+    out = {tid: [] for tid in task_ids}
+    if not task_ids:
+        return out
+    placeholders = ", ".join("?" * len(task_ids))
+    rows = conn.execute(
+        "SELECT task_id, rel_path FROM task_attachments"
+        " WHERE task_id IN (%s) ORDER BY rel_path" % placeholders,
+        tuple(task_ids),
+    ).fetchall()
+    for row in rows:
+        tid = row[0] if isinstance(row, tuple) else row["task_id"]
+        rel = row[1] if isinstance(row, tuple) else row["rel_path"]
+        if tid in out:
+            out[tid].append(rel)
+    return out
+
+
 def _ensure_followups_table(conn):
    """幂等建后续消息表：server init_db 与各 followup 接口都调它，
    老库第一次跑到这里时自动补表，不丢数据。DOUBLE PRECISION 在
@@ -614,6 +679,7 @@ def init_db(db_path):
            migrate_projects_table(conn)
            _ensure_followups_table(conn)
            _ensure_environments_table(conn)
+            _ensure_uploads_tables(conn)
            conn.commit()
            return
        conn.execute(
@@ -644,6 +710,7 @@ def init_db(db_path):
        migrate_projects_table(conn)
        _ensure_followups_table(conn)
        _ensure_environments_table(conn)
+        _ensure_uploads_tables(conn)
        # worker 每轮 poll 都按 status 取任务（见 load_tasks 的
        # WHERE status IN (...)），无索引时是全表扫描；任务表随
        # 时间变大后索引是必须的。IF NOT EXISTS 保证老库幂等。
@@ -676,7 +743,7 @@ def _row_get(row, name):
        return None


-def _task_to_dict(row):
+def _task_to_dict(row, attachments=None):
    depends_on = row["depends_on"]
    try:
        depends_on = json.loads(depends_on) if depends_on else []
@@ -709,6 +776,7 @@ def _task_to_dict(row):
        "verify_command": _row_get(row, "verify_command"),
        "goal_max_rounds": _row_get(row, "goal_max_rounds"),
        "goal_mode": _row_get(row, "goal_mode") or 0,
+        "attachments": list(attachments or []),
        "worktree_host": _row_get(row, "worktree_host"),
        "model": _row_get(row, "model"),
        "effort": _row_get(row, "effort"),
@@ -1027,6 +1095,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                self._send_json(404, {"error": "not found"})
        elif path == "/tasks" and method == "POST":
            self._create_task(user_id)
+        elif path == "/uploads" and method == "POST":
+            self._create_upload(user_id)
        elif path == "/tasks" and method == "GET":
            self._list_tasks(user_id, query)
        elif path == "/admin/sweep" and method == "POST":
@@ -1765,6 +1835,65 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
        })

    # -- 任务 ----------------------------------------------------------
+    # -- 任务附件 ------------------------------------------------------
+    def _create_upload(self, user_id):
+        """上传一包附件，返回 upload_id（建任务时绑定）。
+
+        权限：任何持有效 key 的用户都可上传（包本身不挂项目，
+        归属在建任务时校验）；包不直接进任务，必须经 _create_task
+        绑定才生效。413 预检在读 body 之前按 Content-Length 卡，
+        防止超大 body 先占内存；文件级上限由 validate_bundle 卡。
+        """
+        content_type = self.headers.get("Content-Type", "")
+        if "multipart/form-data" not in content_type:
+            self._send_json(400, {
+                "error": "Content-Type 必须是 multipart/form-data"})
+            return
+        try:
+            length = int(self.headers.get("Content-Length") or 0)
+        except ValueError:
+            length = 0
+        if length > MAX_RAW_BYTES:
+            self._send_json(413, {"error": "upload too large"})
+            return
+        raw = self._read_body()
+        try:
+            fields, files = parse_multipart(raw, content_type)
+            paths = parse_paths_field(fields, files)
+        except ValueError as exc:
+            self._send_json(400, {"error": str(exc)})
+            return
+        if not files:
+            self._send_json(400, {"error": "no files uploaded"})
+            return
+        if len(paths) != len(files):
+            self._send_json(400, {
+                "error": "附件路径与文件数量不一致"})
+            return
+        err = validate_bundle(paths, [len(data) for _name, data in files])
+        if err:
+            self._send_json(400, {"error": err})
+            return
+        root = _upload_dir_for(self.server)
+        os.makedirs(root, exist_ok=True)
+        bundle_id, saved = save_bundle(
+            root, paths, [data for _name, data in files])
+        conn = self._db()
+        try:
+            conn.execute(
+                "INSERT INTO uploads(id, created_by, created_at, status)"
+                " VALUES(?, ?, ?, 'pending')",
+                (bundle_id, user_id, time.time()),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        self._send_json(201, {
+            "upload_id": bundle_id,
+            "files": [{"path": path, "size": size}
+                      for path, size in saved],
+        })
+
    def _create_task(self, user_id):
        body, err = self._read_json()
        if err:
@@ -1921,6 +2050,55 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                return
            if executor is None and env["executor"]:
                executor = env["executor"]
+        # 附件绑定（upload_id 由 POST /uploads 发出）：包必须存在、
+        # 归属当前用户、未使用、24 小时内、暂存目录还在；cloud
+        # 执行器没有 worktree 可 seed，直接拒绝。校验在 INSERT
+        # 之前做，绑定写与任务 INSERT 同一事务，失败不留半截。
+        bundle_files = []
+        upload_id = body.get("upload_id")
+        if upload_id is not None:
+            if not isinstance(upload_id, str) or not upload_id:
+                self._send_json(400, {
+                    "error": "upload_id must be a non-empty string"})
+                return
+            if executor == "cloud":
+                self._send_json(400, {
+                    "error": "附件不支持 cloud 执行器"})
+                return
+            conn0 = self._db()
+            try:
+                brow = conn0.execute(
+                    "SELECT * FROM uploads WHERE id = ?",
+                    (upload_id,),
+                ).fetchone()
+            finally:
+                conn0.close()
+            if brow is None:
+                self._send_json(400, {"error": "upload_id 不存在"})
+                return
+            if brow["created_by"] != user_id:
+                self._send_json(403, {"error": "forbidden"})
+                return
+            if brow["status"] == "used":
+                self._send_json(409, {"error": "附件包已使用"})
+                return
+            if time.time() - brow["created_at"] > BUNDLE_TTL_SECONDS:
+                self._send_json(400, {"error": "附件包已过期"})
+                return
+            bundle_dir = os.path.join(
+                _upload_dir_for(self.server), upload_id)
+            if not os.path.isdir(bundle_dir):
+                self._send_json(400, {"error": "附件暂存目录缺失"})
+                return
+            for dirpath, _dirnames, filenames in os.walk(bundle_dir):
+                for fn in sorted(filenames):
+                    full = os.path.join(dirpath, fn)
+                    rel = os.path.relpath(full, bundle_dir)
+                    bundle_files.append((rel, os.path.getsize(full)))
+            bundle_files.sort()
+            if not bundle_files:
+                self._send_json(400, {"error": "附件包为空"})
+                return
        now = time.time()
        conn = self._db()
        try:
@@ -1931,7 +2109,9 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                    (project_id, idempotency_key),
                ).fetchone()
                if row:
-                    self._send_json(200, _task_to_dict(row))
+                    self._send_json(200, _task_to_dict(
+                        row, _attachments_map(
+                            conn, [row["id"]])[row["id"]]))
                    return
            task_id = uuid.uuid4().hex
            try:
@@ -1950,6 +2130,18 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                     model, effort, approval, plan, goal_mode,
                     environment_id),
                )
+                for rel_path, size in bundle_files:
+                    conn.execute(
+                        "INSERT INTO task_attachments"
+                        "(task_id, bundle_id, rel_path, size)"
+                        " VALUES(?, ?, ?, ?)",
+                        (task_id, upload_id, rel_path, size),
+                    )
+                if upload_id is not None:
+                    conn.execute(
+                        "UPDATE uploads SET status = 'used' WHERE id = ?",
+                        (upload_id,),
+                    )
                conn.commit()
            except (sqlite3.IntegrityError, _PgIntegrityError):
                # 并发下同幂等键抢建：回读已有任务，不建重复。
@@ -1959,13 +2151,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                    (project_id, idempotency_key),
                ).fetchone()
                if row:
-                    self._send_json(200, _task_to_dict(row))
+                    self._send_json(200, _task_to_dict(
+                        row, _attachments_map(
+                            conn, [row["id"]])[row["id"]]))
                    return
                raise
            row = conn.execute(
                "SELECT * FROM tasks WHERE id = ?", (task_id,)
            ).fetchone()
-            task = _task_to_dict(row)
+            task = _task_to_dict(
+                row, [rel_path for rel_path, _size in bundle_files])
        finally:
            conn.close()
        with _auth_store(self.server.db_path) as store:
@@ -1989,9 +2184,11 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                " ORDER BY created_at, id",
                (project_id,),
            ).fetchall()
+            amap = _attachments_map(conn, [r["id"] for r in rows])
        finally:
            conn.close()
-        self._send_json(200, {"tasks": [_task_to_dict(r) for r in rows]})
+        self._send_json(200, {"tasks": [
+            _task_to_dict(r, amap[r["id"]]) for r in rows]})

    def _fetch_task(self, task_id):
        conn = self._db()
@@ -2007,7 +2204,12 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
        if row is None:
            self._send_json(404, {"error": "task not found"})
            return
-        task = _task_to_dict(row)
+        conn = self._db()
+        try:
+            atts = _attachments_map(conn, [task_id])[task_id]
+        finally:
+            conn.close()
+        task = _task_to_dict(row, atts)
        with _auth_store(self.server.db_path) as store:
            if not store.check_permission(user_id, task["project_id"], "read"):
                self._send_json(403, {"error": "forbidden"})
@@ -2290,11 +2492,21 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 task["effort"], task["approval"], task["plan"] or 0,
                 task_id),
            )
+            # 分支带走附件清单（worktree 本身已含这些文件，分支
+            # 不再 seed；清单只作展示，bundle 删了也不影响分支）。
+            conn.execute(
+                "INSERT INTO task_attachments"
+                "(task_id, bundle_id, rel_path, size)"
+                " SELECT ?, bundle_id, rel_path, size"
+                " FROM task_attachments WHERE task_id = ?",
+                (new_id, task_id),
+            )
            conn.commit()
            new_task = _task_to_dict(
                conn.execute(
                    "SELECT * FROM tasks WHERE id = ?", (new_id,)
-                ).fetchone()
+                ).fetchone(),
+                _attachments_map(conn, [new_id])[new_id],
            )
        finally:
            conn.close()
@@ -2331,6 +2543,23 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                    "SELECT * FROM tasks WHERE id = ?", (task_id,)
                ).fetchone()
            )
+            # 取消后 worker 不会再跑它：暂存包没人 seed 了，顺手删掉
+            # （任务终局后 worker 侧也会删，两边删同一目录都容忍
+            # 不存在；删失败不挡取消本身）。
+            try:
+                brow = conn.execute(
+                    "SELECT bundle_id FROM task_attachments"
+                    " WHERE task_id = ? LIMIT 1",
+                    (task_id,),
+                ).fetchone()
+                if brow is not None:
+                    bundle_id = brow[0]
+                    shutil.rmtree(
+                        os.path.join(
+                            _upload_dir_for(self.server), bundle_id),
+                        ignore_errors=True)
+            except Exception:
+                pass
        finally:
            conn.close()
        with _auth_store(self.server.db_path) as store: