goal 模式：输入框内容即目标，不需要专用 prompt 与验收命令；不限轮次，模型确认达成后在回复末尾输出“目标已达成”即结束，未达成会一直运行，可用取消停止。
+
@@ -378,7 +384,7 @@ async function loadTasks() {
var pr = t.pr_url ? "
#" + esc(t.pr_number) + "" : "";
return "
" +
"| " + esc(shortId(t.id)) + " | " +
- "" + (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)) + " | " +
+ "" + (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)) + " | " +
"" + esc(t.status) + " | " +
"" + esc(t.executor || "") + " | " +
"" + esc(t.worktree_host || "") + " | " +
@@ -505,6 +511,7 @@ async function showDetail(taskId) {
(t.plan ? " ｜ 计划模式（只读）" : "") +
(t.forked_from ? " ｜ 分支自 " + esc(shortId(t.forked_from)) : "") +
(t.goal_mode ? " ｜ goal 模式（不限轮次，模型宣告完成）" : (t.verify_command ? " ｜ goal 模式（旧语义，验收命令）" : "")) +
+ ((t.attachments && t.attachments.length) ? " ｜ 附件：" + t.attachments.map(esc).join("、") : "") +
(t.pr_url ? " ｜ PR #" + esc(t.pr_number) + "" : "") +
"
prompt：
" + esc(t.prompt || "");
var body = "";
@@ -614,6 +621,41 @@ $("#createEnvironment").onclick = async function () {
loadEnvironments();
} catch (e) { $("#envMsg").innerHTML = "新建失败：" + esc(e.message) + ""; }
};
+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 = "附件上传失败：" + esc(e.message) + "";
+ return;
+ }
+ }
try {
var t = await api("/tasks", "POST", body);
$("#taskMsg").innerHTML = "任务已建：" + esc(shortId(t.id)) + "";
$("#prompt").value = "";
+ $("#attFiles").value = ""; $("#attDir").value = ""; $("#attImg").value = "";
loadTasks();
} catch (e) { $("#taskMsg").innerHTML = "提交失败：" + esc(e.message) + ""; }
};
diff --git a/cloud/db.py b/cloud/db.py
index c4a5c98..ae18cfd 100644
--- a/cloud/db.py
+++ b/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, "
diff --git a/cloud/sandbox.py b/cloud/sandbox.py
index ce209f0..e5bdb69 100644
--- a/cloud/sandbox.py
+++ b/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
@@ -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
@@ -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,
- "--max-rounds", str(rounds)]
+ "--task", goal_text]
+ 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:
- """容器内调 host.py dispatch：建 host 任务并跑一轮 turn。"""
+ network=None, extra_env=None,
+ 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:
diff --git a/cloud/server.py b/cloud/server.py
index 9fddfda..ec5101f 100644
--- a/cloud/server.py
+++ b/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:
diff --git a/cloud/tests/test_task_attachments.py b/cloud/tests/test_task_attachments.py
new file mode 100644
index 0000000..a28e001
--- /dev/null
+++ b/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()
diff --git a/cloud/uploads.py b/cloud/uploads.py
new file mode 100644
index 0000000..add0720
--- /dev/null
+++ b/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
diff --git a/cloud/worker.py b/cloud/worker.py
index 384e8de..5530b73 100644
--- a/cloud/worker.py
+++ b/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/
diff --git a/host.py b/host.py
index a527c7f..fd4df36 100644
--- a/host.py
+++ b/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,