Merge pull request #69 from openqj/feat-attachments

feat: 任务附件——文件/文件夹/图片进任务工作区

cloud/README.md

@@ -472,6 +472,29 @@ owner_user_id)` 只写 `project_members` 的 owner 成员记录，不建项目
 location，被吃时从备份贴回再 reload。数据仍凭 X-API-Key，
 `/health` 本来就是无鉴权设计且不含敏感信息。
 
+## 任务附件（放进任务工作区）
+
+命令台新建任务时可附文件、文件夹、图片（三路输入），提交时先
+`POST /uploads`（multipart：`paths` JSON 与 `file0…` 一一对应）
+拿 `upload_id`，再 `POST /tasks` 带 `upload_id` 绑定；图片与普通
+文件不区分，一律是 worktree 里的文件（模型能否“看懂”图片取决于
+模型多模态能力，本链路不保证）。派发时 worker 经 host.py
+`--seed-dir` 把暂存目录拷进 worktree（dispatch/goal 首轮才 seed，
+run 续跑不重复拷，文件已在 worktree 里）；Docker 任务把暂存目录
+只读挂载进容器（`/tmp/codex-seed:ro`），cloud 执行器拒绝带附件的
+任务。限制：单包最多 200 文件、单文件 20MB、合计 100MB、整个
+multipart body 200MB（超了 413）；路径拒绝对路径、`..`、空段、
+反斜杠、NUL、盘符；包 24 小时过期、建任务绑定后置 used（一包
+只能用一次，他人包 403、cloud 任务 400）；任务终局后 worker 删
+暂存目录，取消时 server 侧也删（删不到不报错）。暂存根默认
+`/opt/codex-app-server/work/uploads`，`CODEX_UPLOAD_DIR` 可覆盖，
+server 与 worker 两边必须一致（否则 worker 找不到包，派发静默无
+附件）。公网 nginx 须代理 `/uploads`（与 `/tasks` 同批），否则页
+面上传 404。注意：popen 路径的 goal 命令此前强制要求
+verify_command，goal 新语义（无验收）任务到本地 worker 会直接
+报错，本批顺手修成与同步 goal_task 一致（无验收不传 --verify，
+轮次 0 原样传）。
+
 ## 测试
 
 ```bash

cloud/console.html

@@ -173,6 +173,12 @@
   </select>
 </div>
 <div id="goalHint" style="display:none" class="muted">goal 模式：输入框内容即目标，不需要专用 prompt 与验收命令；不限轮次，模型确认达成后在回复末尾输出“目标已达成”即结束，未达成会一直运行，可用取消停止。</div>
+<div class="bar">
+  <span class="muted">附件（放进任务工作区，图片与普通文件不区分）：</span>
+  <input type="file" id="attFiles" multiple title="文件">
+  <input type="file" id="attDir" webkitdirectory title="文件夹">
+  <input type="file" id="attImg" accept="image/*" multiple title="图片">
+</div>
 <div class="bar">
   <button id="createTask" class="primary">提交任务</button>
   <span id="taskMsg" class="muted"></span>
@@ -378,7 +384,7 @@ async function loadTasks() {
       var pr = t.pr_url ? "<a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>#" + esc(t.pr_number) + "</a>" : "";
       return "<tr data-id='" + esc(t.id) + "' class='" + (currentTask && currentTask.id === t.id ? "selected" : "") + "'>" +
         "<td>" + esc(shortId(t.id)) + "</td>" +
-        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + (t.environment_id ? "[环境 " + esc(shortId(t.environment_id)) + "] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
+        "<td>" + (t.plan ? "[计划] " : "") + (t.forked_from ? "[分支] " : "") + ((t.goal_mode || t.verify_command) ? "[目标] " : "") + (t.environment_id ? "[环境 " + esc(shortId(t.environment_id)) + "] " : "") + ((t.attachments && t.attachments.length) ? "[附件×" + t.attachments.length + "] " : "") + esc(String(t.prompt || "").slice(0, 60)) + "</td>" +
         "<td class='st-" + esc(t.status) + "'>" + esc(t.status) + "</td>" +
         "<td>" + esc(t.executor || "") + "</td>" +
         "<td>" + esc(t.worktree_host || "") + "</td>" +
@@ -505,6 +511,7 @@ async function showDetail(taskId) {
       (t.plan ? " ｜ <b>计划模式（只读）</b>" : "") +
       (t.forked_from ? " ｜ 分支自 " + esc(shortId(t.forked_from)) : "") +
       (t.goal_mode ? " ｜ <b>goal 模式（不限轮次，模型宣告完成）</b>" : (t.verify_command ? " ｜ goal 模式（旧语义，验收命令）" : "")) +
+      ((t.attachments && t.attachments.length) ? " ｜ 附件：" + t.attachments.map(esc).join("、") : "") +
       (t.pr_url ? " ｜ <a href='" + esc(t.pr_url) + "' target='_blank' rel='noopener'>PR #" + esc(t.pr_number) + "</a>" : "") +
       "<br>prompt：<br>" + esc(t.prompt || "");
     var body = "";
@@ -614,6 +621,41 @@ $("#createEnvironment").onclick = async function () {
     loadEnvironments();
   } catch (e) { $("#envMsg").innerHTML = "<span class='err'>新建失败：" + esc(e.message) + "</span>"; }
 };
+function gatherAttachments() {
+  // 三路输入合并成 [{path, file}]：文件路用文件名，文件夹路用
+  // webkitRelativePath（含子目录），图片路用文件名；同路径只留
+  // 第一个（跨路重复以先选为准，服务端还会再校验一次）。
+  var out = [], seen = {};
+  function add(list, pathOf) {
+    for (var i = 0; i < list.length; i++) {
+      var p = pathOf(list[i]);
+      if (!p || seen[p]) continue;
+      seen[p] = true;
+      out.push({ path: p, file: list[i] });
+    }
+  }
+  add($("#attFiles").files, function (f) { return f.name; });
+  add($("#attDir").files, function (f) { return f.webkitRelativePath || f.name; });
+  add($("#attImg").files, function (f) { return f.name; });
+  return out;
+}
+async function uploadAttachments(items) {
+  // 先 POST /uploads 拿 upload_id，建任务时绑定；上传失败直接
+  // 抛错，任务不建（避免无附件的任务先跑起来）。注意不能复用
+  // api()（它会固定写 Content-Type: application/json，multipart
+  // 的 boundary 必须由浏览器自己生成）。
+  var fd = new FormData();
+  fd.append("paths", JSON.stringify(items.map(function (x) { return x.path; })));
+  items.forEach(function (x, i) { fd.append("file" + i, x.file); });
+  var r = await fetch("/uploads", {
+    method: "POST",
+    headers: { "X-API-Key": getKey() },
+    body: fd
+  });
+  var d = await r.json().catch(function () { return {}; });
+  if (!r.ok) throw new Error((d && d.error) || ("HTTP " + r.status));
+  return d.upload_id;
+}
 $("#createTask").onclick = async function () {
   var pid = $("#projectSelect").value;
   var prompt = $("#prompt").value.trim();
@@ -643,10 +685,21 @@ $("#createTask").onclick = async function () {
     // 验收命令与轮次（服务端 goal_mode=1 时一律不限轮次）。
     body.goal_mode = 1;
   }
+  var atts = gatherAttachments();
+  if (atts.length) {
+    try {
+      $("#taskMsg").textContent = "上传附件中…";
+      body.upload_id = await uploadAttachments(atts);
+    } catch (e) {
+      $("#taskMsg").innerHTML = "<span class='err'>附件上传失败：" + esc(e.message) + "</span>";
+      return;
+    }
+  }
   try {
     var t = await api("/tasks", "POST", body);
     $("#taskMsg").innerHTML = "<span class='ok'>任务已建：" + esc(shortId(t.id)) + "</span>";
     $("#prompt").value = "";
+    $("#attFiles").value = ""; $("#attDir").value = ""; $("#attImg").value = "";
     loadTasks();
   } catch (e) { $("#taskMsg").innerHTML = "<span class='err'>提交失败：" + esc(e.message) + "</span>"; }
 };

cloud/db.py

@@ -840,6 +840,21 @@ POSTGRES_SCHEMA = (
     "environment_id TEXT, "
     "UNIQUE(project_id, idempotency_key))",
     "CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status)",
+    # 附件包与任务附件：与 server._ensure_uploads_tables 同款，
+    # POSTGRES_SCHEMA 在每次初始化时全量执行，新表自动补建。
+    "CREATE TABLE IF NOT EXISTS uploads("
+    "id TEXT PRIMARY KEY, "
+    "created_by TEXT, "
+    "created_at DOUBLE PRECISION, "
+    "status TEXT)",
+    "CREATE TABLE IF NOT EXISTS task_attachments("
+    "task_id TEXT, "
+    "bundle_id TEXT, "
+    "rel_path TEXT, "
+    "size INTEGER, "
+    "PRIMARY KEY(task_id, rel_path))",
+    "CREATE INDEX IF NOT EXISTS idx_attachments_task"
+    " ON task_attachments(task_id)",
     "CREATE TABLE IF NOT EXISTS task_followups("
     "id TEXT PRIMARY KEY, "
     "task_id TEXT, "

cloud/sandbox.py

@@ -270,7 +270,7 @@ class LocalExecutor(Executor):
         raw = str(task.get("id"))
         return re.sub(r"[^A-Za-z0-9_.-]", "-", raw)
 
-    def dispatch_task(self, task) -> dict:
+    def dispatch_task(self, task, seed_dir=None) -> dict:
         """调 host.py dispatch：建 host 任务并跑一轮 turn。
 
         命令：host.py [--config CFG] [--model M] dispatch --name <host 名>
@@ -281,6 +281,8 @@ class LocalExecutor(Executor):
         model 字段，没有则取构造时传的 model，都没有则不传 --model，
         host.py 用配置文件里的 model（fallback 时 worker 会把任务的
         model 换成 fallback 模型，第二次 dispatch/run 就跑在新模型上）。
+        seed_dir 非空时追加 --seed-dir：host.py 在建 worktree 后、
+        跑 turn 前把暂存附件拷进去（run 续跑不走这条路，不重复拷）。
         """
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
@@ -296,6 +298,8 @@ class LocalExecutor(Executor):
         repo = task.get("repo") or self.repo
         if repo:
             cmd += ["--repo", repo]
+        if seed_dir:
+            cmd += ["--seed-dir", seed_dir]
         return self._result_from_runner(
             task["id"], self.runner(cmd))
 
@@ -342,7 +346,7 @@ class LocalExecutor(Executor):
         return self._result_from_runner(
             task["id"], self.runner(cmd))
 
-    def goal_task(self, task) -> dict:
+    def goal_task(self, task, seed_dir=None) -> dict:
         """调 host.py goal：目标循环（turn→验测→喂回重试）。
 
         命令：host.py [--config CFG] [--model M] goal --name <host 名>
@@ -353,6 +357,7 @@ class LocalExecutor(Executor):
         0 是合法值不能用 or 兜底成 3；旧形态仍按原样传验收命令
         与轮次。轮次缺省 3（任务行没写时），与 host.py 默认一致；
         repo 取任务的 repo 字段，没有则取构造时传的 repo。
+        seed_dir 非空时追加 --seed-dir（同 dispatch 语义）。
         结论解析见 _goal_result_from_runner：host.py goal 两种结局
         都 exit 0，不能沿用 _result_from_runner。
         """
@@ -376,6 +381,8 @@ class LocalExecutor(Executor):
         repo = task.get("repo") or self.repo
         if repo:
             cmd += ["--repo", repo]
+        if seed_dir:
+            cmd += ["--seed-dir", seed_dir]
         return self._goal_result_from_runner(
             task["id"], self.runner(cmd))
 
@@ -424,7 +431,7 @@ class LocalExecutor(Executor):
             "usage": _parse_goal_usage(stdout_full),
         }
 
-    def popen_task(self, task, kind="dispatch"):
+    def popen_task(self, task, kind="dispatch", seed_dir=None):
         """进程组启动 host.py 子进程，返回 Popen（供真取消用）。
 
         start_new_session=True 让子进程自成进程组，worker 取消时
@@ -432,6 +439,10 @@ class LocalExecutor(Executor):
         subprocess）时才用本方法；单测注入的假 runner 走同步路径，
         不走这里。kind=goal 时建 goal 命令（结论解析由 worker 的
         _execute_via_process 按 kind 分发到 _goal_result_from_runner）。
+        goal 语义与同步 goal_task 一致：新形态（goal_mode）无验收
+        命令时 --verify 不传、--max-rounds 传 0（0 不能用 or 兜底
+        成 3）；旧形态仍传验收命令与轮次。seed_dir 只对 dispatch/
+        goal 追加（run 续跑不传，文件已在 worktree 里）。
         """
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
@@ -446,25 +457,31 @@ class LocalExecutor(Executor):
             cmd += ["run", "--task", self._host_name(task),
                     "--prompt", task["prompt"]]
         elif kind == "goal":
-            verify = task.get("verify_command")
-            if not verify:
-                raise ValueError("goal 任务缺少 verify_command")
             goal_text = task.get("goal") or task.get("prompt")
             if not goal_text:
                 raise ValueError("任务缺少 prompt")
-            rounds = task.get("goal_max_rounds") or 3
+            verify = task.get("verify_command")
+            rounds = task.get("goal_max_rounds")
+            if rounds is None:
+                rounds = 3
             cmd += ["goal", "--name", self._host_name(task),
-                    "--task", goal_text, "--verify", verify,
+                    "--task", goal_text]
-                    "--max-rounds", str(rounds)]
+            if verify:
+                cmd += ["--verify", verify]
+            cmd += ["--max-rounds", str(rounds)]
             repo = task.get("repo") or self.repo
             if repo:
                 cmd += ["--repo", repo]
+            if seed_dir:
+                cmd += ["--seed-dir", seed_dir]
         else:
             cmd += ["dispatch", "--name", self._host_name(task),
                     "--task", task["prompt"]]
             repo = task.get("repo") or self.repo
             if repo:
                 cmd += ["--repo", repo]
+            if seed_dir:
+                cmd += ["--seed-dir", seed_dir]
         return subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE, text=True,
                                 start_new_session=True)
@@ -859,8 +876,14 @@ class DockerExecutor(Executor):
         return parsed
 
     def dispatch_task(self, task, timeout=None, extra_mounts=None,
-                      network=None, extra_env=None) -> dict:
+                      network=None, extra_env=None,
-        """容器内调 host.py dispatch：建 host 任务并跑一轮 turn。"""
+                      seed_host_dir=None) -> dict:
+        """容器内调 host.py dispatch：建 host 任务并跑一轮 turn。
+
+        seed_host_dir 非空时把暂存目录只读挂载进容器
+        （/tmp/codex-seed），内层 host.py 用 --seed-dir
+        /tmp/codex-seed 拷进 worktree；挂载整个目录（不是单个
+        文件）是为了一包多文件与子目录。"""
         if not isinstance(task, dict) or task.get("id") is None:
             raise ValueError("任务缺少 id")
         if not task.get("prompt"):
@@ -873,8 +896,12 @@ class DockerExecutor(Executor):
         repo = task.get("repo")
         if repo:
             inner += ["--repo", repo]
+        mounts = list(extra_mounts or [])
+        if seed_host_dir:
+            inner += ["--seed-dir", "/tmp/codex-seed"]
+            mounts.append("%s:/tmp/codex-seed:ro" % seed_host_dir)
         return self._run_in_container(task, inner, timeout=timeout,
-                                      extra_mounts=extra_mounts,
+                                      extra_mounts=mounts or None,
                                       network=network,
                                       extra_env=extra_env)
 
@@ -913,9 +940,13 @@ class DockerExecutor(Executor):
                                       extra_env=extra_env)
 
     def goal_task(self, task, timeout=None, extra_mounts=None,
-                  network=None, extra_env=None) -> dict:
+                  network=None, extra_env=None,
+                  seed_host_dir=None) -> dict:
         """容器内调 host.py goal：目标循环（turn→验测→喂回重试）。
 
+        seed_host_dir 语义同 dispatch_task（只读挂载 + 内层
+        --seed-dir /tmp/codex-seed）。
+
         不走 _run_in_container：它经 _parse_host_result 截头
         stdout，而 goal 结论在末尾，截头会把结论截掉；这里直接
         调 build_run_command + _call_runner，结论用完整 stdout
@@ -944,8 +975,12 @@ class DockerExecutor(Executor):
         repo = task.get("repo")
         if repo:
             inner += ["--repo", repo]
+        mounts = list(extra_mounts or [])
+        if seed_host_dir:
+            inner += ["--seed-dir", "/tmp/codex-seed"]
+            mounts.append("%s:/tmp/codex-seed:ro" % seed_host_dir)
         cmd = self.build_run_command(task, inner,
-                                     extra_mounts=extra_mounts,
+                                     extra_mounts=mounts or None,
                                      network=network,
                                      extra_env=extra_env)
         try:

cloud/server.py

@@ -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:

cloud/tests/test_task_attachments.py

@@ -0,0 +1,418 @@
+"""任务附件单测：上传包校验、建任务绑定、派发 seed、终局清理。
+
+不跑真实模型回合、不联网：server 侧走本地临时端口 + 临时
+SQLite；worker/sandbox 侧注入假 runner 录命令；host seed 侧用
+临时目录。图片与普通文件不区分，一律是 worktree 里的文件。
+"""
+
+import http.client
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+try:
+    from cloud.auth import AuthStore
+    from cloud.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server, init_db
+    from cloud.uploads import (
+        MAX_FILES,
+        parse_multipart,
+        parse_paths_field,
+        save_bundle,
+        validate_bundle,
+        validate_rel_path,
+    )
+    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.sandbox import DockerExecutor, LocalExecutor
+    from cloud.server import create_server, init_db
+    from cloud.uploads import (
+        MAX_FILES,
+        parse_multipart,
+        parse_paths_field,
+        save_bundle,
+        validate_bundle,
+        validate_rel_path,
+    )
+    from cloud.worker import Worker
+
+
+def _multipart_body(boundary, fields, files):
+    """拼一个 multipart body：fields {名: 文本}，files
+    [(part 名, 文件名, bytes)]。"""
+    out = b""
+    for name, value in fields.items():
+        out += ("--%s\r\nContent-Disposition: form-data; name=\"%s\"\r\n\r\n"
+                % (boundary, name)).encode("utf-8")
+        out += value.encode("utf-8") + b"\r\n"
+    for part_name, filename, data in files:
+        out += ("--%s\r\nContent-Disposition: form-data; name=\"%s\"; "
+                "filename=\"%s\"\r\nContent-Type: application/octet-stream"
+                "\r\n\r\n" % (boundary, part_name, filename)).encode("utf-8")
+        out += data + b"\r\n"
+    out += ("--%s--\r\n" % boundary).encode("utf-8")
+    return out
+
+
+class UploadsModuleTestCase(unittest.TestCase):
+    def test_parse_roundtrip_nested_paths(self):
+        boundary = "TESTBOUNDARY"
+        body = _multipart_body(
+            boundary,
+            {"paths": json.dumps(["a.txt", "sub/b.png"])},
+            [("file0", "a.txt", b"hello"),
+             ("file1", "b.png", b"\x89PNG")])
+        fields, files = parse_multipart(
+            body, "multipart/form-data; boundary=" + boundary)
+        self.assertEqual(fields["paths"],
+                         json.dumps(["a.txt", "sub/b.png"]))
+        self.assertEqual([n for n, _d in files], ["a.txt", "b.png"])
+        paths = parse_paths_field(fields, files)
+        self.assertEqual(paths, ["a.txt", "sub/b.png"])
+
+    def test_parse_paths_count_mismatch(self):
+        with self.assertRaises(ValueError):
+            parse_paths_field({"paths": json.dumps(["a.txt"])},
+                              [("file0", b"x"), ("file1", b"y")])
+
+    def test_validate_rel_path(self):
+        for good in ["a.txt", "sub/b/c.png", "a-b_c.d/e f.txt", ".hidden/x"]:
+            self.assertIsNone(validate_rel_path(good), good)
+        for bad in ["/abs", "../x", "a/../b", "a//b", "", "a\\b",
+                    "C:/x", "a\x00b", ".", "..", "a/./b"]:
+            self.assertIsNotNone(validate_rel_path(bad), bad)
+
+    def test_validate_bundle_limits(self):
+        self.assertIsNotNone(validate_bundle(
+            ["f%d.txt" % i for i in range(MAX_FILES + 1)],
+            [1] * (MAX_FILES + 1)))
+        self.assertIsNotNone(validate_bundle(
+            ["big.bin"], [21 * 1024 * 1024]))
+        self.assertIsNotNone(validate_bundle(
+            ["a.bin", "b.bin"], [60 * 1024 * 1024, 60 * 1024 * 1024]))
+        self.assertIsNone(validate_bundle(["a.txt"], [5]))
+
+    def test_save_bundle_roundtrip_and_traversal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            bundle_id, saved = save_bundle(
+                tmp, ["a.txt", "sub/b.txt"], [b"hello", b"world"])
+            self.assertEqual([p for p, _s in saved],
+                             ["a.txt", "sub/b.txt"])
+            with open(os.path.join(tmp, bundle_id, "sub", "b.txt"),
+                      "rb") as f:
+                self.assertEqual(f.read(), b"world")
+            with self.assertRaises(ValueError):
+                save_bundle(tmp, ["../escape.txt"], [b"x"])
+
+
+class ServerAttachmentsTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.upload_dir = os.path.join(self.tmp.name, "uploads")
+        self.auth = AuthStore(self.db_path)
+        self.owner_id, self.owner_key = self.auth.create_user("owner")
+        self.other_id, self.other_key = self.auth.create_user("other")
+        self.server = create_server(self.db_path, "127.0.0.1", 0)
+        self.server.upload_dir = self.upload_dir
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(
+            target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        status, self.project = self._req(
+            "POST", "/projects", {"name": "demo"}, self.owner_key)
+        self.assertEqual(status, 201)
+
+    def tearDown(self):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        try:
+            self.auth.conn.close()
+        except Exception:
+            pass
+        self.tmp.cleanup()
+
+    def _req(self, method, path, body=None, key=None,
+             content_type="application/json"):
+        conn = http.client.HTTPConnection(
+            "127.0.0.1", self.port, timeout=10)
+        hdrs = {}
+        payload = None
+        if key:
+            hdrs["X-API-Key"] = key
+        if body is not None:
+            if isinstance(body, bytes):
+                payload = body
+            else:
+                payload = json.dumps(body)
+            hdrs["Content-Type"] = content_type
+        conn.request(method, path, body=payload, headers=hdrs)
+        resp = conn.getresponse()
+        raw = resp.read()
+        conn.close()
+        try:
+            return resp.status, json.loads(raw.decode("utf-8"))
+        except (ValueError, UnicodeDecodeError):
+            return resp.status, {}
+
+    def _upload(self, paths, contents, key=None):
+        body = _multipart_body(
+            "TESTBOUNDARY", {"paths": json.dumps(paths)},
+            [("file%d" % i, os.path.basename(p) or "f", data)
+             for i, (p, data) in enumerate(zip(paths, contents))])
+        return self._req("POST", "/uploads", body,
+                         key or self.owner_key,
+                         content_type="multipart/form-data;"
+                         " boundary=TESTBOUNDARY")
+
+    def _create_task(self, upload_id=None, executor=None):
+        body = {"project_id": self.project["id"], "prompt": "do work"}
+        if upload_id is not None:
+            body["upload_id"] = upload_id
+        if executor is not None:
+            body["executor"] = executor
+        return self._req("POST", "/tasks", body, self.owner_key)
+
+    def test_upload_and_bind(self):
+        status, up = self._upload(["a.txt", "sub/b.png"],
+                                  [b"hello", b"\x89PNG"])
+        self.assertEqual(status, 201, up)
+        self.assertEqual([f["path"] for f in up["files"]],
+                         ["a.txt", "sub/b.png"])
+        status, task = self._create_task(up["upload_id"])
+        self.assertEqual(status, 201, task)
+        self.assertEqual(task["attachments"], ["a.txt", "sub/b.png"])
+        status, got = self._req(
+            "GET", "/tasks/" + task["id"], key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(got["attachments"], ["a.txt", "sub/b.png"])
+        status, listed = self._req(
+            "GET", "/tasks?project_id=" + self.project["id"],
+            key=self.owner_key)
+        self.assertEqual(status, 200)
+        self.assertEqual(listed["tasks"][0]["attachments"],
+                         ["a.txt", "sub/b.png"])
+
+    def test_upload_rejects_bad_path_and_empty(self):
+        status, body = self._upload(["../escape.txt"], [b"x"])
+        self.assertEqual(status, 400)
+        status, body = self._req("POST", "/uploads",
+                                 {"paths": "[]"}, self.owner_key)
+        self.assertEqual(status, 400)
+
+    def test_bind_rejects_used_unknown_other_cloud(self):
+        status, up = self._upload(["a.txt"], [b"hi"])
+        self.assertEqual(status, 201)
+        status, _task = self._create_task(up["upload_id"])
+        self.assertEqual(status, 201)
+        status, body = self._create_task(up["upload_id"])
+        self.assertEqual(status, 409)
+        status, body = self._create_task("no-such-bundle")
+        self.assertEqual(status, 400)
+        status, up2 = self._upload(["b.txt"], [b"hi"])
+        self.assertEqual(status, 201)
+        status, body = self._req(
+            "POST", "/tasks",
+            {"project_id": self.project["id"], "prompt": "x",
+             "upload_id": up2["upload_id"]}, self.other_key)
+        self.assertEqual(status, 403)
+        status, up3 = self._upload(["c.txt"], [b"hi"])
+        self.assertEqual(status, 201)
+        status, body = self._create_task(up3["upload_id"],
+                                         executor="cloud")
+        self.assertEqual(status, 400)
+
+    def test_cancel_deletes_bundle_dir(self):
+        status, up = self._upload(["a.txt"], [b"hi"])
+        self.assertEqual(status, 201)
+        bundle_dir = os.path.join(self.upload_dir, up["upload_id"])
+        self.assertTrue(os.path.isdir(bundle_dir))
+        status, task = self._create_task(up["upload_id"])
+        self.assertEqual(status, 201)
+        status, _body = self._req(
+            "POST", "/tasks/%s/cancel" % task["id"], {},
+            self.owner_key)
+        self.assertEqual(status, 200, _body)
+        self.assertFalse(os.path.exists(bundle_dir))
+
+    def test_fork_copies_attachment_rows(self):
+        status, up = self._upload(["a.txt"], [b"hi"])
+        self.assertEqual(status, 201)
+        status, task = self._create_task(up["upload_id"])
+        self.assertEqual(status, 201)
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("UPDATE tasks SET status = 'completed' WHERE id = ?",
+                     (task["id"],))
+        conn.commit()
+        conn.close()
+        status, forked = self._req(
+            "POST", "/tasks/%s/fork" % task["id"],
+            {"message": "continue"}, self.owner_key)
+        self.assertEqual(status, 201, forked)
+        self.assertEqual(forked["attachments"], ["a.txt"])
+
+
+class SandboxSeedTestCase(unittest.TestCase):
+    def _recording(self, calls):
+        def fake(cmd):
+            calls.append(list(cmd))
+            return SimpleNamespace(returncode=0, stdout="ok", stderr="")
+        return fake
+
+    def test_local_dispatch_and_goal_seed_flags(self):
+        calls = []
+        ex = LocalExecutor("/fake/host.py", runner=self._recording(calls))
+        ex.dispatch_task({"id": "t1", "prompt": "do"},
+                         seed_dir="/tmp/seed")
+        self.assertIn("--seed-dir", calls[0])
+        self.assertEqual(calls[0][calls[0].index("--seed-dir") + 1],
+                         "/tmp/seed")
+        calls.clear()
+        ex.dispatch_task({"id": "t1", "prompt": "do"})
+        self.assertNotIn("--seed-dir", calls[0])
+        calls.clear()
+        ex.goal_task({"id": "t2", "prompt": "build it", "goal_mode": 1,
+                      "goal": "build it", "goal_max_rounds": 0},
+                     seed_dir="/tmp/seed")
+        self.assertIn("--seed-dir", calls[0])
+        self.assertNotIn("--verify", calls[0])
+        self.assertEqual(calls[0][calls[0].index("--max-rounds") + 1],
+                         "0")
+
+    def test_popen_goal_new_semantics_and_seed(self):
+        import cloud.sandbox as sandbox_mod
+        calls = []
+
+        class FakePopen(object):
+            def __init__(self, cmd, **kwargs):
+                calls.append(list(cmd))
+
+        orig = sandbox_mod.subprocess.Popen
+        sandbox_mod.subprocess.Popen = FakePopen
+        try:
+            ex = LocalExecutor("/fake/host.py")
+            ex.popen_task({"id": "t2", "prompt": "build it",
+                           "goal_mode": 1, "goal": "build it",
+                           "goal_max_rounds": 0},
+                          kind="goal", seed_dir="/tmp/seed")
+            ex.popen_task({"id": "t1", "prompt": "do"}, kind="run",
+                          seed_dir="/tmp/seed")
+        finally:
+            sandbox_mod.subprocess.Popen = orig
+        goal_cmd, run_cmd = calls
+        self.assertNotIn("--verify", goal_cmd)
+        self.assertEqual(goal_cmd[goal_cmd.index("--max-rounds") + 1],
+                         "0")
+        self.assertIn("--seed-dir", goal_cmd)
+        self.assertNotIn("--seed-dir", run_cmd)
+
+    def test_docker_seed_mount(self):
+        from unittest import mock
+        calls = []
+        ex = DockerExecutor(
+            "/fake/host.py", runner=self._recording(calls),
+            env={"XCM_API_KEY": "placeholder-not-a-real-key"})
+        with tempfile.TemporaryDirectory() as seed_dir, \
+                mock.patch("cloud.sandbox.shutil.which",
+                           return_value="/usr/bin/docker"):
+            ex.dispatch_task({"id": "t1", "prompt": "do"},
+                             seed_host_dir=seed_dir)
+            flat = " ".join(calls[0])
+            self.assertIn("%s:/tmp/codex-seed:ro" % seed_dir, flat)
+            self.assertIn("/tmp/codex-seed", flat)
+            calls.clear()
+            ex.dispatch_task({"id": "t1", "prompt": "do"})
+            flat = " ".join(calls[0])
+            self.assertNotIn("codex-seed", flat)
+
+
+class WorkerSeedTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self.tmp.name, "control.db")
+        self.upload_dir = os.path.join(self.tmp.name, "uploads")
+        init_db(self.db_path)
+        self.env = {"XCM_API_KEY": "test-placeholder"}
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _insert_task(self, task_id, **extra):
+        now = time.time()
+        cols = ["id", "project_id", "prompt", "status",
+                "idempotency_key", "depends_on", "priority",
+                "created_by", "created_at", "updated_at",
+                "last_heartbeat"]
+        vals = [task_id, "p1", "do work", "queued", None, "[]", 0,
+                "tester", now, now, None]
+        for k, v in extra.items():
+            cols.append(k)
+            vals.append(v)
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO tasks(%s) VALUES(%s)"
+                % (", ".join(cols), ", ".join("?" * len(cols))), vals)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def _insert_bundle(self, task_id, bundle_id):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO task_attachments"
+                "(task_id, bundle_id, rel_path, size) VALUES(?, ?, ?, ?)",
+                (task_id, bundle_id, "a.txt", 2))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def test_task_seed_dir_resolution(self):
+        worker = Worker(self.db_path, "/fake/host.py", env=self.env,
+                        upload_dir=self.upload_dir)
+        self.assertIsNone(worker._task_seed_dir({"id": "nope"}))
+        self._insert_bundle("t1", "b1")
+        self.assertIsNone(worker._task_seed_dir({"id": "t1"}))
+        bundle_dir = os.path.join(self.upload_dir, "b1")
+        os.makedirs(bundle_dir)
+        self.assertEqual(worker._task_seed_dir({"id": "t1"}), bundle_dir)
+
+    def test_run_one_passes_seed_and_cleans_bundle(self):
+        self._insert_task("t1")
+        self._insert_bundle("t1", "b1")
+        bundle_dir = os.path.join(self.upload_dir, "b1")
+        os.makedirs(bundle_dir)
+        with open(os.path.join(bundle_dir, "a.txt"), "w") as f:
+            f.write("hi")
+        seen = []
+
+        def fake(cmd):
+            seen.append(list(cmd))
+            return SimpleNamespace(returncode=0, stdout="ok", stderr="")
+
+        worker = Worker(self.db_path, "/fake/host.py", runner=fake,
+                        env=self.env, upload_dir=self.upload_dir)
+        result = worker.run_one()
+        self.assertEqual(result["status"], "completed")
+        dispatch_cmds = [c for c in seen if "dispatch" in c]
+        self.assertTrue(dispatch_cmds)
+        self.assertIn("--seed-dir", dispatch_cmds[0])
+        self.assertEqual(
+            dispatch_cmds[0][dispatch_cmds[0].index("--seed-dir") + 1],
+            bundle_dir)
+        self.assertFalse(os.path.exists(bundle_dir))
+
+
+if __name__ == "__main__":
+    unittest.main()

cloud/uploads.py

@@ -0,0 +1,180 @@
+"""任务附件：multipart 解析、相对路径校验、暂存包落盘。
+
+控制台上传流程（见 README）：先 POST /uploads 拿 upload_id，
+再 POST /tasks 带 upload_id 绑定；worker 派发时把暂存目录经
+host.py --seed-dir 拷进 worktree。图片与普通文件不区分对待，
+一律是 worktree 里的文件（模型能否“看懂”图片取决于模型本身，
+本模块不承诺多模态理解）。
+
+上限：最多 200 个文件，单文件 20MB，文件合计 100MB。路径
+只允许 worktree 内的相对路径，绝不接受绝对路径、..、空段、
+反斜杠与 NUL（防止写出暂存目录或 worktree）。
+"""
+
+import json
+import os
+import re
+import uuid
+
+MAX_FILES = 200
+MAX_FILE_BYTES = 20 * 1024 * 1024
+MAX_TOTAL_BYTES = 100 * 1024 * 1024
+# multipart 信封本身的开销：413 预检按这个上限放行原始 body，
+# 真正的文件合计仍按 MAX_TOTAL_BYTES 卡。
+MAX_RAW_BYTES = MAX_TOTAL_BYTES + 10 * 1024 * 1024
+BUNDLE_TTL_SECONDS = 86400
+DEFAULT_UPLOAD_ROOT = "/opt/codex-app-server/work/uploads"
+
+
+def upload_root(upload_dir=None):
+    """暂存根目录：显式传入 > 环境变量 > 默认值（与 server 侧一致）。"""
+    if upload_dir:
+        return str(upload_dir)
+    return os.environ.get("CODEX_UPLOAD_DIR") or DEFAULT_UPLOAD_ROOT
+
+
+def parse_multipart(body, content_type):
+    """解析 multipart/form-data，返回 (fields, files)。
+
+    fields：name -> 文本（UTF-8 解码失败即 ValueError）。
+    files：[(filename, bytes)]，按 part 出现顺序。
+    只要结构对不上（缺 boundary、缺头、缺分隔）就 ValueError，
+    调用方按 400 处理，不静默丢 part。"""
+    if not isinstance(body, (bytes, bytearray)):
+        raise ValueError("multipart body 必须是 bytes")
+    m = re.search(r"boundary=([^;]+)", content_type or "")
+    if not m:
+        raise ValueError("multipart 缺 boundary")
+    boundary = m.group(1).strip().strip('"').encode("utf-8")
+    if not boundary:
+        raise ValueError("multipart boundary 为空")
+    delimiter = b"--" + boundary
+    chunks = body.split(delimiter)
+    # 首段是信封前导（空），末段是 -- 收尾；中间每段一个 part。
+    fields = {}
+    files = []
+    for chunk in chunks[1:]:
+        if chunk.startswith(b"--"):
+            break
+        if chunk.startswith(b"\r\n"):
+            chunk = chunk[2:]
+        elif chunk.startswith(b"\n"):
+            chunk = chunk[1:]
+        if chunk.endswith(b"\r\n"):
+            chunk = chunk[:-2]
+        elif chunk.endswith(b"\n"):
+            chunk = chunk[:-1]
+        head, sep, payload = chunk.partition(b"\r\n\r\n")
+        if not sep:
+            head, sep, payload = chunk.partition(b"\n\n")
+            if not sep:
+                raise ValueError("multipart part 缺头部分隔")
+        headers = head.decode("utf-8", errors="replace")
+        disp = None
+        for line in headers.splitlines():
+            if line.lower().startswith("content-disposition:"):
+                disp = line.split(":", 1)[1].strip()
+                break
+        if disp is None:
+            raise ValueError("multipart part 缺 Content-Disposition")
+        name_m = re.search(r'name="([^"]*)"', disp)
+        if not name_m:
+            raise ValueError("multipart part 缺 name")
+        name = name_m.group(1)
+        file_m = re.search(r'filename="([^"]*)"', disp)
+        if file_m is not None:
+            files.append((file_m.group(1), bytes(payload)))
+        else:
+            try:
+                fields[name] = payload.decode("utf-8")
+            except UnicodeDecodeError:
+                raise ValueError("multipart 文本字段不是 UTF-8: %s" % name)
+    return fields, files
+
+
+def validate_rel_path(path):
+    """校验 worktree 内相对路径；非法返回错误文本，合法返回 None。"""
+    if not isinstance(path, str) or not path:
+        return "附件路径不能为空"
+    if len(path) > 512:
+        return "附件路径过长: %s" % path[:50]
+    if path.startswith("/") or path.startswith("\\"):
+        return "附件路径不能是绝对路径: %s" % path
+    if re.match(r"^[A-Za-z]:", path):
+        return "附件路径不能带盘符: %s" % path
+    if "\x00" in path or "\\" in path:
+        return "附件路径含非法字符: %s" % path
+    parts = path.split("/")
+    for part in parts:
+        if part in ("", ".", ".."):
+            return "附件路径含非法段: %s" % path
+    return None
+
+
+def validate_bundle(paths, sizes):
+    """校验一包附件：数量、单文件、合计、路径合法、路径不重复。"""
+    if len(paths) > MAX_FILES:
+        return "附件最多 %d 个" % MAX_FILES
+    if len(paths) != len(sizes):
+        return "附件路径与文件数量不一致"
+    seen = set()
+    total = 0
+    for path, size in zip(paths, sizes):
+        err = validate_rel_path(path)
+        if err:
+            return err
+        if path in seen:
+            return "附件路径重复: %s" % path
+        seen.add(path)
+        if size > MAX_FILE_BYTES:
+            return "单个附件超过 %dMB: %s" % (
+                MAX_FILE_BYTES // (1024 * 1024), path)
+        total += size
+    if total > MAX_TOTAL_BYTES:
+        return "附件合计超过 %dMB" % (MAX_TOTAL_BYTES // (1024 * 1024))
+    return None
+
+
+def save_bundle(root, paths, datas):
+    """把一包附件写进暂存目录，返回 (bundle_id, [(rel_path, size)])。
+
+    调用方须先过 validate_bundle；这里再逐路径校验一次（纵深
+    防御：直接调本函数的调用方漏了校验也不至于把 ../ 写出包外）。
+    目录是全新 uuid 子目录，里面不会有符号链接，写盘按相对路径
+    逐级建目录。"""
+    bundle_id = uuid.uuid4().hex
+    bundle_dir = os.path.join(root, bundle_id)
+    os.makedirs(bundle_dir, exist_ok=False)
+    out = []
+    for path, data in zip(paths, datas):
+        err = validate_rel_path(path)
+        if err:
+            raise ValueError(err)
+        dest = os.path.join(bundle_dir, path)
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        with open(dest, "wb") as f:
+            f.write(data)
+        out.append((path, len(data)))
+    return bundle_id, out
+
+
+def bundle_dir_for(root, bundle_id):
+    return os.path.join(root, bundle_id)
+
+
+def parse_paths_field(fields, files):
+    """控制台传 paths JSON（与 file part 一一对应）；不传时退回
+    各 part 自带的 filename（API 直调可用）。"""
+    raw = fields.get("paths")
+    if raw is None:
+        return [name for name, _data in files]
+    try:
+        paths = json.loads(raw)
+    except ValueError:
+        raise ValueError("paths 不是合法 JSON")
+    if not isinstance(paths, list) or not all(
+            isinstance(p, str) for p in paths):
+        raise ValueError("paths 必须是字符串数组")
+    if len(paths) != len(files):
+        raise ValueError("附件路径与文件数量不一致")
+    return paths

cloud/worker.py

@@ -122,6 +122,7 @@ import argparse
 import json
 import os
 import re
+import shutil
 import signal
 import socket
 import subprocess
@@ -140,6 +141,7 @@ from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job)
 from cloud.queue import budget_exceeded, ready_tasks
 from cloud.repos import RepoError, checkout_dir, resolve_repo
 from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor
+from cloud.uploads import upload_root as _upload_root
 from cloud.sandbox import _default_runner as _sandbox_default_runner
 from cloud.workers import heartbeat_worker, register_worker
 
@@ -268,7 +270,8 @@ class Worker:
                  tenant_isolation=False, tenants_root=None,
                  tenant_template_home=None, tenant_template_repo=None,
                  egress_proxy_port=0, claim_executor_only=False,
-                 host_name=None, pr_jobs_only=False):
+                 host_name=None, pr_jobs_only=False,
+                 upload_dir=None):
         self.db_path = str(db_path)
         self.host_py_path = host_py_path
         self.config_path = config_path
@@ -378,6 +381,9 @@ class Worker:
         # 只跑 pr 任务（HA 阶段 1 的跨主机 /pr 演练与专用 pr-runner
         # 用）：不认领普通任务，只处理本主机的 pr_jobs。
         self.pr_jobs_only = bool(pr_jobs_only)
+        # 附件暂存根：显式传入 > 环境变量 > 默认路径（与 server 侧
+        # 同款解析；两边不一致时 worker 找不到包，派发会静默无附件）。
+        self.upload_dir = str(upload_dir) if upload_dir else _upload_root()
         # 运维心跳：路径优先显式参数，其次环境变量，默认当前目录下
         # 的 worker-heartbeat.json；主机路径不在代码里硬编码，主机
         # 部署时用 --heartbeat-file 或环境变量传入（见 deploy/README）。
@@ -928,8 +934,31 @@ class Worker:
                 is _sandbox_default_runner
                 and hasattr(self.executor, "popen_task"))
 
+    def _task_seed_dir(self, task):
+        """任务附件暂存目录（派发时 seed 用）：无附件返回 None；
+        有附件行但目录已不在（如终局后已清理）也返回 None——
+        续跑任务不需要重 seed（文件已在 worktree 里），静默跳过
+        比报错更正确（重试走已存在续跑时包早已删了）。"""
+        try:
+            conn = _db_connect(self.db_path)
+            try:
+                row = conn.execute(
+                    "SELECT bundle_id FROM task_attachments"
+                    " WHERE task_id = ? LIMIT 1",
+                    (task["id"],),
+                ).fetchone()
+            finally:
+                conn.close()
+        except Exception:
+            return None
+        if row is None:
+            return None
+        bundle_id = row[0] if isinstance(row, tuple) else row["bundle_id"]
+        bundle_dir = os.path.join(self.upload_dir, bundle_id)
+        return bundle_dir if os.path.isdir(bundle_dir) else None
+
     def _execute_via_process(self, task, kind="dispatch",
-                             deadline_seconds=None):
+                             deadline_seconds=None, seed_host_dir=None):
         """进程模式执行一轮：启动、轮询、取消时杀进程组。
 
         返回与同步路径同款 result dict（status/stdout/stderr/
@@ -942,7 +971,11 @@ class Worker:
         if self.process_factory is not None:
             proc = self.process_factory(task)
         else:
-            proc = self.executor.popen_task(task, kind=kind)
+            # seed 只对 dispatch/goal 首轮有意义：run 续跑不传
+            # （文件已在 worktree 里）；popen_task 内部同样只在
+            # dispatch/goal 命令上追加 --seed-dir。
+            proc = self.executor.popen_task(
+                task, kind=kind, seed_dir=seed_host_dir)
         self._running_set(task_id, proc=proc)
         try:
             self._set_child_pid(task_id, getattr(proc, "pid", None))
@@ -1564,6 +1597,16 @@ class Worker:
         # 的已存在续跑（run）对它没有意义（goal 没有 run 语义）。
         goal_mode = bool(task.get("goal_mode")
                          or task.get("verify_command"))
+        # 附件 seed：dispatch/goal 首轮才用；run 续跑不传（文件已
+        # 在 worktree 里）。cloud 执行器没有 worktree 可 seed，
+        # 创建时本应已拒绝，这里是兜底防线（直接失败，不静默跑）。
+        seed_host_dir = self._task_seed_dir(task)
+        if seed_host_dir and isinstance(self.executor, CloudExecutor):
+            return ({"executor": "cloud", "task_id": task_id,
+                     "status": "failed", "stdout": "",
+                     "returncode": None, "host_status": None,
+                     "stderr": ""},
+                    "cloud 执行器不支持附件")
         # Docker 同步路径：dispatch 阻塞在 docker run 里时，取消
         # 线程靠容器名杀容器（见 cancel()）。容器名按任务派生，
         # 与执行器本身无关，所以这里只记名不碰执行器状态。
@@ -1593,7 +1636,8 @@ class Worker:
             if self._use_popen():
                 result = self._execute_via_process(
                     task, kind="goal" if goal_mode else "dispatch",
-                    deadline_seconds=self.task_timeout)
+                    deadline_seconds=self.task_timeout,
+                    seed_host_dir=seed_host_dir)
                 combined = (result.get("stdout", "") or "") + \
                     (result.get("stderr", "") or "")
                 if not goal_mode and result["status"] not in (
@@ -1623,18 +1667,37 @@ class Worker:
                             task, timeout=effective_docker_timeout,
                             extra_mounts=tenant_mounts,
                             network=tenant_network,
-                            extra_env=proxy_env)
+                            extra_env=proxy_env,
+                            seed_host_dir=seed_host_dir)
                     else:
                         result = self.executor.dispatch_task(
                             task, timeout=effective_docker_timeout,
                             extra_mounts=tenant_mounts,
                             network=tenant_network,
-                            extra_env=proxy_env)
+                            extra_env=proxy_env,
+                            seed_host_dir=seed_host_dir)
                 else:
+                    # seed 参数只传给 LocalExecutor：CloudExecutor
+                    # 基类与测试假执行器的 goal_task/dispatch_task
+                    # 只有 (task) 一个参数，多传 seed_dir 会直接
+                    # TypeError（2026-10-27 实测炸掉 3 个旧测试）；
+                    # Docker 路径上面已有 is_docker 守卫，不走这里。
                     if goal_mode:
-                        result = self.executor.goal_task(task)
+                        if task.get("_resume") and isinstance(
+                                self.executor, LocalExecutor):
+                            result = self.executor.goal_resume_task(
+                                task, task["prompt"])
+                        elif isinstance(self.executor, LocalExecutor):
+                            result = self.executor.goal_task(
+                                task, seed_dir=seed_host_dir)
+                        else:
+                            result = self.executor.goal_task(task)
                     else:
-                        result = self.executor.dispatch_task(task)
+                        if isinstance(self.executor, LocalExecutor):
+                            result = self.executor.dispatch_task(
+                                task, seed_dir=seed_host_dir)
+                        else:
+                            result = self.executor.dispatch_task(task)
                 combined = (result.get("stdout", "") or "") + \
                     (result.get("stderr", "") or "")
                 if not goal_mode and result["status"] != "completed" and \
@@ -2030,6 +2093,15 @@ class Worker:
             task_id, result_text=stdout, error_text=error,
             actual_model=actual_model, started_at=started_at,
             finished_at=finished_at)
+        # 附件暂存清理：任务终局后删 bundle 目录（重试走已存在
+        # 续跑，文件已在 worktree 里，不再需要暂存；取消走 server
+        # 侧删，这里删不到也不报错；删失败不影响任务结果本身）。
+        try:
+            _seed = self._task_seed_dir(task)
+            if _seed:
+                shutil.rmtree(_seed, ignore_errors=True)
+        except Exception:
+            pass
         with self._db_lock:
             self.usage_turns += 1
             # 成本：执行器带回上游真实用量时记 actual（input/output/

host.py

@@ -1560,9 +1560,32 @@ def dispatch_from_branch(server, state, name, prompt, repo, dep_branch,
     return task, ""
 
 
+def _seed_worktree(cwd, seed_dir):
+    """把暂存附件拷进 worktree（dispatch/goal 共用，turn 开始前）。
+
+    seed_dir 是服务端写好的暂存目录（只含常规文件，无符号链接）：
+    顶层 .git 与符号链接一律跳过；子目录整树拷入，同名文件覆盖
+    （worktree 刚从 repo 建出来，覆盖即用户文件优先）。"""
+    import shutil
+    seed_dir = os.path.abspath(seed_dir)
+    if not os.path.isdir(seed_dir):
+        raise RuntimeError(f"seed-dir 不是目录: {seed_dir}")
+    for entry in sorted(os.listdir(seed_dir)):
+        if entry == ".git":
+            continue
+        src = os.path.join(seed_dir, entry)
+        dst = os.path.join(cwd, entry)
+        if os.path.islink(src):
+            continue
+        if os.path.isdir(src):
+            shutil.copytree(src, dst, dirs_exist_ok=True, symlinks=False)
+        else:
+            shutil.copy2(src, dst)
+
+
 def dispatch(server, state, name, prompt, repo, model, effort, sandbox,
              approval_policy, use_worktree=True, parent_task=None,
-             depends_on=None, priority=0, max_retries=0):
+             depends_on=None, priority=0, max_retries=0, seed_dir=None):
     if name in state["tasks"]:
         raise RuntimeError(f"task 已存在: {name}，先换名或 discard 它")
     if repo and not os.path.isabs(repo) and not os.path.exists(repo):
@@ -1575,6 +1598,13 @@ def dispatch(server, state, name, prompt, repo, model, effort, sandbox,
     branch = None
     if use_worktree and repo:
         cwd, branch = make_worktree(repo, name)
+    if seed_dir:
+        # 附件只能进 worktree：直接以 repo 为 cwd 时拷进去等于污染
+        # 仓库，必须 loud fail，绝不静默放行。
+        if not use_worktree or not repo or not cwd:
+            raise RuntimeError(
+                "seed-dir 需要 worktree，当前派发没有建 worktree")
+        _seed_worktree(cwd, seed_dir)
     thread_id = server.create_thread(cwd, model=model, sandbox=sandbox,
                                      approval_policy=approval_policy)
     task = {"name": name, "thread_id": thread_id, "cwd": cwd,
@@ -1905,6 +1935,8 @@ def main():
     p.add_argument("--name", required=True)
     p.add_argument("--task", required=True)
     p.add_argument("--repo", default=None)
+    p.add_argument("--seed-dir", default=None,
+                   help="附件暂存目录：派发时拷进 worktree 再跑 turn")
     p.add_argument("--no-worktree", action="store_true")
     p.add_argument("--depends-on", default=None,
                    help="逗号分隔的依赖任务名，依赖未完成前不跑")
@@ -1948,6 +1980,8 @@ def main():
     p = sub.add_parser("goal")
     p.add_argument("--name", required=True); p.add_argument("--task", required=True)
     p.add_argument("--repo", default=None)
+    p.add_argument("--seed-dir", default=None,
+                   help="附件暂存目录：派发时拷进 worktree 再跑 turn")
     p.add_argument("--verify", default=None,
                    help="验收命令；不传时走模型自决：模型在回合末尾"
                         "单独一行输出“目标已达成”即达标")
@@ -2277,7 +2311,8 @@ def main():
                             args.approval_policy,
                             use_worktree=not args.no_worktree,
                             depends_on=deps, priority=args.priority,
-                            max_retries=args.max_retries)
+                            max_retries=args.max_retries,
+                            seed_dir=args.seed_dir)
             # 生产教训：先落盘任务登记再跑 turn，中途被杀/超时后
             # 任务还在 host-state 里，可用 run 续跑；否则 turn 中断
             # 就等于任务丢了（2026-10-25 委派评审 300 秒超时实测）。
@@ -2538,7 +2573,8 @@ def main():
         elif args.cmd == "goal":
             task = dispatch(srv, state, args.name, args.task, args.repo,
                             args.model, args.effort, args.sandbox,
-                            args.approval_policy)
+                            args.approval_policy,
+                            seed_dir=args.seed_dir)
             save_state(args.state, state)
             run_goal_loop(srv, state, args.state, task, args.task,
                           args.verify, args.max_rounds, args.model,