From 91e03629e457119ab0ddaef45684870f0aa12848 Mon Sep 17 00:00:00 2001 From: Xian Xu Date: Mon, 20 Jul 2026 17:48:29 +0800 Subject: [PATCH 1/2] add safe run-scoped workspaces --- README.md | 32 +++-- README_zh.md | 24 ++-- dashboard.py | 23 ++- main.py | 88 +++++++----- src/domain_knowledge.py | 8 +- src/entry_reasoning_pipeline.py | 10 +- src/incremental_reasoner.py | 75 ++++++---- src/opencode_trace.py | 22 +-- src/pipeline_setup.py | 41 +++--- src/run_workspace.py | 240 ++++++++++++++++++++++++++++++++ src/verification.py | 23 +-- tests/__init__.py | 1 + tests/test_dashboard_workdir.py | 22 +++ tests/test_run_workspace.py | 126 +++++++++++++++++ 14 files changed, 599 insertions(+), 136 deletions(-) create mode 100644 src/run_workspace.py create mode 100644 tests/__init__.py create mode 100644 tests/test_dashboard_workdir.py create mode 100644 tests/test_run_workspace.py diff --git a/README.md b/README.md index ac7a7319..80f804d8 100644 --- a/README.md +++ b/README.md @@ -160,13 +160,13 @@ You can open your oh-my-openagent config file (typically ~/.config/opencode/oh-m ### Structured Trace -FM-Agent always writes structured execution traces under `fm_agent/trace/`: +FM-Agent writes structured execution traces under each run's `trace/` directory (`fm_agent/runs//trace/`): | Path | Content | |---|---| -| `fm_agent/trace/events.jsonl` | Structured events for OpenCode calls and verification LLM calls | -| `fm_agent/trace/payloads/` | Event payloads such as OpenCode stdout and selected LLM messages | -| `fm_agent/trace/opencode/` | Optional raw OpenCode LLM request/response JSONL files | +| `trace/events.jsonl` | Structured events for OpenCode calls and verification LLM calls | +| `trace/payloads/` | Event payloads such as OpenCode stdout and selected LLM messages | +| `trace/opencode/` | Optional raw OpenCode LLM request/response JSONL files | To capture raw OpenCode LLM traffic, install the OpenCode trace plugin manually by adding it to `~/.config/opencode/opencode.json`: @@ -177,7 +177,7 @@ To capture raw OpenCode LLM traffic, install the OpenCode trace plugin manually } ``` -FM-Agent automatically passes `TRACE_DIR` and `TRACE_FILENAME` to each OpenCode process. The plugin writes `fm_agent/trace/opencode/.jsonl`, where `` matches the corresponding `opencode_call` event in `events.jsonl`. +FM-Agent automatically passes `TRACE_DIR` and `TRACE_FILENAME` to each OpenCode process. The plugin writes `trace/opencode/.jsonl` inside the selected run, where `` matches the corresponding `opencode_call` event in `events.jsonl`. OpenCode may cache the `@latest` package; to force a refresh, remove `~/.cache/opencode/packages/@lucentia/opencode-trace@latest`. @@ -204,7 +204,7 @@ To provide project-specific domain knowledge without editing FM-Agent's built-in uv run python main.py --domain-knowledge docs/invariants.md docs/protocol.md ``` -FM-Agent stages these files under `fm_agent/spec_prompts/domain_context/user_knowledge/` for the current run. You can also set `FM_AGENT_DOMAIN_KNOWLEDGE` to an `os.pathsep`-separated list of Markdown files. +FM-Agent stages these files under `fm_agent/runs//spec_prompts/domain_context/user_knowledge/` for the current run. You can also set `FM_AGENT_DOMAIN_KNOWLEDGE` to an `os.pathsep`-separated list of Markdown files. Use `--submodule` to limit a full or incremental run to selected project subdirectories: @@ -215,21 +215,23 @@ uv run python main.py --incremental intent.md --submodule src/core sr `--submodule` paths must point to directories inside `proj_dir`. The option can be combined with `--resume`, `--isolate`, and `--incremental`, but not with `--entry-func`. -By default, every invocation wipes the existing `fm_agent/` directory and restarts from scratch, so an interrupted run loses all prior progress. Pass `--resume` (or set the environment variable `FM_AGENT_RESUME=1`) to continue where the previous run left off. In resume mode FM-Agent keeps the existing `fm_agent/` directory and only does the remaining work. +Each new invocation writes to a timestamped directory such as `fm_agent/runs/20260717-143000/`; FM-Agent never deletes the entire `fm_agent/` root automatically. `fm_agent/current_run.json` identifies the active run. When results already exist, an interactive terminal offers four choices: resume the current run, archive it by leaving it under `runs/` and create a new run, overwrite only the current run, or exit without changes. Pass `--resume` (or set `FM_AGENT_RESUME=1`) to select the current run directly. In a non-interactive terminal, existing results are preserved and FM-Agent exits unless `--resume` is explicit. + +Older flat `fm_agent/` workspaces are migrated to `fm_agent/runs/legacy-/` when resumed or archived. ### Incremental Mode -In incremental mode, FM-Agent reuses the results of a previous run and only re-checks what changed. It diffs the current code against the commit recorded by the previous run in `fm_agent/version.log`. Each run records the processed commit id to that file, so a subsequent `--incremental` run automatically picks it up: +In incremental mode, FM-Agent reuses the selected run and only re-checks what changed. It diffs the current code against the commit recorded in `fm_agent/runs//version.log`. ```bash python3 main.py --incremental ``` -If `fm_agent/version.log` does not exist (no previous run to compare against), FM-Agent falls back to a full run. +If the selected run has no `version.log`, FM-Agent falls back to a full run. ### Live Dashboard -FM-Agent ships a standalone real-time TUI dashboard ([dashboard.py](dashboard.py)) that visualizes a run as it progresses: per-stage progress, token usage and cost, prompt-cache hit rate, and bug-validation verdicts. It reads the trace files FM-Agent writes under `fm_agent/`, so run it in a second terminal while `main.py` is going: +FM-Agent ships a standalone real-time TUI dashboard ([dashboard.py](dashboard.py)) that visualizes a run as it progresses: per-stage progress, token usage and cost, prompt-cache hit rate, and bug-validation verdicts. Given a project root, it follows `current_run.json`; you can also pass a specific run directory. ```bash uv run python dashboard.py @@ -237,15 +239,15 @@ uv run python dashboard.py | Argument | Description | | ----------- | ----------------------------------------------------------- | -| `proj_dir` | Same codebase directory passed to `main.py` (monitors `/fm_agent/`). You can also point it directly at any workspace directory containing a `trace/` subdir, e.g. an archived run | +| `proj_dir` | Same codebase directory passed to `main.py` (monitors the current run), or a specific `fm_agent/runs//` directory containing `trace/` | Press `Ctrl-C` to exit the dashboard; it does not affect the running pipeline. ### Output -FM-Agent creates an `fm_agent/` directory under your codebase directory. The key outputs are: +FM-Agent creates each run under `fm_agent/runs//`. The key outputs below are relative to that run directory: -#### Bug Reports (`fm_agent/bug_validation/.md`) +#### Bug Reports (`bug_validation/.md`) Each confirmed or investigated bug produces a Markdown report containing: @@ -259,11 +261,11 @@ Each confirmed or investigated bug produces a Markdown report containing: | Probe Script | The full test script used to confirm the bug | | Probe Output | Raw stdout from executing the probe script | -A `summary.json` file in `fm_agent/bug_validation/` aggregates all bug results with counts of total reported, confirmed, not confirmed bugs. +A `summary.json` file in `bug_validation/` aggregates all bug results with counts of total reported, confirmed, not confirmed bugs. ## Important Notes -1. FM-Agent will create an `fm_agent/` directory under your codebase directory. Make sure there is no name conflict. +1. FM-Agent keeps independent run directories under `fm_agent/runs/`; use `current_run.json` or the startup prompt to select the active run. 2. The markdown files under `md/` provide general instructions that guide the agent's reasoning process. Prefer `--domain-knowledge` for project-specific context such as invariants, protocols, encoding rules, and domain terminology. For reusable framework behavior, customize the built-in prompts; for example, if you are reasoning about a compiler, modify `md/bug_validator.md` to instruct the agent to compare outputs against a reference implementation (e.g., GCC). 3. **Supported languages**: Rust, C, C++, Python, Java, Go, CUDA, JavaScript, TypeScript, ArkTS, Erlang. Erlang function extraction and call graphs require ELP; if ELP is unavailable, Erlang files are skipped with a warning. diff --git a/README_zh.md b/README_zh.md index 9ba39a5b..90c2a912 100644 --- a/README_zh.md +++ b/README_zh.md @@ -183,7 +183,7 @@ uv run python main.py [--resume] [--domain-knowledge FILE ...] [--sub uv run python main.py --domain-knowledge docs/invariants.md docs/protocol.md ``` -FM-Agent 会将这些文件暂存到 `fm_agent/spec_prompts/domain_context/user_knowledge/`，并在本次运行中让相关 Agent 读取。也可以通过 `FM_AGENT_DOMAIN_KNOWLEDGE` 提供使用 `os.pathsep` 分隔的 Markdown 文件列表。 +FM-Agent 会将这些文件暂存到 `fm_agent/runs//spec_prompts/domain_context/user_knowledge/`，并在本次运行中让相关 Agent 读取。也可以通过 `FM_AGENT_DOMAIN_KNOWLEDGE` 提供使用 `os.pathsep` 分隔的 Markdown 文件列表。 使用 `--submodule` 可以把完整运行或增量运行限制到指定项目子目录： @@ -194,21 +194,23 @@ uv run python main.py --incremental intent.md --submodule src/core sr `--submodule` 路径必须是 `proj_dir` 内部目录。该参数可与 `--resume`、`--isolate` 和 `--incremental` 一起使用，但不能与 `--entry-func` 一起使用。 -默认情况下，每次运行都会清空已有的 `fm_agent/` 目录并从头开始，因此一旦运行中断，之前的所有进度都会丢失。可通过 `--resume` 参数（或设置环境变量 `FM_AGENT_RESUME=1`）从上一次中断处继续。在续跑模式下，FM-Agent 会保留已有的 `fm_agent/` 目录，只执行剩余的工作。 +每次新任务默认写入带时间戳的独立目录，例如 `fm_agent/runs/20260717-143000/`；FM-Agent 不会再自动删除整个 `fm_agent/` 根目录。`fm_agent/current_run.json` 指向当前 Run。检测到已有结果时，交互式终端会提供四个选项：续跑当前 Run、保留归档后创建新 Run、仅覆盖当前 Run、退出且不做修改。使用 `--resume`（或设置 `FM_AGENT_RESUME=1`）可直接续跑当前 Run。非交互终端检测到旧结果时会安全退出，除非显式传入 `--resume`。 + +旧版扁平的 `fm_agent/` 工作区在续跑或归档时会迁移至 `fm_agent/runs/legacy-/`。 ### 增量模式 -增量模式会复用上一次运行的结果，仅重新检测发生变化的部分。它将当前代码与上一次运行记录在 `fm_agent/version.log` 中的提交进行 diff。每次运行都会把所处理的提交 id 写入该文件，因此后续的 `--incremental` 运行会自动读取它： +增量模式会复用所选 Run 的结果，仅重新检测发生变化的部分。它将当前代码与 `fm_agent/runs//version.log` 中记录的提交进行 diff： ```bash python3 main.py --incremental ``` -如果 `fm_agent/version.log` 不存在（没有可供比较的历史运行），FM-Agent 会回退为完整运行。 +如果所选 Run 中不存在 `version.log`，FM-Agent 会回退为完整运行。 ### 实时监控面板 -FM-Agent 自带一个独立的实时 TUI 监控面板（[dashboard.py](dashboard.py)），用于在运行过程中可视化展示：各阶段进度、Token 用量与花费、prompt 缓存命中率，以及 Bug 验证结果。它读取 FM-Agent 写入 `fm_agent/` 目录下的 trace 文件，因此可在 `main.py` 运行期间于另一个终端中启动： +FM-Agent 自带一个独立的实时 TUI 监控面板（[dashboard.py](dashboard.py)），用于在运行过程中可视化展示：各阶段进度、Token 用量与花费、prompt 缓存命中率，以及 Bug 验证结果。传入项目根目录时，面板会根据 `current_run.json` 监控当前 Run；也可以直接传入指定 Run 目录。 ```bash uv run python dashboard.py @@ -216,15 +218,15 @@ uv run python dashboard.py | 参数 | 描述 | |---|---| -| `proj_dir` | 与 `main.py` 相同的代码库目录（监控 `/fm_agent/`）。也可直接指向任意包含 `trace/` 子目录的工作区目录，例如已归档的运行 | +| `proj_dir` | 与 `main.py` 相同的代码库目录（监控当前 Run），或包含 `trace/` 的指定 `fm_agent/runs//` 目录 | 按 `Ctrl-C` 退出监控面板，不会影响正在运行的流水线。 ### 输出说明 -FM-Agent 会在代码库目录下创建 `fm_agent/` 目录，主要输出内容如下： +FM-Agent 会把每次运行创建在 `fm_agent/runs//` 下。以下路径均相对于该 Run 目录： -#### Bug 报告（`fm_agent/bug_validation/.md`） +#### Bug 报告（`bug_validation/.md`） 每个已确认或经过排查的 Bug 都会生成一份 Markdown 报告，包含以下内容： @@ -238,15 +240,15 @@ FM-Agent 会在代码库目录下创建 `fm_agent/` 目录，主要输出内容 | Probe Script | 用于触发 Bug 的完整测试脚本 | | Probe Output | 执行测试脚本的输出 | -`fm_agent/bug_validation/` 目录下的 `summary.json` 文件汇总了所有 Bug 结果，包括报告的Bug总数、已确认Bug数、未确认Bug数。 +`bug_validation/` 目录下的 `summary.json` 文件汇总了所有 Bug 结果，包括报告的Bug总数、已确认Bug数、未确认Bug数。 -#### 日志文件（`fm_agent/fm_agent.log`） +#### 日志文件（`fm_agent.log`） 单一日志文件记录完整的流水线执行过程，包括文件提取进度、推理任务的提交与完成情况、网络错误与重试，以及最终的推理统计摘要。日志级别为 `INFO`，格式为 `%(asctime)s [%(levelname)s] %(message)s`。 ## 注意事项 -1. FM-Agent 会在代码库目录下创建 `fm_agent/` 目录，请确保不存在命名冲突。 +1. FM-Agent 将独立运行保存在 `fm_agent/runs/` 下，可通过 `current_run.json` 或启动时的交互选项选择当前 Run。 2. `md/` 目录下的 Markdown 文件提供了引导 Agent 推理过程的通用说明。针对项目特定的上下文（如不变量、协议、编码规则、领域术语），优先使用 `--domain-knowledge`。对于可复用的框架行为，可定制内置提示词；例如，若正在推理编译器的正确性，可修改 `md/bug_validator.md`，指示 Agent 将输出与参考实现（如 GCC）进行对比。 3. **支持的编程语言**：Rust、C、C++、Python、Java、Go、CUDA、JavaScript、TypeScript、ArkTS、Erlang。Erlang 的函数抽取与调用图需要 ELP；ELP 不可用时会给出警告并跳过 Erlang 文件。 diff --git a/dashboard.py b/dashboard.py index 188029ca..5e1c4b87 100644 --- a/dashboard.py +++ b/dashboard.py @@ -2,7 +2,7 @@ """Real-time TUI dashboard for an FM-Agent run. Usage: - uv run python dashboard.py # live: /fm_agent/ + uv run python dashboard.py # live: current run uv run python dashboard.py /fm_agent.archived_xx # any workspace dir (auto-detected by trace/ subdir) uv run python dashboard.py --refresh 1.0 # refresh every 1.0s @@ -178,14 +178,29 @@ def _locate_workdir(proj_dir): """Resolve which fm_agent workdir to monitor. Accepts either: - - A project root: dashboard looks for /fm_agent/ (the live workspace). + - A project root: dashboard follows /fm_agent/current_run.json. - A workspace directly (any name like fm_agent.opus_partial_*): detected by the presence of a `trace/` subdir, used as-is. """ p = Path(proj_dir).resolve() if (p / "trace").is_dir(): return p - return p / "fm_agent" + root = p / "fm_agent" + marker = root / "current_run.json" + try: + run_id = json.loads(marker.read_text()).get("run_id", "") + except (OSError, ValueError, AttributeError): + run_id = "" + if run_id and Path(run_id).name == run_id: + current = root / "runs" / run_id + if current.is_dir(): + return current + runs_dir = root / "runs" + if runs_dir.is_dir(): + runs = [candidate for candidate in runs_dir.iterdir() if candidate.is_dir()] + if runs: + return max(runs, key=lambda candidate: candidate.stat().st_mtime) + return root class State: @@ -740,7 +755,7 @@ def build_layout(state): def main(): ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("proj_dir", - help=("Either a target codebase (monitors /fm_agent/) " + help=("Either a target codebase (monitors its current fm_agent run) " "or a workspace directly (any dir containing a trace/ subdir)")) ap.add_argument("--refresh", type=float, default=1.5, help="Refresh seconds (default 1.5)") args = ap.parse_args() diff --git a/main.py b/main.py index cf31d154..fe67d450 100644 --- a/main.py +++ b/main.py @@ -40,6 +40,11 @@ list_staged_domain_knowledge_relpaths, stage_domain_knowledge_files, ) +from src.run_workspace import ( + RunSelectionCancelled, + select_run_workspace, + workdir_relpath, +) import os import sys import argparse @@ -52,12 +57,6 @@ import concurrent.futures -def _clean_previous_run(work_dir): - """Remove the fm_agent working directory from the previous pipeline run.""" - if os.path.isdir(work_dir): - shutil.rmtree(work_dir) - - def _get_pending_batches(batches, proj_dir): """Return batches that still have at least one function without specs.""" pending = [] @@ -125,12 +124,13 @@ def _run_spec_generation_batch( function_id_from_extracted_path(func_rel) for func_rel in function_files ] - fm_reminder = ("IMPORTANT: fm_agent/ is your output workspace, not project source. " + work_rel = workdir_relpath(proj_dir, work_dir) + fm_reminder = (f"IMPORTANT: {work_rel}/ is your output workspace, not project source. " "Do NOT modify any existing project files.") if attempt == 1: prompt = ( f"Process the batch prompt file at {batch_prompt_rel}. " - f"Read it and fm_agent/spec_prompts/system_prompt.md, " + f"Read it and {work_rel}/spec_prompts/system_prompt.md, " f"generate behavioral specs for each function listed, " f"and write the complete specced files directly. {fm_reminder}" ) @@ -140,9 +140,9 @@ def _run_spec_generation_batch( f"Some functions may already have specs from a previous attempt. " f"Check each function file — only generate specs for those " f"that don't have [SPEC] blocks yet. " - f"Read fm_agent/spec_prompts/system_prompt.md for the format rules. {fm_reminder}" + f"Read {work_rel}/spec_prompts/system_prompt.md for the format rules. {fm_reminder}" ) - prompt_file = os.path.join(proj_dir, "fm_agent", "workflow_spec_step4_batch.md") + prompt_file = os.path.join(work_dir, "workflow_spec_step4_batch.md") if is_cli_backend_enabled(): command = build_agent_command( model=OPENCODE_SPEC_MODEL, @@ -165,9 +165,9 @@ def _run_spec_generation_batch( stage="spec_generation", function_ids=function_ids, input_files=[ - "fm_agent/workflow_spec_step4_batch.md", + f"{work_rel}/workflow_spec_step4_batch.md", batch_prompt_rel, - "fm_agent/spec_prompts/system_prompt.md", + f"{work_rel}/spec_prompts/system_prompt.md", *list_staged_domain_knowledge_relpaths(work_dir), ], output_files=function_files, @@ -187,6 +187,7 @@ def _run_spec_generation_batch( def run_pipeline( proj_dir, resume=False, + work_dir=None, required_source_files=None, domain_knowledge_files=None, submodules=None, @@ -200,23 +201,19 @@ def run_pipeline( f"Supported extensions: {', '.join(sorted(EXT_TO_LANG.keys()))}") sys.exit(1) - work_dir = os.path.join(proj_dir, "fm_agent") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") input_dir = os.path.join(work_dir, "extracted_functions") output_dir = os.path.join(work_dir, "logic_verification_results") script_dir = os.path.dirname(os.path.abspath(__file__)) - # Clean files from the previous run — unless resuming, where we keep all - # prior progress (phases.json, generated specs, verification results) and - # only do the remaining work. if resume: if os.path.isdir(work_dir): print(f"[Pipeline] RESUME: keeping existing {os.path.relpath(work_dir, proj_dir)}/ — only remaining work will run.") else: - print("[Pipeline] RESUME requested but no previous fm_agent/ found — starting fresh.") + print("[Pipeline] RESUME requested but the selected run was not found — starting fresh.") resume = False - else: - _clean_previous_run(work_dir) os.makedirs(work_dir, exist_ok=True) + work_rel = workdir_relpath(proj_dir, work_dir) domain_knowledge_relpaths = stage_domain_knowledge_files( proj_dir, work_dir, domain_knowledge_files ) @@ -244,8 +241,8 @@ def run_pipeline( try_codegraph_init(proj_dir, force=not resume) # Run function extraction using extract.py - # force=False on resume preserves already-specced extracted files; on a fresh - # run fm_agent/ was just wiped so it is equivalent to force=True. + # force=False on resume preserves already-specced extracted files. A fresh + # run uses a newly created, empty run directory. print("[Pipeline] Extracting functions from source files...") run_extraction(proj_dir, work_dir=work_dir, force=not resume, verbose=True) @@ -290,7 +287,10 @@ def run_pipeline( print("[Pipeline] Stage 4/4: Generating specs & verification...") batch_md_src = os.path.join(script_dir, "md", "workflow_spec_step4_batch.md") batch_md_dst = os.path.join(work_dir, "workflow_spec_step4_batch.md") - shutil.copy2(batch_md_src, batch_md_dst) + with open(batch_md_src, "r") as f: + batch_md = f.read().replace("fm_agent/", f"{work_rel}/") + with open(batch_md_dst, "w") as f: + f.write(batch_md) all_processed = set() num_phases = len(phases_data["phases"]) @@ -325,7 +325,7 @@ def run_pipeline( # Generate batch prompts for this layer. On resume, skip functions # that were already specced in a previous run. - batch_cmd = ["python3", "fm_agent/spec_prompts/generate_batch_prompts.py", + batch_cmd = ["python3", f"{work_rel}/spec_prompts/generate_batch_prompts.py", "--phase", str(phase_num), "--layers", str(layer_idx)] if resume: batch_cmd.append("--resume") @@ -455,7 +455,7 @@ def run_pipeline( f"[Pipeline] ERROR: Stage 4 Phase {phase_num} Layer {layer_idx} failed " f"after {OPENCODE_MAX_RETRIES} attempts. " f"No specs were generated. " - f"Check {os.path.basename(proj_dir)}/fm_agent/trace/ for details." + f"Check {work_rel}/trace/ for details." ) sys.exit(1) @@ -486,7 +486,7 @@ def run_pipeline( parser.add_argument( "--resume", action="store_true", - help="continue a previous run in /fm_agent instead of wiping it: " + help="continue the run selected by /fm_agent/current_run.json: " "keeps phases.json, generated specs, and existing verification results; " "only does the remaining work.", ) @@ -510,7 +510,7 @@ def run_pipeline( nargs="+", default=[], help="additional Markdown domain-knowledge file(s) to copy into " - "fm_agent/spec_prompts/domain_context/user_knowledge/ and provide to " + "the selected run's spec_prompts/domain_context/user_knowledge/ and provide to " "setup, spec generation, and validation agents. May be repeated. " "FM_AGENT_DOMAIN_KNOWLEDGE can also provide os.pathsep-separated files.", ) @@ -556,12 +556,27 @@ def run_pipeline( if submodules and args.entry_func is not None: parser.error("--submodule cannot be combined with --entry-func.") + if not os.path.isdir(proj_dir): + parser.error(f"project directory does not exist: {proj_dir}") + # ---- pre-flight environment check (shared by all pipeline modes) ---- import config from src.env_check import run as env_check_run if not env_check_run(proj_dir, config): sys.exit(0) + try: + run_selection = select_run_workspace( + proj_dir, + resume_requested=resume, + ) + except RunSelectionCancelled as exc: + print(f"[Pipeline] {exc}") + sys.exit(0) + resume = run_selection.resume + selected_work_dir = run_selection.work_dir + selected_work_rel = os.path.relpath(selected_work_dir, proj_dir) + start_time = time.time() # Entry-point mode: reason only about the call graph reachable from a specific @@ -573,6 +588,7 @@ def run_pipeline( entry_func=args.entry_func, end_funcs=args.end_func, resume=resume, + work_dir=selected_work_dir, domain_knowledge_files=domain_knowledge_files, ) end_time = time.time() @@ -597,7 +613,7 @@ def run_pipeline( # the real project before snapshotting. old_commit = None if args.incremental: - version_path = os.path.join(proj_dir, "fm_agent", "version.log") + version_path = os.path.join(selected_work_dir, "version.log") if os.path.exists(version_path): with open(version_path, "r") as f: commits = [line.strip() for line in f if line.strip()] @@ -620,6 +636,7 @@ def run_pipeline( else contextlib.nullcontext(proj_dir) ) with run_ctx as run_dir: + run_work_dir = os.path.join(run_dir, selected_work_rel) try: # Incremental mode requires a recorded commit to diff against; without a # version.log from a previous run, fall back to the full pipeline. @@ -628,6 +645,7 @@ def run_pipeline( run_dir, intent_path, old_commit, + work_dir=run_work_dir, domain_knowledge_files=domain_knowledge_files, submodules=submodules, ) @@ -635,6 +653,7 @@ def run_pipeline( run_pipeline( run_dir, resume=resume, + work_dir=run_work_dir, domain_knowledge_files=domain_knowledge_files, submodules=submodules, ) @@ -642,7 +661,7 @@ def run_pipeline( # it recreates fm_agent/; with --isolate it lives in the snapshot and is # copied back to the real project below. Only recorded on success so a # partial run does not advance the version baseline. - _record_version(new_commit, os.path.join(run_dir, "fm_agent")) + _record_version(new_commit, run_work_dir) finally: # With --isolate the pipeline ran against a throwaway snapshot, so its # fm_agent/ results live in the snapshot. Copy them back into the real @@ -650,12 +669,11 @@ def run_pipeline( # even when the pipeline crashes or is interrupted mid-run, so partial # progress survives and can be resumed with --resume. if args.isolate: - src_fm = os.path.join(run_dir, "fm_agent") - dst_fm = os.path.join(proj_dir, "fm_agent") - if os.path.isdir(src_fm): - if os.path.isdir(dst_fm): - shutil.rmtree(dst_fm) - shutil.copytree(src_fm, dst_fm, symlinks=True) - print(f"[Pipeline] Copied results back to {dst_fm}") + if os.path.isdir(run_work_dir): + if os.path.isdir(selected_work_dir): + shutil.rmtree(selected_work_dir) + os.makedirs(os.path.dirname(selected_work_dir), exist_ok=True) + shutil.copytree(run_work_dir, selected_work_dir, symlinks=True) + print(f"[Pipeline] Copied results back to {selected_work_dir}") end_time = time.time() logging.info(f"Total time: {end_time - start_time:.2f} seconds") diff --git a/src/domain_knowledge.py b/src/domain_knowledge.py index c6f0071c..7f7f3476 100644 --- a/src/domain_knowledge.py +++ b/src/domain_knowledge.py @@ -5,6 +5,8 @@ import re import shutil +from .run_workspace import inferred_workdir_relpath + VALID_DOMAIN_KNOWLEDGE_EXTENSIONS = {".md", ".markdown"} USER_KNOWLEDGE_REL_DIR = os.path.join( @@ -104,8 +106,10 @@ def _safe_staged_name(source_path, used_names): index += 1 -def list_staged_domain_knowledge_relpaths(work_dir, prefix="fm_agent"): +def list_staged_domain_knowledge_relpaths(work_dir, prefix=None): """Return project-relative staged markdown paths, sorted for stable prompts.""" + if prefix is None: + prefix = inferred_workdir_relpath(work_dir) knowledge_dir = os.path.join(work_dir, USER_KNOWLEDGE_REL_DIR) if not os.path.isdir(knowledge_dir): return [] @@ -156,7 +160,7 @@ def stage_domain_knowledge_files(proj_dir, work_dir, markdown_paths=None): ) entries.append({ "source_path": source_path, - "staged_path": f"fm_agent/{rel_to_work}", + "staged_path": f"{inferred_workdir_relpath(work_dir)}/{rel_to_work}", }) manifest_path = os.path.join(tmp_dir, USER_KNOWLEDGE_MANIFEST) diff --git a/src/entry_reasoning_pipeline.py b/src/entry_reasoning_pipeline.py index 0856d272..1c824efa 100644 --- a/src/entry_reasoning_pipeline.py +++ b/src/entry_reasoning_pipeline.py @@ -218,6 +218,7 @@ def run_entry_pipeline( entry_func=None, end_funcs=None, resume=False, + work_dir=None, domain_knowledge_files=None, ): """Run the entry-point-scoped reasoning pipeline. @@ -256,7 +257,7 @@ def run_entry_pipeline( raise ValueError("entry_func is required to run the entry pipeline") proj_dir = os.path.abspath(proj_dir) - work_dir = os.path.join(proj_dir, "fm_agent") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") config.BUG_VALIDATION_MAX_RETRIES = 0 # The entry_func's source file may match the test-file heuristics (a test @@ -419,7 +420,8 @@ def _run_entry_pipeline_inner( # 2. Copy the sources into a separate run directory, then trim that copy. # proj_dir is left untouched throughout. run_dir = proj_dir + ".fm-entry-run" - run_work_dir = os.path.join(run_dir, "fm_agent") + work_rel = os.path.relpath(work_dir, proj_dir) + run_work_dir = os.path.join(run_dir, work_rel) # _make_run_copy brings along an existing fm_agent/, so a resumed run finds # the prior state in run_dir without any extra seeding here. _make_run_copy(proj_dir, run_dir) @@ -447,6 +449,7 @@ def _run_entry_pipeline_inner( run_pipeline( run_dir, resume=resume, + work_dir=run_work_dir, required_source_files=[_entry_func_source_rel(entry_func)], domain_knowledge_files=domain_knowledge_files, ) @@ -456,8 +459,9 @@ def _run_entry_pipeline_inner( if os.path.isdir(run_work_dir): if os.path.isdir(work_dir): shutil.rmtree(work_dir) + os.makedirs(os.path.dirname(work_dir), exist_ok=True) shutil.copytree(run_work_dir, work_dir, symlinks=True) - print(f"[EntryPipeline] Copied generated fm_agent/ to {work_dir}.") + print(f"[EntryPipeline] Copied generated run to {work_dir}.") shutil.rmtree(run_dir, ignore_errors=True) # Report the bug count: the number of MISMATCH verdicts the reasoner wrote diff --git a/src/incremental_reasoner.py b/src/incremental_reasoner.py index 67b30601..29846d56 100644 --- a/src/incremental_reasoner.py +++ b/src/incremental_reasoner.py @@ -58,6 +58,7 @@ load_staged_domain_knowledge_text, stage_domain_knowledge_files, ) +from .run_workspace import inferred_workdir_relpath class _StdoutTee: @@ -154,7 +155,7 @@ def _setup_incremental_logging(work_dir): return log_path -def check_last_run_existence(proj_dir, submodules=None): +def check_last_run_existence(proj_dir, submodules=None, work_dir=None): """ Return whether a full pipeline run (run_pipeline) has already completed under proj_dir. @@ -174,7 +175,7 @@ def check_last_run_existence(proj_dir, submodules=None): selected scope has at least one ready function and no selected function is incomplete; otherwise False (so the caller can fall back to a scoped full run). """ - work_dir = os.path.join(proj_dir, "fm_agent") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") if not os.path.isfile(os.path.join(work_dir, "phases.json")): return False @@ -196,7 +197,7 @@ def check_last_run_existence(proj_dir, submodules=None): return saw_function -def extract_existing_specs(proj_dir): +def extract_existing_specs(proj_dir, work_dir=None): """ Collect the leading [SPEC]/[INFO] blocks from every specced file produced by a previous full run (the extracted_functions tree that _run_setup_extract + @@ -217,7 +218,8 @@ def extract_existing_specs(proj_dir): block. Files without a [SPEC] block are skipped. Returns an empty dict when the extracted_functions directory does not exist. """ - extracted_dir = os.path.join(proj_dir, "fm_agent", "extracted_functions") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") + extracted_dir = os.path.join(work_dir, "extracted_functions") if not os.path.isdir(extracted_dir): return {} @@ -248,7 +250,7 @@ def extract_existing_specs(proj_dir): return specs -def _reapply_existing_specs(proj_dir, specs): +def _reapply_existing_specs(proj_dir, specs, work_dir=None): """ Prepend previously captured [SPEC]/[INFO] header blocks back onto the freshly re-extracted function files. @@ -265,7 +267,8 @@ def _reapply_existing_specs(proj_dir, specs): Returns the number of files to which a spec block was (re)applied. """ - extracted_dir = os.path.join(proj_dir, "fm_agent", "extracted_functions") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") + extracted_dir = os.path.join(work_dir, "extracted_functions") for rel_path, entry in specs.items(): spec_block = entry.get("spec") if not spec_block: @@ -402,7 +405,8 @@ def _funcs_from_commit(rel_path, lang_key, ext): def _modified_function_targets( - proj_dir, modified_functions, classes=("added", "removed", "modified") + proj_dir, modified_functions, classes=("added", "removed", "modified"), + work_dir=None, ): """ Map the functions recorded in modified_functions to (FQN, extracted-file path). @@ -418,7 +422,8 @@ def _modified_function_targets( Returns a dict mapping FQN -> absolute extracted-file path. """ - extracted_base = os.path.join(proj_dir, "fm_agent", "extracted_functions") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") + extracted_base = os.path.join(work_dir, "extracted_functions") targets = {} for abs_src, changes in modified_functions.items(): rel = os.path.relpath(abs_src, proj_dir) @@ -438,12 +443,12 @@ def _modified_function_targets( for name in names: fname = f"{name}.{ext}" if ext else name path = os.path.join(func_dir, fname) - fqn = _file_to_fqn(path, os.path.join(proj_dir, "fm_agent")) + fqn = _file_to_fqn(path, work_dir) targets[fqn] = path return targets -def _remove_stale_extracted(proj_dir, modified_functions): +def _remove_stale_extracted(proj_dir, modified_functions, work_dir=None): """ Delete extracted-function files for functions reported as removed (including every function of a deleted source file), and prune any function directory left empty as @@ -451,7 +456,7 @@ def _remove_stale_extracted(proj_dir, modified_functions): stale specs under fm_agent/extracted_functions/. """ removed = _modified_function_targets( - proj_dir, modified_functions, classes=("removed",) + proj_dir, modified_functions, classes=("removed",), work_dir=work_dir ) for path in removed.values(): if os.path.isfile(path): @@ -576,6 +581,7 @@ def run_incremental_pipeline( proj_dir, intent_file_path, old_commit_id, + work_dir=None, domain_knowledge_files=None, submodules=None, ): @@ -592,7 +598,7 @@ def run_incremental_pipeline( # import them lazily here to avoid a src -> main import cycle at module load time. from main import run_pipeline, _run_setup_extract - work_dir = os.path.join(proj_dir, "fm_agent") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) input_dir = os.path.join(work_dir, "extracted_functions") output_dir = os.path.join(work_dir, "logic_verification_results") @@ -618,13 +624,16 @@ def run_incremental_pipeline( # 1. Check whether there is a last run to compare against; if not, fall back to a full run since we have no basis for incremental analysis. logging.info("[Stage 1/10] Checking for a previous full run to compare against...") - has_last_run = check_last_run_existence(proj_dir, submodules=submodules) + has_last_run = check_last_run_existence( + proj_dir, submodules=submodules, work_dir=work_dir + ) if not has_last_run: logging.warning( "No previous full run detected (phases.json missing or incomplete extracted_functions), so falling back to a full run rather than incremental." ) run_pipeline( proj_dir, + work_dir=work_dir, domain_knowledge_files=domain_knowledge_files, submodules=submodules, ) @@ -691,7 +700,7 @@ def run_incremental_pipeline( # added or whose extraction path changed are left unspecced for the spec-update # stage to handle; unchanged functions keep their previous specs verbatim. logging.info("[Stage 4/10] Re-extracting functions and restoring previous specs...") - old_spec = extract_existing_specs(proj_dir) + old_spec = extract_existing_specs(proj_dir, work_dir=work_dir) logging.info(" -> captured %d existing spec block(s) before re-extraction.", len(old_spec)) # Rebuild the codegraph index before re-extraction. The index still reflects the code as # of the previous full run, but the working tree has changed since then; run_extraction @@ -700,7 +709,7 @@ def run_incremental_pipeline( # default; no-op when codegraph is uninstalled (extraction then falls back to regex). try_codegraph_init(proj_dir) run_extraction(proj_dir, work_dir=work_dir, force=True, verbose=True) - _reapply_existing_specs(proj_dir, old_spec) + _reapply_existing_specs(proj_dir, old_spec, work_dir=work_dir) logging.info(" -> functions re-extracted and prior [SPEC]/[INFO] headers reapplied.") # 5. Collect changed functions by comparing against the old version of functions in commit_id @@ -719,7 +728,7 @@ def run_incremental_pipeline( # 5b. Delete extracted-function files for functions (or whole source files) that were # removed since old_commit_id. Re-extraction never rewrites these, so without this # they linger as stale specs and would pollute the file list and call graph below. - _remove_stale_extracted(proj_dir, changed_functions) + _remove_stale_extracted(proj_dir, changed_functions, work_dir=work_dir) logging.info(" -> stale extracted-function files for removed functions deleted.") # 6. Update file list @@ -743,7 +752,9 @@ def run_incremental_pipeline( # 8. Collect the scope of functions relevant to the developer intent (the intent file defines the goal of modification). logging.info("[Stage 8/10] Collecting functions relevant to the developer intent...") - spec_files = collect_relevent_function_scope(proj_dir, developer_intent, changed_functions) + spec_files = collect_relevent_function_scope( + proj_dir, developer_intent, changed_functions, work_dir=work_dir + ) logging.info(" -> %d function(s) judged relevant to the intent.", len(spec_files)) # 9. Re-generate the spec of functions if it satisfies one of the following conditions: 1) the function is changed; 2) the function is relevant to the developer intent. @@ -921,7 +932,9 @@ def _domain_knowledge_prompt_section(work_dir): return f"## User-provided domain knowledge\n\n{text}\n\n" if text else "" -def collect_relevent_function_scope(proj_dir, developer_intent, changed_functions, range=None): +def collect_relevent_function_scope( + proj_dir, developer_intent, changed_functions, range=None, work_dir=None +): """ Select the functions relevant to developer_intent and return the most relevant ones. @@ -946,7 +959,8 @@ def collect_relevent_function_scope(proj_dir, developer_intent, changed_function relevance score and truncated to the first `range` entries. Returns an empty list when phases.json has no modules or opencode selects none / fails to produce a result. """ - work_dir = os.path.join(proj_dir, "fm_agent") + work_dir = work_dir or os.path.join(proj_dir, "fm_agent") + work_rel = inferred_workdir_relpath(work_dir) extracted_dir = os.path.join(work_dir, "extracted_functions") phases_data = _load_phases(work_dir) @@ -1049,7 +1063,7 @@ def collect_relevent_function_scope(proj_dir, developer_intent, changed_function "1. Read each of the module source files listed below.\n" "2. Decide which files are relevant to the developer intent -- a file is relevant " "if the developer intent is likely to affect it or depend on its behavior.\n" - f"3. Write your answer to `fm_agent/relevant_files_{idx}.json` as a JSON array of " + f"3. Write your answer to `{work_rel}/relevant_files_{idx}.json` as a JSON array of " "the relevant file paths, each copied verbatim from the list below. Write `[]` if " "no file is relevant. Write ONLY that file; do not modify any other project " "files.\n\n" @@ -1061,11 +1075,11 @@ def collect_relevent_function_scope(proj_dir, developer_intent, changed_function file_selection = _opencode_select_json( proj_dir, work_dir, - os.path.join("fm_agent", f"select_relevant_files_{idx}.md"), + os.path.join(work_rel, f"select_relevant_files_{idx}.md"), file_prompt, - os.path.join("fm_agent", f"relevant_files_{idx}.json"), + os.path.join(work_rel, f"relevant_files_{idx}.json"), stage="select_relevant_files", - input_files=[f"fm_agent/select_relevant_files_{idx}.md", *source_files], + input_files=[f"{work_rel}/select_relevant_files_{idx}.md", *source_files], ) if isinstance(file_selection, list): @@ -1381,8 +1395,9 @@ def _opencode_generate_spec(proj_dir, work_dir, idx, fqn, lang_key, comment_pref produced), "new_spec" (str), "info_updated" (bool), "new_info" (str), "updated_callees" (list[str]) — or None when opencode produced nothing usable. """ - result_relpath = os.path.join("fm_agent", f"spec_generate_{idx}.json") - prompt_relpath = os.path.join("fm_agent", f"spec_generate_{idx}.md") + work_rel = inferred_workdir_relpath(work_dir) + result_relpath = os.path.join(work_rel, f"spec_generate_{idx}.json") + prompt_relpath = os.path.join(work_rel, f"spec_generate_{idx}.md") # Caller context (callers' own specs + what each caller's [INFO] expects from this # function), mirroring run_pipeline's "EARLIER-LAYER CALLER SPECS" / "CALLEE EXPECTATIONS @@ -1443,7 +1458,7 @@ def _opencode_generate_spec(proj_dir, work_dir, idx, fqn, lang_key, comment_pref f"```{lang_key}\n{source.strip()}\n```\n\n" f"{caller_section}" "## Steps\n\n" - "1. Read `fm_agent/spec_prompts/system_prompt.md` for the exact [SPEC]/[INFO] format " + f"1. Read `{work_rel}/spec_prompts/system_prompt.md` for the exact [SPEC]/[INFO] format " "rules used by this project.\n" f"{user_knowledge_step}" f"{2 + step_offset}. Produce the COMPLETE [SPEC] block describing this function's behavior — the " @@ -1468,7 +1483,7 @@ def _opencode_generate_spec(proj_dir, work_dir, idx, fqn, lang_key, comment_pref stage="generate_function_spec", input_files=[ prompt_relpath, - "fm_agent/spec_prompts/system_prompt.md", + f"{work_rel}/spec_prompts/system_prompt.md", *user_knowledge_paths, ], ) @@ -1505,7 +1520,8 @@ def _update_specs_for_intent(proj_dir, work_dir, developer_intent, changed_funct # exist on disk) plus functions relevant to the developer intent. seed = set() changed_targets = _modified_function_targets( - proj_dir, changed_functions, classes=("added", "modified") + proj_dir, changed_functions, classes=("added", "modified"), + work_dir=work_dir, ) seed.update(changed_targets.keys()) for rel in relevant_rel_files: @@ -1799,7 +1815,8 @@ def _verify_incremental_functions( # (1) Functions changed in the working tree (added/modified; removed ones are gone). verify_targets.update( _modified_function_targets( - proj_dir, changed_functions, classes=("added", "modified") + proj_dir, changed_functions, classes=("added", "modified"), + work_dir=work_dir, ).values() ) diff --git a/src/opencode_trace.py b/src/opencode_trace.py index 7b80498a..807a5e45 100644 --- a/src/opencode_trace.py +++ b/src/opencode_trace.py @@ -15,18 +15,19 @@ def function_id_from_extracted_path(path): rel = path.replace("\\", "/") - for prefix in ("fm_agent/extracted_functions/", "extracted_functions/"): - if rel.startswith(prefix): - rel = rel[len(prefix):] - break + marker = "/extracted_functions/" + if marker in f"/{rel}": + rel = f"/{rel}".split(marker, 1)[1] + elif rel.startswith("extracted_functions/"): + rel = rel[len("extracted_functions/"):] return os.path.splitext(rel)[0].replace("/", "::") def function_id_from_result_path(path): rel = path.replace("\\", "/") - prefix = "fm_agent/logic_verification_results/" - if rel.startswith(prefix): - rel = rel[len(prefix):] + marker = "/logic_verification_results/" + if marker in f"/{rel}": + rel = f"/{rel}".split(marker, 1)[1] return os.path.splitext(rel)[0].replace("/", "::") @@ -72,7 +73,7 @@ def _opencode_trace_path(work_dir, event_id): return os.path.join(_trace_dir(work_dir), "opencode", f"{event_id}.jsonl") -def _opencode_env(work_dir, event_id): +def _opencode_env(proj_dir, work_dir, event_id): env = os.environ.copy() trace_dir = os.path.abspath(os.path.join(_trace_dir(work_dir), "opencode")) os.makedirs(trace_dir, exist_ok=True) @@ -83,8 +84,7 @@ def _opencode_env(work_dir, event_id): # fm-agent repo's own AGENTS.md instead of the target's, baking ~10K bytes # of repo docs into every system prompt and invalidating the cache prefix # on every edit. - proj_dir = os.path.dirname(os.path.abspath(work_dir)) - env["PWD"] = proj_dir + env["PWD"] = os.path.abspath(proj_dir) return env @@ -118,7 +118,7 @@ def _start_opencode_process(proj_dir, work_dir, event_id, command, trace_log_pat proc = subprocess.Popen( command_argv(command), cwd=proj_dir, - env=_opencode_env(work_dir, event_id), + env=_opencode_env(proj_dir, work_dir, event_id), stdin=subprocess.PIPE if stdin_text is not None else None, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, diff --git a/src/pipeline_setup.py b/src/pipeline_setup.py index 0d359cf7..8d5f26a7 100644 --- a/src/pipeline_setup.py +++ b/src/pipeline_setup.py @@ -31,6 +31,7 @@ format_domain_knowledge_bullets, list_staged_domain_knowledge_relpaths, ) +from .run_workspace import workdir_relpath def _merge_descriptions(target_desc, source_desc): @@ -197,8 +198,11 @@ def _sync_domain_context(proj_dir, work_dir, changed_phases, phase_cleanup=None) ) return - prompt = _build_domain_context_regen_prompt(regenerate, phase_cleanup) - fm_reminder = ("IMPORTANT: fm_agent/ is your output workspace, not project source. " + work_rel = workdir_relpath(proj_dir, work_dir) + prompt = _build_domain_context_regen_prompt(regenerate, phase_cleanup).replace( + "fm_agent/", f"{work_rel}/" + ) + fm_reminder = (f"IMPORTANT: {work_rel}/ is your output workspace, not project source. " "Do NOT modify any existing project files.") prompt = f"{prompt}\n\n{fm_reminder}" @@ -220,11 +224,11 @@ def _sync_domain_context(proj_dir, work_dir, changed_phases, phase_cleanup=None) command=command, stage="sync_domain_context", input_files=[ - "fm_agent/phases.json", + f"{work_rel}/phases.json", *list_staged_domain_knowledge_relpaths(work_dir), ], output_files=[ - "fm_agent/spec_prompts/domain_context/engine_overview.txt", + f"{work_rel}/spec_prompts/domain_context/engine_overview.txt", ], summary=f"Regenerate domain_context for changed phases (attempt {attempt})", metadata={"attempt": attempt}, @@ -535,13 +539,15 @@ def _update_module_description(proj_dir, work_dir, modified_modules): logging.info("No phases.json to update module descriptions in; skipping.") return + work_rel = workdir_relpath(proj_dir, work_dir) prompt = _build_module_description_prompt(modified_modules, phases_path) if not prompt: logging.info( "No changed module still owns source files; skipping description update." ) return - fm_reminder = ("IMPORTANT: fm_agent/ is your output workspace, not project source. " + prompt = prompt.replace("fm_agent/", f"{work_rel}/") + fm_reminder = (f"IMPORTANT: {work_rel}/ is your output workspace, not project source. " "Do NOT modify any existing project files.") prompt = f"{prompt}\n\n{fm_reminder}" @@ -562,8 +568,8 @@ def _update_module_description(proj_dir, work_dir, modified_modules): work_dir=work_dir, command=command, stage="update_module_description", - input_files=["fm_agent/phases.json"], - output_files=["fm_agent/phases.json"], + input_files=[f"{work_rel}/phases.json"], + output_files=[f"{work_rel}/phases.json"], summary=f"Update module descriptions after phase dedup (attempt {attempt})", metadata={"attempt": attempt}, ) @@ -784,6 +790,8 @@ def _prepare_setup_workflow_file(proj_dir, work_dir, script_dir): proj_dir_name = os.path.basename(proj_dir_abs) with open(workflow_dst, "r") as _f: md = _f.read() + work_rel = workdir_relpath(proj_dir, work_dir) + md = md.replace("fm_agent/", f"{work_rel}/") old = ("- `phases[*].modules[*].source_files` — relative paths from repo root of all source files " "that belong to this module.") new = (f"- `phases[*].modules[*].source_files` — relative paths from the project root " @@ -827,13 +835,14 @@ def _run_setup_extract(proj_dir, work_dir, script_dir, is_incremental=False, _prepare_setup_workflow_file(proj_dir, work_dir, script_dir) - fm_reminder = ("IMPORTANT: The fm_agent/ directory is NOT part of the project source code. " + work_rel = workdir_relpath(proj_dir, work_dir) + fm_reminder = (f"IMPORTANT: The {work_rel}/ directory is NOT part of the project source code. " "It is a workspace for storing your output files only. " "Do NOT include fm_agent/ paths in phases.json. " "Do NOT modify any existing project files.") - incremental_reminder = ("IMPORTANT: An existing fm_agent/phases.json from a previous run is already " + incremental_reminder = (f"IMPORTANT: An existing {work_rel}/phases.json from a previous run is already " "present. Do NOT regenerate it from scratch. Instead, inspect the current " - "state of the source code and UPDATE the existing fm_agent/phases.json so it " + f"state of the source code and UPDATE the existing {work_rel}/phases.json so it " "reflects the current version of the code: add modules and source files that " "are new, remove entries whose files no longer exist, and adjust phases as " "needed. Preserve entries that are still accurate.") @@ -862,13 +871,13 @@ def _run_setup_extract(proj_dir, work_dir, script_dir, is_incremental=False, # regenerating everything and overwriting valid work. prompt = ("A previous setup attempt was interrupted and may have already produced some of the " "required output files. Follow the instructions in the attached file, but FIRST " - "check the current progress in fm_agent/ (e.g. phases.json and the " + f"check the current progress in {work_rel}/ (e.g. phases.json and the " "spec_prompts/domain_context/ files). Keep any existing valid output as-is and only " "generate the files that are missing or incomplete — do NOT regenerate or overwrite " f"work that is already done. {fm_reminder} {submodule_reminder}") if is_incremental: prompt = f"{prompt} {incremental_reminder}" - prompt_file = os.path.join(proj_dir, "fm_agent", "workflow_setup_extract.md") + prompt_file = os.path.join(work_dir, "workflow_setup_extract.md") if is_cli_backend_enabled(): command = build_agent_command( model=OPENCODE_SETUP_MODEL, @@ -886,12 +895,12 @@ def _run_setup_extract(proj_dir, work_dir, script_dir, is_incremental=False, command=command, stage="setup_context", input_files=[ - "fm_agent/workflow_setup_extract.md", + f"{work_rel}/workflow_setup_extract.md", *list_staged_domain_knowledge_relpaths(work_dir), ], output_files=[ - "fm_agent/phases.json", - "fm_agent/spec_prompts/domain_context/engine_overview.txt", + f"{work_rel}/phases.json", + f"{work_rel}/spec_prompts/domain_context/engine_overview.txt", ], summary=f"OpenCode setup context attempt {attempt}", metadata={"attempt": attempt}, @@ -937,7 +946,7 @@ def _run_setup_extract(proj_dir, work_dir, script_dir, is_incremental=False, print( f"[Pipeline] ERROR: Stage 1 failed after {OPENCODE_MAX_RETRIES} attempts. " f"{missing}. " - f"Check {os.path.basename(proj_dir)}/fm_agent/trace/ for details." + f"Check {work_rel}/trace/ for details." ) sys.exit(1) diff --git a/src/run_workspace.py b/src/run_workspace.py new file mode 100644 index 00000000..948e02f1 --- /dev/null +++ b/src/run_workspace.py @@ -0,0 +1,240 @@ +"""Run workspace selection and lifecycle management. + +FM-Agent stores each execution below ``fm_agent/runs/``. The small +``current_run.json`` marker points commands such as ``--resume`` and the +dashboard at the active run without making ``fm_agent/`` itself disposable. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +import json +import os +import shutil +import sys +import tempfile + + +WORKSPACE_DIRNAME = "fm_agent" +RUNS_DIRNAME = "runs" +CURRENT_RUN_FILENAME = "current_run.json" +_ROOT_METADATA = {RUNS_DIRNAME, CURRENT_RUN_FILENAME, ".env_check_memory"} + + +class RunSelectionCancelled(RuntimeError): + """Raised when the user exits instead of selecting a run workspace.""" + + +@dataclass(frozen=True) +class RunSelection: + work_dir: str + run_id: str + resume: bool + action: str + + +def workspace_root(proj_dir): + return os.path.join(os.path.abspath(proj_dir), WORKSPACE_DIRNAME) + + +def workdir_relpath(proj_dir, work_dir): + """Return an agent-facing, slash-separated path for ``work_dir``.""" + return os.path.relpath(work_dir, proj_dir).replace(os.sep, "/") + + +def inferred_workdir_relpath(work_dir): + """Infer the project-relative workspace path from an absolute work dir.""" + parts = os.path.normpath(os.path.abspath(work_dir)).split(os.sep) + try: + index = len(parts) - 1 - parts[::-1].index(WORKSPACE_DIRNAME) + except ValueError: + return WORKSPACE_DIRNAME + return "/".join(parts[index:]) + + +def _timestamp(now=None): + return (now or datetime.now()).strftime("%Y%m%d-%H%M%S") + + +def _unique_run_id(runs_dir, prefix=None, now=None): + base = prefix or _timestamp(now) + candidate = base + suffix = 2 + while os.path.exists(os.path.join(runs_dir, candidate)): + candidate = f"{base}-{suffix}" + suffix += 1 + return candidate + + +def _write_current(root, run_id): + os.makedirs(root, exist_ok=True) + marker = os.path.join(root, CURRENT_RUN_FILENAME) + fd, tmp = tempfile.mkstemp(prefix=".current_run.", dir=root, text=True) + try: + with os.fdopen(fd, "w") as f: + json.dump({"run_id": run_id}, f, indent=2) + f.write("\n") + os.replace(tmp, marker) + finally: + if os.path.exists(tmp): + os.unlink(tmp) + + +def current_run_dir(proj_dir): + """Return the active run directory, falling back to the newest run.""" + root = workspace_root(proj_dir) + runs_dir = os.path.join(root, RUNS_DIRNAME) + marker = os.path.join(root, CURRENT_RUN_FILENAME) + try: + with open(marker, "r") as f: + run_id = json.load(f).get("run_id", "") + except (OSError, ValueError, AttributeError): + run_id = "" + if run_id and os.path.basename(run_id) == run_id: + candidate = os.path.join(runs_dir, run_id) + if os.path.isdir(candidate): + return candidate + if not os.path.isdir(runs_dir): + return None + candidates = [ + os.path.join(runs_dir, name) + for name in os.listdir(runs_dir) + if os.path.isdir(os.path.join(runs_dir, name)) + ] + return max(candidates, key=os.path.getmtime) if candidates else None + + +def _legacy_entries(root): + if not os.path.isdir(root): + return [] + return [ + os.path.join(root, name) + for name in os.listdir(root) + if name not in _ROOT_METADATA + ] + + +def _move_legacy_run(root, now=None): + entries = _legacy_entries(root) + if not entries: + return None + runs_dir = os.path.join(root, RUNS_DIRNAME) + os.makedirs(runs_dir, exist_ok=True) + run_id = _unique_run_id(runs_dir, prefix=f"legacy-{_timestamp(now)}") + target = os.path.join(runs_dir, run_id) + os.makedirs(target) + for entry in entries: + shutil.move(entry, os.path.join(target, os.path.basename(entry))) + return target + + +def _new_run(root, now=None): + runs_dir = os.path.join(root, RUNS_DIRNAME) + os.makedirs(runs_dir, exist_ok=True) + run_id = _unique_run_id(runs_dir, now=now) + work_dir = os.path.join(runs_dir, run_id) + os.makedirs(work_dir) + _write_current(root, run_id) + return work_dir, run_id + + +def _activate(root, work_dir): + run_id = os.path.basename(work_dir) + _write_current(root, run_id) + return run_id + + +def _prompt_action(input_fn, output_fn): + output_fn("[Pipeline] Existing FM-Agent results were found. Choose an action:") + output_fn(" 1. Resume the current run") + output_fn(" 2. Archive it and start a new run") + output_fn(" 3. Overwrite it and start a new run") + output_fn(" 4. Exit") + aliases = { + "1": "resume", "r": "resume", "resume": "resume", + "2": "archive", "a": "archive", "archive": "archive", + "3": "overwrite", "o": "overwrite", "overwrite": "overwrite", + "4": "exit", "e": "exit", "q": "exit", "exit": "exit", + } + while True: + answer = input_fn("Select [1-4]: ").strip().lower() + if answer in aliases: + return aliases[answer] + output_fn("Please enter 1, 2, 3, or 4.") + + +def select_run_workspace( + proj_dir, + resume_requested=False, + *, + input_fn=input, + output_fn=print, + interactive=None, + now=None, +): + """Select or create a safe run workspace for one invocation. + + Legacy flat ``fm_agent/`` results are moved into ``runs/legacy-*`` before + they are resumed or archived. In a non-interactive terminal, existing + results are never modified unless ``--resume`` was explicitly requested. + """ + root = workspace_root(proj_dir) + os.makedirs(root, exist_ok=True) + active = current_run_dir(proj_dir) + legacy = _legacy_entries(root) + existing = active + if not existing and legacy: + existing = root + + if not existing: + work_dir, run_id = _new_run(root, now=now) + output_fn(f"[Pipeline] New run: {workdir_relpath(proj_dir, work_dir)}/") + return RunSelection(work_dir, run_id, False, "new") + + if resume_requested: + if existing == root: + existing = _move_legacy_run(root, now=now) + run_id = _activate(root, existing) + output_fn(f"[Pipeline] RESUME: {workdir_relpath(proj_dir, existing)}/") + return RunSelection(existing, run_id, True, "resume") + + if interactive is None: + interactive = sys.stdin.isatty() + if not interactive: + raise RunSelectionCancelled( + "Existing FM-Agent results found. Re-run with --resume in a " + "non-interactive terminal, or run interactively to archive, overwrite, or exit." + ) + + action = _prompt_action(input_fn, output_fn) + if action == "exit": + raise RunSelectionCancelled("Run cancelled; existing results were not changed.") + if action == "resume": + if existing == root: + existing = _move_legacy_run(root, now=now) + run_id = _activate(root, existing) + return RunSelection(existing, run_id, True, action) + if action == "archive": + if existing == root: + archived = _move_legacy_run(root, now=now) + else: + archived = existing + output_fn(f"[Pipeline] Archived previous run at {workdir_relpath(proj_dir, archived)}/") + work_dir, run_id = _new_run(root, now=now) + output_fn(f"[Pipeline] New run: {workdir_relpath(proj_dir, work_dir)}/") + return RunSelection(work_dir, run_id, False, action) + + # Overwrite is the only path that removes prior results, and it is reachable + # only through an explicit interactive choice. + if existing == root: + for entry in _legacy_entries(root): + if os.path.isdir(entry) and not os.path.islink(entry): + shutil.rmtree(entry) + else: + os.unlink(entry) + else: + shutil.rmtree(existing) + work_dir, run_id = _new_run(root, now=now) + output_fn(f"[Pipeline] Previous run overwritten; new run: {workdir_relpath(proj_dir, work_dir)}/") + return RunSelection(work_dir, run_id, False, action) diff --git a/src/verification.py b/src/verification.py index 9078ad9b..f3e305fe 100644 --- a/src/verification.py +++ b/src/verification.py @@ -10,6 +10,7 @@ list_staged_domain_knowledge_relpaths, load_staged_domain_knowledge_text, ) +from .run_workspace import inferred_workdir_relpath import os import re import json @@ -172,11 +173,12 @@ def streaming_reasoner(input_dir, output_dir, file_list=None, proj_dir=None, wor future.result() # Read validation result to check confirmation parts = result_json_rel - prefix = os.path.join("fm_agent", "logic_verification_results") + os.sep + result_prefix = os.path.relpath(output_dir, proj_dir) + prefix = result_prefix + os.sep if parts.startswith(prefix): parts = parts[len(prefix):] - elif parts.startswith("fm_agent/logic_verification_results/"): - parts = parts[len("fm_agent/logic_verification_results/"):] + elif parts.startswith(result_prefix.replace(os.sep, "/") + "/"): + parts = parts[len(result_prefix.replace(os.sep, "/") + "/"):] bug_id = os.path.splitext(parts)[0].replace(os.sep, "--").replace("/", "--") result_path = os.path.join(work_dir, "bug_validation", f"{bug_id}.result.json") confirmed = False @@ -336,18 +338,19 @@ def _validate_single_bug(result_json_rel, proj_dir, work_dir=None, resume=False) # Derive bug id from result path relative to results dir # e.g. "fm_agent/logic_verification_results/mod/func.json" -> "mod--func" parts = result_json_rel - prefix = os.path.join("fm_agent", "logic_verification_results") + os.sep + work_rel = inferred_workdir_relpath(work_dir) + prefix = os.path.join(work_rel, "logic_verification_results") + os.sep if parts.startswith(prefix): parts = parts[len(prefix):] - elif parts.startswith("fm_agent/logic_verification_results/"): - parts = parts[len("fm_agent/logic_verification_results/"):] + elif parts.startswith(f"{work_rel}/logic_verification_results/"): + parts = parts[len(f"{work_rel}/logic_verification_results/"):] bug_id = os.path.splitext(parts)[0].replace(os.sep, "--").replace("/", "--") function_id = function_id_from_result_path(result_json_rel) # Read the base bug_validator.md base_md_path = os.path.join(script_dir, "md", "bug_validator.md") with open(base_md_path, "r") as f: - base_content = f.read() + base_content = f.read().replace("fm_agent/", f"{work_rel}/") user_knowledge_paths = list_staged_domain_knowledge_relpaths(work_dir) if user_knowledge_paths: @@ -373,7 +376,7 @@ def _validate_single_bug(result_json_rel, proj_dir, work_dir=None, resume=False) os.makedirs(os.path.join(work_dir, "bug_validation"), exist_ok=True) prompt_filename = os.path.join( - "fm_agent", "bug_validation", f"bug_validator_{bug_id}.md" + work_rel, "bug_validation", f"bug_validator_{bug_id}.md" ) prompt_path = os.path.join(proj_dir, prompt_filename) @@ -394,7 +397,7 @@ def _validate_single_bug(result_json_rel, proj_dir, work_dir=None, resume=False) command = ["opencode", "run", "--model", f"{OPENCODE_MODEL_PROVIDER}/{OPENCODE_BUG_VALIDATION_MODEL}", "--file", prompt_path, "--", prompt] - result_relpath = os.path.join("fm_agent", "bug_validation", f"{bug_id}.result.json") + result_relpath = os.path.join(work_rel, "bug_validation", f"{bug_id}.result.json") result_path = os.path.join(proj_dir, result_relpath) # Resume idempotency: if resuming and this bug was already validated, don't pay for it again. if resume and os.path.exists(result_path): @@ -422,7 +425,7 @@ def _validate_single_bug(result_json_rel, proj_dir, work_dir=None, resume=False) *user_knowledge_paths, ], output_files=[ - os.path.join("fm_agent", "bug_validation", f"{bug_id}.md"), + os.path.join(work_rel, "bug_validation", f"{bug_id}.md"), result_relpath, ], summary=f"OpenCode bug validation for {bug_id}", diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..26430da6 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""FM-Agent test suite.""" diff --git a/tests/test_dashboard_workdir.py b/tests/test_dashboard_workdir.py new file mode 100644 index 00000000..8e38a492 --- /dev/null +++ b/tests/test_dashboard_workdir.py @@ -0,0 +1,22 @@ +import json +import os +import tempfile +import unittest + +from dashboard import _locate_workdir + + +class DashboardWorkdirTests(unittest.TestCase): + def test_project_root_follows_current_run_marker(self): + with tempfile.TemporaryDirectory() as project: + run_dir = os.path.join(project, "fm_agent", "runs", "run-2") + os.makedirs(run_dir) + marker = os.path.join(project, "fm_agent", "current_run.json") + with open(marker, "w") as f: + json.dump({"run_id": "run-2"}, f) + self.assertEqual(str(_locate_workdir(project)), os.path.realpath(run_dir)) + + def test_specific_run_directory_is_used_directly(self): + with tempfile.TemporaryDirectory() as run_dir: + os.makedirs(os.path.join(run_dir, "trace")) + self.assertEqual(str(_locate_workdir(run_dir)), os.path.realpath(run_dir)) diff --git a/tests/test_run_workspace.py b/tests/test_run_workspace.py new file mode 100644 index 00000000..2702670d --- /dev/null +++ b/tests/test_run_workspace.py @@ -0,0 +1,126 @@ +import json +import os +import tempfile +import unittest +from datetime import datetime + +from src.run_workspace import ( + RunSelectionCancelled, + current_run_dir, + inferred_workdir_relpath, + select_run_workspace, +) + + +NOW = datetime(2026, 7, 17, 14, 30, 0) + + +class RunWorkspaceTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.project = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + + def select(self, answers=(), **kwargs): + answers = iter(answers) + return select_run_workspace( + self.project, + input_fn=lambda _prompt: next(answers), + output_fn=lambda _line: None, + interactive=True, + now=NOW, + **kwargs, + ) + + def test_first_run_uses_timestamped_directory(self): + selected = self.select() + self.assertEqual(selected.run_id, "20260717-143000") + self.assertTrue(selected.work_dir.endswith("fm_agent/runs/20260717-143000")) + self.assertEqual(current_run_dir(self.project), selected.work_dir) + self.assertEqual( + inferred_workdir_relpath(selected.work_dir), + "fm_agent/runs/20260717-143000", + ) + + def test_first_noninteractive_run_is_created_without_prompt(self): + selected = select_run_workspace( + self.project, + interactive=False, + output_fn=lambda _line: None, + now=NOW, + ) + self.assertEqual(selected.action, "new") + self.assertTrue(os.path.isdir(selected.work_dir)) + + def test_archive_preserves_previous_run_and_creates_another(self): + first = self.select() + marker = os.path.join(first.work_dir, "result.txt") + with open(marker, "w") as f: + f.write("keep me") + + second = self.select(["2"]) + self.assertEqual(second.action, "archive") + self.assertTrue(os.path.isfile(marker)) + self.assertNotEqual(first.work_dir, second.work_dir) + self.assertTrue(second.work_dir.endswith("20260717-143000-2")) + + def test_resume_reuses_current_run(self): + first = self.select() + with open(os.path.join(first.work_dir, "phases.json"), "w") as f: + json.dump({}, f) + resumed = self.select(resume_requested=True) + self.assertTrue(resumed.resume) + self.assertEqual(resumed.work_dir, first.work_dir) + + def test_overwrite_requires_explicit_choice(self): + archived = self.select() + archived_file = os.path.join(archived.work_dir, "archived.txt") + with open(archived_file, "w") as f: + f.write("preserve me") + first = self.select(["2"]) + old_file = os.path.join(first.work_dir, "result.txt") + with open(old_file, "w") as f: + f.write("delete me") + replacement = self.select(["3"]) + self.assertEqual(replacement.action, "overwrite") + self.assertFalse(os.path.exists(old_file)) + self.assertTrue(os.path.isfile(archived_file)) + self.assertTrue(os.path.isdir(replacement.work_dir)) + + def test_exit_preserves_results(self): + first = self.select() + marker = os.path.join(first.work_dir, "result.txt") + with open(marker, "w") as f: + f.write("keep me") + with self.assertRaises(RunSelectionCancelled): + self.select(["4"]) + self.assertTrue(os.path.isfile(marker)) + + def test_noninteractive_existing_run_fails_safe(self): + first = self.select() + with open(os.path.join(first.work_dir, "result.txt"), "w") as f: + f.write("keep me") + with self.assertRaises(RunSelectionCancelled): + select_run_workspace( + self.project, + interactive=False, + output_fn=lambda _line: None, + now=NOW, + ) + + def test_legacy_layout_is_migrated_when_resumed(self): + root = os.path.join(self.project, "fm_agent") + os.makedirs(root) + legacy_file = os.path.join(root, "phases.json") + with open(legacy_file, "w") as f: + f.write("{}") + selected = self.select(resume_requested=True) + self.assertTrue(selected.run_id.startswith("legacy-20260717-143000")) + self.assertTrue(os.path.isfile(os.path.join(selected.work_dir, "phases.json"))) + self.assertFalse(os.path.exists(legacy_file)) + + +if __name__ == "__main__": + unittest.main() From e9f45ad5c1ba214db3671465a2403bd05e1f8cf5 Mon Sep 17 00:00:00 2001 From: Xian Xu Date: Tue, 21 Jul 2026 19:40:30 +0800 Subject: [PATCH 2/2] fix nested run batch prompt paths --- main.py | 3 +- src/generate_batch_prompts.py | 22 +++++- .../test_generate_batch_prompts_workspace.py | 79 +++++++++++++++++++ 3 files changed, 100 insertions(+), 4 deletions(-) create mode 100644 tests/test_generate_batch_prompts_workspace.py diff --git a/main.py b/main.py index b02d8360..cded03cd 100644 --- a/main.py +++ b/main.py @@ -342,7 +342,8 @@ def run_pipeline( # Generate batch prompts for this layer. On resume, skip functions # that were already specced in a previous run. batch_cmd = ["python3", f"{work_rel}/spec_prompts/generate_batch_prompts.py", - "--phase", str(phase_num), "--layers", str(layer_idx)] + "--phase", str(phase_num), "--layers", str(layer_idx), + "--repo-root", proj_dir] if resume: batch_cmd.append("--resume") subprocess.run(batch_cmd, cwd=proj_dir, check=True) diff --git a/src/generate_batch_prompts.py b/src/generate_batch_prompts.py index 8d3910f2..a17bab76 100644 --- a/src/generate_batch_prompts.py +++ b/src/generate_batch_prompts.py @@ -62,6 +62,15 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--layers", required=True, help="Layer index or inclusive range, e.g. 0 or 0-5") parser.add_argument("--batch-size", type=int, default=2, help="Functions per prompt file") parser.add_argument("--output-dir", default=None, help="Output directory for batch prompt files") + parser.add_argument( + "--repo-root", + default=None, + help=( + "Project repository root. Required when the copied script lives in " + "a nested run workspace; defaults to the work directory's parent " + "for compatibility with the legacy fm_agent/ layout." + ), + ) parser.add_argument("--dry-run", action="store_true", help="Show plan without writing files") parser.add_argument( "--resume", @@ -383,9 +392,16 @@ def main() -> int: # work_dir is the fm_agent/ directory (parent of spec_prompts/ where this script lives) work_dir = Path(__file__).resolve().parent.parent - # fm_agent_prefix is the relative path from the project root to work_dir - repo_root = work_dir.parent - fm_agent_prefix = str(work_dir.relative_to(repo_root)) + "/" + # fm_agent_prefix is the relative path from the project root to work_dir. + # A run-scoped workspace is nested under fm_agent/runs/, so its + # parent is not the repository root; the caller passes that root explicitly. + repo_root = Path(args.repo_root).resolve() if args.repo_root else work_dir.parent + try: + fm_agent_prefix = work_dir.relative_to(repo_root).as_posix() + "/" + except ValueError as exc: + raise ValueError( + f"work directory {work_dir} is not inside --repo-root {repo_root}" + ) from exc phases_json = read_json(work_dir / "phases.json") project = phases_json["project"] diff --git a/tests/test_generate_batch_prompts_workspace.py b/tests/test_generate_batch_prompts_workspace.py new file mode 100644 index 00000000..2f8f8515 --- /dev/null +++ b/tests/test_generate_batch_prompts_workspace.py @@ -0,0 +1,79 @@ +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +class GenerateBatchPromptsWorkspaceTests(unittest.TestCase): + def test_manifest_paths_are_relative_to_repo_root_for_nested_run(self): + repo_source = Path(__file__).resolve().parents[1] + with tempfile.TemporaryDirectory() as tmp: + project = Path(tmp, "project") + work_dir = project / "fm_agent" / "runs" / "run-1" + spec_dir = work_dir / "spec_prompts" + spec_dir.mkdir(parents=True) + + shutil.copy2( + repo_source / "src" / "generate_batch_prompts.py", + spec_dir / "generate_batch_prompts.py", + ) + shutil.copy2(repo_source / "src" / "file_utils.py", spec_dir / "file_utils.py") + + (work_dir / "phases.json").write_text( + json.dumps( + { + "project": "sample", + "languages": ["python"], + "file_extensions": ["py"], + } + ) + ) + function_path = "extracted_functions/sample.py/function.json" + (spec_dir / "phase_01_topdown_layers.json").write_text( + json.dumps( + { + "layers": [ + { + "layer": 0, + "functions": [ + {"name": "sample", "file": function_path} + ], + } + ] + } + ) + ) + + subprocess.run( + [ + sys.executable, + str(spec_dir / "generate_batch_prompts.py"), + "--phase", + "1", + "--layers", + "0", + "--repo-root", + str(project), + ], + cwd=project, + check=True, + capture_output=True, + text=True, + ) + + manifest_path = ( + spec_dir / "batch_prompts_sample_phase01" / "manifest.json" + ) + manifest = json.loads(manifest_path.read_text()) + expected = os.path.join( + "fm_agent", "runs", "run-1", function_path + ).replace(os.sep, "/") + self.assertEqual(manifest["batches"][0]["functions"], [expected]) + + +if __name__ == "__main__": + unittest.main()