# 工作区隔离机制

本文档说明 `parallel-vibe` 为什么能够“并行地、独立地”运行同一条用户指令，以及这种做法的优劣与边界。

## 术语

- **thread**：在本 skill 中指一个“并行执行单元”。智能模式下通常对应宿主 subagent / 独立上下文；代码模式下通常对应一个独立子进程（不是 Python 线程，也不是模型内部的线程）。
- **workspace**：某个 thread 的专属工作目录（文件系统隔离边界），通常位于 `.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/{run_id}/{thread_id}/workspace/`。
- **runner**：代码模式中实际执行用户指令的 CLI 引擎，例如 `codex exec`、`claude -p`、或用于测试的 `local` runner。

## 机制概览

`parallel-vibe` 的目录管理与模式无关：智能模式和代码模式都使用 `.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/{run_id}/{thread_id}/workspace/` 作为 thread 产物边界。两者的差别只在底层执行机制：

- **智能模式**：每个 thread 通过宿主原生 subagent / 独立上下文执行，主 agent 负责把 thread 结果落到同一套目录结构中。
- **代码模式**：每个 thread 通过“一条 CLI 命令”启动一个独立子进程（runner），并以该 thread 的 `workspace/` 作为 cwd。
- **默认串行、可选并行**：为节省系统资源与降低 API 限流风险，默认串行执行 threads；用户明确要求时才开启并行。
- **目录隔离**：每个 thread 有自己的 `workspace/`。代码模式强制 runner 在该目录作为 cwd 运行；智能模式要求 subagent 只把产物写入自己分配到的 `workspace/`，能否强制 cwd 取决于宿主能力。
- **结果落盘**：每个 thread 的日志、退出码、产物都写入该 thread 的目录中；最后把各 thread 的结果汇总到 `@main/`。

这是一种“工程隔离”而非“安全隔离”：它主要解决“写文件互相覆盖、相对路径混乱、缓存污染”等工程问题，但不是容器/沙箱那种强安全边界。

## 为什么可以做到“开多个独立线程来工作”

### 智能模式的独立上下文

宿主原生 subagent / 独立上下文让每个 thread 独立读取任务、形成判断并输出结果。它解决的是“思考上下文互不污染”的问题。为了让产物管理和代码模式完全一致，主 agent 仍要为每个 thread 分配 `.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/{run_id}/{thread_id}/workspace/`，并把结果同步到该 thread 的 `RESULT.md`、`runner.log`、`done.json` 等固定文件。

智能模式的限制是：独立上下文不必然等于独立 cwd。若宿主不能把 subagent 的文件读写绑定到各自 `workspace/`，实现型任务应让 subagent 输出方案或 patch 建议，再由主 agent 单点落地；需要真实 cwd 隔离时切到代码模式。

### 代码模式的操作系统并行

现代操作系统允许同时运行多个进程。每个进程拥有：

- 独立的地址空间（内存隔离）
- 独立的进程状态（退出码、信号、资源统计）
- 独立的当前工作目录（cwd，可由父进程设置）

因此，代码模式只要把每个 thread 的 runner 放在不同子进程里运行，就能做到“同时推进、互不阻塞”（至少在 CPU/IO 资源允许的前提下）。

### 文件系统层面的互不覆盖

在 Vibe Coding 里，大量操作都是“读写文件”。如果多个执行单元共享同一个工作目录，会遇到典型冲突：

- 同名文件覆盖（例如同时写 `output.txt`、`report.md`）
- 相对路径歧义（`./data` 在不同执行单元里本应指向不同目标）
- 中间产物混在一起，难以追溯与回滚

`parallel-vibe` 通过为每个 thread 创建独立 `workspace/`，让产物天然分区。代码模式进一步强制 `cwd=workspace/`，让相对路径读写天然隔离；智能模式则依赖宿主 subagent 能力和 prompt 约束来遵守同一边界。

## “互不干扰”能保证到什么程度

在默认策略下，`parallel-vibe` 能较可靠保证：

- **产物目录互不覆盖**：因为每个 thread 的 `RESULT.md`、`runner.log`、`workspace/` 分开保存。
- **代码模式相对路径读写互不覆盖**：因为每个 thread 的 cwd 不同。
- **thread 产物与日志互不混淆**：因为输出目录按 thread 分开写。
- **并行执行互不阻塞**：单个 thread 卡住不会直接卡死其他 thread（除非系统资源耗尽）。

但它不能强保证：

- **强安全隔离/强不可见性**：同一用户权限下，进程理论上可以访问工作区之外的路径（例如通过绝对路径）。是否真的会访问，取决于 runner 的能力与指令约束。
- **外部资源互不影响**：例如共享的 API 速率限制、同一数据库、同一端口占用、同一个远程锁、同一云端工作区等。
- **用户家目录缓存完全隔离**：一些工具会写 `~/.cache`、`~/.config` 等全局目录；默认策略不会隔离这些位置。

## Prompt 约束（软隔离护栏）

由于不使用容器/沙箱，`parallel-vibe` 会在每个 thread 的实际执行 prompt 前追加“软护栏前缀”。代码模式由脚本自动拼接；智能模式由主 agent 在分配 subagent 任务时写入同等约束。其目标是：

- 让 AI 明确“你只在当前工作目录（cwd）工作”
- 明确禁止访问父目录与绝对路径写入
- 明确禁止访问 `.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/{run_id}` 的其他 thread 目录
- 要求所有产物落盘到当前 `workspace/`（便于汇总复制）

这类约束属于“软护栏”：它依赖 runner 的行为规范与模型遵守规则，主要降低误操作概率，而非提供强安全保证。

## 优势

- **实现简单（KISS）**：不引入容器/VM/复杂权限控制，跨平台更容易落地。
- **可追溯**：每个 thread 的日志与产物独立保存，便于对比、复盘与回滚。
- **减少互相覆盖**：相对路径隔离让多数“并行写文件冲突”自然消失。
- **易扩展**：可逐步增加 runner 类型、超时、重跑/续跑、汇总策略，而不改变核心目录模型。

## 劣势与成本

- **磁盘与 IO 成本**：每个 thread 复制一份 workspace，目录越大，启动越慢、占用越高。
- **结果不一定可直接合并**：并行产物可能相互冲突，`@main` 汇总更多是“收集与对比”，不是自动合并。
- **非确定性与外部依赖**：并行执行可能触发不同的竞态（例如构建缓存/端口/网络资源），导致结果差异。
- **不是安全沙箱**：对“信息隔离/防止越权访问”的保证有限。

## 适用与不适用场景

适用：

- 想让多个 thread 各自独立尝试实现/改写/分析同一需求，然后对比结果再汇总
- 想隔离相对路径产物与中间文件，避免污染主工作目录

不适用（或需谨慎）：

- 需要强安全隔离（例如严格防止互相读取、处理敏感数据）
- 强依赖共享外部资源且容易冲突（同一数据库写入、端口绑定服务）
- 工作目录非常大且复制成本不可接受（可考虑后续引入 git worktree 或更精细的复制策略）

## 实践建议

- 默认线程数见 `config.yaml:defaults.n_threads`（当前默认 5）。一般建议先保持默认与串行执行；只有用户明确要求并行/你确认资源与限流风险可控时再开启并行并调大并发。
- 把 `.bensz-api/` 加入 `.gitignore`，避免误提交；`.parallel-vibe/` 和 `.parallel_vibe/` 仅作为历史遗留目录继续忽略。
- 默认拒绝 `src_dir` 中的 symlink（可用 `--symlink-policy skip|keep` 覆盖，但会引入越界风险）。
- `copy_exclude` 建议保持以“缓存/构建产物”为主（见 `config.yaml:defaults.copy_exclude`），避免把大体积依赖与缓存复制进每个 thread/workspace（如 `node_modules`、`.venv`、`dist/build/target` 等）。
- 如使用 `codex` 作为 runner，建议通过 `config.yaml:cli.codex.global_args` 固化 `--ask-for-approval never --sandbox workspace-write`，避免子进程等待确认而卡住，并把模型生成的 shell 执行约束在 workspace-write。
- 汇总阶段优先做“收集与索引”（`@main/index.json` + `@main/summary.md`），不要在早期做复杂的自动合并策略。

# CLI Prompt 用法速查（Codex / Claude）

本文档用于 `parallel-vibe` 的“可执行命令规划”阶段：把每个 thread 落到“一条命令一次执行”的 CLI 形态。

注意：不同版本 CLI 的参数可能不同；以你本机 `--help` 输出为准。

## Claude（claude）

官方文档（Claude Code SDK / CLI reference）：

```
https://docs.anthropic.com/en/docs/claude-code/sdk
```

常用形态：

```bash
# 打印（非交互）模式
claude -p "你的指令内容"

# 指定模型（示例形态；可用参数以 --help 为准）
claude --model <model_id> --effort medium -p "你的指令内容"

# 从 stdin 提供上下文（文件内容），再给一个 query
cat some_context.md | claude -p "请基于以上内容完成任务"
```

## OpenAI Codex CLI（codex）

官方文档（Codex CLI 非交互模式 / CLI 参考）：

```
https://developers.openai.com/codex/
```

官方入门文章中常见的模型指定形态：

```bash
codex -m <model_id>
```

非交互执行（一次命令一次执行）：

```bash
codex exec "你的指令内容"

# 指定模型（常见形态；也可能支持 --model）
codex -m <model_id> exec "你的指令内容"

# 追加配置（示例：推理强度；注意 -c/--config 通常是全局参数，建议放在 exec 之前）
codex -m <model_id> -c 'reasoning_effort="medium"' exec "你的指令内容"

# 从 stdin 读取 prompt（PROMPT 为 "-"）
cat synthesis_input.md | codex exec -
```

## 参数放置约定（parallel-vibe 的计划字段）

为避免“参数放错位置导致 CLI 不识别/把参数当作 prompt”的问题，`parallel-vibe` 约定：

- `runner.args`：全局参数（子命令前）
- `runner.sub_args`：子命令参数（子命令后、prompt 前）

# parallel-vibe 智能模式协议

智能模式是 `parallel-vibe` 的默认交互路径。它依赖宿主工具的原生 subagent / 独立上下文能力，让多个 thread 独立分析同一任务，再由主 agent 汇总。智能模式和代码模式共享同一套目录契约，区别只在 thread 的底层执行机制。

## 适用边界

使用智能模式：

- 多个独立 agent 审查同一份代码、PR、文档、方案或研究假设
- 目标是方案探索、风险评估、文档优化、测试边界补充或路线对比
- 用户要求 `.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/`、`plan.json`、`RESULT.md`、`runner.log` 等固定目录产物，但不要求脚本 runner 或真实退出码
- 宿主工具明确支持 subagent 或等价独立上下文能力

切到代码模式：

- 用户要求脚本模式、CLI runner、`--plan-file`、`--resume`、`--dry-run`、真实退出码或可复跑批处理
- 需要跨 `codex` / `claude` / `shell` runner 批量执行
- 宿主没有可用 subagent 能力
- 多个 agent 需要实际并行修改文件，且宿主不能把 subagent 绑定到各自 `workspace/`

## 共享目录契约

智能模式必须和代码模式使用同一套目录管理。默认 run id 为 `{yyyy-mm-dd-hh-mm}`，同一分钟重复运行时追加 `-02` 等后缀；代码模式显式传 `--project-id` 时可复用该值作为 run/project id：

```text
.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/<run_id>/
  project.json
  @main/
    plan.json
    plan.md
    summary.md
  <thread_id>/
    workspace/
      RESULT.md
    RESULT.md
    runner.log
    prompt.txt
    thread.json
    done.json
    exit_code.txt
```

执行含义：

- `@main/plan.json` / `@main/plan.md`：记录 thread 拆分、角色、输入 prompt 和执行策略。
- `<thread_id>/workspace/`：该 thread 的专属工作目录；智能模式下也要把该路径作为 subagent 的产物边界。
- `<thread_id>/workspace/RESULT.md`：thread 首选结果产物。
- `<thread_id>/RESULT.md`：供主 agent 汇总读取的规范副本；如果 subagent 不能直接落盘，由主 agent 把 subagent 返回内容写入此文件。
- `<thread_id>/runner.log`：代码模式保存 CLI stdout/stderr；智能模式保存宿主 subagent 返回内容、摘要或“宿主不暴露 transcript”的说明。
- `<thread_id>/done.json` / `exit_code.txt`：代码模式保存真实进程状态；智能模式成功用 `0`，失败或未完成用 `1`，并在 `done.json` 标明 `mode: smart`。

## Thread 规划

主 agent 先规划 `n` 个 thread。用户未指定数量时：

- 轻量审查：2-3 个 thread
- 中等方案对比：3-5 个 thread
- 高风险或多维审查：5-7 个 thread

常见角色：

- `conservative_approach`：保守、最小改动路线
- `ambitious_approach`：更彻底的重构或替代路线
- `risk_review`：bug、回归、安全、边界条件
- `test_strategy`：验证路径、测试缺口、可复现命令
- `ux_or_docs_review`：用户体验、文档、可维护性

## Subagent 输出 schema

每个 subagent 使用独立上下文，不读取其他 thread 的结果，并优先把以下 schema 写入自己的 `<thread_id>/workspace/RESULT.md`。如果宿主不允许 subagent 直接写文件，主 agent 用同一内容兜底写入 `<thread_id>/RESULT.md`：

```markdown
## Thread Result

- thread_id:
- role:
- conclusion:
- evidence:
- recommended_changes:
- risks:
- verification:
```

字段要求：

- `conclusion`：一句话结论，说明该 thread 选择或反对什么。
- `evidence`：引用本地文件、命令输出、用户材料或明确推理依据。
- `recommended_changes`：可执行建议。实现型任务默认输出 patch 建议，不直接并行写同一工作区。
- `risks`：仍未解决的不确定性、潜在回归或依赖假设。
- `verification`：最小验证命令或人工检查点。

## 主 agent 汇总 schema

主 agent 读取各 thread 的 `RESULT.md`，写入 `.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/<run_id>/@main/summary.md`，并在面向用户的最终输出包含：

```markdown
运行模式：智能模式
project 目录：.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/<run_id>/
thread 数与策略：

## Thread 摘要

- <thread_id> / <role>：<一句话结论>

## 综合结论

- 推荐路线：
- 共识：
- 主要分歧：
- 风险：

## 验证步骤

- <命令或检查点>
```

## 串行与并行策略

- 分析、审查、方案对比类任务可以并行启动 subagent。
- 依赖前一轮结论的任务应串行：先让一组 subagent 审查，再由主 agent 生成下一轮输入。
- 实现型任务先给每个 thread 分配自己的 `workspace/`。只有在宿主能把 subagent 的读写边界绑定到该 `workspace/` 时，才让多个 subagent 并行实改；否则让 subagent 输出方案或 patch 建议，再由主 agent 统一落地。

## 文件系统边界

共享目录契约不等于强安全隔离。

智能模式保证各 subagent 在上下文上独立思考，并要求它们把产物写入同一套 thread `workspace/` 结构；但是否拥有真实独立 cwd、独立 Git checkout 或安全沙箱，取决于宿主能力。需要并行写文件时，必须满足其一：

- 宿主提供独立 worktree / workspace；
- 使用代码模式复制独立 workspace；
- 改为只输出方案、diff 或 patch 建议，由主 agent 单点落地。

不要把智能模式描述成脚本级可复跑批处理系统；需要真实进程日志、失败退出码和 CLI runner 复跑时，使用代码模式。

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import datetime as _dt
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple


SKILL_NAME = "parallel-vibe"
WORK_DIR_NAME = ".bensz-api"
RUN_ID_TIMESTAMP_FORMAT = "%Y%m%d-%H%M"
LEGACY_WORK_DIR_NAMES = [".parallel-vibe", ".parallel_vibe"]
SAFE_RUN_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")

IgnoreFunc = Callable[[str, List[str]], set[str]]


def _now_iso() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat()


def _is_safe_run_id(s: str) -> bool:
    if not SAFE_RUN_ID_RE.fullmatch(s):
        return False
    return s not in {".", ".."} and "/" not in s and "\\" not in s


def _now_run_id() -> str:
    return _dt.datetime.now().strftime(RUN_ID_TIMESTAMP_FORMAT)


def _validate_existing_dir(path: Path, *, label: str) -> None:
    if not path.exists():
        raise ValueError(f"{label} does not exist: {path}")
    if not path.is_dir():
        raise ValueError(f"{label} is not a directory: {path}")


def _require_within(base_dir: Path, target: Path) -> None:
    base = base_dir.resolve()
    tgt = target.resolve()
    try:
        tgt.relative_to(base)
    except Exception as e:
        raise ValueError(f"path escapes base_dir: base={base} target={tgt}") from e


def _load_yaml_if_possible(path: Path) -> Optional[dict]:
    # Optional dependency: keep KISS; fall back to hardcoded defaults if PyYAML isn't available.
    try:
        import yaml  # type: ignore
    except Exception:
        return None
    try:
        with path.open("r", encoding="utf-8") as f:
            return yaml.safe_load(f) or {}
    except Exception:
        return None


def _default_config() -> dict:
    return {
        "defaults": {
            "n_threads": 5,
            "thread_id_width": 3,
            "execution": "serial",  # serial|parallel
            "max_parallel": 3,
            "symlink_policy": "error",  # error|skip|keep
            "copy_exclude": [
                ".bensz-api",
                WORK_DIR_NAME,
                *LEGACY_WORK_DIR_NAMES,
                ".git",
                "node_modules",
                "__pycache__",
                ".DS_Store",
                "Thumbs.db",
                ".venv",
                "venv",
                ".pytest_cache",
                ".mypy_cache",
                ".ruff_cache",
                ".cache",
                ".tox",
                ".coverage",
                "dist",
                "build",
                "target",
            ],
            "synthesize": True,
        },
        "cli": {
            # Prefer simple, stable command composition:
            # - codex: global flags first, then "exec"
            # - claude: global flags first, then "-p"
            "codex": {
                "cmd": ["codex"],
                "global_args": ["--ask-for-approval", "never", "--sandbox", "workspace-write"],
                "exec_subcommand": ["exec"],
                "subcommand_args": [],
                "model_flag": "-m",
                "profile_args": {
                    "default": [],
                    "fast": ["-c", 'reasoning_effort="low"'],
                    "deep": ["-c", 'reasoning_effort="medium"'],
                },
            },
            "claude": {
                "cmd": ["claude"],
                # --dangerously-skip-permissions: bypass tool permission prompts in non-interactive -p mode
                # --no-session-persistence: avoid polluting session history with parallel thread runs
                "global_args": ["--dangerously-skip-permissions", "--no-session-persistence"],
                "print_subcommand": ["-p"],
                "subcommand_args": [],
                "model_flag": "--model",
                "profile_args": {"default": [], "fast": ["--effort", "low"], "deep": ["--effort", "medium"]},
            },
        },
        "models": {
            # Keep defaults empty to avoid hardcoding model IDs that may not exist in a user's environment.
            # Users can fill these in config.yaml if they want deterministic routing.
            "codex": {"default": "", "fast": "", "deep": ""},
            "claude": {"default": "", "fast": "", "deep": ""},
        },
    }


def _deep_merge_dict(dst: dict, src: dict) -> dict:
    """
    Deep-merge dicts (dict values only). Lists/atoms are overwritten.

    This keeps KISS while making config overrides robust:
    users can override only one field (e.g. cli.codex.model_flag) without
    losing newly-added defaults (e.g. cli.codex.global_args).
    """
    for k, v in (src or {}).items():
        if isinstance(v, dict) and isinstance(dst.get(k), dict):
            _deep_merge_dict(dst[k], v)  # type: ignore[index]
        else:
            dst[k] = v
    return dst


def load_config() -> dict:
    skill_root = Path(__file__).resolve().parents[1]
    cfg_path = skill_root / "config.yaml"
    cfg = _load_yaml_if_possible(cfg_path)
    if isinstance(cfg, dict):
        merged = _default_config()
        return _deep_merge_dict(merged, cfg)
    return _default_config()


def wrap_thread_prompt(user_prompt: str, thread_id: str, project_id: str) -> str:
    prefix = (
        "你正在一个并行 thread 的独立工作区中执行任务。\n"
        f"- thread_id: {thread_id}\n"
        f"- project_id: {project_id}\n"
        "- 你的工作目录是当前目录（cwd）。只允许读写当前目录及其子目录。\n"
        "- 禁止访问父目录（..），禁止任何绝对路径写入。\n"
        f"- 禁止访问 {WORK_DIR_NAME}/{project_id} 下的其他 thread 目录。\n"
        "- 所有产物必须落盘到当前目录（workspace）。\n\n"
        "用户指令如下（请严格在本工作区内完成）：\n"
    )
    return prefix + user_prompt


def _parse_copy_exclude(arg: Optional[str], default_list: List[str]) -> List[str]:
    if not arg:
        return list(default_list)
    parts = [p.strip() for p in arg.split(",")]
    return [p for p in parts if p]


def _find_symlinks(src_dir: Path, *, ignore: Optional[IgnoreFunc], max_found: int = 50) -> List[str]:
    found: List[str] = []
    for root, dirs, files in os.walk(src_dir, topdown=True, followlinks=False):
        names = list(dirs) + list(files)
        ignored = set(ignore(root, names)) if ignore else set()

        # Prevent walking into ignored directories to keep scan/copy consistent.
        dirs[:] = [d for d in dirs if d not in ignored]

        for name in names:
            if name in ignored:
                continue
            p = Path(root) / name
            try:
                if p.is_symlink():
                    found.append(str(p.relative_to(src_dir)))
                    if len(found) >= max_found:
                        return found
            except Exception:
                # If we cannot stat an entry, ignore it; copy will likely fail anyway.
                continue
    return found


def copy_workspace(
    src_dir: Path,
    dst_dir: Path,
    exclude_names: Sequence[str],
    *,
    symlink_policy: str,
) -> None:
    base_ignore: Optional[IgnoreFunc] = shutil.ignore_patterns(*exclude_names) if exclude_names else None

    policy = (symlink_policy or "error").strip().lower()
    if policy not in {"error", "skip", "keep"}:
        policy = "error"

    def ignore_with_symlink_skip(path: str, names: List[str]) -> set[str]:
        ignored = set(base_ignore(path, names)) if base_ignore else set()
        if policy != "skip":
            return ignored
        root = Path(path)
        for n in names:
            if n in ignored:
                continue
            try:
                if (root / n).is_symlink():
                    ignored.add(n)
            except Exception:
                continue
        return ignored

    ignore = ignore_with_symlink_skip if (base_ignore or policy == "skip") else None

    if policy == "error":
        syms = _find_symlinks(src_dir, ignore=ignore)
        if syms:
            preview = "\n".join([f"- {s}" for s in syms[:20]])
            more = "" if len(syms) <= 20 else f"\n- ...(and {len(syms) - 20} more)"
            raise ValueError(
                "src_dir contains symlinks; this breaks the workspace boundary assumption.\n"
                "Refuse to proceed by default.\n"
                f"Symlinks (relative to src_dir):\n{preview}{more}\n\n"
                "Use --symlink-policy skip (drop symlinks) or --symlink-policy keep (copy symlinks as symlinks) if you understand the risk."
            )

    if dst_dir.exists():
        shutil.rmtree(dst_dir)
    # Always copy symlinks as symlinks (never dereference), to avoid accidentally pulling in files
    # from outside src_dir via symlink targets.
    shutil.copytree(src_dir, dst_dir, symlinks=True, ignore=ignore)


def _build_shell_cmd_from_template(template: str, wrapped_prompt: str) -> List[str]:
    t = str(template or "")
    if not t.strip():
        raise ValueError("runner-cmd is empty")
    if "{prompt}" not in t:
        raise ValueError("runner-cmd must include a '{prompt}' placeholder")
    cmd_str = t.replace("{prompt}", shlex.quote(wrapped_prompt))
    return shlex.split(cmd_str)


def _cmd_with_optional_model(
    *,
    base_cmd: Sequence[str],
    global_args: Sequence[str],
    model_flag: str,
    model: str,
    subcommand: Sequence[str],
    subcommand_args: Sequence[str],
    prompt_arg: str,
) -> List[str]:
    cmd: List[str] = list(base_cmd)
    cmd.extend(list(global_args))
    m = str(model or "").strip()
    if m:
        cmd.extend([model_flag, m])
    cmd.extend(list(subcommand))
    cmd.extend(list(subcommand_args))
    cmd.append(prompt_arg)
    return cmd


def _resolve_model_id(cfg: dict, *, runner_type: str, model: str, profile: str) -> str:
    """
    Prefer an explicit model_id from the plan. If absent, fall back to
    config.yaml:models.{runner_type}.{profile|default}.
    """
    m = str(model or "").strip()
    if m:
        return m
    rt = str(runner_type or "").strip().lower()
    prof = str(profile or "").strip() or "default"
    models = (cfg.get("models", {}) or {}).get(rt, {}) or {}
    return str(models.get(prof) or models.get("default") or "").strip()


def _resolve_profile_args(cfg: dict, *, runner_type: str, profile: str) -> List[str]:
    rt = str(runner_type or "").strip().lower()
    prof = str(profile or "").strip() or "default"
    c = (cfg.get("cli", {}) or {}).get(rt, {}) or {}
    profile_args = c.get("profile_args", {}) or {}
    pa = profile_args.get(prof, None)
    if pa is None:
        pa = profile_args.get("default", None)
    if isinstance(pa, list):
        return [str(x) for x in pa if str(x).strip()]
    return []


def build_runner_cmd(
    *,
    runner_type: str,
    wrapped_prompt: str,
    profile: str,
    model: str,
    runner_args: Optional[Sequence[str]],
    runner_sub_args: Optional[Sequence[str]],
    runner_cmd_template: Optional[str],
    cfg: dict,
) -> List[str]:
    """
    Build a *single* command for one thread.

    runner_type:
      - codex: codex exec "..."
      - claude: claude -p "..."
      - shell: user-provided template with {prompt}
      - local: deterministic local runner for tests
    """
    t = runner_type.strip().lower()
    extra = [str(x) for x in (runner_args or [])]
    sub_extra = [str(x) for x in (runner_sub_args or [])]

    if t == "codex":
        c = cfg.get("cli", {}).get("codex", {}) or {}
        base_cmd = list(c.get("cmd") or ["codex"])
        sub = list(c.get("exec_subcommand") or ["exec"])
        model_flag = str(c.get("model_flag") or "-m")
        global_args = [str(x) for x in (c.get("global_args") or []) if str(x).strip()]
        global_args.extend(_resolve_profile_args(cfg, runner_type="codex", profile=profile))
        global_args.extend([x for x in extra if str(x).strip()])
        sub_args = [str(x) for x in (c.get("subcommand_args") or []) if str(x).strip()]
        sub_args.extend([x for x in sub_extra if str(x).strip()])
        return _cmd_with_optional_model(
            base_cmd=base_cmd,
            global_args=global_args,
            model_flag=model_flag,
            model=model,
            subcommand=sub,
            subcommand_args=sub_args,
            prompt_arg=wrapped_prompt,
        )

    if t == "claude":
        c = cfg.get("cli", {}).get("claude", {}) or {}
        base_cmd = list(c.get("cmd") or ["claude"])
        sub = list(c.get("print_subcommand") or ["-p"])
        model_flag = str(c.get("model_flag") or "--model")
        global_args = [str(x) for x in (c.get("global_args") or []) if str(x).strip()]
        global_args.extend(_resolve_profile_args(cfg, runner_type="claude", profile=profile))
        global_args.extend([x for x in extra if str(x).strip()])
        sub_args = [str(x) for x in (c.get("subcommand_args") or []) if str(x).strip()]
        sub_args.extend([x for x in sub_extra if str(x).strip()])
        return _cmd_with_optional_model(
            base_cmd=base_cmd,
            global_args=global_args,
            model_flag=model_flag,
            model=model,
            subcommand=sub,
            subcommand_args=sub_args,
            prompt_arg=wrapped_prompt,
        )

    if t == "shell":
        if not runner_cmd_template:
            raise ValueError("--runner-cmd is required when runner_type is shell")
        # Shell template can include a model placeholder too, but we keep it minimal.
        return _build_shell_cmd_from_template(runner_cmd_template, wrapped_prompt)

    if t == "local":
        # Deterministic runner for tests: writes a RESULT.md to the workspace.
        code = (
            "import os, pathlib, textwrap\n"
            "tid=os.environ.get('PARALLEL_VIBE_THREAD_ID','')\n"
            "pid=os.environ.get('PARALLEL_VIBE_PROJECT_ID','')\n"
            "p=pathlib.Path('RESULT.md')\n"
            "p.write_text(textwrap.dedent(f'''\\\n"
            "# Local Runner Result\\n\\n"
            "- thread_id: `{tid}`\\n"
            "- project_id: `{pid}`\\n"
            "'''), encoding='utf-8')\n"
            "print('wrote', str(p))\n"
        )
        return [sys.executable, "-c", code]

    raise ValueError(f"unknown runner_type: {runner_type}")


@dataclass(frozen=True)
class ThreadDirs:
    thread_id: str
    thread_root: Path
    workspace: Path
    runner_log: Path
    exit_code_txt: Path
    done_json: Path
    thread_json: Path
    prompt_txt: Path
    result_md: Path


def make_thread_dirs(project_root: Path, thread_id: str) -> ThreadDirs:
    thread_root = project_root / thread_id
    workspace = thread_root / "workspace"
    return ThreadDirs(
        thread_id=thread_id,
        thread_root=thread_root,
        workspace=workspace,
        runner_log=thread_root / "runner.log",
        exit_code_txt=thread_root / "exit_code.txt",
        done_json=thread_root / "done.json",
        thread_json=thread_root / "thread.json",
        prompt_txt=thread_root / "prompt.txt",
        result_md=thread_root / "RESULT.md",
    )


def _write_json(path: Path, obj: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False, indent=2)
        f.write("\n")


def _write_text(path: Path, s: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(s, encoding="utf-8")


def ensure_project_root(workdir: Path, work_dir_name: str, project_id: str) -> Path:
    base = (workdir.resolve() / work_dir_name).resolve()
    task_root = (base / f"task-{project_id}-{SKILL_NAME}").resolve()
    project_root = (task_root / SKILL_NAME).resolve()
    _require_within(base, project_root)
    project_root.mkdir(parents=True, exist_ok=True)
    for category in ("input", "output", "log"):
        (project_root / category).mkdir(exist_ok=True)
    _write_text(
        task_root / "README.md",
        "# BenszAPI 任务工作区\n\n"
        f"- 本轮 skill：`{SKILL_NAME}`\n"
        "- 并行线程、计划和日志均位于该 skill 的 input/output/log 子目录或其专用子目录。\n",
    )
    return project_root


def allocate_project_root(workdir: Path, work_dir_name: str, base_run_id: str) -> tuple[str, Path]:
    base = (workdir.resolve() / work_dir_name).resolve()
    _require_within(workdir.resolve(), base)
    base.mkdir(parents=True, exist_ok=True)
    for idx in range(1, 100):
        run_id = base_run_id if idx == 1 else f"{base_run_id}-{idx:02d}"
        task_root = (base / f"task-{run_id}-{SKILL_NAME}").resolve()
        project_root = (task_root / SKILL_NAME).resolve()
        _require_within(base, project_root)
        if not project_root.exists():
            project_root.mkdir(parents=True, exist_ok=True)
            for category in ("input", "output", "log"):
                (project_root / category).mkdir(exist_ok=True)
            _write_text(
                task_root / "README.md",
                "# BenszAPI 任务工作区\n\n"
                f"- 本轮 skill：`{SKILL_NAME}`\n"
                "- 并行线程、计划和日志均位于该 skill 的 input/output/log 子目录或其专用子目录。\n",
            )
            return run_id, project_root
    raise RuntimeError(f"failed to allocate unique run directory under {base}: {base_run_id}")


def ensure_project_json(project_root: Path, meta: dict) -> Path:
    path = project_root / "project.json"
    _write_json(path, meta)
    return path


def _read_text_maybe(path: Path, *, max_chars: int = 80_000) -> str:
    try:
        s = path.read_text(encoding="utf-8")
    except Exception:
        return ""
    if len(s) > max_chars:
        return s[:max_chars] + f"\n\n...(truncated, total_chars={len(s)})\n"
    return s


def _build_split_plan(
    *,
    user_prompt: str,
    n_threads: int,
    thread_id_width: int,
    cfg: dict,
) -> dict:
    """
    Deterministic plan template (no LLM calls). The host AI can still override
    by passing --plan-file.
    """
    def model_for(runner: str, profile: str) -> str:
        models = cfg.get("models", {}).get(runner, {}) or {}
        return str(models.get(profile) or models.get("default") or "").strip()

    def mk_thread(i: int, title: str, runner: str, profile: str, extra: str) -> dict:
        tid = str(i).zfill(thread_id_width)
        # Each thread must always produce a small, structured artifact in workspace root.
        prompt = (
            f"{user_prompt.strip()}\n\n"
            "你需要以本 thread 的角色独立完成上述任务，并遵守以下交付要求：\n"
            "- 优先：在当前工作目录写出 `RESULT.md`（Markdown）。\n"
            "- 如果你所在 runner 处于“只打印不落盘”的模式（例如 `claude -p`），请直接在输出中给出 `RESULT.md` 的完整内容（Markdown），以便上层把它落盘。\n"
            "- RESULT.md 必须包含：你做了什么、关键决策、你运行了哪些命令（如有）、如何验证、风险与下一步。\n"
            "- 如果你修改了代码：只在当前工作区内修改，并在 RESULT.md 里列出你改动的关键文件路径。\n"
            f"\n你的角色要求：{extra.strip()}\n"
        )
        return {
            "thread_id": tid,
            "title": title,
            "runner": {
                "type": runner,
                "profile": profile,
                "model": model_for(runner, profile),
                "args": [],
            },
            "prompt": prompt,
        }

    threads: List[dict] = []
    if n_threads <= 1:
        threads.append(
            mk_thread(
                1,
                "Full Task",
                "codex",
                "deep",
                "一次性完成：方案设计 + 实现/修改 + 自检 + 轻量验证。优先最小可用，避免过度设计。",
            )
        )
    elif n_threads == 2:
        threads.append(
            mk_thread(1, "Implementation", "codex", "deep", "聚焦实现/修改出一个可用版本。")
        )
        threads.append(
            mk_thread(2, "Review & Risks", "claude", "deep", "聚焦挑错：边界/安全/一致性/可维护性，给出可执行改进清单。")
        )
    elif n_threads == 3:
        threads.append(
            mk_thread(1, "Approach A", "codex", "deep", "实现/修改：路线 A（偏保守、最小改动）。")
        )
        threads.append(
            mk_thread(2, "Approach B", "codex", "deep", "实现/修改：路线 B（偏激进/重构，但仍保持可交付）。")
        )
        threads.append(
            mk_thread(3, "Tests & Edge Cases", "codex", "fast", "聚焦测试与边界：补充最小验证、列出高风险边界。")
        )
    else:
        # Default (>=4): start with planning, then implementations, then tests/review.
        threads.append(
            mk_thread(
                1,
                "Planner",
                "claude",
                "deep",
                "聚焦规划：澄清需求边界，提出结构化执行计划与验收标准。不要假设你能访问其他 thread 的产物。",
            )
        )
        # Implementation threads
        threads.append(
            mk_thread(2, "Implementation A", "codex", "deep", "实现/修改：按计划落地，优先正确性与可验证性。")
        )
        threads.append(
            mk_thread(3, "Implementation B", "codex", "deep", "实现/修改：尝试不同实现或重构路径，争取更高质量。")
        )
        threads.append(
            mk_thread(
                4,
                "Tests & Verification",
                "codex",
                "fast",
                "聚焦测试与验证：跑/补最小测试，列出如何复现与验证。",
            )
        )
        if n_threads >= 5:
            threads.append(
                mk_thread(
                    5,
                    "Code Review",
                    "claude",
                    "deep",
                    "聚焦 code review：指出潜在 bug/回归/安全风险，并给出最小修复建议。",
                )
            )
        # Extra threads beyond 5: keep them useful and diverse.
        for i in range(6, n_threads + 1):
            threads.append(
                mk_thread(
                    i,
                    f"Extra {i}",
                    "codex",
                    "fast",
                    "聚焦补充视角：性能/可读性/文档/错误处理（任选最关键点），输出可执行建议。",
                )
            )

    plan = {
        "plan_version": 1,
        "prompt": user_prompt,
        "threads": threads[:n_threads],
        "synthesis": {
            "enabled": bool(cfg.get("defaults", {}).get("synthesize", True)),
            "runner": {
                "type": "claude",
                "profile": "deep",
                "model": str((cfg.get("models", {}).get("claude", {}) or {}).get("deep") or "").strip(),
                "args": [],
            },
            "prompt": (
                "请综合输入中的多 thread 产物，生成一份可执行的最终结论（面向用户）。\n"
                "要求：\n"
                "- 先给结论，再给取舍理由。\n"
                "- 如果存在互相矛盾的建议，明确你选择哪一个，并说明依据。\n"
                "- 给出最小验证步骤（命令/检查点）。\n"
                "- 输出为 Markdown。\n"
            ),
        },
    }
    return plan


def _validate_plan_obj(plan: dict, *, thread_id_width: int) -> None:
    if not isinstance(plan, dict):
        raise ValueError("plan must be a JSON object")
    threads = plan.get("threads")
    if not isinstance(threads, list) or not threads:
        raise ValueError("plan.threads must be a non-empty list")
    seen: set[str] = set()
    for t in threads:
        if not isinstance(t, dict):
            raise ValueError("each thread must be an object")
        tid = str(t.get("thread_id") or "").strip()
        if not tid.isdigit() or len(tid) != thread_id_width:
            raise ValueError(f"invalid thread_id: {tid!r} (expected {thread_id_width}-digit string)")
        if tid in seen:
            raise ValueError(f"duplicate thread_id in plan: {tid}")
        seen.add(tid)
        runner = t.get("runner")
        if not isinstance(runner, dict):
            raise ValueError(f"thread {tid}: runner must be an object")
        rtype = str(runner.get("type") or "").strip().lower()
        if rtype not in {"codex", "claude", "shell", "local"}:
            raise ValueError(f"thread {tid}: unsupported runner.type: {rtype!r}")
        if rtype == "shell":
            # shell runner needs a template somewhere (thread or global)
            if not str(runner.get("cmd_template") or "").strip():
                raise ValueError(f"thread {tid}: runner.cmd_template required when runner.type=shell")
        if "sub_args" in runner and not isinstance(runner.get("sub_args"), list):
            raise ValueError(f"thread {tid}: runner.sub_args must be a list when provided")
        if not str(t.get("prompt") or "").strip():
            raise ValueError(f"thread {tid}: prompt is required")


def _run_threads(
    *,
    thread_dirs: List[ThreadDirs],
    cfg: dict,
    plan: dict,
    project_id: str,
    timeout_seconds: int,
    parallel: bool,
    max_parallel: int,
) -> List[dict]:
    plan_threads = {str(t.get("thread_id")): t for t in (plan.get("threads") or []) if isinstance(t, dict)}

    def start_thread(td: ThreadDirs) -> Tuple[subprocess.Popen, Any, dict]:
        t = plan_threads.get(td.thread_id) or {}
        runner = t.get("runner") or {}

        env = os.environ.copy()
        env["PARALLEL_VIBE_THREAD_ID"] = td.thread_id
        env["PARALLEL_VIBE_PROJECT_ID"] = project_id

        start_at = _now_iso()
        wrapped_prompt = wrap_thread_prompt(str(t.get("prompt") or ""), td.thread_id, project_id)

        td.workspace.mkdir(parents=True, exist_ok=True)
        _write_text(td.prompt_txt, wrapped_prompt)
        _write_json(td.thread_json, t)

        cmd = build_runner_cmd(
            runner_type=str(runner.get("type") or ""),
            wrapped_prompt=wrapped_prompt,
            profile=str(runner.get("profile") or "").strip(),
            model=_resolve_model_id(
                cfg,
                runner_type=str(runner.get("type") or ""),
                model=str(runner.get("model") or ""),
                profile=str(runner.get("profile") or ""),
            ),
            runner_args=runner.get("args") if isinstance(runner.get("args"), list) else [],
            runner_sub_args=runner.get("sub_args") if isinstance(runner.get("sub_args"), list) else [],
            runner_cmd_template=str(runner.get("cmd_template") or "").strip() or None,
            cfg=cfg,
        )

        log_f = td.runner_log.open("w", encoding="utf-8")
        try:
            p = subprocess.Popen(
                cmd,
                cwd=str(td.workspace),
                stdout=log_f,
                stderr=subprocess.STDOUT,
                env=env,
            )
        except Exception:
            try:
                log_f.close()
            except Exception:
                pass
            raise
        # Keep internal start time for enforcing timeouts in parallel mode.
        try:
            import time as _time

            start_mono = _time.monotonic()
        except Exception:
            start_mono = 0.0
        meta = {
            "thread_id": td.thread_id,
            "runner": runner.get("type"),
            "cmd": cmd,
            "start_at": start_at,
            "_start_mono": start_mono,
        }
        return p, log_f, meta

    def finish_thread(td: ThreadDirs, p: subprocess.Popen, log_f: Any, meta: dict) -> dict:
        err: Optional[str] = None
        exit_code: int = 1
        end_at: str = _now_iso()
        # Internal-only fields (do not persist into done.json).
        meta.pop("_start_mono", None)
        timeout_killed = bool(meta.pop("_timeout_killed", False))
        try:
            polled = p.poll()
            if polled is not None:
                exit_code = int(polled)
                end_at = _now_iso()
            elif timeout_seconds and timeout_seconds > 0:
                exit_code = p.wait(timeout=timeout_seconds)
                end_at = _now_iso()
            else:
                exit_code = p.wait()
                end_at = _now_iso()
        except subprocess.TimeoutExpired:
            err = f"timeout after {timeout_seconds}s"
            try:
                p.kill()
            except Exception:
                pass
            exit_code = 124
            end_at = _now_iso()
        except Exception as e:
            err = f"{type(e).__name__}: {e}"
            try:
                p.kill()
            except Exception:
                pass
            exit_code = 1
            end_at = _now_iso()
        finally:
            try:
                log_f.close()
            except Exception:
                pass

        # Pull RESULT.md from workspace root if present.
        ws_result = td.workspace / "RESULT.md"
        if ws_result.exists():
            try:
                td.result_md.write_text(_read_text_maybe(ws_result), encoding="utf-8")
            except Exception:
                pass
        elif td.runner_log.exists():
            # Fallback: make sure we always have a per-thread RESULT.md even for "print-only" runners (e.g. claude -p).
            try:
                log_txt = _read_text_maybe(td.runner_log, max_chars=120_000)
                td.result_md.write_text(
                    "# Thread Result (Fallback)\n\n"
                    "本 thread 未生成 `workspace/RESULT.md`，以下为 runner 的标准输出/错误（见 `runner.log`）。\n\n"
                    "```text\n"
                    f"{log_txt.rstrip()}\n"
                    "```\n",
                    encoding="utf-8",
                )
            except Exception:
                pass

        _write_text(td.exit_code_txt, str(exit_code) + "\n")
        done = dict(meta)
        done.update({"end_at": end_at, "exit_code": exit_code})
        if timeout_killed and not err:
            # Normalize timeout semantics even if the OS exit code is e.g. SIGKILL.
            done["exit_code"] = 124
            done["error"] = f"timeout after {timeout_seconds}s"
        elif err:
            done["error"] = err
        _write_json(td.done_json, done)
        return done

    metas: List[dict] = []
    running: List[Tuple[ThreadDirs, subprocess.Popen, Any, dict]] = []

    if not parallel:
        for td in thread_dirs:
            try:
                p, log_f, meta = start_thread(td)
            except FileNotFoundError as e:
                done = {"thread_id": td.thread_id, "exit_code": 127, "error": f"FileNotFoundError: {e}", "start_at": _now_iso(), "end_at": _now_iso()}
                _write_text(td.exit_code_txt, "127\n")
                _write_json(td.done_json, done)
                _write_text(td.result_md, f"# Thread Result\n\n- exit_code: 127\n- error: {done['error']}\n")
                metas.append(done)
                continue
            except Exception as e:
                done = {"thread_id": td.thread_id, "exit_code": 1, "error": f"{type(e).__name__}: {e}", "start_at": _now_iso(), "end_at": _now_iso()}
                _write_text(td.exit_code_txt, "1\n")
                _write_json(td.done_json, done)
                _write_text(td.result_md, f"# Thread Result\n\n- exit_code: 1\n- error: {done['error']}\n")
                metas.append(done)
                continue
            metas.append(finish_thread(td, p, log_f, meta))
        return metas

    # Parallel with a simple fixed-size pool (max_parallel).
    cap = max(1, int(max_parallel or 1))
    q = list(thread_dirs)
    while q or running:
        while q and len(running) < cap:
            td = q.pop(0)
            try:
                p, log_f, meta = start_thread(td)
                running.append((td, p, log_f, meta))
            except FileNotFoundError as e:
                done = {"thread_id": td.thread_id, "exit_code": 127, "error": f"FileNotFoundError: {e}", "start_at": _now_iso(), "end_at": _now_iso()}
                _write_text(td.exit_code_txt, "127\n")
                _write_json(td.done_json, done)
                _write_text(td.result_md, f"# Thread Result\n\n- exit_code: 127\n- error: {done['error']}\n")
                metas.append(done)
            except Exception as e:
                done = {"thread_id": td.thread_id, "exit_code": 1, "error": f"{type(e).__name__}: {e}", "start_at": _now_iso(), "end_at": _now_iso()}
                _write_text(td.exit_code_txt, "1\n")
                _write_json(td.done_json, done)
                _write_text(td.result_md, f"# Thread Result\n\n- exit_code: 1\n- error: {done['error']}\n")
                metas.append(done)

        # Poll running processes; finish those that exited.
        still: List[Tuple[ThreadDirs, subprocess.Popen, Any, dict]] = []
        progressed = False
        for td, p, log_f, meta in running:
            if p.poll() is None:
                # Enforce per-thread timeout in parallel mode (best-effort).
                if timeout_seconds and timeout_seconds > 0:
                    try:
                        import time as _time

                        start_mono = float(meta.get("_start_mono") or 0.0)
                        if start_mono > 0.0 and (_time.monotonic() - start_mono) > float(timeout_seconds):
                            try:
                                p.kill()
                            except Exception:
                                pass
                            meta["_timeout_killed"] = True
                    except Exception:
                        pass
                still.append((td, p, log_f, meta))
                continue
            metas.append(finish_thread(td, p, log_f, meta))
            progressed = True
        running = still
        if not progressed:
            # Avoid busy-wait.
            try:
                import time

                time.sleep(0.2)
            except Exception:
                pass

    return metas


def render_summary(project_meta: dict, plan: dict, thread_metas: List[dict]) -> str:
    lines: List[str] = []
    lines.append("# parallel-vibe Summary")
    lines.append("")
    lines.append("## Project")
    lines.append(f"- project_id: `{project_meta.get('project_id','')}`")
    lines.append(f"- created_at: `{project_meta.get('created_at','')}`")
    if project_meta.get("last_run_at"):
        lines.append(f"- last_run_at: `{project_meta.get('last_run_at','')}`")
    lines.append(f"- n_threads: `{project_meta.get('n_threads','')}`")
    lines.append(f"- execution: `{project_meta.get('execution','')}`")
    lines.append("")
    lines.append("## Prompt")
    lines.append("")
    lines.append("```text")
    lines.append(str(project_meta.get("last_run_prompt") or project_meta.get("prompt") or ""))
    lines.append("```")
    lines.append("")
    lines.append("## Plan")
    lines.append("")
    for t in plan.get("threads") or []:
        if not isinstance(t, dict):
            continue
        tid = str(t.get("thread_id") or "")
        title = str(t.get("title") or "").strip()
        runner = t.get("runner") or {}
        rtype = str(runner.get("type") or "")
        profile = str(runner.get("profile") or "").strip()
        model = str(runner.get("model") or "").strip()
        extra: List[str] = [f"runner={rtype}"]
        if profile:
            extra.append(f"profile={profile}")
        if model:
            extra.append(f"model={model}")
        if len(extra) > 1:
            lines.append(f"- {tid}: {title} ({', '.join(extra)})")
        else:
            lines.append(f"- {tid}: {title} (runner={rtype})")
    lines.append("")
    lines.append("## Threads")
    lines.append("")
    for tm in sorted(thread_metas, key=lambda x: x.get("thread_id", "")):
        tid = tm.get("thread_id", "")
        ec = tm.get("exit_code", "")
        err = str(tm.get("error") or "").strip()
        if err:
            err = err.replace("\n", " ")
            if len(err) > 140:
                err = err[:140] + "..."
            lines.append(f"- {tid}: exit_code={ec} error={err} (see `{tid}/RESULT.md` and `{tid}/runner.log`)")
        else:
            lines.append(f"- {tid}: exit_code={ec} (see `{tid}/RESULT.md` and `{tid}/runner.log`)")
    lines.append("")
    lines.append("## Where To Look")
    lines.append("")
    lines.append("- 汇总（索引）：`@main/summary.md`")
    lines.append("- 规划：`@main/plan.json`（机器可读） / `@main/plan.md`（人类可读）")
    lines.append("- 各 thread 结果：`{thread_id}/RESULT.md`（如果 runner 生成）")
    lines.append("")
    return "\n".join(lines)


def _preflight_check_runners(cfg: dict, plan: dict, *, synth_enabled: bool) -> Optional[str]:
    """
    Return an error message if required runner executables are missing.
    """
    required: set[str] = set()
    for t in plan.get("threads") or []:
        if not isinstance(t, dict):
            continue
        runner = t.get("runner") or {}
        rtype = str((runner or {}).get("type") or "").strip().lower()
        if rtype and rtype not in {"shell", "local"}:
            required.add(rtype)

    if synth_enabled:
        synth = plan.get("synthesis") or {}
        sr = synth.get("runner") or {}
        rtype = str((sr or {}).get("type") or "").strip().lower()
        if rtype and rtype not in {"shell", "local"}:
            required.add(rtype)

    missing: List[str] = []
    for rt in sorted(required):
        c = (cfg.get("cli", {}) or {}).get(rt, {}) or {}
        cmd = c.get("cmd") or []
        exe = str(cmd[0]) if isinstance(cmd, list) and cmd else rt
        if not shutil.which(exe):
            missing.append(f"- runner={rt}: executable not found in PATH: {exe}")

    if missing:
        return (
            "required runner executable(s) not found:\n"
            + "\n".join(missing)
            + "\n\n"
            "Fix:\n"
            "- install the missing CLI(s), or\n"
            "- edit `@main/plan.json` and switch runner.type to an available runner (e.g. codex/shell/local).\n"
        )
    return None


def main(argv: Optional[Sequence[str]] = None) -> int:
    cfg = load_config()
    defaults = cfg.get("defaults", {})

    p = argparse.ArgumentParser(prog="parallel_vibe.py")
    p.add_argument("--prompt", default="", help="用户指令原文（用于生成计划并传给各 thread runner）")
    p.add_argument("--plan-file", default="", help="自定义计划文件（JSON）。如提供，则忽略 --prompt 的自动拆分逻辑。")
    p.add_argument("--plan-only", action="store_true", help="只生成 project 目录与 plan.json，不运行 threads")
    p.add_argument("--n", type=int, default=int(defaults.get("n_threads", 5)), help="线程数（1-9；仅在未提供 --plan-file 时生效）")
    p.add_argument("--src-dir", default=".", help="复制到各 thread/workspace 的源目录（默认当前目录）")
    p.add_argument("--out-dir", default=".", help="创建 .bensz-api/task-{timestamp}-parallel-vibe/parallel-vibe 的项目根目录（默认当前目录）")
    # Backward-compatible alias (old versions used --workdir as out-dir).
    p.add_argument("--workdir", default="", help=argparse.SUPPRESS)
    p.add_argument("--project-id", default="", help="指定或复用 run/project id；未指定时默认使用 yyyy-mm-dd-hh-mm")
    p.add_argument(
        "--resume",
        action="store_true",
        help="项目目录存在时复用（保留 @main 与 project.json；每次运行仍会重建各 thread/workspace）",
    )
    p.add_argument("--timeout-seconds", type=int, default=0, help="0 表示不超时")
    p.add_argument(
        "--copy-exclude",
        default=",".join(list(defaults.get("copy_exclude", []))),
        help="逗号分隔的排除项（用于复制 workspace）",
    )
    p.add_argument(
        "--symlink-policy",
        default=str(defaults.get("symlink_policy") or "error"),
        choices=["error", "skip", "keep"],
        help="src_dir 中遇到 symlink 的处理策略：error(默认拒绝)/skip(剔除)/keep(保留为 symlink；有越界风险)",
    )
    p.add_argument("--parallel", action="store_true", help="并行运行 threads（默认串行）")
    p.add_argument("--max-parallel", type=int, default=int(defaults.get("max_parallel", 3)), help="并行上限（仅 --parallel 时生效）")
    p.add_argument("--synthesize", action="store_true", help="运行汇总 synth（会额外调用一次 runner；默认由 config.yaml 控制）")
    p.add_argument("--no-synthesize", action="store_true", help="禁用 synth（覆盖 config.yaml）")
    p.add_argument("--dry-run", action="store_true", help="只打印/落盘计划与命令，不实际运行")
    args = p.parse_args(argv)

    # Resolve out-dir with backward-compatible behavior.
    out_dir_arg = str(args.out_dir or "").strip()
    if not out_dir_arg and str(args.workdir or "").strip():
        out_dir_arg = str(args.workdir).strip()
    out_dir = Path(out_dir_arg or ".").resolve()
    src_dir = Path(str(args.src_dir)).resolve()

    try:
        _validate_existing_dir(out_dir, label="out_dir")
        _validate_existing_dir(src_dir, label="src_dir")
    except ValueError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2

    plan_file = str(args.plan_file or "").strip()
    user_prompt = str(args.prompt or "").strip()
    if not plan_file and not user_prompt:
        print("error: either --prompt or --plan-file is required", file=sys.stderr)
        return 2

    work_dir_name = WORK_DIR_NAME
    # Prevent writing outside out_dir if ".bensz-api" or a nested workspace path is a symlink to elsewhere.
    try:
        base = (out_dir.resolve() / WORK_DIR_NAME).resolve()
        _require_within(out_dir.resolve(), base)
    except ValueError as e:
        print(f"error: invalid {WORK_DIR_NAME} directory under out_dir: {e}", file=sys.stderr)
        return 2

    thread_id_width = int(defaults.get("thread_id_width", 3))
    exclude_names = _parse_copy_exclude(str(args.copy_exclude), list(defaults.get("copy_exclude", [])))
    for name in [".bensz-api", work_dir_name, *LEGACY_WORK_DIR_NAMES]:
        if name not in exclude_names:
            exclude_names.append(name)

    project_id = str(args.project_id).strip()
    if project_id:
        if not _is_safe_run_id(project_id):
            print("error: --project-id must be a safe run id using letters, digits, dot, underscore or hyphen", file=sys.stderr)
            return 2
        project_root = ensure_project_root(out_dir, work_dir_name, project_id)
    else:
        if args.resume:
            print("error: --resume requires --project-id when using timestamped run directories", file=sys.stderr)
            return 2
        try:
            project_id, project_root = allocate_project_root(out_dir, work_dir_name, _now_run_id())
        except RuntimeError as e:
            print(f"error: {e}", file=sys.stderr)
            return 2

    existing_entries = [
        entry for entry in project_root.iterdir()
        if entry.name not in {"input", "output", "log"}
    ]
    if existing_entries and not args.resume:
        # Avoid surprising overwrites; use --resume to reuse an existing project directory.
        print(f"error: project already exists: {project_root} (use --resume)", file=sys.stderr)
        return 2

    project_json_path = project_root / "project.json"
    created_at = _now_iso()
    base_prompt = user_prompt
    if args.resume and project_json_path.exists():
        try:
            existing = json.loads(project_json_path.read_text(encoding="utf-8"))
            if isinstance(existing, dict):
                created_at = str(existing.get("created_at") or created_at)
                base_prompt = str(existing.get("prompt") or base_prompt)
        except Exception:
            pass

    if args.n < 1 or args.n > 9:
        print("error: --n must be in [1, 9]", file=sys.stderr)
        return 2

    # Load or build plan.
    plan: dict
    if plan_file:
        try:
            plan = json.loads(Path(plan_file).read_text(encoding="utf-8"))
        except Exception as e:
            print(f"error: failed to read plan file: {e}", file=sys.stderr)
            return 2
    else:
        plan = _build_split_plan(
            user_prompt=user_prompt,
            n_threads=int(args.n),
            thread_id_width=thread_id_width,
            cfg=cfg,
        )

    try:
        _validate_plan_obj(plan, thread_id_width=thread_id_width)
    except ValueError as e:
        print(f"error: invalid plan: {e}", file=sys.stderr)
        return 2

    # Determine execution mode (default serial).
    exec_mode = "parallel" if bool(args.parallel) else str(defaults.get("execution") or "serial")
    if exec_mode not in {"serial", "parallel"}:
        exec_mode = "serial"

    # Synthesis toggle: config default -> CLI overrides.
    synth_enabled = bool(plan.get("synthesis", {}).get("enabled", False))
    if bool(args.synthesize):
        synth_enabled = True
    if bool(args.no_synthesize):
        synth_enabled = False

    preflight_err = _preflight_check_runners(cfg, plan, synth_enabled=synth_enabled)
    if preflight_err:
        print(f"error: {preflight_err}", file=sys.stderr)
        return 127

    project_meta = {
        "project_id": project_id,
        "created_at": created_at,
        "prompt": base_prompt,
        "last_run_at": _now_iso(),
        "last_run_prompt": user_prompt or base_prompt,
        "n_threads": len(list(plan.get("threads") or [])),
        "execution": exec_mode,
        "src_dir": str(src_dir),
        "out_dir": str(out_dir),
    }
    ensure_project_json(project_root, project_meta)

    # Threads
    thread_dirs: List[ThreadDirs] = []
    for t in plan.get("threads") or []:
        tid = str((t or {}).get("thread_id") or "").strip()
        td = make_thread_dirs(project_root, tid)
        td.thread_root.mkdir(parents=True, exist_ok=True)
        thread_dirs.append(td)

    # Copy workspaces
    for td in thread_dirs:
        try:
            copy_workspace(
                src_dir,
                td.workspace,
                exclude_names,
                symlink_policy=str(args.symlink_policy or "error"),
            )
        except ValueError as e:
            print(f"error: failed to copy workspace for thread {td.thread_id}: {e}", file=sys.stderr)
            return 2

    main_dir = project_root / "@main"
    main_dir.mkdir(parents=True, exist_ok=True)
    plan_json_path = main_dir / "plan.json"
    plan_md_path = main_dir / "plan.md"
    _write_json(plan_json_path, plan)
    # A small human-readable plan rendering.
    plan_lines: List[str] = ["# parallel-vibe Plan", ""]
    plan_lines.append("## Threads")
    plan_lines.append("")
    for t in plan.get("threads") or []:
        if not isinstance(t, dict):
            continue
        tid = str(t.get("thread_id") or "")
        title = str(t.get("title") or "").strip()
        runner = t.get("runner") or {}
        rtype = str(runner.get("type") or "")
        profile = str(runner.get("profile") or "").strip()
        model = str(runner.get("model") or "").strip()
        extra: List[str] = [f"runner={rtype}"]
        if profile:
            extra.append(f"profile={profile}")
        if model:
            extra.append(f"model={model}")
        if len(extra) > 1:
            plan_lines.append(f"- {tid}: {title} ({', '.join(extra)})")
        else:
            plan_lines.append(f"- {tid}: {title} (runner={rtype})")
    plan_lines.append("")
    plan_md_path.write_text("\n".join(plan_lines) + "\n", encoding="utf-8")

    if bool(args.plan_only):
        print(str(project_root))
        return 0

    if bool(args.dry_run):
        # We still create plan + workspaces for traceability.
        print(str(project_root))
        return 0

    # Run threads.
    thread_metas = _run_threads(
        thread_dirs=thread_dirs,
        cfg=cfg,
        plan=plan,
        project_id=project_id,
        timeout_seconds=int(args.timeout_seconds),
        parallel=(exec_mode == "parallel"),
        max_parallel=int(args.max_parallel),
    )

    summary_path = main_dir / "summary.md"
    summary_path.write_text(render_summary(project_meta, plan, thread_metas), encoding="utf-8")

    # Optional synthesis step: run one more CLI call in @main, feeding it thread RESULT.md files.
    if synth_enabled:
        synth = plan.get("synthesis") or {}
        synth_runner = synth.get("runner") or {}
        synth_prompt = str(synth.get("prompt") or "").strip()
        synth_runner_type = str(synth_runner.get("type") or "claude").strip().lower()
        # Build input
        input_lines: List[str] = []
        input_lines.append("# parallel-vibe Synthesis Input")
        input_lines.append("")
        input_lines.append("## Original Prompt")
        input_lines.append("")
        input_lines.append(user_prompt or base_prompt)
        input_lines.append("")
        input_lines.append("## Thread Results")
        input_lines.append("")
        for td in sorted(thread_dirs, key=lambda x: x.thread_id):
            input_lines.append(f"### Thread {td.thread_id}")
            input_lines.append("")
            if td.result_md.exists():
                input_lines.append(_read_text_maybe(td.result_md, max_chars=40_000))
            else:
                input_lines.append("_RESULT.md not found; see runner.log_")
            input_lines.append("")
        synth_input_path = main_dir / "synthesis_input.md"
        synth_input_path.write_text("\n".join(input_lines) + "\n", encoding="utf-8")

        try:
            # If we synthesize via codex, use stdin prompt ("-") and embed synth_prompt into stdin.
            stdin_path = synth_input_path
            wrapped_prompt = synth_prompt
            if synth_runner_type == "codex":
                stdin_lines = ["# Synthesis Instructions", "", synth_prompt, "", "---", ""]
                stdin_lines.extend(input_lines)
                stdin_path = main_dir / "synthesis_stdin.md"
                stdin_path.write_text("\n".join(stdin_lines) + "\n", encoding="utf-8")
                wrapped_prompt = "-"

            cmd = build_runner_cmd(
                runner_type=synth_runner_type,
                wrapped_prompt=wrapped_prompt,
                profile=str(synth_runner.get("profile") or "").strip(),
                model=_resolve_model_id(
                    cfg,
                    runner_type=synth_runner_type,
                    model=str(synth_runner.get("model") or ""),
                    profile=str(synth_runner.get("profile") or ""),
                ),
                runner_args=synth_runner.get("args") if isinstance(synth_runner.get("args"), list) else [],
                runner_sub_args=synth_runner.get("sub_args") if isinstance(synth_runner.get("sub_args"), list) else [],
                runner_cmd_template=str(synth_runner.get("cmd_template") or "").strip() or None,
                cfg=cfg,
            )
            _write_json(
                main_dir / "synthesis_meta.json",
                {"runner": synth_runner_type, "cmd": cmd, "start_at": _now_iso()},
            )
            summary_ai_path = main_dir / "summary_ai.md"
            synth_out = summary_ai_path.open("w", encoding="utf-8")
            p2 = subprocess.Popen(
                cmd,
                cwd=str(main_dir),
                stdin=stdin_path.open("r", encoding="utf-8"),
                stdout=synth_out,
                stderr=subprocess.STDOUT,
                env=os.environ.copy(),
            )
            if int(args.timeout_seconds) > 0:
                rc = p2.wait(timeout=int(args.timeout_seconds))
            else:
                rc = p2.wait()
            try:
                synth_out.close()
            except Exception:
                pass
            _write_text(main_dir / "synthesis_exit_code.txt", str(rc) + "\n")
            # Add a pointer in the main summary for discoverability.
            if summary_ai_path.exists():
                try:
                    with summary_path.open("a", encoding="utf-8") as f:
                        f.write("\n## Synthesized Result (AI)\n\n")
                        f.write("见：`@main/summary_ai.md`\n")
                except Exception:
                    pass
        except Exception as e:
            _write_text(main_dir / "synthesis_error.txt", f"{type(e).__name__}: {e}\n")

    print(str(project_root))
    return 0 if all(int(tm.get("exit_code", 1)) == 0 for tm in thread_metas) else 1


if __name__ == "__main__":
    raise SystemExit(main())