diff --git a/README.md b/README.md index 55f1843..a6d6414 100644 --- a/README.md +++ b/README.md @@ -65,13 +65,28 @@ python -m venv .venv 模型与凭证配置完成后，运行 pytest 修复闭环： ```sh +# 一次性构建无网络运行所需的基础镜像；项目有额外依赖时应基于此镜像预装依赖。 +docker build -f docker/pytest-sandbox.Dockerfile \ + -t firstcoder-pytest-sandbox:py311 . + .venv/bin/firstcoder pytest-fix \ --project /path/to/python-project \ --test-command "python -m pytest -q" \ + --execution-backend docker \ --json-out runs/pytest-fix-result.json ``` -`pytest-fix` 默认最多执行两次修复尝试，不重试 Provider 请求。Qdrant 不可用时，Parser 与确定性候选仍会继续工作。 +`pytest-fix` 默认使用 Docker，最多执行两次修复尝试，不重试 Provider 请求。Docker 运行参数固定关闭网络、使用只读基础文件系统、丢弃 capabilities、启用 `no-new-privileges`，并限制 CPU、内存、pids、单文件大小、输出和时间。目标仓库必须是干净 Git worktree；每次 Attempt 从同一 baseline commit 建立独立 worktree，只有通过 focused 和 full pytest 的候选才会回写原仓库。依赖必须预装进 sandbox 镜像，因为测试阶段没有网络。 + +仅对可信项目或本地单元测试，可显式使用宿主进程： + +```sh +.venv/bin/firstcoder pytest-fix \ + --project /path/to/trusted-project \ + --execution-backend local +``` + +Local backend 没有容器级文件系统和网络隔离，CLI 会输出安全警告。写工具则由 Runtime `FreshSourceGuard` 强制要求同路径、未过期、SHA-256 未变化的 `read_token`；读取其他文件或使用 stale token 无法通过写入校验。Qdrant 不可用时，Parser 与确定性候选仍会继续工作。详细威胁模型见 `docs/PYTESTPILOT_SECURITY.md`。 ## 离线验证 diff --git a/docker/pytest-sandbox.Dockerfile b/docker/pytest-sandbox.Dockerfile new file mode 100644 index 0000000..34adca7 --- /dev/null +++ b/docker/pytest-sandbox.Dockerfile @@ -0,0 +1,8 @@ +FROM python:3.11-slim + +RUN python -m pip install --no-cache-dir pytest \ + && groupadd --gid 65532 firstcoder \ + && useradd --uid 65532 --gid 65532 --no-create-home --shell /usr/sbin/nologin firstcoder + +WORKDIR /workspace +USER 65532:65532 diff --git a/docs/DEEPSEEK_BENCHMARK_AUDIT.md b/docs/DEEPSEEK_BENCHMARK_AUDIT.md index 0313cfd..25943a2 100644 --- a/docs/DEEPSEEK_BENCHMARK_AUDIT.md +++ b/docs/DEEPSEEK_BENCHMARK_AUDIT.md @@ -37,7 +37,7 @@ Streaming 只在最终 `message_completed` usage 提交一次。准确边界是 ## 路径级 source-read 与 retrieval policy -Evaluator 按 Transcript 执行顺序提取 `view/read_multi` 与 `edit/write/delete/apply_patch` 的规范化路径。每个既有被修改文件必须在首次修改前被准确读取；新文件、越界路径、未读路径和 stale-read 字段分别输出。测试文件修改和 editable scope 外写入仍直接失败。本轮没有实现运行时 FreshSourceGuard；`stale_read_paths` 预留但尚未做内容 hash 对比。 +Evaluator 按 Transcript 执行顺序提取 `view/read_multi` 与 `edit/write/delete/apply_patch` 的规范化路径。每个既有被修改文件必须在首次修改前被准确读取；新文件、越界路径、未读路径和 stale-read 字段分别输出。测试文件修改和 editable scope 外写入仍直接失败。该次历史 Benchmark 尚未实现运行时 FreshSourceGuard；2026-07-25 后的 `pytest-fix` 已增加基于 path、SHA-256、TTL 和一次性 read token 的 Runtime 强制校验，旧结果不追溯重标。 Baseline 不注册 `code_search`。Vector 的 retrieval-required 任务注册该 Tool，并通过通用首 Tool 约束在第一次请求强制选择它；这段逻辑位于 Benchmark Provider 装饰层，不修改 AgentLoop，也不包含 DeepSeek 分支。向量 preview 不算 source read，必须再次 `view/read_multi`。 diff --git a/docs/MVP_GOAL.md b/docs/MVP_GOAL.md index 4c5bf5e..dad9862 100644 --- a/docs/MVP_GOAL.md +++ b/docs/MVP_GOAL.md @@ -27,7 +27,6 @@ - 新的完整 Trace 子系统； - `archive_read`（复用已有 `retrieve_archive`）； -- FreshSourceGuard； - 专用 TUI 面板； - 14 个 Benchmark； - 多 Agent； @@ -112,3 +111,4 @@ - 两个语义任务的受控小样本中，Baseline 6/6、合规 Vector 6/6；Vector 实际调用 `code_search` 6 次并重新读取候选。两组通过率相同，不能声称检索提升准确率；Vector 平均 Token、Tool Call 与耗时更高。 - 初始 12 项中有两个 Vector 运行测试虽通过但未调用 `code_search`，被标记并排除；策略门加固后仅补跑这两项并通过。本轮 Smoke、12 项及 2 个补跑累计保守成本 `$0.08710954`，低于 `$0.25`，无 usage 缺失或预算中止。 - 加固后完整测试：`.venv/bin/python -m pytest tests -q` 得到 891 passed、2 skipped、0 failed（37.61 秒）；文档完成后仍需执行最终一次完整验证。 +- 2026-07-25 根据生产安全审计扩展原 MVP 范围：`pytest-fix` 增加 Docker Sandbox Backend、Runtime `FreshSourceGuard` 和逐 Attempt Git worktree 隔离；Docker 成为 CLI 默认，Local backend 仅用于可信项目与单元测试。 diff --git a/docs/PYTESTPILOT_SECURITY.md b/docs/PYTESTPILOT_SECURITY.md new file mode 100644 index 0000000..9259b3d --- /dev/null +++ b/docs/PYTESTPILOT_SECURITY.md @@ -0,0 +1,44 @@ +# PytestPilot 执行安全与 Attempt 隔离 + +## 执行边界 + +`ExecutionBackend` 接收参数数组、工作目录和 `ResourceLimits`。当前实现： + +- `LocalProcessBackend`：仅用于可信项目与单元测试；使用环境白名单、进程组超时、输出上限以及平台支持的 rlimit，但不提供网络或宿主文件系统隔离。 +- `DockerSandboxBackend`：用于不可信仓库；关闭网络，根文件系统只读，worktree 单独以可写 bind mount 暴露，使用非 root 用户，丢弃 Linux capabilities，启用 `no-new-privileges`，限制 CPU、内存、pids、单文件大小、tmpfs、输出与墙钟时间。 + +Docker 使用 `--pull=never`，运行时不会联网拉取镜像。先构建 `docker/pytest-sandbox.Dockerfile`；若目标项目依赖 pytest 之外的第三方包，应制作预装且版本锁定的派生镜像。当前实现限制单文件大小，但 bind mount 的总磁盘配额仍由宿主文件系统或 Docker 运行环境负责。 + +相同 Backend 同时用于 Workflow 的 baseline/focused/full pytest，以及 Agent 的 `diagnostics`、`shell` 和 `python_exec`，避免验证命令被隔离但 Agent 内部命令仍在宿主机执行。 + +## FreshSourceGuard + +启用 Guard 的读取返回： + +```text +path + sha256 + size + read_at + read_token +``` + +已有文件的 `edit`、`write`、`delete` 和 `apply_patch` 在 mutation 前强制校验： + +1. token 存在且属于同一路径； +2. 路径位于 `editable_paths`； +3. 当前内容 SHA-256 与 size 等于读取版本； +4. token 未超过 TTL； +5. 成功写入后 token 立即消费，不能重放。 + +新文件不需要伪造读取，但路径必须提前列入 `editable_paths`。Guard 在 Tool executor 内校验，失败时不会写文件；Transcript 审计继续保留为二次证据，不再是主要执行约束。 + +## Attempt 隔离 + +运行前要求用户 Git worktree 干净。每轮从同一个 `base_commit` 创建 detached Git worktree： + +```text +baseline commit +├── attempt-1 worktree → patch + focused/full evidence +└── attempt-2 worktree → baseline + 显式的 attempt-1 patch/失败证据 +``` + +第一轮文件状态不会隐式进入第二轮。只有 focused 和 full pytest 都通过的 Attempt 标记为 `selected`，随后仅将 `editable_paths` 回写原仓库；所有 Attempt 失败时原仓库保持不变。非 Git 目录只为本地 fixture 兼容，会先复制到临时 Git snapshot，再以相同 worktree 流程执行。 + +每轮结果记录 `base_commit`、`attempt_patch`、focused/full 结果、`introduced_failures` 和 `selected`。 diff --git a/docs/PYTEST_FIX_DEMO_RUNBOOK.md b/docs/PYTEST_FIX_DEMO_RUNBOOK.md index 1d11dfc..e37231d 100644 --- a/docs/PYTEST_FIX_DEMO_RUNBOOK.md +++ b/docs/PYTEST_FIX_DEMO_RUNBOOK.md @@ -128,9 +128,13 @@ Vector 六个合规运行均有一次 `code_search`，且候选随后被 `read_m 只有用户明确确认凭证、额度和预算后才运行真实模型。命令会使用现有 Provider 配置，不应复制或打印 API Key；DeepSeek 审计实验使用 `benchmark.deepseek_paired` 的请求预算入口，不使用未包装的默认 Runner： ```sh +docker build -f docker/pytest-sandbox.Dockerfile \ + -t firstcoder-pytest-sandbox:py311 . + .venv/bin/firstcoder pytest-fix \ --project /path/to/python-repo \ --test-command "python -m pytest -q --tb=short" \ + --execution-backend docker \ --json-out runs/pytest-fix-result.json .venv/bin/python -m benchmark.deepseek_paired \ diff --git a/firstcoder/cli.py b/firstcoder/cli.py index 5f632af..916245f 100644 --- a/firstcoder/cli.py +++ b/firstcoder/cli.py @@ -77,6 +77,19 @@ def build_parser() -> argparse.ArgumentParser: fix_parser.add_argument("--failure-log", default=None) fix_parser.add_argument("--max-attempts", type=_positive_int, default=2) fix_parser.add_argument("--json-out", default="runs/pytest-fix-result.json") + fix_parser.add_argument( + "--execution-backend", + choices=("docker", "local"), + default="docker", + help="Use Docker for untrusted projects; local is trusted-only.", + ) + fix_parser.add_argument("--docker-image", default="firstcoder-pytest-sandbox:py311") + fix_parser.add_argument("--timeout-seconds", type=_positive_int, default=300) + fix_parser.add_argument("--cpu-count", type=float, default=1.0) + fix_parser.add_argument("--memory-mb", type=_positive_int, default=1024) + fix_parser.add_argument("--pids-limit", type=_positive_int, default=128) + fix_parser.add_argument("--max-output-chars", type=_positive_int, default=100000) + fix_parser.add_argument("--max-file-size-mb", type=_positive_int, default=64) parser.add_argument("--project", default=".", help="Project root for tools and AGENTS.md.") parser.add_argument("--data-root", default=None, help="Directory for FirstCoder session data.") @@ -277,6 +290,7 @@ def run_index_command(args: argparse.Namespace) -> int: def run_pytest_fix_command(args: argparse.Namespace) -> int: + from firstcoder.execution import DockerSandboxBackend, LocalProcessBackend, ResourceLimits from firstcoder.retrieval import ( FastEmbedProvider, QdrantLocalVectorStore, @@ -323,11 +337,32 @@ def run_pytest_fix_command(args: argparse.Namespace) -> int: extra_tools=extra_tools, ) failure_log = Path(args.failure_log).read_text(encoding="utf-8") if args.failure_log else None + limits = ResourceLimits( + timeout_seconds=args.timeout_seconds, + cpu_count=args.cpu_count, + memory_mb=args.memory_mb, + pids=args.pids_limit, + max_output_chars=args.max_output_chars, + max_file_size_mb=args.max_file_size_mb, + ) + if args.execution_backend == "docker": + uid = os.getuid() if hasattr(os, "getuid") and os.getuid() != 0 else 65532 + gid = os.getgid() if hasattr(os, "getgid") and os.getgid() != 0 else 65532 + execution_backend = DockerSandboxBackend(image=args.docker_image, uid=uid, gid=gid) + else: + execution_backend = LocalProcessBackend() + print( + "warning: --execution-backend local runs project code with host permissions; " + "use it only for trusted repositories", + file=sys.stderr, + ) try: result = PytestFixWorkflow( project, adapter=adapter, semantic_search=semantic_search, + execution_backend=execution_backend, + resource_limits=limits, max_attempts=args.max_attempts, ).run( test_command=args.test_command, diff --git a/firstcoder/eval/adapter.py b/firstcoder/eval/adapter.py index a50beb9..1a482bb 100644 --- a/firstcoder/eval/adapter.py +++ b/firstcoder/eval/adapter.py @@ -26,6 +26,7 @@ from firstcoder.providers.factory import create_provider from firstcoder.providers.types import ChatRequest, ChatResponse, ChatStreamEvent, ToolChoiceFunction from firstcoder.tools.builtin import create_builtin_registry +from firstcoder.tools.fresh_source import FreshSourceGuard from firstcoder.tools.types import Tool from firstcoder.utils.sandbox_access import SandboxAccess @@ -115,6 +116,12 @@ def _session_root_for_task(self, task: CodingTask) -> Path: def _create_loop(self, task: CodingTask, session_root: Path) -> AgentLoop: sandbox_access = SandboxAccess() + fresh_source_guard = None + if task.metadata.get("enforce_fresh_source_guard"): + fresh_source_guard = FreshSourceGuard( + task.repo_path, + editable_paths=task.metadata.get("editable_paths") or (), + ) registry = create_builtin_registry( task.repo_path, include_mutation_tools=True, @@ -122,6 +129,9 @@ def _create_loop(self, task: CodingTask, session_root: Path) -> AgentLoop: include_network_tools=False, include_interactive_tools=False, access=sandbox_access, + fresh_source_guard=fresh_source_guard, + execution_backend=task.metadata.get("execution_backend"), + resource_limits=task.metadata.get("resource_limits"), ) task_tools = [*self.extra_tools] if self.extra_tools_factory is not None: @@ -298,6 +308,8 @@ def _build_task_prompt(task: CodingTask) -> str: "Use the diagnostics tool for pytest so it runs with FirstCoder's active Python environment. " "Start with stack-trace paths, the failing test module, exact symbols, and grep. " "Before changing an existing file, read that exact file with view or read_multi. " + "When a source read returns a read_token, pass that token to edit/write/delete; " + "for apply_patch pass a read_tokens mapping keyed by every existing path. " "When this task is marked retrieval_required and code_search is available, call code_search before the first " "mutation, then read at least one returned candidate with view or read_multi. If code_search is unavailable, " "continue with deterministic grep/glob/view tools without asking the user. Never access /workspace. " diff --git a/firstcoder/execution.py b/firstcoder/execution.py new file mode 100644 index 0000000..887c47f --- /dev/null +++ b/firstcoder/execution.py @@ -0,0 +1,295 @@ +"""Command execution backends for trusted and untrusted repositories.""" + +from __future__ import annotations + +import os +import signal +import subprocess +import sys +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol + +from firstcoder.utils.subprocess import CommandResult + + +_DEFAULT_ENV_ALLOWLIST = ( + "LANG", + "LC_ALL", + "TZ", +) + + +@dataclass(frozen=True, slots=True) +class ResourceLimits: + """Hard and host-side limits applied to one command.""" + + timeout_seconds: float = 300.0 + cpu_count: float = 1.0 + memory_mb: int = 1024 + pids: int = 128 + max_output_chars: int = 100_000 + max_file_size_mb: int = 64 + tmpfs_mb: int = 256 + + def __post_init__(self) -> None: + for name in ( + "timeout_seconds", + "cpu_count", + "memory_mb", + "pids", + "max_output_chars", + "max_file_size_mb", + "tmpfs_mb", + ): + if getattr(self, name) <= 0: + raise ValueError(f"{name} must be greater than zero") + + +class ExecutionBackend(Protocol): + """Runs an argv command inside an explicit execution boundary.""" + + def run( + self, + command: list[str], + cwd: Path, + limits: ResourceLimits, + ) -> CommandResult: + ... + + +class LocalProcessBackend: + """Host process backend for trusted projects and unit tests only.""" + + trusted_only = True + + def __init__( + self, + *, + environment: dict[str, str] | None = None, + env_allowlist: tuple[str, ...] = _DEFAULT_ENV_ALLOWLIST, + ) -> None: + source = os.environ if environment is None else environment + self.environment = _whitelisted_environment(source, env_allowlist) + self.environment["PYTHONDONTWRITEBYTECODE"] = "1" + + def run(self, command: list[str], cwd: Path, limits: ResourceLimits) -> CommandResult: + return _run_bounded( + command, + cwd=cwd.resolve(), + env=self.environment, + limits=limits, + local_resource_limits=True, + ) + + +class DockerSandboxBackend: + """Docker boundary for running tests from an isolated attempt worktree.""" + + trusted_only = False + + def __init__( + self, + *, + image: str = "python:3.11-slim", + docker_binary: str = "docker", + environment: dict[str, str] | None = None, + env_allowlist: tuple[str, ...] = _DEFAULT_ENV_ALLOWLIST, + uid: int = 65532, + gid: int = 65532, + ) -> None: + if not image.strip(): + raise ValueError("Docker image cannot be empty") + self.image = image + self.docker_binary = docker_binary + source = os.environ if environment is None else environment + self.environment = _whitelisted_environment(source, env_allowlist) + self.environment["PYTHONDONTWRITEBYTECODE"] = "1" + self.uid = uid + self.gid = gid + + def run(self, command: list[str], cwd: Path, limits: ResourceLimits) -> CommandResult: + root = cwd.resolve() + container_command = _container_command(command) + docker_command = [ + self.docker_binary, + "run", + "--rm", + "--pull=never", + "--network=none", + "--read-only", + "--cap-drop=ALL", + "--security-opt=no-new-privileges", + f"--cpus={limits.cpu_count}", + f"--memory={limits.memory_mb}m", + "--memory-swap", + f"{limits.memory_mb}m", + f"--pids-limit={limits.pids}", + f"--user={self.uid}:{self.gid}", + f"--ulimit=fsize={limits.max_file_size_mb * 1024}:{limits.max_file_size_mb * 1024}", + f"--tmpfs=/tmp:rw,noexec,nosuid,nodev,size={limits.tmpfs_mb}m", + "--mount", + f"type=bind,src={root},dst=/workspace,rw", + "--workdir=/workspace", + ] + for key, value in sorted(self.environment.items()): + docker_command.extend(["--env", f"{key}={value}"]) + docker_command.extend([self.image, *container_command]) + return _run_bounded( + docker_command, + cwd=root, + env=_docker_client_environment(os.environ), + limits=limits, + local_resource_limits=False, + ) + + +def _container_command(command: list[str]) -> list[str]: + if not command: + raise ValueError("command cannot be empty") + converted = list(command) + if Path(converted[0]).resolve() == Path(sys.executable).resolve(): + converted[0] = "python" + return converted + + +def _whitelisted_environment(source: dict[str, str], allowlist: tuple[str, ...]) -> dict[str, str]: + return {key: source[key] for key in allowlist if key in source} + + +def _docker_client_environment(source: dict[str, str]) -> dict[str, str]: + """Keep only variables needed to locate and talk to the Docker daemon.""" + + keys = ("PATH", "DOCKER_HOST", "DOCKER_CONTEXT", "DOCKER_CONFIG", "HOME") + return {key: source[key] for key in keys if key in source} + + +def _run_bounded( + command: list[str], + *, + cwd: Path, + env: dict[str, str], + limits: ResourceLimits, + local_resource_limits: bool, +) -> CommandResult: + """Stream output into bounded buffers and kill the whole process group on timeout.""" + + preexec_fn = _local_preexec(limits) if local_resource_limits and os.name == "posix" else None + try: + process = subprocess.Popen( + command, + cwd=cwd, + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + start_new_session=preexec_fn is None, + preexec_fn=preexec_fn, + ) + except OSError as exc: + return CommandResult( + exit_code=-1, + stdout="", + stderr="", + stdout_truncated=False, + stderr_truncated=False, + ok=False, + error=f"Command execution failed: {exc}", + ) + + stdout_parts: list[str] = [] + stderr_parts: list[str] = [] + stdout_state = [0, False] + stderr_state = [0, False] + readers = [ + threading.Thread( + target=_drain_stream, + args=(process.stdout, stdout_parts, stdout_state, limits.max_output_chars), + daemon=True, + ), + threading.Thread( + target=_drain_stream, + args=(process.stderr, stderr_parts, stderr_state, limits.max_output_chars), + daemon=True, + ), + ] + for reader in readers: + reader.start() + + error: str | None = None + try: + process.wait(timeout=limits.timeout_seconds) + except subprocess.TimeoutExpired: + error = "Command execution timed out" + _kill_process_group(process) + for reader in readers: + reader.join(timeout=1) + + stdout = "".join(stdout_parts) + stderr = "".join(stderr_parts) + returncode = process.returncode if process.returncode is not None else -1 + return CommandResult( + exit_code=returncode, + stdout=stdout, + stderr=stderr, + stdout_truncated=bool(stdout_state[1]), + stderr_truncated=bool(stderr_state[1]), + ok=returncode == 0 and error is None, + error=error, + ) + + +def _drain_stream(stream, parts: list[str], state: list[int | bool], limit: int) -> None: + if stream is None: + return + for chunk in iter(lambda: stream.read(4096), ""): + remaining = limit - int(state[0]) + if remaining > 0: + kept = chunk[:remaining] + parts.append(kept) + state[0] = int(state[0]) + len(kept) + if len(chunk) > max(remaining, 0): + state[1] = True + stream.close() + + +def _kill_process_group(process: subprocess.Popen[str]) -> None: + if process.poll() is not None: + return + try: + os.killpg(process.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): + process.kill() + process.wait(timeout=5) + + +def _local_preexec(limits: ResourceLimits): + def apply_limits() -> None: + import resource + + os.setsid() + cpu_seconds = max(1, int(limits.timeout_seconds)) + _try_setrlimit(resource, resource.RLIMIT_CPU, cpu_seconds) + memory_bytes = limits.memory_mb * 1024 * 1024 + _try_setrlimit(resource, resource.RLIMIT_AS, memory_bytes) + _try_setrlimit(resource, resource.RLIMIT_NPROC, limits.pids) + file_bytes = limits.max_file_size_mb * 1024 * 1024 + _try_setrlimit(resource, resource.RLIMIT_FSIZE, file_bytes) + + return apply_limits + + +def _try_setrlimit(resource_module, resource_kind: int, requested: int) -> None: + """Apply the tightest supported limit without making trusted-local startup brittle.""" + + try: + _, hard = resource_module.getrlimit(resource_kind) + value = requested if hard == resource_module.RLIM_INFINITY else min(requested, hard) + resource_module.setrlimit(resource_kind, (value, value)) + except (OSError, ValueError): + return diff --git a/firstcoder/tools/apply_patch.py b/firstcoder/tools/apply_patch.py index c33905f..a80c6e7 100644 --- a/firstcoder/tools/apply_patch.py +++ b/firstcoder/tools/apply_patch.py @@ -7,6 +7,7 @@ from firstcoder.permissions.types import PermissionAction from firstcoder.tools.types import Tool, ToolPermissionSpec, ToolResult, make_error_result, make_text_result +from firstcoder.tools.fresh_source import FreshSourceGuard, FreshSourceViolation from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.sandbox_access import SandboxAccess @@ -43,19 +44,28 @@ class PatchPlan: operations: list[PatchOperation] -def create_apply_patch_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_apply_patch_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, +) -> Tool: """创建多文件文本补丁工具。""" sandbox = PathSandbox(root, access=access) - def apply_patch(patch: str, dry_run: bool = False) -> ToolResult: + def apply_patch(patch: str, dry_run: bool = False, read_tokens: dict = None) -> ToolResult: """按 patch 语法新增、更新、删除或移动项目内文本文件。""" try: plan = parse_patch(patch) + consumed_tokens = _validate_fresh_patch(fresh_source_guard, sandbox, plan, read_tokens or {}) outcome = _apply_plan(sandbox, plan, dry_run=dry_run) - except ValueError as exc: + except (ValueError, FreshSourceViolation) as exc: return make_error_result("apply_patch", str(exc)) + if fresh_source_guard is not None and not dry_run: + for token in consumed_tokens: + fresh_source_guard.consume(token) return make_text_result( "apply_patch", @@ -78,6 +88,28 @@ def apply_patch(patch: str, dry_run: bool = False) -> ToolResult: return tool +def _validate_fresh_patch( + guard: FreshSourceGuard | None, + sandbox: PathSandbox, + plan: PatchPlan, + read_tokens: dict, +) -> list[str]: + if guard is None: + return [] + consumed: list[str] = [] + for operation in plan.operations: + target = sandbox.resolve(operation.path) + if operation.action == "add": + guard.validate_new(operation.path) + else: + token = str(read_tokens.get(operation.path) or "") + guard.validate_existing(operation.path, token) + consumed.append(token) + if operation.move_to: + guard.validate_new(operation.move_to) + return consumed + + def _permission_target_for_patch(arguments: dict[str, object]) -> str: patch = str(arguments.get("patch") or "") plan = parse_patch(patch) diff --git a/firstcoder/tools/builtin.py b/firstcoder/tools/builtin.py index 5457852..197bd0d 100644 --- a/firstcoder/tools/builtin.py +++ b/firstcoder/tools/builtin.py @@ -27,6 +27,8 @@ from firstcoder.tools.web_search import create_web_search_tool from firstcoder.tools.write import create_write_tool from firstcoder.tools.descriptions import apply_agent_tool_description +from firstcoder.tools.fresh_source import FreshSourceGuard +from firstcoder.execution import ExecutionBackend, ResourceLimits from firstcoder.utils.sandbox_access import SandboxAccess @@ -37,6 +39,9 @@ def create_builtin_registry( include_network_tools: bool = False, include_interactive_tools: bool = True, access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, + execution_backend: ExecutionBackend | None = None, + resource_limits: ResourceLimits | None = None, ) -> ToolRegistry: """创建第一阶段默认可用工具。 @@ -45,16 +50,21 @@ def create_builtin_registry( tools = [ create_ls_tool(root, access=access), - create_view_tool(root, access=access), + create_view_tool(root, access=access, fresh_source_guard=fresh_source_guard), create_grep_tool(root, access=access), create_glob_tool(root, access=access), create_tree_tool(root, access=access), create_git_status_tool(root, access=access), create_git_diff_tool(root, access=access), create_git_log_tool(root, access=access), - create_diagnostics_tool(root, access=access), + create_diagnostics_tool( + root, + access=access, + execution_backend=execution_backend, + resource_limits=resource_limits, + ), create_think_tool(), - create_read_multi_tool(root, access=access), + create_read_multi_tool(root, access=access, fresh_source_guard=fresh_source_guard), create_todo_tool(), ] if include_interactive_tools: @@ -62,17 +72,27 @@ def create_builtin_registry( if include_mutation_tools: tools.extend( [ - create_write_tool(root, access=access), - create_edit_tool(root, access=access), - create_delete_tool(root, access=access), - create_apply_patch_tool(root, access=access), + create_write_tool(root, access=access, fresh_source_guard=fresh_source_guard), + create_edit_tool(root, access=access, fresh_source_guard=fresh_source_guard), + create_delete_tool(root, access=access, fresh_source_guard=fresh_source_guard), + create_apply_patch_tool(root, access=access, fresh_source_guard=fresh_source_guard), ] ) if include_execution_tools: tools.extend( [ - create_shell_tool(root, access=access), - create_python_exec_tool(root, access=access), + create_shell_tool( + root, + access=access, + execution_backend=execution_backend, + resource_limits=resource_limits, + ), + create_python_exec_tool( + root, + access=access, + execution_backend=execution_backend, + resource_limits=resource_limits, + ), ] ) if include_network_tools: diff --git a/firstcoder/tools/delete.py b/firstcoder/tools/delete.py index c5be661..3262322 100644 --- a/firstcoder/tools/delete.py +++ b/firstcoder/tools/delete.py @@ -7,17 +7,23 @@ from firstcoder.permissions.types import PermissionAction from firstcoder.tools.types import Tool, ToolPermissionSpec, ToolResult, make_error_result, make_text_result +from firstcoder.tools.fresh_source import FreshSourceGuard, FreshSourceViolation from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.sandbox_access import SandboxAccess -def create_delete_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_delete_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, +) -> Tool: """创建删除文件或目录的工具。""" sandbox = PathSandbox(root, access=access) - def delete(path: str, recursive: bool = False) -> ToolResult: + def delete(path: str, recursive: bool = False, read_token: str = "") -> ToolResult: """删除项目内文件或目录；目录删除必须 recursive=true。""" try: @@ -26,6 +32,13 @@ def delete(path: str, recursive: bool = False) -> ToolResult: return make_error_result("delete", str(exc)) if target.resolve() == sandbox.root: return make_error_result("delete", "不能删除项目根目录") + if fresh_source_guard is not None: + if target.is_dir() and not target.is_symlink(): + return make_error_result("delete", "FreshSourceGuard 不允许递归目录删除", path=path) + try: + fresh_source_guard.validate_existing(path, read_token) + except FreshSourceViolation as exc: + return make_error_result("delete", f"FreshSourceGuard: {exc}", path=path) relative = sandbox.relative(target) if target.is_dir() and not target.is_symlink(): @@ -35,6 +48,8 @@ def delete(path: str, recursive: bool = False) -> ToolResult: return make_text_result("delete", f"已删除目录：{relative}", path=relative, type="dir") target.unlink() + if fresh_source_guard is not None: + fresh_source_guard.consume(read_token) return make_text_result("delete", f"已删除文件：{relative}", path=relative, type="file") tool = tool_from_function(delete) diff --git a/firstcoder/tools/diagnostics.py b/firstcoder/tools/diagnostics.py index 4793e0f..46def78 100644 --- a/firstcoder/tools/diagnostics.py +++ b/firstcoder/tools/diagnostics.py @@ -7,16 +7,28 @@ from pathlib import Path from firstcoder.ci.pytest_parser import parse_pytest_output +from firstcoder.execution import ExecutionBackend, ResourceLimits from firstcoder.tools.types import Tool, ToolResult, make_error_result, make_text_result from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.execution_sandbox import ExecutionSandbox from firstcoder.utils.sandbox_access import SandboxAccess -def create_diagnostics_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_diagnostics_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + execution_backend: ExecutionBackend | None = None, + resource_limits: ResourceLimits | None = None, +) -> Tool: """创建项目诊断工具。""" - sandbox = ExecutionSandbox(root, access=access) + sandbox = ExecutionSandbox( + root, + access=access, + backend=execution_backend, + resource_limits=resource_limits, + ) def diagnostics(command: str = "python -m pytest -q", timeout_seconds: int = 120, max_output_chars: int = 20000) -> ToolResult: """运行项目诊断命令，适合测试、lint、类型检查。""" @@ -26,7 +38,11 @@ def diagnostics(command: str = "python -m pytest -q", timeout_seconds: int = 120 if max_output_chars <= 0: return make_error_result("diagnostics", "max_output_chars 必须大于 0") - normalized_command = command.replace("python", sys.executable, 1) if command.startswith("python ") else command + normalized_command = ( + command + if execution_backend is not None + else command.replace("python", sys.executable, 1) if command.startswith("python ") else command + ) result = sandbox.run( normalized_command, cwd=".", diff --git a/firstcoder/tools/edit.py b/firstcoder/tools/edit.py index bb84b24..0c70c54 100644 --- a/firstcoder/tools/edit.py +++ b/firstcoder/tools/edit.py @@ -6,24 +6,35 @@ from firstcoder.permissions.types import PermissionAction from firstcoder.tools.types import Tool, ToolPermissionSpec, ToolResult, make_error_result, make_text_result +from firstcoder.tools.fresh_source import FreshSourceGuard, FreshSourceViolation from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.sandbox_access import SandboxAccess from firstcoder.utils.text import safe_read_text -def create_edit_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_edit_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, +) -> Tool: """创建替换文本片段的工具。""" sandbox = PathSandbox(root, access=access) - def edit(path: str, old: str, new: str, replace_all: bool = False) -> ToolResult: + def edit(path: str, old: str, new: str, replace_all: bool = False, read_token: str = "") -> ToolResult: """替换项目内 UTF-8 文本片段；默认只替换唯一匹配。""" try: target = sandbox.resolve_validated(path, expect="file") except ValueError as exc: return make_error_result("edit", str(exc)) + if fresh_source_guard is not None: + try: + fresh_source_guard.validate_existing(path, read_token) + except FreshSourceViolation as exc: + return make_error_result("edit", f"FreshSourceGuard: {exc}", path=path) if old == "": return make_error_result("edit", "old 不能为空") @@ -41,6 +52,8 @@ def edit(path: str, old: str, new: str, replace_all: bool = False) -> ToolResult new_text = text.replace(old, new) if replace_all else text.replace(old, new, 1) replacements = count if replace_all else 1 target.write_text(new_text, encoding="utf-8") + if fresh_source_guard is not None: + fresh_source_guard.consume(read_token) return make_text_result( "edit", diff --git a/firstcoder/tools/fresh_source.py b/firstcoder/tools/fresh_source.py new file mode 100644 index 0000000..fb906c1 --- /dev/null +++ b/firstcoder/tools/fresh_source.py @@ -0,0 +1,103 @@ +"""Runtime-enforced fresh-source revisions for mutation tools.""" + +from __future__ import annotations + +import hashlib +import secrets +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from firstcoder.utils.sandbox import PathSandbox + + +@dataclass(frozen=True, slots=True) +class SourceRevision: + path: str + sha256: str + size: int + read_at: datetime + token: str + + def to_dict(self) -> dict[str, object]: + return { + "path": self.path, + "sha256": self.sha256, + "size": self.size, + "read_at": self.read_at.isoformat(), + "token": self.token, + } + + +class FreshSourceViolation(ValueError): + """Raised before a mutation that lacks current, scoped source evidence.""" + + +class FreshSourceGuard: + """Issues short-lived revisions and validates them immediately before writes.""" + + def __init__( + self, + root: str | Path, + *, + editable_paths: list[str] | tuple[str, ...], + token_ttl_seconds: float = 900.0, + ) -> None: + if token_ttl_seconds <= 0: + raise ValueError("token_ttl_seconds must be greater than zero") + self.sandbox = PathSandbox(root) + self.editable_paths = frozenset(self._normalize(path) for path in editable_paths) + self.token_ttl = timedelta(seconds=token_ttl_seconds) + self._revisions: dict[str, SourceRevision] = {} + + def issue(self, path: str | Path) -> SourceRevision: + target = self.sandbox.resolve_validated(path, expect="file") + relative = self.sandbox.relative(target) + content = target.read_bytes() + revision = SourceRevision( + path=relative, + sha256=hashlib.sha256(content).hexdigest(), + size=len(content), + read_at=datetime.now(timezone.utc), + token=secrets.token_urlsafe(24), + ) + self._revisions[revision.token] = revision + return revision + + def validate_existing(self, path: str | Path, token: str) -> SourceRevision: + relative = self._require_editable(path) + revision = self._revisions.get(token) + if revision is None: + raise FreshSourceViolation("missing or unknown read_token") + if revision.path != relative: + raise FreshSourceViolation( + f"read_token belongs to {revision.path}, not {relative}" + ) + if datetime.now(timezone.utc) - revision.read_at > self.token_ttl: + self._revisions.pop(token, None) + raise FreshSourceViolation(f"read_token expired for {relative}") + target = self.sandbox.resolve_validated(relative, expect="file") + content = target.read_bytes() + current_hash = hashlib.sha256(content).hexdigest() + if current_hash != revision.sha256 or len(content) != revision.size: + raise FreshSourceViolation(f"source changed after read: {relative}") + return revision + + def validate_new(self, path: str | Path) -> str: + relative = self._require_editable(path) + target = self.sandbox.resolve(relative) + if target.exists(): + raise FreshSourceViolation(f"existing file requires read_token: {relative}") + return relative + + def consume(self, token: str) -> None: + self._revisions.pop(token, None) + + def _require_editable(self, path: str | Path) -> str: + relative = self._normalize(path) + if relative not in self.editable_paths: + raise FreshSourceViolation(f"path is outside editable_paths: {relative}") + return relative + + def _normalize(self, path: str | Path) -> str: + return self.sandbox.relative(self.sandbox.resolve(path)) diff --git a/firstcoder/tools/python_exec.py b/firstcoder/tools/python_exec.py index ee2db22..1b3cea7 100644 --- a/firstcoder/tools/python_exec.py +++ b/firstcoder/tools/python_exec.py @@ -7,16 +7,28 @@ from pathlib import Path from firstcoder.permissions.types import PermissionAction +from firstcoder.execution import ExecutionBackend, ResourceLimits from firstcoder.tools.types import Tool, ToolPermissionSpec, ToolResult, make_error_result, make_text_result from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.execution_sandbox import ExecutionSandbox from firstcoder.utils.sandbox_access import SandboxAccess -def create_python_exec_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_python_exec_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + execution_backend: ExecutionBackend | None = None, + resource_limits: ResourceLimits | None = None, +) -> Tool: """创建 Python 代码执行工具。""" - sandbox = ExecutionSandbox(root, access=access) + sandbox = ExecutionSandbox( + root, + access=access, + backend=execution_backend, + resource_limits=resource_limits, + ) def python_exec(code: str, cwd: str = ".", timeout_seconds: int = 30, max_output_chars: int = 20000) -> ToolResult: """在项目内执行 Python 代码；高风险，需显式启用。""" diff --git a/firstcoder/tools/read_multi.py b/firstcoder/tools/read_multi.py index d413d85..b67d18f 100644 --- a/firstcoder/tools/read_multi.py +++ b/firstcoder/tools/read_multi.py @@ -8,13 +8,19 @@ from pathlib import Path from firstcoder.tools.types import Tool, ToolResult, make_error_result, make_text_result +from firstcoder.tools.fresh_source import FreshSourceGuard from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.sandbox_access import SandboxAccess from firstcoder.utils.text import safe_read_text -def create_read_multi_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_read_multi_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, +) -> Tool: """创建批量文件读取工具。""" sandbox = PathSandbox(root, access=access) @@ -54,8 +60,15 @@ def read_multi(paths: list[str], max_total_chars: int = 100000) -> ToolResult: continue relative = sandbox.relative(target) + revision = fresh_source_guard.issue(relative) if fresh_source_guard is not None else None file_header = f"=== {relative} ===\n" - file_text = file_header + text + "\n" + revision_text = ( + f"\n[source_revision path={revision.path} sha256={revision.sha256} " + f"size={revision.size} read_token={revision.token}]\n" + if revision is not None + else "\n" + ) + file_text = file_header + text + revision_text # 检查总长度限制 if total_chars + len(file_text) > max_total_chars: @@ -70,7 +83,13 @@ def read_multi(paths: list[str], max_total_chars: int = 100000) -> ToolResult: contents.append(file_text) total_chars += len(file_text) - file_data.append({"path": relative, "lines": text.count("\n") + 1}) + file_data.append( + { + "path": relative, + "lines": text.count("\n") + 1, + "source_revision": revision.to_dict() if revision is not None else None, + } + ) content = "".join(contents).rstrip("\n") diff --git a/firstcoder/tools/shell.py b/firstcoder/tools/shell.py index 2753be7..50713ca 100644 --- a/firstcoder/tools/shell.py +++ b/firstcoder/tools/shell.py @@ -6,6 +6,7 @@ from pathlib import Path from firstcoder.permissions.types import PermissionAction +from firstcoder.execution import ExecutionBackend, ResourceLimits from firstcoder.tools.types import Tool, ToolPermissionSpec, ToolResult, make_error_result, make_text_result from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.execution_sandbox import ExecutionSandbox @@ -16,13 +17,24 @@ DEFAULT_MAX_OUTPUT_CHARS = 20000 -def create_shell_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_shell_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + execution_backend: ExecutionBackend | None = None, + resource_limits: ResourceLimits | None = None, +) -> Tool: """创建命令执行工具。 这是高风险工具：调用方必须在用户明确开启执行权限后才能注册它。 """ - sandbox = ExecutionSandbox(root, access=access) + sandbox = ExecutionSandbox( + root, + access=access, + backend=execution_backend, + resource_limits=resource_limits, + ) def shell( command: str, diff --git a/firstcoder/tools/view.py b/firstcoder/tools/view.py index eb46d23..8254389 100644 --- a/firstcoder/tools/view.py +++ b/firstcoder/tools/view.py @@ -5,13 +5,19 @@ from pathlib import Path from firstcoder.tools.types import Tool, ToolResult, make_error_result, make_text_result +from firstcoder.tools.fresh_source import FreshSourceGuard from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.sandbox_access import SandboxAccess from firstcoder.utils.text import safe_read_text -def create_view_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_view_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, +) -> Tool: """创建读取文本文件的工具。""" sandbox = PathSandbox(root, access=access) @@ -39,14 +45,22 @@ def view(path: str, offset: int = 0, limit: int = 200) -> ToolResult: content = "\n".join(f"{line_number}: {line}" for line_number, line in enumerate(selected, start=offset + 1)) truncated = offset + limit < len(lines) + revision = fresh_source_guard.issue(path) if fresh_source_guard is not None else None + revision_text = ( + f"\n\n[source_revision path={revision.path} sha256={revision.sha256} " + f"size={revision.size} read_token={revision.token}]" + if revision is not None + else "" + ) return make_text_result( "view", - content or "没有可显示内容。", + (content or "没有可显示内容。") + revision_text, path=sandbox.relative(target), start_line=start_line, end_line=end_line, truncated=truncated, total_lines=len(lines), + source_revision=revision.to_dict() if revision is not None else None, ) return tool_from_function(view) diff --git a/firstcoder/tools/write.py b/firstcoder/tools/write.py index 96592a3..4ffe681 100644 --- a/firstcoder/tools/write.py +++ b/firstcoder/tools/write.py @@ -6,17 +6,29 @@ from firstcoder.permissions.types import PermissionAction from firstcoder.tools.types import Tool, ToolPermissionSpec, ToolResult, make_error_result, make_text_result +from firstcoder.tools.fresh_source import FreshSourceGuard, FreshSourceViolation from firstcoder.utils.introspection import tool_from_function from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.sandbox_access import SandboxAccess -def create_write_tool(root: str | Path, *, access: SandboxAccess | None = None) -> Tool: +def create_write_tool( + root: str | Path, + *, + access: SandboxAccess | None = None, + fresh_source_guard: FreshSourceGuard | None = None, +) -> Tool: """创建写入文本文件的工具。""" sandbox = PathSandbox(root, access=access) - def write(path: str, content: str, create_dirs: bool = True, overwrite: bool = True) -> ToolResult: + def write( + path: str, + content: str, + create_dirs: bool = True, + overwrite: bool = True, + read_token: str = "", + ) -> ToolResult: """写入项目内 UTF-8 文本文件；可创建目录或覆盖文件。""" target = sandbox.resolve(path) @@ -24,6 +36,14 @@ def write(path: str, content: str, create_dirs: bool = True, overwrite: bool = T return make_error_result("write", f"路径是目录，不能写入文件：{path}") if target.exists() and not overwrite: return make_error_result("write", f"文件已存在且 overwrite 为 False：{path}") + if fresh_source_guard is not None: + try: + if target.exists(): + fresh_source_guard.validate_existing(path, read_token) + else: + fresh_source_guard.validate_new(path) + except FreshSourceViolation as exc: + return make_error_result("write", f"FreshSourceGuard: {exc}", path=path) parent = target.parent if not parent.exists(): @@ -33,6 +53,8 @@ def write(path: str, content: str, create_dirs: bool = True, overwrite: bool = T created = not target.exists() target.write_text(content, encoding="utf-8") + if fresh_source_guard is not None and read_token: + fresh_source_guard.consume(read_token) return make_text_result( "write", f"已写入文件：{sandbox.relative(target)}", diff --git a/firstcoder/utils/execution_sandbox.py b/firstcoder/utils/execution_sandbox.py index 39f8e4b..c3630c5 100644 --- a/firstcoder/utils/execution_sandbox.py +++ b/firstcoder/utils/execution_sandbox.py @@ -3,9 +3,11 @@ from __future__ import annotations import os +import shlex from pathlib import Path from firstcoder.agent.cancellation import current_cancellation_token +from firstcoder.execution import ExecutionBackend, ResourceLimits from firstcoder.utils.sandbox_access import SandboxAccess from firstcoder.utils.sandbox import PathSandbox from firstcoder.utils.subprocess import CommandResult, run_command @@ -21,9 +23,18 @@ class ExecutionSandbox: a command may run; this class constrains how approved subprocesses run. """ - def __init__(self, root: str | Path, *, access: SandboxAccess | None = None) -> None: + def __init__( + self, + root: str | Path, + *, + access: SandboxAccess | None = None, + backend: ExecutionBackend | None = None, + resource_limits: ResourceLimits | None = None, + ) -> None: self.path_sandbox = PathSandbox(root, access=access) self.root = self.path_sandbox.root + self.backend = backend + self.resource_limits = resource_limits def resolve_cwd(self, cwd: str | Path | None = ".") -> Path: return self.path_sandbox.resolve_validated(cwd, expect="dir") @@ -60,6 +71,21 @@ def run( ok=False, error=str(exc), ) + if self.backend is not None: + args = ["/bin/sh", "-lc", command] if isinstance(command, str) and shell else ( + shlex.split(command) if isinstance(command, str) else list(command) + ) + base = self.resource_limits or ResourceLimits() + limits = ResourceLimits( + timeout_seconds=timeout_seconds, + cpu_count=base.cpu_count, + memory_mb=base.memory_mb, + pids=base.pids, + max_output_chars=max_output_chars, + max_file_size_mb=base.max_file_size_mb, + tmpfs_mb=base.tmpfs_mb, + ) + return self.backend.run(args, workdir, limits) return run_command( command, cwd=workdir, diff --git a/firstcoder/workflows/attempt_workspace.py b/firstcoder/workflows/attempt_workspace.py new file mode 100644 index 0000000..21a69f9 --- /dev/null +++ b/firstcoder/workflows/attempt_workspace.py @@ -0,0 +1,121 @@ +"""Independent workspaces for bounded pytest repair attempts.""" + +from __future__ import annotations + +import shutil +import subprocess +import tempfile +from pathlib import Path + + +class AttemptWorkspaceManager: + """Creates clean Git worktrees, with a copy fallback for non-Git unit fixtures.""" + + def __init__(self, project_root: str | Path) -> None: + self.project_root = Path(project_root).resolve() + self._temporary = tempfile.TemporaryDirectory(prefix="firstcoder-pytest-attempts-") + self.root = Path(self._temporary.name) + self.git_backed = _is_git_worktree(self.project_root) + if self.git_backed: + self.source_repo = self.project_root + self.base_commit = _git_stdout(["rev-parse", "HEAD"], self.project_root) + dirty = _git_stdout(["status", "--porcelain"], self.project_root) + if dirty: + self.close() + raise RuntimeError( + "pytest-fix requires a clean Git worktree so attempts can be isolated without " + "overwriting user changes" + ) + else: + self.source_repo = self.root / "baseline" + shutil.copytree( + self.project_root, + self.source_repo, + ignore=shutil.ignore_patterns(".git", ".firstcoder", "__pycache__", ".pytest_cache"), + ) + _initialize_snapshot_repo(self.source_repo) + self.base_commit = _git_stdout(["rev-parse", "HEAD"], self.source_repo) + self._worktrees: list[Path] = [] + + def create(self, number: int) -> Path: + destination = self.root / f"attempt-{number}" + subprocess.run( + ["git", "worktree", "add", "--detach", str(destination), str(self.base_commit)], + cwd=self.source_repo, + check=True, + text=True, + capture_output=True, + ) + self._worktrees.append(destination) + return destination + + def close(self) -> None: + for worktree in reversed(getattr(self, "_worktrees", [])): + subprocess.run( + ["git", "worktree", "remove", "--force", str(worktree)], + cwd=self.source_repo, + check=False, + text=True, + capture_output=True, + ) + self._worktrees = [] + self._temporary.cleanup() + + def __enter__(self) -> "AttemptWorkspaceManager": + return self + + def __exit__(self, exc_type, exc, traceback) -> None: + self.close() + + +def apply_selected_files( + source: Path, + destination: Path, + *, + editable_paths: list[str], +) -> None: + """Copy only the explicitly editable result set back to the user's clean tree.""" + + for relative in editable_paths: + source_path = source / relative + destination_path = destination / relative + if source_path.is_file(): + destination_path.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source_path, destination_path) + elif destination_path.is_file() or destination_path.is_symlink(): + destination_path.unlink() + + +def _is_git_worktree(root: Path) -> bool: + completed = subprocess.run( + ["git", "rev-parse", "--is-inside-work-tree"], + cwd=root, + text=True, + capture_output=True, + ) + return completed.returncode == 0 and completed.stdout.strip() == "true" + + +def _git_stdout(args: list[str], cwd: Path) -> str: + completed = subprocess.run( + ["git", *args], + cwd=cwd, + check=True, + text=True, + capture_output=True, + ) + return completed.stdout.strip() + + +def _initialize_snapshot_repo(root: Path) -> None: + subprocess.run(["git", "init"], cwd=root, check=True, text=True, capture_output=True) + subprocess.run(["git", "config", "user.email", "firstcoder@local"], cwd=root, check=True) + subprocess.run(["git", "config", "user.name", "FirstCoder Snapshot"], cwd=root, check=True) + subprocess.run(["git", "add", "-A"], cwd=root, check=True) + subprocess.run( + ["git", "commit", "-m", "FirstCoder attempt baseline"], + cwd=root, + check=True, + text=True, + capture_output=True, + ) diff --git a/firstcoder/workflows/models.py b/firstcoder/workflows/models.py index a2b3183..c7b15b1 100644 --- a/firstcoder/workflows/models.py +++ b/firstcoder/workflows/models.py @@ -32,23 +32,31 @@ def to_dict(self) -> dict[str, Any]: @dataclass(frozen=True, slots=True) class PytestFixAttempt: number: int + base_commit: str | None + attempt_patch: str deterministic_candidates: list[str] semantic_candidates: list[dict[str, Any]] transcript_path: str | None focused_result: PytestCommandResult full_result: PytestCommandResult | None source_read_policy_violation: bool + introduced_failures: list[str] = field(default_factory=list) + selected: bool = False runtime_metrics: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "number": self.number, + "base_commit": self.base_commit, + "attempt_patch": self.attempt_patch, "deterministic_candidates": self.deterministic_candidates, "semantic_candidates": self.semantic_candidates, "transcript_path": self.transcript_path, "focused_result": self.focused_result.to_dict(), "full_result": self.full_result.to_dict() if self.full_result is not None else None, "source_read_policy_violation": self.source_read_policy_violation, + "introduced_failures": self.introduced_failures, + "selected": self.selected, "runtime_metrics": self.runtime_metrics, } diff --git a/firstcoder/workflows/pytest_fix.py b/firstcoder/workflows/pytest_fix.py index 46e0bc0..a89d815 100644 --- a/firstcoder/workflows/pytest_fix.py +++ b/firstcoder/workflows/pytest_fix.py @@ -3,11 +3,8 @@ from __future__ import annotations import json -import os import shlex -import subprocess import sys -import tempfile import time from collections import Counter from pathlib import Path @@ -17,8 +14,10 @@ from firstcoder.eval.metrics import collect_diff_metrics, empty_runtime_metrics from firstcoder.eval.patch import collect_git_diff from firstcoder.eval.tasks import CodingTask, CodingTaskResult +from firstcoder.execution import ExecutionBackend, LocalProcessBackend, ResourceLimits from firstcoder.retrieval.models import RetrievalUnavailableError from firstcoder.workflows.models import PytestCommandResult, PytestFixAttempt, PytestFixResult +from firstcoder.workflows.attempt_workspace import AttemptWorkspaceManager, apply_selected_files from firstcoder.workflows.prompts import build_pytest_repair_prompt @@ -37,39 +36,40 @@ def search(self, query: str, *, top_k: int = 5, path_prefix: str | None = None, class PytestCommandRunner(Protocol): - def run(self, command: str) -> PytestCommandResult: + def run(self, command: str, *, cwd: Path | None = None) -> PytestCommandResult: ... class SubprocessPytestCommandRunner: - def __init__(self, root: str | Path, *, timeout_seconds: float = 300.0) -> None: + """Compatibility wrapper around an explicit ExecutionBackend.""" + + def __init__( + self, + root: str | Path, + *, + backend: ExecutionBackend | None = None, + limits: ResourceLimits | None = None, + timeout_seconds: float | None = None, + ) -> None: self.root = Path(root).resolve() - self.timeout_seconds = timeout_seconds + self.backend = backend or LocalProcessBackend() + self.limits = limits or ResourceLimits(timeout_seconds=timeout_seconds or 300.0) - def run(self, command: str) -> PytestCommandResult: + def run(self, command: str, *, cwd: Path | None = None) -> PytestCommandResult: started = time.perf_counter() args = shlex.split(command) if not args: raise ValueError("pytest command cannot be empty") if args[:2] in (["python", "-m"], ["python3", "-m"]): args[0] = sys.executable - with tempfile.TemporaryDirectory(prefix="firstcoder-pycache-") as pycache: - env = os.environ.copy() - env["PYTHONPYCACHEPREFIX"] = pycache - env["PYTHONDONTWRITEBYTECODE"] = "1" - completed = subprocess.run( - args, - cwd=self.root, - env=env, - text=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - timeout=self.timeout_seconds, - ) + completed = self.backend.run(args, (cwd or self.root).resolve(), self.limits) + output = "\n".join(part for part in (completed.stdout, completed.stderr) if part) + if completed.error: + output = "\n".join(part for part in (output, completed.error) if part) return PytestCommandResult( command=command, - exit_code=completed.returncode, - output=completed.stdout or "", + exit_code=completed.exit_code, + output=output, elapsed_seconds=round(time.perf_counter() - started, 6), ) @@ -82,12 +82,24 @@ def __init__( adapter: RepairAdapter, semantic_search: SemanticSearchLike | None = None, command_runner: PytestCommandRunner | None = None, + execution_backend: ExecutionBackend | None = None, + resource_limits: ResourceLimits | None = None, + editable_paths: list[str] | tuple[str, ...] | None = None, max_attempts: int = 2, ) -> None: self.project_root = Path(project_root).resolve() self.adapter = adapter self.semantic_search = semantic_search - self.command_runner = command_runner or SubprocessPytestCommandRunner(self.project_root) + self.execution_backend = execution_backend or LocalProcessBackend() + self.resource_limits = resource_limits or ResourceLimits() + self.command_runner = command_runner or SubprocessPytestCommandRunner( + self.project_root, + backend=self.execution_backend, + limits=self.resource_limits, + ) + self.editable_paths = list(editable_paths) if editable_paths is not None else _default_editable_paths( + self.project_root + ) self.max_attempts = min(2, max(1, max_attempts)) def run( @@ -99,7 +111,9 @@ def run( ) -> PytestFixResult: started = time.perf_counter() if failure_log is None: - baseline_command = self.command_runner.run(test_command) + with AttemptWorkspaceManager(self.project_root) as baseline_workspaces: + baseline_root = baseline_workspaces.create(0) + baseline_command = self.command_runner.run(test_command, cwd=baseline_root) else: baseline_command = PytestCommandResult( command="", @@ -122,54 +136,91 @@ def run( _write_result(json_out, result) return result - for number in range(1, self.max_attempts + 1): - deterministic = _deterministic_candidates(self.project_root, current_report) - semantic = self._semantic_candidates(current_report) - focused_command = _focused_command(test_command, current_report) - task = CodingTask( - instance_id=f"pytest-fix-attempt-{number}", - repo_path=self.project_root, - problem_statement=build_pytest_repair_prompt( + with AttemptWorkspaceManager(self.project_root) as workspaces: + previous_evidence = "" + for number in range(1, self.max_attempts + 1): + attempt_root = workspaces.create(number) + deterministic = _deterministic_candidates(attempt_root, current_report) + semantic = self._semantic_candidates(current_report) + focused_command = _focused_command(test_command, current_report) + problem_statement = build_pytest_repair_prompt( current_report, deterministic_candidates=deterministic, semantic_candidates=semantic, focused_command=focused_command, full_command=test_command, - ), - metadata={"workflow": "pytest_fix", "attempt": number, "test_command": test_command}, - ) - agent_result = self.adapter.run_task(task) - violation = _source_read_policy_violation( - agent_result.transcript_path, - mutation_observed=bool(agent_result.model_patch), - ) - focused = self.command_runner.run(focused_command) - full = self.command_runner.run(test_command) if focused.passed else None - attempts.append( - PytestFixAttempt( - number=number, - deterministic_candidates=deterministic, - semantic_candidates=semantic, - transcript_path=str(agent_result.transcript_path) if agent_result.transcript_path else None, - focused_result=focused, - full_result=full, - source_read_policy_violation=violation, - runtime_metrics=dict(agent_result.runtime_metrics), ) - ) - validation = full or focused - current_report = _report(validation) - if full is not None and full.passed: - result = self._result( - status="passed", - test_command=test_command, - baseline=baseline_report, - final=current_report, - attempts=attempts, - started=started, + if previous_evidence: + problem_statement += previous_evidence + task = CodingTask( + instance_id=f"pytest-fix-attempt-{number}", + repo_path=attempt_root, + problem_statement=problem_statement, + base_commit=workspaces.base_commit, + metadata={ + "workflow": "pytest_fix", + "attempt": number, + "test_command": test_command, + "existing_paths": self.editable_paths, + "editable_paths": self.editable_paths, + "enforce_fresh_source_guard": True, + "execution_backend": self.execution_backend, + "resource_limits": self.resource_limits, + }, + ) + agent_result = self.adapter.run_task(task) + attempt_patch = collect_git_diff(attempt_root, include_untracked=True) + transcript_path = _preserve_transcript( + agent_result.transcript_path, + project_root=self.project_root, + attempt_number=number, + ) + violation = _source_read_policy_violation( + transcript_path, + mutation_observed=bool(attempt_patch), + ) + focused = self.command_runner.run(focused_command, cwd=attempt_root) + full = self.command_runner.run(test_command, cwd=attempt_root) if focused.passed else None + validation = full or focused + current_report = _report(validation) + selected = full is not None and full.passed + introduced = _introduced_failures(baseline_report, current_report) + attempts.append( + PytestFixAttempt( + number=number, + base_commit=workspaces.base_commit, + attempt_patch=attempt_patch, + deterministic_candidates=deterministic, + semantic_candidates=semantic, + transcript_path=str(transcript_path) if transcript_path else None, + focused_result=focused, + full_result=full, + source_read_policy_violation=violation, + introduced_failures=introduced, + selected=selected, + runtime_metrics=dict(agent_result.runtime_metrics), + ) + ) + if selected: + apply_selected_files( + attempt_root, + self.project_root, + editable_paths=self.editable_paths, + ) + result = self._result( + status="passed", + test_command=test_command, + baseline=baseline_report, + final=current_report, + attempts=attempts, + started=started, + ) + _write_result(json_out, result) + return result + previous_evidence = _previous_attempt_evidence( + attempt_patch=attempt_patch, + validation=current_report, ) - _write_result(json_out, result) - return result result = self._result( status="failed", @@ -248,6 +299,7 @@ def _focused_command(command: str, report: PytestRunReport) -> str: def _deterministic_candidates(root: Path, report: PytestRunReport) -> list[str]: + root = root.resolve() candidates: list[str] = [] for failure in report.failures: for location in failure.source_locations: @@ -255,6 +307,12 @@ def _deterministic_candidates(root: Path, report: PytestRunReport) -> list[str]: node_path = failure.node_id.split("::", 1)[0] _append_repo_path(root, node_path, candidates) test_name = Path(node_path).name + if not test_name.startswith("test_"): + for location in failure.source_locations: + location_name = Path(location.path).name + if location_name.startswith("test_"): + test_name = location_name + break if test_name.startswith("test_"): target_name = test_name.removeprefix("test_") for path in sorted(root.rglob(target_name)): @@ -338,3 +396,56 @@ def _write_result(path: str | Path | None, result: PytestFixResult) -> None: output = Path(path) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result.to_dict(), ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +def _default_editable_paths(root: Path) -> list[str]: + ignored_parts = {".git", ".firstcoder", ".venv", "__pycache__", ".pytest_cache", "tests", "test"} + return [ + path.relative_to(root).as_posix() + for path in sorted(root.rglob("*")) + if path.is_file() + and path.suffix == ".py" + and not ignored_parts.intersection(path.relative_to(root).parts) + ] + + +def _preserve_transcript( + transcript_path: str | Path | None, + *, + project_root: Path, + attempt_number: int, +) -> Path | None: + if transcript_path is None: + return None + source = Path(transcript_path) + if not source.is_file(): + return source + try: + source.relative_to(project_root) + return source + except ValueError: + destination = project_root / ".firstcoder" / "pytest-fix-transcripts" / f"attempt-{attempt_number}.jsonl" + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(source.read_bytes()) + return destination + + +def _introduced_failures(baseline: PytestRunReport, validation: PytestRunReport) -> list[str]: + baseline_fingerprints = {failure.fingerprint for failure in baseline.failures} + return [ + failure.fingerprint + for failure in validation.failures + if failure.fingerprint and failure.fingerprint not in baseline_fingerprints + ] + + +def _previous_attempt_evidence(*, attempt_patch: str, validation: PytestRunReport) -> str: + bounded_patch = attempt_patch[-12_000:] + bounded_output = validation.bounded_raw_output[-8_000:] + return ( + "\n\nPrevious attempt evidence (the current worktree was reset to the original baseline; " + "the patch below is evidence only and is not implicitly applied):\n" + f"Patch:\n{bounded_patch or ''}\n" + f"Validation exit code: {validation.exit_code}\n" + f"Validation output:\n{bounded_output}\n" + ) diff --git a/tests/test_cli.py b/tests/test_cli.py index e3b352f..74f9f30 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -411,6 +411,8 @@ def test_parser_supports_index_and_pytest_fix_subcommands() -> None: assert fix.command == "pytest-fix" assert fix.failure_log == "ci.log" assert fix.max_attempts == 2 + assert fix.execution_backend == "docker" + assert fix.docker_image == "firstcoder-pytest-sandbox:py311" def test_main_routes_pytest_fix_without_starting_chat_provider(monkeypatch, capsys) -> None: diff --git a/tests/test_execution_backends.py b/tests/test_execution_backends.py new file mode 100644 index 0000000..d8634b5 --- /dev/null +++ b/tests/test_execution_backends.py @@ -0,0 +1,78 @@ +import sys +from pathlib import Path + +import firstcoder.execution as execution +from firstcoder.execution import DockerSandboxBackend, LocalProcessBackend, ResourceLimits +from firstcoder.utils.subprocess import CommandResult + + +def test_local_backend_filters_environment_and_bounds_output(tmp_path: Path) -> None: + backend = LocalProcessBackend( + environment={"LANG": "C", "API_TOKEN": "secret", "UNLISTED": "value"}, + env_allowlist=("LANG",), + ) + result = backend.run( + [ + sys.executable, + "-c", + "import os; print(os.getenv('LANG')); print(os.getenv('API_TOKEN')); print('x' * 100)", + ], + tmp_path, + ResourceLimits(timeout_seconds=5, max_output_chars=20), + ) + + assert result.exit_code == 0 + assert result.stdout.startswith("C\nNone\n") + assert len(result.stdout) == 20 + assert result.stdout_truncated is True + + +def test_docker_backend_applies_isolation_and_resource_limits(monkeypatch, tmp_path: Path) -> None: + seen = {} + + def fake_run(command, *, cwd, env, limits, local_resource_limits): + seen.update(command=command, cwd=cwd, env=env, limits=limits, local=local_resource_limits) + return CommandResult(0, "ok", "", False, False, True) + + monkeypatch.setattr(execution, "_run_bounded", fake_run) + backend = DockerSandboxBackend( + image="example/pytest:locked", + environment={"LANG": "C", "API_TOKEN": "secret"}, + ) + limits = ResourceLimits( + timeout_seconds=12, + cpu_count=1.5, + memory_mb=768, + pids=32, + max_file_size_mb=8, + tmpfs_mb=16, + ) + + result = backend.run(["python", "-m", "pytest", "-q"], tmp_path, limits) + + command = seen["command"] + assert result.ok is True + assert "--network=none" in command + assert "--read-only" in command + assert "--cap-drop=ALL" in command + assert "--security-opt=no-new-privileges" in command + assert "--cpus=1.5" in command + assert "--memory=768m" in command + assert "--pids-limit=32" in command + assert "--user=65532:65532" in command + assert "--ulimit=fsize=8192:8192" in command + assert "--tmpfs=/tmp:rw,noexec,nosuid,nodev,size=16m" in command + assert f"type=bind,src={tmp_path.resolve()},dst=/workspace,rw" in command + assert "LANG=C" in command + assert all("secret" not in part for part in command) + assert command[-5:] == ["example/pytest:locked", "python", "-m", "pytest", "-q"] + assert seen["local"] is False + + +def test_resource_limits_reject_non_positive_values() -> None: + try: + ResourceLimits(memory_mb=0) + except ValueError as exc: + assert "memory_mb" in str(exc) + else: + raise AssertionError("invalid resource limits must be rejected") diff --git a/tests/test_fresh_source_guard.py b/tests/test_fresh_source_guard.py new file mode 100644 index 0000000..3905f8a --- /dev/null +++ b/tests/test_fresh_source_guard.py @@ -0,0 +1,130 @@ +from pathlib import Path + +from firstcoder.tools.builtin import create_builtin_registry +from firstcoder.tools.fresh_source import FreshSourceGuard + + +def _guarded_tools(root: Path, editable_paths: list[str]): + guard = FreshSourceGuard(root, editable_paths=editable_paths) + registry = create_builtin_registry( + root, + include_mutation_tools=True, + fresh_source_guard=guard, + ) + return {tool.name: tool for tool in registry.tools()} + + +def test_edit_requires_token_for_the_same_path(tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("A = 1\n", encoding="utf-8") + (tmp_path / "b.py").write_text("B = 1\n", encoding="utf-8") + tools = _guarded_tools(tmp_path, ["a.py", "b.py"]) + + revision = tools["view"].executor(path="a.py").data["source_revision"] + missing = tools["edit"].executor(path="b.py", old="B = 1", new="B = 2") + wrong = tools["edit"].executor( + path="b.py", + old="B = 1", + new="B = 2", + read_token=revision["token"], + ) + + assert missing.ok is False + assert "read_token" in missing.error + assert wrong.ok is False + assert "belongs to a.py" in wrong.error + assert (tmp_path / "b.py").read_text(encoding="utf-8") == "B = 1\n" + + +def test_edit_rejects_stale_hash_and_does_not_mutate(tmp_path: Path) -> None: + target = tmp_path / "a.py" + target.write_text("A = 1\n", encoding="utf-8") + tools = _guarded_tools(tmp_path, ["a.py"]) + revision = tools["view"].executor(path="a.py").data["source_revision"] + target.write_text("A = 3\n", encoding="utf-8") + + result = tools["edit"].executor( + path="a.py", + old="A = 3", + new="A = 2", + read_token=revision["token"], + ) + + assert result.ok is False + assert "source changed after read" in result.error + assert target.read_text(encoding="utf-8") == "A = 3\n" + + +def test_successful_edit_consumes_token(tmp_path: Path) -> None: + target = tmp_path / "a.py" + target.write_text("A = 1\n", encoding="utf-8") + tools = _guarded_tools(tmp_path, ["a.py"]) + revision = tools["view"].executor(path="a.py").data["source_revision"] + + first = tools["edit"].executor( + path="a.py", + old="A = 1", + new="A = 2", + read_token=revision["token"], + ) + replay = tools["edit"].executor( + path="a.py", + old="A = 2", + new="A = 3", + read_token=revision["token"], + ) + + assert first.ok is True + assert replay.ok is False + assert "unknown read_token" in replay.error + + +def test_new_file_must_be_inside_editable_paths(tmp_path: Path) -> None: + tools = _guarded_tools(tmp_path, ["src/new.py"]) + + denied = tools["write"].executor(path="other.py", content="x = 1\n") + allowed = tools["write"].executor(path="src/new.py", content="x = 1\n") + + assert denied.ok is False + assert "outside editable_paths" in denied.error + assert allowed.ok is True + + +def test_read_multi_returns_per_file_revision_tokens(tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("A = 1\n", encoding="utf-8") + (tmp_path / "b.py").write_text("B = 1\n", encoding="utf-8") + tools = _guarded_tools(tmp_path, ["a.py", "b.py"]) + + result = tools["read_multi"].executor(paths=["a.py", "b.py"]) + + revisions = [item["source_revision"] for item in result.data["files"]] + assert [revision["path"] for revision in revisions] == ["a.py", "b.py"] + assert all(revision["token"] in result.content for revision in revisions) + + +def test_apply_patch_requires_token_for_each_existing_path(tmp_path: Path) -> None: + (tmp_path / "a.py").write_text("A = 1\n", encoding="utf-8") + (tmp_path / "b.py").write_text("B = 1\n", encoding="utf-8") + tools = _guarded_tools(tmp_path, ["a.py", "b.py"]) + revision = tools["view"].executor(path="a.py").data["source_revision"] + patch = ( + "*** Begin Patch\n" + "*** Update File: a.py\n" + "@@\n" + "-A = 1\n" + "+A = 2\n" + "*** Update File: b.py\n" + "@@\n" + "-B = 1\n" + "+B = 2\n" + "*** End Patch" + ) + + denied = tools["apply_patch"].executor( + patch=patch, + read_tokens={"a.py": revision["token"]}, + ) + + assert denied.ok is False + assert "read_token" in denied.error + assert (tmp_path / "a.py").read_text(encoding="utf-8") == "A = 1\n" + assert (tmp_path / "b.py").read_text(encoding="utf-8") == "B = 1\n" diff --git a/tests/test_pytest_fix_workflow.py b/tests/test_pytest_fix_workflow.py index 751f510..132fda1 100644 --- a/tests/test_pytest_fix_workflow.py +++ b/tests/test_pytest_fix_workflow.py @@ -1,4 +1,5 @@ import json +import re import subprocess from pathlib import Path @@ -67,6 +68,7 @@ def complete(self, request: ChatRequest) -> ChatResponse: tool_calls=[ToolCall(id="read", name="view", arguments={"path": "src/value.py"})], ) if self.calls == 2: + token = re.search(r"read_token=([A-Za-z0-9_-]+)", request.messages[-1].content).group(1) return ChatResponse( provider=self.name, model=self.model, @@ -76,7 +78,12 @@ def complete(self, request: ChatRequest) -> ChatResponse: ToolCall( id="edit", name="edit", - arguments={"path": "src/value.py", "old": "VALUE = 1", "new": "VALUE = 2"}, + arguments={ + "path": "src/value.py", + "old": "VALUE = 1", + "new": "VALUE = 2", + "read_token": token, + }, ) ], ) @@ -111,6 +118,10 @@ def _noop(root: Path) -> None: return None +def _poison(root: Path) -> None: + (root / "src" / "value.py").write_text("VALUE = 99\n", encoding="utf-8") + + def _write_read_then_mutation_transcript(path: Path, *, read_first: bool = True) -> None: read = { "type": "tool_result", @@ -168,6 +179,58 @@ def test_second_attempt_can_succeed(tmp_path: Path) -> None: assert result.attempts[1].full_result is not None and result.attempts[1].full_result.passed +def test_second_attempt_starts_from_baseline_not_failed_first_patch(tmp_path: Path) -> None: + _write_project(tmp_path) + + def assert_baseline_then_fix(root: Path) -> None: + assert (root / "src" / "value.py").read_text(encoding="utf-8") == "VALUE = 1\n" + _fix(root) + + result = PytestFixWorkflow( + tmp_path, + adapter=ScriptedAdapter([_poison, assert_baseline_then_fix]), + max_attempts=2, + ).run(test_command="python -m pytest -q") + + assert result.status == "passed" + assert "VALUE = 99" in result.attempts[0].attempt_patch + assert "VALUE = 99" not in result.attempts[1].attempt_patch + assert result.attempts[0].selected is False + assert result.attempts[1].selected is True + assert result.attempts[0].base_commit == result.attempts[1].base_commit + assert (tmp_path / "src" / "value.py").read_text(encoding="utf-8") == "VALUE = 2\n" + + +def test_failed_attempts_leave_original_repository_unchanged(tmp_path: Path) -> None: + _write_project(tmp_path) + + result = PytestFixWorkflow( + tmp_path, + adapter=ScriptedAdapter([_poison, _poison]), + max_attempts=2, + ).run(test_command="python -m pytest -q") + + assert result.status == "failed" + assert all(attempt.selected is False for attempt in result.attempts) + assert (tmp_path / "src" / "value.py").read_text(encoding="utf-8") == "VALUE = 1\n" + assert result.final_diff == "" + + +def test_dirty_git_repository_is_rejected_before_attempts(tmp_path: Path) -> None: + _write_project(tmp_path) + _init_git(tmp_path) + (tmp_path / "src" / "value.py").write_text("VALUE = 7\n", encoding="utf-8") + + try: + PytestFixWorkflow(tmp_path, adapter=NeverAdapter(), max_attempts=1).run( + test_command="python -m pytest -q" + ) + except RuntimeError as exc: + assert "clean Git worktree" in str(exc) + else: + raise AssertionError("dirty user worktree must not be used as an attempt baseline") + + def test_attempt_limit_returns_two_when_tests_still_fail(tmp_path: Path) -> None: _write_project(tmp_path) result = PytestFixWorkflow(tmp_path, adapter=ScriptedAdapter([_noop, _noop]), max_attempts=2).run(