From 9537e6f0d2434de19b806e55027b475048fc1c29 Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 03:40:46 +0800 Subject: [PATCH] =?UTF-8?q?WIP:=20=E4=BB=BB=E5=8A=A1=E9=99=84=E4=BB=B6?= =?UTF-8?q?=E2=80=94=E2=80=94host=20seed-dir=20=E4=B8=8E=E4=B8=8A=E4=BC=A0?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=EF=BC=88=E6=9C=AA=E5=AE=8C=EF=BC=8Cserver/wo?= =?UTF-8?q?rker/console=20=E5=BE=85=E7=BB=AD=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cloud/uploads.py | 173 +++++++++++++++++++++++++++++++++++++++++++++++ host.py | 42 +++++++++++- 2 files changed, 212 insertions(+), 3 deletions(-) create mode 100644 cloud/uploads.py diff --git a/cloud/uploads.py b/cloud/uploads.py new file mode 100644 index 0000000..3287be0 --- /dev/null +++ b/cloud/uploads.py @@ -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 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,