WIP: 任务附件——host seed-dir 与上传模块(未完,server/worker/console 待续)

This commit is contained in:
host committed 2026-10-27 04:29:40 +08:00
1 parent c102327040
commit 9537e6f0d2
2 files changed
+212 -3

No files matched your search

+173
View File
@@ -0,0 +1,173 @@
"""任务附件: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):
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 必须是字符串数组")
return paths
+39 -3
View File
@@ -1560,9 +1560,32 @@ def dispatch_from_branch(server, state, name, prompt, repo, dep_branch,
return task, "" 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, def dispatch(server, state, name, prompt, repo, model, effort, sandbox,
approval_policy, use_worktree=True, parent_task=None, 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"]: if name in state["tasks"]:
raise RuntimeError(f"task 已存在: {name},先换名或 discard 它") raise RuntimeError(f"task 已存在: {name},先换名或 discard 它")
if repo and not os.path.isabs(repo) and not os.path.exists(repo): 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 branch = None
if use_worktree and repo: if use_worktree and repo:
cwd, branch = make_worktree(repo, name) 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, thread_id = server.create_thread(cwd, model=model, sandbox=sandbox,
approval_policy=approval_policy) approval_policy=approval_policy)
task = {"name": name, "thread_id": thread_id, "cwd": cwd, task = {"name": name, "thread_id": thread_id, "cwd": cwd,
@@ -1905,6 +1935,8 @@ def main():
p.add_argument("--name", required=True) p.add_argument("--name", required=True)
p.add_argument("--task", required=True) p.add_argument("--task", required=True)
p.add_argument("--repo", default=None) 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("--no-worktree", action="store_true")
p.add_argument("--depends-on", default=None, p.add_argument("--depends-on", default=None,
help="逗号分隔的依赖任务名,依赖未完成前不跑") help="逗号分隔的依赖任务名,依赖未完成前不跑")
@@ -1948,6 +1980,8 @@ def main():
p = sub.add_parser("goal") p = sub.add_parser("goal")
p.add_argument("--name", required=True); p.add_argument("--task", required=True) p.add_argument("--name", required=True); p.add_argument("--task", required=True)
p.add_argument("--repo", default=None) p.add_argument("--repo", default=None)
p.add_argument("--seed-dir", default=None,
help="附件暂存目录:派发时拷进 worktree 再跑 turn")
p.add_argument("--verify", default=None, p.add_argument("--verify", default=None,
help="验收命令;不传时走模型自决:模型在回合末尾" help="验收命令;不传时走模型自决:模型在回合末尾"
"单独一行输出“目标已达成”即达标") "单独一行输出“目标已达成”即达标")
@@ -2277,7 +2311,8 @@ def main():
args.approval_policy, args.approval_policy,
use_worktree=not args.no_worktree, use_worktree=not args.no_worktree,
depends_on=deps, priority=args.priority, depends_on=deps, priority=args.priority,
max_retries=args.max_retries) max_retries=args.max_retries,
seed_dir=args.seed_dir)
# 生产教训:先落盘任务登记再跑 turn,中途被杀/超时后 # 生产教训:先落盘任务登记再跑 turn,中途被杀/超时后
# 任务还在 host-state 里,可用 run 续跑;否则 turn 中断 # 任务还在 host-state 里,可用 run 续跑;否则 turn 中断
# 就等于任务丢了(2026-10-25 委派评审 300 秒超时实测)。 # 就等于任务丢了(2026-10-25 委派评审 300 秒超时实测)。
@@ -2538,7 +2573,8 @@ def main():
elif args.cmd == "goal": elif args.cmd == "goal":
task = dispatch(srv, state, args.name, args.task, args.repo, task = dispatch(srv, state, args.name, args.task, args.repo,
args.model, args.effort, args.sandbox, args.model, args.effort, args.sandbox,
args.approval_policy) args.approval_policy,
seed_dir=args.seed_dir)
save_state(args.state, state) save_state(args.state, state)
run_goal_loop(srv, state, args.state, task, args.task, run_goal_loop(srv, state, args.state, task, args.task,
args.verify, args.max_rounds, args.model, args.verify, args.max_rounds, args.model,