From 44b598e44f6bdcc4427056f8b7e638a73b8a1a9b Mon Sep 17 00:00:00 2001 From: Devan <68340659+DevanMetz@users.noreply.github.com> Date: Sat, 13 Jun 2026 14:26:19 -0500 Subject: [PATCH 1/2] Add auto-evolve and token efficiency controls --- .env.example | 7 + README.md | 14 + devbot/agent.py | 13 + devbot/autopilot.py | 27 +- devbot/cli.py | 14 + devbot/config.py | 5 + devbot/evolve.py | 300 +++++++++++++++++ devbot/evolve_limits.py | 85 +++++ devbot/evolve_planner.py | 164 +++++++++ devbot/gitcheckpoint.py | 169 ++++++++++ devbot/swarm.py | 33 +- devbot/tools.py | 38 ++- tests/test_autopilot.py | 10 +- tests/test_cli.py | 40 +++ tests/test_evolve.py | 483 +++++++++++++++++++++++++++ tests/test_evolve_limits.py | 322 ++++++++++++++++++ tests/test_evolve_planner.py | 332 ++++++++++++++++++ tests/test_gitcheckpoint.py | 298 +++++++++++++++++ tests/test_megaswarm_improvements.py | 16 +- tests/test_phase1.py | 22 ++ tests/test_phase5.py | 14 +- 21 files changed, 2378 insertions(+), 28 deletions(-) create mode 100644 devbot/evolve.py create mode 100644 devbot/evolve_limits.py create mode 100644 devbot/evolve_planner.py create mode 100644 devbot/gitcheckpoint.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_evolve.py create mode 100644 tests/test_evolve_limits.py create mode 100644 tests/test_evolve_planner.py create mode 100644 tests/test_gitcheckpoint.py diff --git a/.env.example b/.env.example index 3b16cea..e8dc3aa 100644 --- a/.env.example +++ b/.env.example @@ -10,3 +10,10 @@ DEEPSEEK_API_KEY=sk-your-key-here # Optional: default model. One of: deepseek-v4-flash (default), deepseek-v4-pro DEVBOT_MODEL=deepseek-v4-flash + +# Optional: long-run token efficiency knobs. +# DEVBOT_VERBOSITY=concise +# DEVBOT_MAX_TOOL_OUTPUT=12000 +# DEVBOT_READ_FILE_LIMIT=800 +# DEVBOT_DIFF_CLIP=1200 +# DEVBOT_SPECIALIST_RESULT_LIMIT=4000 diff --git a/README.md b/README.md index 9ea6604..12591c3 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ task is done. - **Cost estimation** — live USD cost estimate based on per-model pricing. - **Structured JSONL logging** — set `DEVBOT_LOG` to a file path for timestamped tool-call and turn logs. - **Global token budget** — process-wide cap (`DEVBOT_GLOBAL_BUDGET`) shared across all swarm agents. +- **Token-efficiency mode** — compact prompts, smaller tool returns, and clipped agent handoffs for long runs. - **Approval gates** — writes, shell commands, and test runs require confirmation (`y` / `a`lways / decline). - **Sandboxed file access** — tools cannot read or write outside the project root. - **Swarm mode** — a manager agent can delegate subtasks to specialist sub-agents. @@ -185,18 +186,28 @@ integrates the results. | `DEVBOT_TOKEN_BUDGET` | Per-agent token cap (0 = unlimited) | 0 | | `DEVBOT_GLOBAL_BUDGET` | Process-wide token cap (0 = unlimited) | 0 | | `DEVBOT_COMPRESS_MODEL` | Model used for context compression | `deepseek-v4-flash` | +| `DEVBOT_VERBOSITY` | Set to `concise`, `terse`, `compact`, or `caveman` for shorter agent replies | normal | +| `DEVBOT_MAX_TOOL_OUTPUT` | Max chars returned to the model per tool call | 50000 | +| `DEVBOT_READ_FILE_LIMIT` | Default line count for `read_file` when no limit is passed | 2000 | +| `DEVBOT_DIFF_CLIP` | Max chars of edit diff returned to the model | 2000 | +| `DEVBOT_SPECIALIST_RESULT_LIMIT` | Max chars returned from one specialist/pipeline handoff | 8000 | | `DEVBOT_MEGA_WARN_THRESHOLD` | Warn when N > threshold in megadelegate | 5 | | `DEVBOT_PIPELINE_ROUNDS` | Max review→fix rounds in pipeline | 2 | | `DEVBOT_SHOW_REASONING` | Show chain-of-thought (`1`, `true`, `yes`) | off | | `DEVBOT_ALLOW_SHELL` | Skip shell approval for allow-listed commands (`1`, `true`) | off | | `DEVBOT_LOG` | Path for JSONL structured log | off (no logging) | | `DEVBOT_LOOP_LIMIT` | Consecutive identical tool calls / errors before halting (0 = off) | 3 | +| `DEVBOT_EVOLVE_MAX_PHASES` | Max phases per auto-evolve run | 20 | +| `DEVBOT_EVOLVE_TIME_LIMIT` | Max minutes per auto-evolve run (0 = unlimited) | 0 | You can also put these in a `.env` file in your project root instead of exporting them: ``` DEEPSEEK_API_KEY=sk-... DEVBOT_MODEL=deepseek-v4-flash +DEVBOT_VERBOSITY=concise +DEVBOT_MAX_TOOL_OUTPUT=12000 +DEVBOT_SPECIALIST_RESULT_LIMIT=4000 ``` Real environment variables take precedence over `.env`. Add `.env` to your `.gitignore`. @@ -209,6 +220,9 @@ model = "deepseek-v4-pro" max_parallel = 4 token_budget = 100000 loop_limit = 5 +verbosity = "concise" +max_tool_output = 12000 +specialist_result_limit = 4000 ``` Precedence: **environment variables** > `.env` file > `.devbot/config.toml` > defaults. diff --git a/devbot/agent.py b/devbot/agent.py index 26ffdcf..b9c80f2 100644 --- a/devbot/agent.py +++ b/devbot/agent.py @@ -105,6 +105,16 @@ def _load_dotenv(root: Path): - Never run destructive commands (rm -rf, force push, etc.) without explaining why first. """ +CONCISE_ADDENDUM = """\ + +Token-efficiency mode is active. +- Keep replies short: answer first, then only essential details. +- Avoid preamble, repetition, hedging, and long status narration. +- Prefer compact bullets over paragraphs when reporting several facts. +- Do not paste long command output or file contents; cite the relevant file/line or summarize. +- For long tasks, keep a tiny working summary of decisions and next actions. +""" + MANAGER_ADDENDUM = """\ You are running as the MANAGER of an agent swarm. In addition to your own tools, @@ -222,6 +232,9 @@ def __init__(self, root: Path, model: str | None = None, auto_approve: bool = Fa prompt += MANAGER_ADDENDUM + MEGASWARM_ADDENDUM elif swarm: prompt += MANAGER_ADDENDUM + if os.environ.get("DEVBOT_VERBOSITY", "").lower() in ( + "concise", "terse", "compact", "caveman"): + prompt += CONCISE_ADDENDUM self.messages: list[dict] = [{"role": "system", "content": prompt}] # ---- UI hooks (overridden/used by cli.py) ------------------------------- diff --git a/devbot/autopilot.py b/devbot/autopilot.py index 1d3f921..d786341 100644 --- a/devbot/autopilot.py +++ b/devbot/autopilot.py @@ -40,6 +40,24 @@ def parse_phases(plan_text: str) -> list[dict]: return phases +def _plan_outline(phases: list[dict]) -> str: + """Compact phase list used instead of resending the entire plan each time.""" + return "\n".join(f"{i}. {ph['title']}" for i, ph in enumerate(phases, 1)) + + +def _phase_prompt(phases: list[dict], index: int) -> str: + """Build a compact implementation prompt for one phase.""" + phase = phases[index] + return ( + "You are implementing one phase of a multi-phase plan.\n\n" + f"=== PLAN OUTLINE ===\n{_plan_outline(phases)}\n\n" + f"=== IMPLEMENT ONLY PHASE {index + 1} NOW ===\n{phase['body']}\n\n" + "Use the `pipeline` tool for every code change (it enforces review). " + "Do not start other phases. When finished, make sure the test suite " + "passes." + ) + + def _run_tests(root: Path) -> tuple[bool, str]: """Run the project's pytest suite. Returns (passed, tail_of_output).""" try: @@ -90,14 +108,7 @@ def _agent() -> Agent: print(f"\n\x1b[36;1m{'='*70}\n[autopilot] PHASE {idx}/{len(phases)}: " f"{ph['title']}\n{'='*70}\x1b[0m") - _agent().run( - f"You are implementing one phase of a multi-phase plan.\n\n" - f"=== FULL PLAN (for context) ===\n{plan_text}\n\n" - f"=== IMPLEMENT ONLY THIS PHASE NOW ===\n{ph['body']}\n\n" - "Use the `pipeline` tool for every code change (it enforces review). " - "Do not start other phases. When finished, make sure the test suite " - "passes." - ) + _agent().run(_phase_prompt(phases, idx - 1)) ok, out = _run_tests(root) if not ok: diff --git a/devbot/cli.py b/devbot/cli.py index 82b2953..3cddca2 100644 --- a/devbot/cli.py +++ b/devbot/cli.py @@ -124,6 +124,10 @@ def main(): help="Autopilot: implement each '## Phase' of PLAN (default " "plan.md) one at a time, verifying between phases. " "Runs unattended (implies auto-approve).") + parser.add_argument("--auto-evolve", action="store_true", + help="Self-evolving autopilot: generate plans, critique them, " + "implement surviving phases on a dedicated git branch. " + "Runs unattended (implies auto-approve and megaswarm).") parser.add_argument("--version", action="version", version=f"devbot {__version__}") args = parser.parse_args() @@ -137,6 +141,16 @@ def main(): ok = run_plan(root, args.run_plan, model=args.model) sys.exit(0 if ok else 1) + # Autopilot: self-evolving loop that plans, critiques, and implements unattended. + if args.auto_evolve: + from .evolve import run_evolve + print("\x1b[33m[devbot] Auto-evolve runs UNATTENDED on a dedicated git branch " + "with auto-approve and shell access. It will plan, critique, and " + "implement phases, committing each green one. NEVER runs on main/master. " + "Ctrl+C to stop.\x1b[0m") + ok = run_evolve(root, args.model) + sys.exit(0 if ok else 1) + # Handle --resume: restore from a saved session. if args.resume is not None: from .session import restore_agent, list_sessions diff --git a/devbot/config.py b/devbot/config.py index fc718d0..fc9a3bd 100644 --- a/devbot/config.py +++ b/devbot/config.py @@ -22,6 +22,11 @@ "global_budget": "DEVBOT_GLOBAL_BUDGET", "loop_limit": "DEVBOT_LOOP_LIMIT", "compress_model": "DEVBOT_COMPRESS_MODEL", + "verbosity": "DEVBOT_VERBOSITY", + "max_tool_output": "DEVBOT_MAX_TOOL_OUTPUT", + "read_file_limit": "DEVBOT_READ_FILE_LIMIT", + "diff_clip": "DEVBOT_DIFF_CLIP", + "specialist_result_limit": "DEVBOT_SPECIALIST_RESULT_LIMIT", "mega_warn_threshold": "DEVBOT_MEGA_WARN_THRESHOLD", "pipeline_rounds": "DEVBOT_PIPELINE_ROUNDS", } diff --git a/devbot/evolve.py b/devbot/evolve.py new file mode 100644 index 0000000..78942ef --- /dev/null +++ b/devbot/evolve.py @@ -0,0 +1,300 @@ +"""Auto-evolve driver: self-evolving loop that plans, critiques, implements, +and commits improvement phases unattended. + +Orchestrates a loop where an LLM planner proposes phases, a critic filters +them, and implementation agents build each surviving phase — verifying with +the test suite and committing on green. +""" + +from __future__ import annotations + +import datetime +import os +import subprocess +import sys +from pathlib import Path + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _capture_tree(root: Path) -> str: + """Return a file listing of the repo. + + Tries ``git ls-tree -r --name-only HEAD`` first; falls back to a + simple ``Path.rglob`` listing (excluding ``.git`` and hidden dirs). + """ + try: + r = subprocess.run( + ["git", "ls-tree", "-r", "--name-only", "HEAD"], + cwd=root, + capture_output=True, + text=True, + timeout=15, + ) + if r.returncode == 0 and r.stdout.strip(): + return r.stdout.strip() + except Exception: + pass + + # Fallback: collect relative paths, skip .git and dunder dirs. + lines: list[str] = [] + try: + for p in sorted(root.rglob("*")): + if p.is_dir(): + continue + rel = p.relative_to(root) + parts = rel.parts + if any(part == ".git" or part.startswith("__pycache__") + for part in parts): + continue + lines.append(str(rel)) + except Exception: + return "" + return "\n".join(lines) + + +def _read_readme(root: Path) -> str: + """Return the first 200 lines of README.md, or an empty string.""" + readme = root / "README.md" + if not readme.is_file(): + return "" + try: + text = readme.read_text(encoding="utf-8", errors="replace") + lines = text.splitlines()[:200] + return "\n".join(lines) + except Exception: + return "" + + +# --------------------------------------------------------------------------- +# Main driver +# --------------------------------------------------------------------------- + +def run_evolve(root: Path, model: str | None = None) -> bool: + """Run the auto-evolve loop. + + Parameters + ---------- + root : Path + Project root (must be inside a git repository). + model : str | None + Model id to use for all agents (default: ``DEVBOT_MODEL`` env or + the built-in default). + + Returns + ------- + bool + ``True`` if at least one phase was completed and committed, + ``False`` otherwise. + """ + # -- imports (lazy at function level to avoid circular imports at load) -- + from devbot.gitcheckpoint import ( + is_clean, + current_branch, + ensure_branch, + commit_all, + ) + from devbot.evolve_limits import StopController + from devbot.evolve_planner import generate_plan, critique_plan + from devbot.agent import Agent, get_global_token_count + from devbot.autopilot import _run_tests + + # Pre-flight: load .env so the API-key check works for users who keep + # their key in the repo's .env file rather than in the shell environment. + from devbot.agent import _load_dotenv + _load_dotenv(root) + if not os.environ.get("DEEPSEEK_API_KEY"): + print("[auto-evolve] DEEPSEEK_API_KEY is not set. Please set it and " + "try again.") + return False + + root_str = str(root) + + # ---- 1. Pre-flight guards ------------------------------------------------ + try: + if not is_clean(root_str): + print("[auto-evolve] Working tree is NOT clean. Please commit or " + "stash changes before running evolve.") + return False + except Exception as exc: + print(f"[auto-evolve] git error checking clean status: {exc}") + return False + + try: + branch = current_branch(root_str) + except Exception as exc: + print(f"[auto-evolve] git error getting current branch: {exc}") + return False + + branch_name: str = branch + if branch in ("main", "master"): + ts = datetime.datetime.now().strftime("%Y-%m-%d-%H%M") + branch_name = f"autopilot/{ts}" + try: + ensure_branch(root_str, branch_name) + print(f"[auto-evolve] Created/checked out branch: {branch_name}") + except Exception as exc: + print(f"[auto-evolve] Failed to create branch '{branch_name}': {exc}") + return False + + # ---- 2. Setup ------------------------------------------------------------ + sc = StopController() + sc.start() + + commit_shas: list[tuple[str, str]] = [] + total_phases_completed = 0 + cost_estimate = 0.0 + + # ---- 3. Main loop -------------------------------------------------------- + stopped = False + reason = "" + + while True: + # Check stop conditions before generating a new plan. + stopped, reason = sc.should_stop() + if stopped: + break + + # Build context for the planner. + context = { + "tree": _capture_tree(root), + "readme": _read_readme(root), + } + + # --- Generate plan --- + try: + manager = Agent(root=root, model=model, auto_approve=True, + megaswarm=True) + proposed = generate_plan(manager, context) + cost_estimate += manager.estimated_cost() + except (Exception, SystemExit) as exc: + print(f"[auto-evolve] Error generating plan: {exc}") + continue + + if not proposed: + print("[auto-evolve] Planner returned no phases; stopping.") + break + + print(f"[auto-evolve] Planner proposed {len(proposed)} phase(s)") + + # --- Critique plan --- + try: + critic = Agent(root=root, model=model, auto_approve=True, + megaswarm=True) + surviving = critique_plan(critic, proposed) + cost_estimate += critic.estimated_cost() + except (Exception, SystemExit) as exc: + print(f"[auto-evolve] Error during critique: {exc}") + continue + + if not surviving: + print("[auto-evolve] Critic rejected all phases; generating next plan.") + continue + + print(f"[auto-evolve] Critic accepted {len(surviving)} phase(s)") + + # --- Implement each surviving phase --- + for phase in surviving: + # Check stop before each phase. + stopped, reason = sc.should_stop() + if stopped: + break + + title = phase["title"] + body = phase["body"] + print(f"\n[auto-evolve] Implementing phase: {title}") + + try: + impl_agent = Agent(root=root, model=model, auto_approve=True, + megaswarm=True) + prompt = ( + f"You are implementing one phase of a multi-phase plan.\n\n" + f"=== PHASE ===\n# {title}\n{body}\n\n" + "Use the `pipeline` tool for every code change (it enforces " + "review). When finished, make sure the test suite passes." + ) + impl_agent.run(prompt) + cost_estimate += impl_agent.estimated_cost() + except (Exception, SystemExit) as exc: + print(f"[auto-evolve] Error during implementation of " + f"'{title}': {exc}") + sc.set_red() + stopped = True + reason = f"Implementation error in '{title}': {exc}" + break + + # Run tests. + try: + ok, out = _run_tests(root) + except Exception as exc: + print(f"[auto-evolve] Error running tests after '{title}': {exc}") + ok, out = False, str(exc) + + if not ok: + # One fix round. + print(f"[auto-evolve] Tests failing after '{title}' — " + f"one fix round") + try: + fix_agent = Agent(root=root, model=model, auto_approve=True, + megaswarm=True) + fix_agent.run( + f"After implementing this phase, the test suite is " + f"FAILING:\n\n{body}\n\n=== PYTEST OUTPUT ===\n{out}\n\n" + "Diagnose and fix the failure using the `pipeline` tool, " + "then make sure `pytest -q` passes." + ) + cost_estimate += fix_agent.estimated_cost() + except (Exception, SystemExit) as exc: + print(f"[auto-evolve] Error during fix round for " + f"'{title}': {exc}") + + try: + ok, out = _run_tests(root) + except Exception as exc: + ok, out = False, str(exc) + + if not ok: + # Stop-on-red. + print(f"[auto-evolve] STOPPING: '{title}' still failing after " + f"fix round.") + print(out[-2000:]) + sc.set_red() + stopped = True + reason = f"Tests still failing after '{title}'" + break + + # Green — commit to the branch. + print(f"[auto-evolve] '{title}' complete — tests green") + try: + sha = commit_all(root_str, f"evolve: {title}") + if sha: + commit_shas.append((title, sha)) + print(f"[auto-evolve] committed {sha}") + else: + print(f"[auto-evolve] (nothing to commit)") + except Exception as exc: + print(f"[auto-evolve] Commit failed for '{title}': {exc}") + # Don't stop the whole run for a commit failure — just note it. + + sc.record_phase() + total_phases_completed += 1 + + # After all phases in this plan, if stop was hit, break the outer loop. + if stopped: + break + + # ---- 4. Summary ---------------------------------------------------------- + print("\n=== AUTO-EVOLVE SUMMARY ===") + print(f"Branch: {branch_name}") + print(f"Phases completed: {total_phases_completed}") + for title, sha in commit_shas: + print(f" {sha} {title}") + + final_reason = reason if (stopped and reason) else (reason or "completed all plans") + print(f"Stop reason: {final_reason}") + print(f"Total tokens: {get_global_token_count():,}") + print(f"Estimated cost: ${cost_estimate:.2f}") + + return total_phases_completed > 0 diff --git a/devbot/evolve_limits.py b/devbot/evolve_limits.py new file mode 100644 index 0000000..a9277d0 --- /dev/null +++ b/devbot/evolve_limits.py @@ -0,0 +1,85 @@ +"""Hard-stop controller for the auto-evolve loop. + +Centralises all stop conditions so the autopilot can check them in one +place each iteration. +""" + +import os +import time + +from devbot.agent import check_global_budget_exceeded, get_global_token_count + + +class StopController: + """Tracks all stop conditions for the auto-evolve loop.""" + + def __init__(self) -> None: + self._max_phases = int(os.environ.get("DEVBOT_EVOLVE_MAX_PHASES", "20")) + self._time_limit_minutes = int(os.environ.get("DEVBOT_EVOLVE_TIME_LIMIT", "0")) + self._time_limit_seconds = self._time_limit_minutes * 60 + self._start_time: float | None = None + self._phase_count = 0 + self._red = False # stop-on-red flag + + # -- control methods -------------------------------------------------------- + + def start(self) -> None: + """Record the start time for the wall-clock deadline.""" + self._start_time = time.monotonic() + + def record_phase(self) -> None: + """Increment the phase counter.""" + self._phase_count += 1 + + def set_red(self) -> None: + """Set the stop-on-red flag (e.g. from a failed pipeline phase).""" + self._red = True + + # -- properties ------------------------------------------------------------- + + @property + def phase_count(self) -> int: + return self._phase_count + + @property + def max_phases(self) -> int: + return self._max_phases + + @property + def time_limit_minutes(self) -> int: + return self._time_limit_minutes + + # -- stop logic ------------------------------------------------------------- + + def should_stop(self) -> tuple[bool, str]: + """Return (True, reason) if any stop condition is met. + + Checks in order: + + 1. Stop-on-red (``set_red`` was called) + 2. Global token budget exhausted (via ``agent.check_global_budget_exceeded``) + 3. Max phases reached + 4. Time limit exceeded + """ + # 1. Stop-on-red + if self._red: + return True, "Stop-on-red: a phase failed or was rejected" + + # 2. Global token budget + if check_global_budget_exceeded(): + budget = int(os.environ.get("DEVBOT_GLOBAL_BUDGET", "0")) + count = get_global_token_count() + return True, f"Global token budget exhausted ({count:,} >= {budget:,})" + + # 3. Max phases + if self._max_phases > 0 and self._phase_count >= self._max_phases: + return True, f"Max phases reached ({self._phase_count}/{self._max_phases})" + + # 4. Time limit + if self._start_time is not None and self._time_limit_seconds > 0: + elapsed = time.monotonic() - self._start_time + if elapsed >= self._time_limit_seconds: + mins = self._time_limit_minutes + return True, f"Time limit exceeded ({mins} minute{'s' if mins != 1 else ''})" + + return False, "" diff --git a/devbot/evolve_planner.py b/devbot/evolve_planner.py new file mode 100644 index 0000000..638cb80 --- /dev/null +++ b/devbot/evolve_planner.py @@ -0,0 +1,164 @@ +"""Planner + critic for the auto-evolve loop. + +Generates and critiques multi-phase implementation plans using a manager +agent (with swarm/megaswarm tools). Both functions are designed to be +robust — they return [] rather than crashing on malformed model output. +""" + +from __future__ import annotations + +import re + +from .autopilot import parse_phases + +# Regex to extract a justification line from a phase body. +# Matches: "Justification: ..." at the start of a line (case-insensitive, +# optional whitespace after colon). +_JUSTIFICATION_RE = re.compile(r"^Justification:\s*(.*)$", re.MULTILINE | re.IGNORECASE) + + +def generate_plan(manager, context: dict) -> list[dict]: + """Ask the manager to propose up to 5 improvement phases for the repo. + + *manager* is an Agent instance (with swarm/megaswarm tools). + *context* is a dict with optional string keys such as ``'outline'``, + ``'tree'``, ``'readme'``, ``'summary'`` — each describing the current + repo state. + + Returns a list of phase dicts (``{'title': ..., 'body': ...}``) parsed + via ``autopilot.parse_phases``, or ``[]`` if the model returned no + valid phases or an error occurred. + """ + # Build a prompt that includes every available context snippet. + context_blocks: list[str] = [] + for key, value in context.items(): + if value: # skip empty/None values + context_blocks.append(f"=== {key} ===\n{value}") + context_section = "\n\n".join(context_blocks) + + prompt = f"""\ +You are a planning specialist. Analyse the current state of the repo +described below and propose up to 5 phases of improvements. + +{context_section} + +=== INSTRUCTIONS === +1. Propose AT MOST 5 phases. Fewer is fine; quality over quantity. +2. Each phase MUST improve correctness, usefulness, or safety of the + project. Do NOT propose: + - Pure refactors (no functional change) + - Speculative features (vague, no clear benefit) + - Cosmetic changes (whitespace, formatting, naming) +3. Output each phase as a level-2 heading followed by a clear description + of what to implement and why it matters. Use EXACTLY this format: + +## Phase N — Short Title +Detailed description of the change. Include: +- What file(s) to touch +- What the change should accomplish +- Why this improves correctness / usefulness / safety + +Stay focused and concrete. Do NOT include any preamble or commentary +outside the phase headings — your entire output must be parseable.""" + + try: + raw = manager.run(prompt) + if raw is None: + raw = "" + phases = parse_phases(raw) + return phases + except Exception: + return [] + + +def critique_plan(manager, phases: list[dict]) -> list[dict]: + """Critique a list of proposed phases and drop ones that shouldn't run. + + *manager* is an Agent instance. + *phases* is a list of phase dicts (``{'title': ..., 'body': ...}``). + + Returns a (possibly shorter) list of phase dicts, now with an extra + ``'justification'`` key, containing only the phases that pass the + critic's bar. Returns ``[]`` on malformed output or if all phases are + rejected. + """ + if not phases: + return [] + + # Build the list of phases for the prompt. + phase_text_blocks: list[str] = [] + for i, ph in enumerate(phases, 1): + phase_text_blocks.append( + f"### Proposed Phase {i}\n" + f"**Title:** {ph['title']}\n" + f"**Body:** {ph['body']}" + ) + phase_text = "\n\n".join(phase_text_blocks) + + prompt = f"""\ +You are a strict code-review critic. Below are {len(phases)} proposed +improvement phases for a software project. For each phase, score it (1-10) +and decide whether to KEEP or DROP it. + +=== CRITERIA === +DROP any phase that is: +- A pure refactor (no functional change) +- A speculative feature (vague, no clear benefit) +- Not clearly improving correctness, usefulness, or safety +- Cosmetic only (formatting, naming, whitespace) + +KEEP phases that concretely improve the project. + +=== REQUIRED OUTPUT FORMAT === +For each phase you KEEP, output: + +## Phase N — Title +Justification: A brief explanation of why this phase is kept and what score (1-10) it earned. +[original body text] + +Output ONLY the surviving phases — do NOT include dropped phases at all. +If you drop ALL phases, output the single word: NONE + +=== PROPOSED PHASES === +{phase_text}""" + + try: + raw = manager.run(prompt) + if raw is None: + raw = "" + except Exception: + return [] + + raw_stripped = raw.strip() + + # If the model says NONE (or returns empty), there are no survivors. + if not raw_stripped or raw_stripped.upper() == "NONE": + return [] + + # Parse the surviving phases using parse_phases, then extract + # justifications from each body. + try: + surviving = parse_phases(raw) + except Exception: + return [] + + result: list[dict] = [] + for ph in surviving: + body = ph["body"] + just_match = _JUSTIFICATION_RE.search(body) + if just_match: + justification = just_match.group(1).strip() + # Remove the justification line(s) from the body to keep it clean. + # Also remove any blank lines left behind. + clean_body = _JUSTIFICATION_RE.sub("", body).strip() + else: + justification = "" + clean_body = body.strip() + + result.append({ + "title": ph["title"], + "body": clean_body, + "justification": justification, + }) + + return result diff --git a/devbot/gitcheckpoint.py b/devbot/gitcheckpoint.py new file mode 100644 index 0000000..f92d97d --- /dev/null +++ b/devbot/gitcheckpoint.py @@ -0,0 +1,169 @@ +"""Git checkpoint helpers for DevBot. + +Lightweight wrappers around ``git`` subprocess calls so the autopilot can +safely branch, commit, and query repository state without external +dependencies. +""" + +import subprocess +from typing import Optional + + +def current_branch(root: str) -> str: + """Return the current branch name of the git repo at *root*. + + Raises + ------ + RuntimeError + If the command fails (e.g. *root* is not inside a git repository). + """ + proc = subprocess.run( + ["git", "rev-parse", "--abbrev-ref", "HEAD"], + cwd=root, + capture_output=True, + text=True, + ) + if proc.returncode != 0: + raise RuntimeError( + f"git rev-parse failed in {root}: {proc.stderr.strip()}" + ) + return proc.stdout.strip() + + +def is_clean(root: str) -> bool: + """Return ``True`` if there are no uncommitted changes in *root*. + + Uses ``git status --porcelain``: an empty output means the working tree + is clean (no untracked, modified, or staged files). + """ + proc = subprocess.run( + ["git", "status", "--porcelain"], + cwd=root, + capture_output=True, + text=True, + ) + if proc.returncode != 0: + raise RuntimeError( + f"git status failed in {root}: {proc.stderr.strip()}" + ) + return proc.stdout.strip() == "" + + +def ensure_branch(root: str, name: str) -> None: + """Create and checkout branch *name*, or just checkout if it exists. + + Parameters + ---------- + root : str + Path to the git working tree. + name : str + Branch name. Must not be ``"main"`` or ``"master"``. + + Raises + ------ + ValueError + If *name* is ``"main"`` or ``"master"``. + RuntimeError + If any git command fails unexpectedly. + """ + if name in ("main", "master"): + raise ValueError( + f"Refusing to operate on branch '{name}'" + ) + + # Check whether the branch already exists (locale-robust). + proc_ref = subprocess.run( + ["git", "show-ref", "--verify", "-q", f"refs/heads/{name}"], + cwd=root, + capture_output=True, + text=True, + ) + branch_exists = proc_ref.returncode == 0 + + if branch_exists: + proc = subprocess.run( + ["git", "checkout", name], + cwd=root, + capture_output=True, + text=True, + ) + if proc.returncode != 0: + raise RuntimeError( + f"git checkout {name} failed in {root}: {proc.stderr.strip()}" + ) + else: + proc = subprocess.run( + ["git", "checkout", "-b", name], + cwd=root, + capture_output=True, + text=True, + ) + if proc.returncode != 0: + raise RuntimeError( + f"git checkout -b {name} failed in {root}: {proc.stderr.strip()}" + ) + + +def commit_all(root: str, message: str) -> Optional[str]: + """Stage all changes and commit with *message*. + + Parameters + ---------- + root : str + Path to the git working tree. + message : str + Commit message (may contain arbitrary characters). + + Returns + ------- + str | None + The short SHA of the new commit, or ``None`` if there was nothing + to commit (clean tree / no changes). + + Raises + ------ + RuntimeError + If an unexpected git error occurs. + """ + # Stage everything. + proc_add = subprocess.run( + ["git", "add", "-A"], + cwd=root, + capture_output=True, + text=True, + ) + if proc_add.returncode != 0: + raise RuntimeError( + f"git add failed in {root}: {proc_add.stderr.strip()}" + ) + + # Commit. + proc_commit = subprocess.run( + ["git", "commit", "-m", message], + cwd=root, + capture_output=True, + text=True, + ) + if proc_commit.returncode == 0: + # Success — get the short SHA. + proc_sha = subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + cwd=root, + capture_output=True, + text=True, + ) + if proc_sha.returncode != 0: + raise RuntimeError( + f"git rev-parse failed in {root}: {proc_sha.stderr.strip()}" + ) + return proc_sha.stdout.strip() + + # Detect "nothing to commit" — return None. + combined = (proc_commit.stdout + proc_commit.stderr).lower() + if "nothing to commit" in combined: + return None + + # Some other failure. + raise RuntimeError( + f"git commit failed in {root}: {proc_commit.stderr.strip()}" + ) diff --git a/devbot/swarm.py b/devbot/swarm.py index 74047d4..e125f60 100644 --- a/devbot/swarm.py +++ b/devbot/swarm.py @@ -140,7 +140,7 @@ def run_specialist(manager: "Agent", role: str, task: str) -> str: # Roll the sub-agent's token usage up into the manager's session totals. manager.total_tokens += sub.total_tokens manager.delegation_count += 1 - return answer or "(specialist returned no text)" + return _clip_agent_result(answer or "(specialist returned no text)", role) def run_specialist_with_prompt(manager: "Agent", role: str, system_prompt: str, @@ -171,7 +171,7 @@ def run_specialist_with_prompt(manager: "Agent", role: str, system_prompt: str, print(f"\n\x1b[35m╰─ [{lbl}] done in {time.time() - start:.1f}s\x1b[0m") manager.total_tokens += sub.total_tokens manager.delegation_count += 1 - return answer or "(specialist returned no text)" + return _clip_agent_result(answer or "(specialist returned no text)", lbl) # --------------------------------------------------------------------------- @@ -197,6 +197,22 @@ def run_specialist_with_prompt(manager: "Agent", role: str, system_prompt: str, ) +def _env_int(name: str, default: int, minimum: int = 0) -> int: + try: + value = int(os.environ.get(name, str(default)) or str(default)) + except ValueError: + return default + return max(minimum, value) + + +def _clip_agent_result(text: str, label: str = "agent result") -> str: + """Bound sub-agent text before feeding it back into another agent.""" + limit = _env_int("DEVBOT_SPECIALIST_RESULT_LIMIT", 8000, minimum=500) + if len(text) <= limit: + return text + return text[:limit] + f"\n... [{label} truncated, {len(text)} chars total]" + + # --------------------------------------------------------------------------- # ParallelMonitor — live dashboard for megaswarm parallel phase # --------------------------------------------------------------------------- @@ -639,7 +655,11 @@ def _run_one(_role: str, _sem: threading.Semaphore, monitor.update(_label, phase='failed') finally: _sem.release() - return _role, (answer or "(specialist returned no text)"), sub.total_tokens + return ( + _role, + _clip_agent_result(answer or "(specialist returned no text)", _label), + sub.total_tokens, + ) results: dict[str, str] = {} skipped = 0 @@ -748,7 +768,10 @@ def _run_one(_role: str, _sem: threading.Semaphore, print(f"\x1b[35m ╟─ [reviewer] done in {phase2_elapsed:.1f}s\x1b[0m") print(f"\x1b[35;1m╚═ MEGASWARM complete in {total_elapsed:.1f}s\x1b[0m") - return synthesis or "(megaswarm reviewer returned no text)" + return _clip_agent_result( + synthesis or "(megaswarm reviewer returned no text)", + "megaswarm synthesis", + ) # --------------------------------------------------------------------------- @@ -843,4 +866,4 @@ def run_pipeline(manager: "Agent", task: str) -> str: f"({_get_pipeline_rounds()}); some issues may remain\x1b[0m") print(f"\x1b[36;1m╚═ PIPELINE complete\x1b[0m") - return "\n\n".join(transcript) + return _clip_agent_result("\n\n".join(transcript), "pipeline transcript") diff --git a/devbot/tools.py b/devbot/tools.py index cbb245a..3578d5f 100644 --- a/devbot/tools.py +++ b/devbot/tools.py @@ -13,7 +13,8 @@ import time from pathlib import Path -MAX_OUTPUT = 50_000 # chars returned to the model per tool call +MAX_OUTPUT = 50_000 # default chars returned to the model per tool call +DEFAULT_READ_FILE_LIMIT = 2000 _BACKUPS_DIR = ".devbot/backups" _MAX_BACKUP_SETS = 20 @@ -205,13 +206,31 @@ def check_command(command: str) -> tuple: return ("needs_approval", "not on allow-list") +def _env_int(name: str, default: int, minimum: int = 0) -> int: + """Read a non-negative int from env, falling back on bad values.""" + try: + value = int(os.environ.get(name, str(default)) or str(default)) + except ValueError: + return default + return max(minimum, value) + + +def _max_output() -> int: + return _env_int("DEVBOT_MAX_TOOL_OUTPUT", MAX_OUTPUT, minimum=1000) + + +def _default_read_limit() -> int: + return _env_int("DEVBOT_READ_FILE_LIMIT", DEFAULT_READ_FILE_LIMIT, minimum=1) + + def _clip(text: str) -> str: - if len(text) >= MAX_OUTPUT: - return text[:MAX_OUTPUT] + f"\n... [truncated, {len(text)} chars total]" + max_output = _max_output() + if len(text) >= max_output: + return text[:max_output] + f"\n... [truncated, {len(text)} chars total]" return text -DIFF_CLIP = 2000 # chars of unified diff returned to the model +DIFF_CLIP = 2000 # default chars of unified diff returned to the model def _compute_diff(old_content: str, new_content: str, filename: str) -> str: @@ -230,8 +249,9 @@ def _compute_diff(old_content: str, new_content: str, filename: str) -> str: diff_text = "".join(diff_lines) full_block = f"```diff\n{diff_text}```" - if len(full_block) > DIFF_CLIP: - full_block = full_block[:DIFF_CLIP] + "\n... [diff truncated]" + diff_clip = _env_int("DEVBOT_DIFF_CLIP", DIFF_CLIP, minimum=200) + if len(full_block) > diff_clip: + full_block = full_block[:diff_clip] + "\n... [diff truncated]" return full_block @@ -255,13 +275,15 @@ def _resolve(path: str, root: Path) -> Path: return p -def read_file(path: str, root: Path, offset: int = 0, limit: int = 2000) -> str: +def read_file(path: str, root: Path, offset: int = 0, + limit: int | None = None) -> str: p = _resolve(path, root) if not p.is_file(): return f"Error: {p} is not a file" lines = p.read_text(encoding="utf-8", errors="replace").splitlines() total = len(lines) offset = max(0, offset) + limit = _default_read_limit() if limit is None else max(1, limit) if offset >= total: return f"Error: offset {offset} is beyond file end ({total} lines)" chunk = lines[offset : offset + limit] @@ -1022,7 +1044,7 @@ def outline(path: str, root: Path) -> str: # Null-safe arg.get(key) or default handles JSON null coercion (P1-11). _TOOL_HANDLERS = { "read_file": lambda a, r: read_file( - a["path"], r, a.get("offset") or 0, a.get("limit") or 2000), + a["path"], r, a.get("offset") or 0, a.get("limit")), "write_file": lambda a, r: write_file(a["path"], a["content"], r), "edit_file": lambda a, r: edit_file( a["path"], a["old_string"], a["new_string"], r, a.get("replace_all") or False), diff --git a/tests/test_autopilot.py b/tests/test_autopilot.py index a68b5f3..223f709 100644 --- a/tests/test_autopilot.py +++ b/tests/test_autopilot.py @@ -1,7 +1,7 @@ """Tests for the autopilot plan runner. No network calls (Agent is mocked).""" import devbot.autopilot as autopilot -from devbot.autopilot import parse_phases, run_plan +from devbot.autopilot import _phase_prompt, parse_phases, run_plan PLAN = """\ @@ -42,6 +42,14 @@ def test_body_includes_content_until_next_section(self): def test_no_phases_returns_empty(self): assert parse_phases("# Just a title\n\nsome prose") == [] + def test_phase_prompt_uses_outline_not_full_plan_body(self): + phases = parse_phases(PLAN) + prompt = _phase_prompt(phases, 0) + assert "=== PLAN OUTLINE ===" in prompt + assert "Phase 2: Second thing" in prompt + assert "Do the first thing." in prompt + assert "Do the second thing." not in prompt + # --------------------------------------------------------------------------- # run_plan control flow (mock Agent + _run_tests) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..e62be6c --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,40 @@ +"""CLI entry-point tests.""" + +from __future__ import annotations + +import sys + +import pytest + +import devbot.cli as cli + + +def test_auto_evolve_flag_calls_run_evolve(monkeypatch, tmp_path): + calls = [] + + def fake_run_evolve(root, model): + calls.append((root, model)) + return True + + monkeypatch.setattr("devbot.evolve.run_evolve", fake_run_evolve) + monkeypatch.setattr( + sys, + "argv", + ["devbot", "--auto-evolve", "-C", str(tmp_path), "-m", "deepseek-v4-pro"], + ) + + with pytest.raises(SystemExit) as exc: + cli.main() + + assert exc.value.code == 0 + assert calls == [(tmp_path.resolve(), "deepseek-v4-pro")] + + +def test_auto_evolve_flag_exits_nonzero_on_failure(monkeypatch, tmp_path): + monkeypatch.setattr("devbot.evolve.run_evolve", lambda root, model: False) + monkeypatch.setattr(sys, "argv", ["devbot", "--auto-evolve", "-C", str(tmp_path)]) + + with pytest.raises(SystemExit) as exc: + cli.main() + + assert exc.value.code == 1 diff --git a/tests/test_evolve.py b/tests/test_evolve.py new file mode 100644 index 0000000..0b6cb20 --- /dev/null +++ b/tests/test_evolve.py @@ -0,0 +1,483 @@ +"""Comprehensive unit tests for ``devbot.evolve.run_evolve``. + +All external calls (git, network, Agent, subprocess) are mocked so the +suite runs offline and deterministically. +""" + +from __future__ import annotations + +import os +import re +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +import devbot.evolve as evolve_mod +from devbot.evolve import run_evolve + + +# --------------------------------------------------------------------------- +# Fake helpers +# --------------------------------------------------------------------------- + +class _FakeAgent: + """A stand-in for ``devbot.agent.Agent`` that records ``.run()`` calls.""" + + def __init__(self, root=None, model=None, auto_approve=None, megaswarm=None): + # Store constructor args for assertions. + self.root = root + self.model = model + self.auto_approve = auto_approve + self.megaswarm = megaswarm + self.runs: list[str] = [] + + def run(self, prompt: str) -> str: + self.runs.append(prompt) + return "done" + + def estimated_cost(self) -> float: + return 0.01 + + +class _FakeStopController: + """Configurable stand-in for ``devbot.evolve_limits.StopController``.""" + + def __init__(self): + self.phase_count = 0 + self._stopped = False + self._red = False + # List of (stopped: bool, reason: str) to return from should_stop(). + self._should_stop_responses: list[tuple[bool, str]] = [] + self._call_count = 0 + self._started = False + + def start(self) -> None: + self._started = True + + def record_phase(self) -> None: + self.phase_count += 1 + + def set_red(self) -> None: + self._red = True + + def should_stop(self) -> tuple[bool, str]: + self._call_count += 1 + if self._should_stop_responses: + # Pop from front so each call consumes one response. + return self._should_stop_responses.pop(0) + # Default: never stop. + return (False, "") + + +# --------------------------------------------------------------------------- +# Arrange helper +# --------------------------------------------------------------------------- + +def _setup_mocks( + monkeypatch, + tmp_path: Path, + *, + is_clean: bool = True, + branch: str = "feature", + ensure_branch_side_effect=None, + commit_all_return: str | None = "abc1234", + should_stop_responses: list[tuple[bool, str]] | None = None, + generate_plan_phases: list[dict] | None = None, + critique_plan_phases: list[dict] | None = None, + run_tests_responses: list[tuple[bool, str]] | None = None, + fake_agent_class=None, +) -> dict: + """Wire all mocks for ``run_evolve`` and return the fakes for assertions. + + Parameters + ---------- + monkeypatch: + Pytest ``monkeypatch`` fixture. + tmp_path : Path + Temporary directory to use as the repo root (``root`` arg). + is_clean : bool + Return value for the mocked ``is_clean``. + branch : str + Return value for the mocked ``current_branch``. + ensure_branch_side_effect: + Optional side-effect / return for ``ensure_branch``. + commit_all_return : str | None + Return value for ``commit_all``. + should_stop_responses : list[tuple[bool, str]] | None + Responses for ``StopController.should_stop``. If *None*, defaults to + ``[(True, "Max phases reached")]`` (stops on first check). + generate_plan_phases : list[dict] | None + Phases returned by ``generate_plan``. Default: one phase. + critique_plan_phases : list[dict] | None + Phases returned by ``critique_plan``. Default: same as generate. + run_tests_responses : list[tuple[bool, str]] | None + Responses for ``_run_tests``. Default: ``[(True, "")]``. + fake_agent_class: + Class to use in place of ``Agent``. Default: ``_FakeAgent``. + + Returns + ------- + dict + Keys: ``fake_sc``, ``ensure_branch``, ``commit_all``, ``fake_agent_cls``, + ``run_tests``, ``generate_plan``, ``critique_plan``. + """ + root_str = str(tmp_path) + + # ---- API key ----------------------------------------------------------- + monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-test") + + # ---- Neutralise _load_dotenv ------------------------------------------- + monkeypatch.setattr("devbot.agent._load_dotenv", lambda root: None) + + # ---- Git helpers ------------------------------------------------------- + _mock_is_clean = MagicMock(return_value=is_clean) + monkeypatch.setattr("devbot.gitcheckpoint.is_clean", _mock_is_clean) + + _mock_current_branch = MagicMock(return_value=branch) + monkeypatch.setattr("devbot.gitcheckpoint.current_branch", _mock_current_branch) + + _mock_ensure_branch = MagicMock() + if ensure_branch_side_effect is not None: + _mock_ensure_branch.side_effect = ensure_branch_side_effect + monkeypatch.setattr("devbot.gitcheckpoint.ensure_branch", _mock_ensure_branch) + + _mock_commit_all = MagicMock(return_value=commit_all_return) + monkeypatch.setattr("devbot.gitcheckpoint.commit_all", _mock_commit_all) + + # ---- StopController ---------------------------------------------------- + _fake_sc = _FakeStopController() + if should_stop_responses is not None: + _fake_sc._should_stop_responses = list(should_stop_responses) + else: + _fake_sc._should_stop_responses = [(True, "Max phases reached")] + monkeypatch.setattr("devbot.evolve_limits.StopController", lambda: _fake_sc) + + # ---- Planner / critic -------------------------------------------------- + _default_phase = {"title": "Phase 1 — Add tests", "body": "Add unit tests."} + if generate_plan_phases is None: + generate_plan_phases = [_default_phase] + _mock_generate_plan = MagicMock(return_value=generate_plan_phases) + monkeypatch.setattr("devbot.evolve_planner.generate_plan", _mock_generate_plan) + + if critique_plan_phases is None: + critique_plan_phases = list(generate_plan_phases) + _mock_critique_plan = MagicMock(return_value=critique_plan_phases) + monkeypatch.setattr("devbot.evolve_planner.critique_plan", _mock_critique_plan) + + # ---- _run_tests -------------------------------------------------------- + if run_tests_responses is None: + run_tests_responses = [(True, "")] + _mock_run_tests = MagicMock(side_effect=list(run_tests_responses)) + monkeypatch.setattr("devbot.autopilot._run_tests", _mock_run_tests) + + # ---- Agent ------------------------------------------------------------- + _fake_cls = fake_agent_class or _FakeAgent + monkeypatch.setattr("devbot.agent.Agent", _fake_cls) + + # ---- get_global_token_count -------------------------------------------- + monkeypatch.setattr("devbot.agent.get_global_token_count", lambda: 0) + + return { + "fake_sc": _fake_sc, + "ensure_branch": _mock_ensure_branch, + "commit_all": _mock_commit_all, + "fake_agent_cls": _fake_cls, + "run_tests": _mock_run_tests, + "generate_plan": _mock_generate_plan, + "critique_plan": _mock_critique_plan, + } + + +# ============================================================================ +# Tests +# ============================================================================ + +class TestPreflightDirtyTree: + """1. Pre-flight: dirty tree returns False.""" + + def test_dirty_tree_returns_false_and_prints_error(self, monkeypatch, tmp_path): + # Arrange + _setup_mocks(monkeypatch, tmp_path, is_clean=False) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is False + + +class TestPreflightOnMainCreatesAutopilotBranch: + """2. Pre-flight: on main creates autopilot branch.""" + + def test_on_main_creates_autopilot_branch(self, monkeypatch, tmp_path): + # Arrange + fakes = _setup_mocks(monkeypatch, tmp_path, branch="main") + + # Act + run_evolve(tmp_path) + + # Assert + fakes["ensure_branch"].assert_called_once() + call_args = fakes["ensure_branch"].call_args[0] + branch_name = call_args[1] # second positional arg + assert re.match(r"autopilot/\d{4}-\d{2}-\d{2}-\d{4}", branch_name), ( + f"Expected autopilot/YYYY-MM-DD-HHMM, got {branch_name!r}" + ) + assert branch_name not in ("main", "master") + + +class TestPreflightApiKeyMissing: + """3. Pre-flight: API key missing returns False.""" + + def test_api_key_missing_returns_false(self, monkeypatch, tmp_path): + # Arrange + # Delete the API key that conftest.py may have set. + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is False + + +class TestMainLoopPerPhaseCommitsOnGreen: + """4. Main loop: per-phase commits on green.""" + + def test_per_phase_commits_on_green(self, monkeypatch, tmp_path): + # Arrange + phase = {"title": "Phase 1 — Add tests", "body": "Add unit tests."} + + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="feature", + # StopController checks: before plan gen (no stop), before phase (no stop), + # next loop iteration (stop). + should_stop_responses=[ + (False, ""), # 1st: top of while → continue + (False, ""), # 2nd: before phase in for-loop → continue + (True, "Max phases"), # 3rd: top of while after phase → stop + ], + generate_plan_phases=[phase], + critique_plan_phases=[phase], + run_tests_responses=[(True, "")], + commit_all_return="abc1234", + ) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is True + + # commit_all called exactly once with the expected message. + fakes["commit_all"].assert_called_once() + commit_msg = fakes["commit_all"].call_args[0][1] + assert commit_msg == f"evolve: {phase['title']}" + + # record_phase called once. + assert fakes["fake_sc"].phase_count == 1 + + +class TestStopOnRedTestsFailFixAlsoFails: + """5. Stop-on-red: tests fail, fix round also fails.""" + + def test_stop_on_red_no_commit(self, monkeypatch, tmp_path): + # Arrange + phase = {"title": "Phase 1 — Risky change", "body": "A risky change."} + fail_output = "some failure" + + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="feature", + should_stop_responses=[ + (False, ""), # top of while → continue + (False, ""), # before phase → continue + # After break from stop-on-red, we won't reach a 3rd check. + (True, "unused"), + ], + generate_plan_phases=[phase], + critique_plan_phases=[phase], + # Both test runs fail. + run_tests_responses=[(False, fail_output), (False, fail_output)], + ) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is False + # commit_all must NEVER be called. + fakes["commit_all"].assert_not_called() + # set_red must have been called. + assert fakes["fake_sc"]._red is True + + +class TestStopOnCapMaxPhasesViaStopController: + """6. Stop-on-cap: max phases via StopController before first phase.""" + + def test_max_phases_before_implementation(self, monkeypatch, tmp_path): + # Arrange + phase = {"title": "Phase 1 — Never run", "body": "Should not run."} + + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="feature", + should_stop_responses=[ + (False, ""), # 1st: top of while → continue (plan gen) + (True, "Max phases reached"), # 2nd: before phase → stop + ], + generate_plan_phases=[phase], + critique_plan_phases=[phase], + ) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is False + # No phases completed. + assert fakes["fake_sc"].phase_count == 0 + # commit_all never called. + fakes["commit_all"].assert_not_called() + + +class TestBranchIsolationNeverCommitsToMain: + """7. Branch isolation: never commits to main.""" + + def test_never_commits_to_main(self, monkeypatch, tmp_path): + # Arrange + phase = {"title": "Phase 1 — Safe change", "body": "A safe change."} + + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="main", + should_stop_responses=[ + (False, ""), # top of while → continue + (False, ""), # before phase → continue + (True, "Max phases"), # next loop → stop + ], + generate_plan_phases=[phase], + critique_plan_phases=[phase], + run_tests_responses=[(True, "")], + commit_all_return="abc1234", + ) + + # Act + run_evolve(tmp_path) + + # Assert + # ensure_branch called with an autopilot/ branch (not "main"). + fakes["ensure_branch"].assert_called_once() + branch_name = fakes["ensure_branch"].call_args[0][1] + assert branch_name.startswith("autopilot/") + assert branch_name not in ("main", "master") + + # commit_all IS called (since a phase ran) using root_str (tmp_path). + fakes["commit_all"].assert_called_once() + # The root arg passed to commit_all is the stringified tmp_path. + root_arg = fakes["commit_all"].call_args[0][0] + assert root_arg == str(tmp_path) + + +class TestEmptyPlanStopsGracefully: + """8. Empty plan from planner stops gracefully.""" + + def test_empty_plan_returns_false(self, monkeypatch, tmp_path): + # Arrange + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="feature", + # Only one should_stop call happens (top of while, before plan gen). + # We need it to NOT stop there, so the loop can generate the plan. + # Then after empty plan, the loop breaks. + should_stop_responses=[(False, "")], + generate_plan_phases=[], # <-- empty plan + ) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is False + # No phases were implemented. + assert fakes["fake_sc"].phase_count == 0 + # commit_all never called. + fakes["commit_all"].assert_not_called() + + +class TestCriticRejectsAllPhasesContinues: + """9. Critic rejects all phases; loop continues then stops.""" + + def test_critic_rejects_all_then_stops(self, monkeypatch, tmp_path): + # Arrange + phase1 = {"title": "Phase 1 — Bad idea", "body": "Refactor everything."} + phase2 = {"title": "Phase 2 — Also bad", "body": "Rename all vars."} + + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="feature", + should_stop_responses=[ + (False, ""), # 1st: top of while → continue + (True, "Max phases reached"), # 2nd: top of while after critic rejection → stop + ], + generate_plan_phases=[phase1, phase2], + critique_plan_phases=[], # Critic rejects ALL phases + ) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is False + # No phases were implemented. + assert fakes["fake_sc"].phase_count == 0 + # commit_all never called. + fakes["commit_all"].assert_not_called() + # generate_plan was called (once). + fakes["generate_plan"].assert_called_once() + # critique_plan was called (once). + fakes["critique_plan"].assert_called_once() + + +class TestFixRoundSucceeds: + """10. Fix round succeeds — phase commits after fix.""" + + def test_fix_round_succeeds_and_commits(self, monkeypatch, tmp_path): + # Arrange + phase = {"title": "Phase 1 — Tricky change", "body": "A tricky change."} + + fakes = _setup_mocks( + monkeypatch, + tmp_path, + branch="feature", + should_stop_responses=[ + (False, ""), # top of while → continue + (False, ""), # before phase → continue + (True, "Max phases"), # next loop → stop + ], + generate_plan_phases=[phase], + critique_plan_phases=[phase], + # First run fails, second (after fix) passes. + run_tests_responses=[(False, "1 failed"), (True, "")], + commit_all_return="abc1234", + ) + + # Act + result = run_evolve(tmp_path) + + # Assert + assert result is True + # commit_all was called (once, after the fix round made tests green). + fakes["commit_all"].assert_called_once() + # Agent was created 3 times: manager, critic, impl, fix = 4 actually + # Let's verify we got the fix agent run. + # We can check that there was at least one Agent created with megaswarm=True. + assert fakes["fake_sc"].phase_count == 1 diff --git a/tests/test_evolve_limits.py b/tests/test_evolve_limits.py new file mode 100644 index 0000000..6c812c6 --- /dev/null +++ b/tests/test_evolve_limits.py @@ -0,0 +1,322 @@ +"""Tests for devbot.evolve_limits — the auto-evolve StopController.""" + +import os +import time + +import pytest + +from devbot.evolve_limits import StopController + + +# ============================================================================ +# TestDefaultValues +# ============================================================================ + +class TestDefaultValues: + """A fresh StopController has correct defaults.""" + + def test_phase_count_zero(self): + sc = StopController() + assert sc.phase_count == 0 + + def test_max_phases_default(self): + sc = StopController() + assert sc.max_phases == 20 + + def test_time_limit_minutes_default(self): + sc = StopController() + assert sc.time_limit_minutes == 0 + + def test_should_stop_returns_false_initially(self): + sc = StopController() + stopped, reason = sc.should_stop() + assert stopped is False + assert reason == "" + + +# ============================================================================ +# TestMaxPhases +# ============================================================================ + +class TestMaxPhases: + """When DEVBOT_EVOLVE_MAX_PHASES is set, phases are counted.""" + + def test_below_max_no_stop(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "5") + sc = StopController() + for _ in range(3): + sc.record_phase() + stopped, _ = sc.should_stop() + assert stopped is False + + def test_exactly_max_stops(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "5") + sc = StopController() + for _ in range(5): + sc.record_phase() + stopped, reason = sc.should_stop() + assert stopped is True + assert "Max phases reached" in reason + assert "5/5" in reason + + def test_above_max_stops(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "3") + sc = StopController() + for _ in range(7): + sc.record_phase() + stopped, reason = sc.should_stop() + assert stopped is True + assert "Max phases reached" in reason + + def test_max_phases_zero_unlimited(self, monkeypatch): + """max_phases=0 means no cap from phases.""" + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "0") + sc = StopController() + for _ in range(100): + sc.record_phase() + stopped, _ = sc.should_stop() + assert stopped is False # 0 means unlimited + + +# ============================================================================ +# TestTimeLimit +# ============================================================================ + +class TestTimeLimit: + """When DEVBOT_EVOLVE_TIME_LIMIT is set, wall-clock is honoured.""" + + def test_before_time_limit_no_stop(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "10") + sc = StopController() + sc.start() + # Simulate only 1 minute elapsed (limit is 10 min) + fake_now = sc._start_time + 60 # 1 minute + monkeypatch.setattr(time, "monotonic", lambda: fake_now) + stopped, _ = sc.should_stop() + assert stopped is False + + def test_after_time_limit_stops(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "2") + sc = StopController() + sc.start() + # Simulate 2 minutes elapsed exactly (limit is 2 min = 120 s) + fake_now = sc._start_time + 120 + monkeypatch.setattr(time, "monotonic", lambda: fake_now) + stopped, reason = sc.should_stop() + assert stopped is True + assert "Time limit exceeded" in reason + assert "2" in reason + + def test_past_time_limit_stops(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "1") + sc = StopController() + sc.start() + # Way past the limit + fake_now = sc._start_time + 999 + monkeypatch.setattr(time, "monotonic", lambda: fake_now) + stopped, reason = sc.should_stop() + assert stopped is True + assert "Time limit exceeded" in reason + + def test_time_limit_zero_unlimited(self, monkeypatch): + """time_limit=0 means no time cap.""" + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "0") + sc = StopController() + sc.start() + # Simulate a huge elapsed time + fake_now = sc._start_time + 999_999_999 + monkeypatch.setattr(time, "monotonic", lambda: fake_now) + stopped, _ = sc.should_stop() + assert stopped is False # 0 = unlimited + + def test_start_not_called_skips_time_check(self, monkeypatch): + """Without start(), the time check is skipped entirely.""" + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "1") + sc = StopController() + # Never call start() + # Even if we monkeypatch time, should not crash and should not stop + monkeypatch.setattr(time, "monotonic", lambda: 999_999_999.0) + stopped, _ = sc.should_stop() + assert stopped is False # start_time is None → skip + + +# ============================================================================ +# TestGlobalBudget +# ============================================================================ + +class TestGlobalBudget: + """The controller honours the global token budget.""" + + def test_budget_not_exceeded_no_stop(self, monkeypatch): + monkeypatch.setattr( + "devbot.evolve_limits.check_global_budget_exceeded", + lambda: False, + ) + sc = StopController() + stopped, _ = sc.should_stop() + assert stopped is False + + def test_budget_exceeded_stops(self, monkeypatch): + monkeypatch.setattr( + "devbot.evolve_limits.check_global_budget_exceeded", + lambda: True, + ) + monkeypatch.setattr( + "devbot.evolve_limits.get_global_token_count", + lambda: 123_456, + ) + monkeypatch.setenv("DEVBOT_GLOBAL_BUDGET", "100000") + sc = StopController() + stopped, reason = sc.should_stop() + assert stopped is True + assert "Global token budget exhausted" in reason + assert "123,456" in reason + assert "100,000" in reason + + +# ============================================================================ +# TestStopOnRed +# ============================================================================ + +class TestStopOnRed: + """The stop-on-red flag triggers immediately.""" + + def test_after_set_red_stops(self): + sc = StopController() + sc.set_red() + stopped, reason = sc.should_stop() + assert stopped is True + assert "Stop-on-red" in reason + + def test_before_set_red_no_stop(self): + sc = StopController() + stopped, _ = sc.should_stop() + assert stopped is False + + +# ============================================================================ +# TestRecordPhase +# ============================================================================ + +class TestRecordPhase: + """record_phase increments the counter.""" + + def test_zero_calls_count_zero(self): + sc = StopController() + assert sc.phase_count == 0 + + def test_five_calls_count_five(self): + sc = StopController() + for _ in range(5): + sc.record_phase() + assert sc.phase_count == 5 + + +# ============================================================================ +# TestEnvVarConfiguration +# ============================================================================ + +class TestEnvVarConfiguration: + """Custom env vars are read correctly at __init__ time.""" + + def test_max_phases_from_env(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "5") + sc = StopController() + assert sc.max_phases == 5 + + def test_max_phases_zero_means_unlimited(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "0") + sc = StopController() + assert sc.max_phases == 0 + + def test_time_limit_from_env(self, monkeypatch): + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "30") + sc = StopController() + assert sc.time_limit_minutes == 30 + + +# ============================================================================ +# TestOrderOfChecks +# ============================================================================ + +class TestOrderOfChecks: + """The first triggered condition wins in should_stop().""" + + def test_red_wins_over_max_phases(self, monkeypatch): + """If both red and max_phases are hit, red reason is returned.""" + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "1") + monkeypatch.setattr( + "devbot.evolve_limits.check_global_budget_exceeded", + lambda: False, + ) + sc = StopController() + sc.record_phase() # hits max_phases=1 + sc.set_red() + stopped, reason = sc.should_stop() + assert stopped is True + assert "Stop-on-red" in reason # checked first + + def test_budget_wins_over_max_phases(self, monkeypatch): + """Budget is checked before max_phases.""" + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "1") + monkeypatch.setattr( + "devbot.evolve_limits.check_global_budget_exceeded", + lambda: True, + ) + monkeypatch.setattr( + "devbot.evolve_limits.get_global_token_count", + lambda: 100, + ) + monkeypatch.setenv("DEVBOT_GLOBAL_BUDGET", "100") + sc = StopController() + sc.record_phase() # would also trigger max_phases + stopped, reason = sc.should_stop() + assert stopped is True + assert "Global token budget" in reason # checked second, before max_phases + + def test_max_phases_wins_over_time(self, monkeypatch): + """Max phases is checked before time limit.""" + monkeypatch.setenv("DEVBOT_EVOLVE_MAX_PHASES", "1") + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "1") + monkeypatch.setattr( + "devbot.evolve_limits.check_global_budget_exceeded", + lambda: False, + ) + sc = StopController() + sc.start() + sc.record_phase() # triggers max_phases + # Also make time elapsed past limit + fake_now = sc._start_time + 999 + monkeypatch.setattr(time, "monotonic", lambda: fake_now) + stopped, reason = sc.should_stop() + assert stopped is True + assert "Max phases reached" in reason # checked third, before time + + +# ============================================================================ +# TestStartMethod +# ============================================================================ + +class TestStartMethod: + """start() sets the start time so the time check becomes active.""" + + def test_before_start_no_crash_on_should_stop(self): + """should_stop() with a time limit but no start() doesn't crash.""" + sc = StopController() + sc._time_limit_minutes = 1 + sc._time_limit_seconds = 60 + # _start_time is None → time check skipped + stopped, _ = sc.should_stop() + assert stopped is False + + def test_after_start_time_check_becomes_active(self, monkeypatch): + """After start(), the time check actually fires.""" + monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "1") + sc = StopController() + sc.start() + # simulate 2 minutes elapsed + fake_now = sc._start_time + 120 + monkeypatch.setattr(time, "monotonic", lambda: fake_now) + stopped, reason = sc.should_stop() + assert stopped is True + assert "Time limit exceeded" in reason diff --git a/tests/test_evolve_planner.py b/tests/test_evolve_planner.py new file mode 100644 index 0000000..a8b2310 --- /dev/null +++ b/tests/test_evolve_planner.py @@ -0,0 +1,332 @@ +"""Tests for devbot.evolve_planner — the planner + critic for auto-evolve.""" + +import pytest + +from devbot.evolve_planner import generate_plan, critique_plan + + +# ============================================================================ +# _FakeAgent — mimics Agent.run() returning a predetermined response +# ============================================================================ + +class _FakeAgent: + """An Agent stub whose ``run()`` returns a fixed string. + + Stores every call so tests can assert on prompts passed to the manager. + """ + + def __init__(self, response: str = ""): + self.response = response + self.runs: list[str] = [] + + def run(self, prompt: str) -> str: + self.runs.append(prompt) + return self.response + + +# ============================================================================ +# TestGeneratePlan +# ============================================================================ + +class TestGeneratePlan: + """Tests for generate_plan(manager, context).""" + + VALID_PLAN_OUTPUT = """\ +## Phase 1 — Add input validation +Validate all user inputs in cli.py to prevent crashes on malformed data. +This improves correctness and safety. + +## Phase 2 — Fix race condition in session.py +Add a lock around shared state to prevent data corruption under parallel +access. This is a correctness fix. +""" + + def test_proposes_phases(self): + """Mock the manager to return valid phase headings; verify phases.""" + manager = _FakeAgent(response=self.VALID_PLAN_OUTPUT) + context = {"readme": "# My Project\nSome README content."} + + phases = generate_plan(manager, context) + + assert len(phases) == 2 + assert phases[0]["title"] == "Phase 1 — Add input validation" + assert "input validation" in phases[0]["body"] + assert phases[1]["title"] == "Phase 2 — Fix race condition in session.py" + assert "race condition" in phases[1]["body"] + + # Manager.run() must have been called exactly once. + assert len(manager.runs) == 1 + + def test_no_phases_returns_empty(self): + """Model returns text with no phase headings → [].""" + manager = _FakeAgent(response="This is just some random text.\n\nNo phases here.") + context = {"readme": "stuff"} + + phases = generate_plan(manager, context) + assert phases == [] + + def test_malformed_output_returns_empty(self): + """Manager.run() raises an exception → [] (no crash).""" + + class _FailingAgent: + def run(self, prompt): + raise RuntimeError("API exploded") + + manager = _FailingAgent() + phases = generate_plan(manager, {"outline": "x"}) + assert phases == [] + + def test_context_passed_in_prompt(self): + """Verify the prompt string includes context values.""" + manager = _FakeAgent(response=self.VALID_PLAN_OUTPUT) + context = { + "readme": "README: This is a test project.", + "outline": "OUTLINE: src/app.py, tests/", + "tree": "TREE: .\n├── app.py", + } + + generate_plan(manager, context) + + prompt = manager.runs[0] + assert "README: This is a test project." in prompt + assert "OUTLINE: src/app.py, tests/" in prompt + assert "TREE: .\n├── app.py" in prompt + # Context keys should appear as section headers + assert "=== readme ===" in prompt + assert "=== outline ===" in prompt + assert "=== tree ===" in prompt + + def test_empty_context_keys_skipped(self): + """Empty or None context values are not included in the prompt.""" + manager = _FakeAgent(response=self.VALID_PLAN_OUTPUT) + context = { + "readme": "stuff", + "summary": "", # empty → skip + "outline": None, # None → skip + "tree": "tree here", + } + + generate_plan(manager, context) + + prompt = manager.runs[0] + assert "=== readme ===" in prompt + assert "=== tree ===" in prompt + assert "=== summary ===" not in prompt + assert "=== outline ===" not in prompt + + def test_manager_run_returns_none(self): + """If manager.run() returns None, we still handle it (treated as empty).""" + + class _NoneAgent: + def run(self, prompt): + return None + + phases = generate_plan(_NoneAgent(), {"readme": "x"}) + assert phases == [] + + +# ============================================================================ +# TestCritiquePlan +# ============================================================================ + +class TestCritiquePlan: + """Tests for critique_plan(manager, phases).""" + + PHASES_INPUT = [ + { + "title": "Phase 1 — Add input validation", + "body": "Validate all user inputs in cli.py to prevent crashes.", + }, + { + "title": "Phase 2 — Rename variables to camelCase", + "body": "Rename all snake_case locals to camelCase for consistency.", + }, + { + "title": "Phase 3 — Add timeout to network calls", + "body": "Add httpx timeout to prevent hanging connections.", + }, + ] + + CRITIQUE_KEEPS_1_AND_3 = """\ +## Phase 1 — Add input validation +Justification: Improves correctness and safety by preventing crashes from malformed input. Score: 9. +Validate all user inputs in cli.py to prevent crashes. + +## Phase 3 — Add timeout to network calls +Justification: Prevents the agent from hanging indefinitely on network issues. Score: 8. +Add httpx timeout to prevent hanging connections. +""" + + def test_keeps_good_phases(self): + """Critic keeps 2 of 3 phases; verify returned list and keys.""" + manager = _FakeAgent(response=self.CRITIQUE_KEEPS_1_AND_3) + result = critique_plan(manager, self.PHASES_INPUT) + + assert len(result) == 2 + assert result[0]["title"] == "Phase 1 — Add input validation" + assert result[0]["justification"] == ( + "Improves correctness and safety by preventing crashes from " + "malformed input. Score: 9." + ) + assert "input validation" in result[0]["body"] + + assert result[1]["title"] == "Phase 3 — Add timeout to network calls" + assert result[1]["justification"] == ( + "Prevents the agent from hanging indefinitely on network " + "issues. Score: 8." + ) + assert "timeout" in result[1]["body"] + + # Manager should have been called once with all three phases in the prompt + assert len(manager.runs) == 1 + prompt = manager.runs[0] + assert "Phase 1 — Add input validation" in prompt + assert "Phase 2 — Rename variables to camelCase" in prompt + assert "Phase 3 — Add timeout to network calls" in prompt + + def test_drops_all_returns_empty(self): + """Critic returns NONE → [].""" + manager = _FakeAgent(response="NONE") + result = critique_plan(manager, self.PHASES_INPUT) + assert result == [] + + def test_drops_all_empty_string_returns_empty(self): + """Critic returns empty/whitespace string → [].""" + manager = _FakeAgent(response=" \n ") + result = critique_plan(manager, self.PHASES_INPUT) + assert result == [] + + def test_drops_pure_refactors(self): + """The refactor phase (Phase 2) is absent from the surviving list.""" + manager = _FakeAgent(response=self.CRITIQUE_KEEPS_1_AND_3) + result = critique_plan(manager, self.PHASES_INPUT) + + titles = {ph["title"] for ph in result} + assert "Phase 2 — Rename variables to camelCase" not in titles + assert len(result) == 2 + + def test_malformed_output_returns_empty(self): + """Manager.run() raises an exception → [] (no crash).""" + + class _FailingAgent: + def run(self, prompt): + raise RuntimeError("API exploded") + + result = critique_plan(_FailingAgent(), self.PHASES_INPUT) + assert result == [] + + def test_empty_input_returns_empty(self): + """Passing [] as phases returns [] without calling the manager.""" + manager = _FakeAgent(response="should not be called") + result = critique_plan(manager, []) + assert result == [] + assert len(manager.runs) == 0 + + def test_critique_output_without_justification(self): + """If a surviving phase has no Justification line, justification is ''.""" + output = """\ +## Phase 1 — Add input validation +Validate all user inputs in cli.py to prevent crashes. +""" + manager = _FakeAgent(response=output) + phases = [self.PHASES_INPUT[0]] # just Phase 1 + + result = critique_plan(manager, phases) + + assert len(result) == 1 + assert result[0]["title"] == "Phase 1 — Add input validation" + assert result[0]["justification"] == "" + assert "input validation" in result[0]["body"] + + def test_justification_removed_from_body(self): + """The Justification line is stripped from the returned body.""" + output = """\ +## Phase 3 — Add timeout to network calls +Justification: Prevents hanging. Score: 8. +Add httpx timeout to prevent hanging connections. +""" + manager = _FakeAgent(response=output) + phases = [self.PHASES_INPUT[2]] # Phase 3 + + result = critique_plan(manager, phases) + + assert len(result) == 1 + assert result[0]["justification"] == "Prevents hanging. Score: 8." + # Body should NOT contain the justification line + assert "Justification:" not in result[0]["body"] + # Body should still contain the original description + assert "Add httpx timeout" in result[0]["body"] + + def test_parse_error_returns_empty(self): + """If parse_phases raises (unlikely), we still return [] gracefully.""" + manager = _FakeAgent(response="## Not a Phase — just some heading\nbody") + + result = critique_plan(manager, self.PHASES_INPUT) + # parse_phases would find 0 phases (heading doesn't match "Phase N" pattern) + assert result == [] + + +# ============================================================================ +# TestIntegration +# ============================================================================ + +class TestIntegration: + """End-to-end: generate a plan then critique it — all with mocks.""" + + def test_generate_then_critique_flow(self): + """Full flow: planner proposes, critic filters — verify end-to-end.""" + plan_output = """\ +## Phase 1 — Add input validation +Validate all user inputs to stop crashes. Improves correctness. + +## Phase 2 — Rename utils.py to helpers.py +Pure rename for consistency. No functional change. + +## Phase 3 — Add retry logic to API calls +Add exponential backoff for transient failures. Improves reliability. +""" + + critique_output = """\ +## Phase 1 — Add input validation +Justification: Concrete correctness improvement. Score: 8. +Validate all user inputs to stop crashes. Improves correctness. + +## Phase 3 — Add retry logic to API calls +Justification: Improves reliability under real-world conditions. Score: 7. +Add exponential backoff for transient failures. Improves reliability. +""" + + context = {"readme": "# Test Project", "summary": "A small CLI tool."} + + # Step 1: generate plan + planner = _FakeAgent(response=plan_output) + proposed = generate_plan(planner, context) + assert len(proposed) == 3 + assert proposed[1]["title"] == "Phase 2 — Rename utils.py to helpers.py" + + # Step 2: critique the proposed phases + critic = _FakeAgent(response=critique_output) + surviving = critique_plan(critic, proposed) + assert len(surviving) == 2 + + titles = {ph["title"] for ph in surviving} + assert "Phase 1 — Add input validation" in titles + assert "Phase 3 — Add retry logic to API calls" in titles + # The refactor phase was dropped + assert "Phase 2 — Rename utils.py to helpers.py" not in titles + + # Both survivors have justifications + for ph in surviving: + assert ph["justification"] != "" + assert "Justification:" not in ph["body"] + + def test_planner_returns_empty_then_critique_returns_empty(self): + """If the planner finds no phases, critique gets [] and returns [].""" + planner = _FakeAgent(response="Nothing to improve here.") + proposed = generate_plan(planner, {"readme": "x"}) + assert proposed == [] + + critic = _FakeAgent(response="should not be called") + surviving = critique_plan(critic, proposed) + assert surviving == [] + assert len(critic.runs) == 0 diff --git a/tests/test_gitcheckpoint.py b/tests/test_gitcheckpoint.py new file mode 100644 index 0000000..a863aa9 --- /dev/null +++ b/tests/test_gitcheckpoint.py @@ -0,0 +1,298 @@ +"""Tests for devbot.gitcheckpoint — git checkpoint helpers.""" + +import subprocess + +import pytest + +from devbot.gitcheckpoint import current_branch, is_clean, ensure_branch, commit_all + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _init_repo(path, branch="main"): + """Create a git repo at *path* with one commit so HEAD is defined. + + Uses ``git init --initial-branch=``, configures a dummy + user, and commits an empty ``.gitkeep`` so the repo is never in a + detached/empty-HEAD state. + """ + subprocess.run( + ["git", "init", "--initial-branch", branch, str(path)], + capture_output=True, + text=True, + check=True, + ) + # Configure a fake identity so commits work. + subprocess.run( + ["git", "config", "user.email", "test@example.com"], + cwd=str(path), + capture_output=True, + text=True, + check=True, + ) + subprocess.run( + ["git", "config", "user.name", "Test User"], + cwd=str(path), + capture_output=True, + text=True, + check=True, + ) + # In a truly fresh repo without any commits, ``git rev-parse HEAD`` + # fails and ``git status --porcelain`` shows everything as untracked. + # Give the repo one commit so the distinction between "clean" and + # "dirty" is predictable. + (path / ".gitkeep").write_text("") + subprocess.run( + ["git", "add", "-A"], + cwd=str(path), + capture_output=True, + text=True, + check=True, + ) + subprocess.run( + ["git", "commit", "-m", "initial"], + cwd=str(path), + capture_output=True, + text=True, + check=True, + ) + + +def _branch(path) -> str: + """Return the current branch name (raw git output).""" + return subprocess.run( + ["git", "rev-parse", "--abbrev-ref", "HEAD"], + cwd=str(path), + capture_output=True, + text=True, + check=True, + ).stdout.strip() + + +# --------------------------------------------------------------------------- +# current_branch +# --------------------------------------------------------------------------- + + +class TestCurrentBranch: + """Tests for ``current_branch``.""" + + def test_returns_main_on_fresh_repo(self, tmp_path): + _init_repo(tmp_path) + assert current_branch(str(tmp_path)) == "main" + + def test_returns_master_when_init_with_master(self, tmp_path): + _init_repo(tmp_path, branch="master") + assert current_branch(str(tmp_path)) == "master" + + def test_returns_new_branch_after_checkout(self, tmp_path): + _init_repo(tmp_path) + subprocess.run( + ["git", "checkout", "-b", "feature-x"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + assert current_branch(str(tmp_path)) == "feature-x" + + def test_raises_on_non_git_directory(self, tmp_path): + non_git = tmp_path / "not_a_repo" + non_git.mkdir() + with pytest.raises(RuntimeError, match="git rev-parse failed"): + current_branch(str(non_git)) + + +# --------------------------------------------------------------------------- +# is_clean +# --------------------------------------------------------------------------- + + +class TestIsClean: + """Tests for ``is_clean``.""" + + def test_clean_after_commit(self, tmp_path): + _init_repo(tmp_path) + # Only the initial .gitkeep commit — clean. + assert is_clean(str(tmp_path)) is True + + def test_dirty_with_untracked_file(self, tmp_path): + _init_repo(tmp_path) + (tmp_path / "new_file.txt").write_text("hello") + assert is_clean(str(tmp_path)) is False + + def test_dirty_with_modified_tracked_file(self, tmp_path): + _init_repo(tmp_path) + (tmp_path / "tracked.txt").write_text("original") + subprocess.run( + ["git", "add", "-A"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + subprocess.run( + ["git", "commit", "-m", "add tracked"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + assert is_clean(str(tmp_path)) is True + # Now modify + (tmp_path / "tracked.txt").write_text("modified") + assert is_clean(str(tmp_path)) is False + + def test_raises_on_non_git_directory(self, tmp_path): + non_git = tmp_path / "not_a_repo" + non_git.mkdir() + with pytest.raises(RuntimeError, match="git status failed"): + is_clean(str(non_git)) + + def test_becomes_clean_after_commit(self, tmp_path): + _init_repo(tmp_path) + (tmp_path / "file.txt").write_text("content") + assert is_clean(str(tmp_path)) is False + subprocess.run( + ["git", "add", "-A"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + subprocess.run( + ["git", "commit", "-m", "commit file"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + assert is_clean(str(tmp_path)) is True + + +# --------------------------------------------------------------------------- +# ensure_branch +# --------------------------------------------------------------------------- + + +class TestEnsureBranch: + """Tests for ``ensure_branch``.""" + + def test_creates_and_checks_out_new_branch(self, tmp_path): + _init_repo(tmp_path) + ensure_branch(str(tmp_path), "feature") + assert _branch(tmp_path) == "feature" + + def test_already_existing_branch_just_checks_out(self, tmp_path): + _init_repo(tmp_path) + # Create feature branch & commit something, then go back to main. + subprocess.run( + ["git", "checkout", "-b", "feature"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + (tmp_path / "on_feature.txt").write_text("data") + subprocess.run( + ["git", "add", "-A"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + subprocess.run( + ["git", "commit", "-m", "feature work"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + subprocess.run( + ["git", "checkout", "main"], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ) + assert _branch(tmp_path) == "main" + + # Now call ensure_branch with the already-existing name. + ensure_branch(str(tmp_path), "feature") + assert _branch(tmp_path) == "feature" + + def test_refuses_main(self, tmp_path): + _init_repo(tmp_path) + with pytest.raises(ValueError, match="Refusing to operate on branch"): + ensure_branch(str(tmp_path), "main") + + def test_refuses_master(self, tmp_path): + _init_repo(tmp_path, branch="master") + with pytest.raises(ValueError, match="Refusing to operate on branch"): + ensure_branch(str(tmp_path), "master") + + +# --------------------------------------------------------------------------- +# commit_all +# --------------------------------------------------------------------------- + + +class TestCommitAll: + """Tests for ``commit_all``.""" + + def test_returns_short_sha_on_success(self, tmp_path): + _init_repo(tmp_path) + (tmp_path / "work.txt").write_text("hello") + sha = commit_all(str(tmp_path), "first real commit") + assert isinstance(sha, str) + assert len(sha) >= 3 # short SHA is typically 7+ chars, but be flexible + # Verify it's really a commit SHA. + stdout = subprocess.run( + ["git", "rev-parse", sha], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ).stdout.strip() + assert len(stdout) == 40 # full SHA + + def test_returns_none_when_nothing_to_commit(self, tmp_path): + _init_repo(tmp_path) + # Already clean — nothing to commit. + result = commit_all(str(tmp_path), "should be none") + assert result is None + + def test_new_sha_after_modification(self, tmp_path): + _init_repo(tmp_path) + (tmp_path / "data.txt").write_text("v1") + sha1 = commit_all(str(tmp_path), "v1") + assert sha1 is not None + + # Modify and commit again. + (tmp_path / "data.txt").write_text("v2") + sha2 = commit_all(str(tmp_path), "v2") + assert sha2 is not None + assert sha2 != sha1 + + def test_special_characters_in_message(self, tmp_path): + """Commit messages with quotes, newlines, and other special chars work.""" + _init_repo(tmp_path) + (tmp_path / "file.txt").write_text("test") + msg = '''Fix "the thing" -- it's done + +More detail on line two with 'single quotes' and $dollar signs.''' + sha = commit_all(str(tmp_path), msg) + assert sha is not None + # Verify the message survived the trip. + stdout = subprocess.run( + ["git", "log", "-1", "--format=%B", sha], + cwd=str(tmp_path), + capture_output=True, + text=True, + check=True, + ).stdout.strip() + # Git normalises trailing newlines; compare with trailing ws stripped. + assert stdout.strip() == msg.strip() diff --git a/tests/test_megaswarm_improvements.py b/tests/test_megaswarm_improvements.py index 9622038..62a6cbd 100644 --- a/tests/test_megaswarm_improvements.py +++ b/tests/test_megaswarm_improvements.py @@ -3,7 +3,13 @@ These mock the sub-agent runners so no real API calls / tokens are used. """ import devbot.swarm as swarm -from devbot.swarm import megadelegate_schema, pipeline_schema, run_megaswarm, run_pipeline +from devbot.swarm import ( + _clip_agent_result, + megadelegate_schema, + pipeline_schema, + run_megaswarm, + run_pipeline, +) # --------------------------------------------------------------------------- @@ -39,6 +45,14 @@ def test_unknown_role_returns_error(self): assert "unknown specialist" in out.lower() +class TestTokenEfficiency: + def test_specialist_result_clip_is_configurable(self, monkeypatch): + monkeypatch.setenv("DEVBOT_SPECIALIST_RESULT_LIMIT", "500") + out = _clip_agent_result("x" * 650, "coder") + assert len(out) < 650 + assert "coder truncated" in out + + # --------------------------------------------------------------------------- # run_pipeline loop logic (mock the runners) # --------------------------------------------------------------------------- diff --git a/tests/test_phase1.py b/tests/test_phase1.py index a7d9604..6bf07c4 100644 --- a/tests/test_phase1.py +++ b/tests/test_phase1.py @@ -228,6 +228,23 @@ def test_limit_works_correctly(self, tmp_path): assert "9\tline9" not in result assert "[showing lines 6-8 of 20" in result + def test_dispatch_uses_env_default_read_limit(self, tmp_path, monkeypatch): + monkeypatch.setenv("DEVBOT_READ_FILE_LIMIT", "2") + root = tmp_path + (root / "test.txt").write_text("a\nb\nc\n") + result = dispatch("read_file", {"path": "test.txt"}, root) + assert "1\ta" in result + assert "2\tb" in result + assert "3\tc" not in result + assert "[showing lines 1-2 of 3" in result + + def test_tool_output_clip_is_configurable(self, tmp_path, monkeypatch): + monkeypatch.setenv("DEVBOT_MAX_TOOL_OUTPUT", "1000") + root = tmp_path + (root / "large.txt").write_text("x" * 1200) + result = read_file("large.txt", root) + assert "[truncated" in result + # ============================================================================ # 4. grep / find_files — SKIP_DIRS and caps @@ -485,6 +502,11 @@ def test_compress_conversation_returns_false_when_nothing_to_compress(self, tmp_ # verify _keep_index returns 3. assert agent._keep_index() == 3 + def test_concise_verbosity_adds_prompt_policy(self, tmp_path, monkeypatch): + monkeypatch.setenv("DEVBOT_VERBOSITY", "concise") + agent = self._make_agent(tmp_path) + assert "Token-efficiency mode is active." in agent.messages[0]["content"] + # ============================================================================ # 9. Recursive glob (**) support diff --git a/tests/test_phase5.py b/tests/test_phase5.py index 69060fa..9f2e720 100644 --- a/tests/test_phase5.py +++ b/tests/test_phase5.py @@ -54,11 +54,9 @@ def _readme_table_vars(readme_path: Path) -> set[str]: def _source_vars(file_paths: list[Path]) -> set[str]: """Scan Python files for os.environ[...] / os.environ.get(...) of DEVBOT_* vars.""" names: set[str] = set() - # Matches: os.environ.get("DEVBOT_X", ...) or os.environ["DEVBOT_X"] - # Also int(os.environ.get(...)) - pattern = re.compile( - r'os\.environ(?:\.get)?\(\s*["\'](DEVBOT_[A-Z_]+)["\']' - ) + # Match literal DEVBOT_* strings in source, covering direct os.environ + # access plus central config maps and shared env helper call sites. + pattern = re.compile(r'["\'](DEVBOT_[A-Z_]+)["\']') for fp in file_paths: if not fp.is_file(): continue @@ -80,7 +78,10 @@ def test_every_readme_env_var_exists_in_code(): src_files = [ PROJECT_ROOT / "devbot" / "agent.py", PROJECT_ROOT / "devbot" / "swarm.py", + PROJECT_ROOT / "devbot" / "tools.py", + PROJECT_ROOT / "devbot" / "config.py", PROJECT_ROOT / "devbot" / "devlog.py", + PROJECT_ROOT / "devbot" / "evolve_limits.py", ] code_vars = _source_vars(src_files) @@ -102,7 +103,10 @@ def test_every_code_env_var_exists_in_readme(): src_files = [ PROJECT_ROOT / "devbot" / "agent.py", PROJECT_ROOT / "devbot" / "swarm.py", + PROJECT_ROOT / "devbot" / "tools.py", + PROJECT_ROOT / "devbot" / "config.py", PROJECT_ROOT / "devbot" / "devlog.py", + PROJECT_ROOT / "devbot" / "evolve_limits.py", ] code_vars = _source_vars(src_files) From 2ac465f726b75b77b61ed46f4fb357f3fe288c04 Mon Sep 17 00:00:00 2001 From: DevanMetz <68340659+DevanMetz@users.noreply.github.com> Date: Tue, 16 Jun 2026 20:16:58 -0700 Subject: [PATCH 2/2] Add local VibeThinker provider, health check, and reasoning controls --- .env.example | 8 ++ README.md | 70 +++++++++- devbot/agent.py | 252 ++++++++++++++++++++++++++++++++--- devbot/autopilot.py | 5 +- devbot/cli.py | 96 ++++++++++--- devbot/config.py | 5 + devbot/evolve.py | 25 ++-- devbot/local_llm_health.py | 59 ++++++++ pyproject.toml | 1 + tests/test_cli.py | 77 ++++++++++- tests/test_evolve.py | 4 +- tests/test_evolve_limits.py | 15 ++- tests/test_local_provider.py | 243 +++++++++++++++++++++++++++++++++ 13 files changed, 799 insertions(+), 61 deletions(-) create mode 100644 devbot/local_llm_health.py create mode 100644 tests/test_local_provider.py diff --git a/.env.example b/.env.example index e8dc3aa..6d0dac6 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,14 @@ DEEPSEEK_API_KEY=sk-your-key-here # Optional: default model. One of: deepseek-v4-flash (default), deepseek-v4-pro DEVBOT_MODEL=deepseek-v4-flash +# Optional: local VibeThinker through llama.cpp/OpenAI-compatible server. +# To use it, start the server and set: +# DEVBOT_PROVIDER=local-vibethinker +# LOCAL_LLM_BASE_URL=http://127.0.0.1:8092/v1 +# LOCAL_LLM_MODEL=vibethinker-q4-vulkan +# LOCAL_LLM_API_KEY=local +# LOCAL_LLM_MAX_TOKENS=600 + # Optional: long-run token efficiency knobs. # DEVBOT_VERBOSITY=concise # DEVBOT_MAX_TOOL_OUTPUT=12000 diff --git a/README.md b/README.md index 12591c3..9f5e85e 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # DevBot -A Claude Code–style CLI coding agent powered by the [DeepSeek API](https://platform.deepseek.com). +A Claude Code-style CLI coding agent powered by OpenAI-compatible chat APIs. DevBot runs an agentic loop in your terminal: you describe a task, the model reads your code with tools, edits files, and runs commands (with your approval) until the @@ -8,7 +8,7 @@ task is done. ## Features -- **Streaming agentic loop** with OpenAI-compatible function calling against DeepSeek. +- **Streaming agentic loop** with OpenAI-compatible function calling against DeepSeek or a local llama.cpp server. - **Twelve built-in tools** for reading, searching, editing, running, web-searching, and testing code. - **Pipeline mode** — mandatory review→fix loop so every code change is audited before it ships. - **Session persistence** — sessions auto-save after each turn; resume later with `--resume` or `/resume`. @@ -64,6 +64,50 @@ $env:DEEPSEEK_API_KEY = "sk-..." # PowerShell # export DEEPSEEK_API_KEY=sk-... # bash/zsh ``` +### Local VibeThinker + +DevBot can use the local VibeThinker-3B llama.cpp/Vulkan server as another +OpenAI-compatible provider. + +Start the model server: + +```powershell +& "C:\Users\Devan\Documents\Codex\2026-06-16\hey-how-can-i-run-this\outputs\vibethinker-vulkan\start-vibethinker-q4-vulkan.ps1" +``` + +Then start DevBot with the local provider: + +```sh +devbot --local +devbot --local "fix the failing test" +``` + +Inside the REPL, you can switch providers without restarting: + +```text +/local +/cloud +``` + +The default local connection is `http://127.0.0.1:8092/v1` with model +`vibethinker-q4-vulkan`, API key `local`, and a 600-token local output cap. +For custom local servers, set: + +```powershell +$env:LOCAL_LLM_BASE_URL = "http://127.0.0.1:8092/v1" +$env:LOCAL_LLM_MODEL = "vibethinker-q4-vulkan" +$env:LOCAL_LLM_API_KEY = "local" +$env:LOCAL_LLM_MAX_TOKENS = "600" +``` + +You can check the server with: + +```sh +python -m devbot.local_llm_health +# or, after reinstalling with pip install -e . +devbot-local-health +``` + ## Usage ```sh @@ -74,6 +118,7 @@ devbot -m deepseek-v4-pro # use the more capable model for harder tasks devbot -C path/to/project # operate on another directory devbot -s # swarm mode devbot -M # megaswarm mode (3 parallel agents + reviewer) +devbot --local # use local VibeThinker instead of DeepSeek devbot --run-plan # autopilot: implement each `## Phase` from plan.md devbot --run-plan my-plan.md # use a custom plan file devbot --resume # resume the latest saved session @@ -86,6 +131,8 @@ In the REPL: - `/clear` — reset the conversation (keeps system prompt) - `/stats` — show token usage (last prompt + session total) and message count - `/model ` — switch model (`deepseek-v4-flash` or `deepseek-v4-pro`; warns on unknown names) +- `/local` — switch to VibeThinker Local (resets the conversation) +- `/cloud` — switch back to DeepSeek (resets the conversation) - `/auto` — toggle auto-approve - `/think` — toggle display of chain-of-thought reasoning - `/swarm` — toggle swarm mode (resets the conversation) @@ -179,8 +226,13 @@ integrates the results. | Variable | Purpose | Default | |---|---|---| -| `DEEPSEEK_API_KEY` | API key (required) | — | +| `DEEPSEEK_API_KEY` | DeepSeek API key (required for the default provider) | — | +| `DEVBOT_PROVIDER` | LLM provider (`deepseek` or `local-vibethinker`) | `deepseek` | | `DEVBOT_MODEL` | Model id | `deepseek-v4-flash` | +| `LOCAL_LLM_BASE_URL` | Local OpenAI-compatible base URL | `http://127.0.0.1:8092/v1` | +| `LOCAL_LLM_MODEL` | Local model id | `vibethinker-q4-vulkan` | +| `LOCAL_LLM_API_KEY` | Local server API key placeholder | `local` | +| `LOCAL_LLM_MAX_TOKENS` | Max output tokens for local model calls | 600 | | `DEVBOT_MAX_TURNS` | Max tool iterations per message | 200 | | `DEVBOT_MAX_PARALLEL` | Max parallel agents in megaswarm | 8 | | `DEVBOT_TOKEN_BUDGET` | Per-agent token cap (0 = unlimited) | 0 | @@ -205,6 +257,12 @@ You can also put these in a `.env` file in your project root instead of exportin ``` DEEPSEEK_API_KEY=sk-... DEVBOT_MODEL=deepseek-v4-flash +# For local VibeThinker instead: +# DEVBOT_PROVIDER=local-vibethinker +# LOCAL_LLM_BASE_URL=http://127.0.0.1:8092/v1 +# LOCAL_LLM_MODEL=vibethinker-q4-vulkan +# LOCAL_LLM_API_KEY=local +# LOCAL_LLM_MAX_TOKENS=600 DEVBOT_VERBOSITY=concise DEVBOT_MAX_TOOL_OUTPUT=12000 DEVBOT_SPECIALIST_RESULT_LIMIT=4000 @@ -217,6 +275,12 @@ You can also put these settings in a `.devbot/config.toml` file in your project ```toml # .devbot/config.toml (all keys optional) model = "deepseek-v4-pro" +# Or use local VibeThinker: +# provider = "local-vibethinker" +# local_llm_base_url = "http://127.0.0.1:8092/v1" +# local_llm_model = "vibethinker-q4-vulkan" +# local_llm_api_key = "local" +# local_llm_max_tokens = 600 max_parallel = 4 token_budget = 100000 loop_limit = 5 diff --git a/devbot/agent.py b/devbot/agent.py index b9c80f2..b5746a5 100644 --- a/devbot/agent.py +++ b/devbot/agent.py @@ -2,9 +2,11 @@ import json import os +import re import sys import threading import time +from dataclasses import dataclass from pathlib import Path import httpx @@ -78,6 +80,136 @@ def _load_dotenv(root: Path): DEEPSEEK_BASE_URL = "https://api.deepseek.com" DEFAULT_MODEL = "deepseek-v4-flash" # use "deepseek-v4-pro" for harder tasks DEFAULT_MAX_TURNS = 200 # tool-loop cap per message; override with DEVBOT_MAX_TURNS +LOCAL_LLM_DEFAULT_BASE_URL = "http://127.0.0.1:8092/v1" +LOCAL_LLM_DEFAULT_MODEL = "vibethinker-q4-vulkan" +LOCAL_LLM_DEFAULT_API_KEY = "local" +LOCAL_LLM_DEFAULT_MAX_TOKENS = 600 +LOCAL_PROVIDER_NAME = "local-vibethinker" +LOCAL_PROVIDER_ALIASES = { + "local-vibethinker", + "vibethinker-local", + "vibethinker", + "vibethinker local", +} +LOCAL_MODEL_ALIASES = LOCAL_PROVIDER_ALIASES | {"vibethinker-local-3b"} +_THINK_BLOCK_RE = re.compile(r"]*>.*?(?:|$)", re.IGNORECASE | re.DOTALL) +_TEXT_TOOL_BLOCK_RE = re.compile( + r"]*>.*?(?:|$)|]*>.*?(?:|$)", + re.IGNORECASE | re.DOTALL, +) +_TRIVIAL_GREETING_RE = re.compile( + r"^\s*(hi|hello|hey|yo|howdy|sup|what'?s up|good (morning|afternoon|evening))[\s!.?]*$", + re.IGNORECASE, +) + + +@dataclass(frozen=True) +class LLMProviderSettings: + provider: str + display_name: str + base_url: str + api_key: str | None + model: str + requires_deepseek_key: bool = False + + +def _normalise_provider_name(value: str | None) -> str | None: + if not value: + return None + normalised = value.strip().lower() + if normalised in LOCAL_PROVIDER_ALIASES: + return LOCAL_PROVIDER_NAME + if normalised in {"deepseek", "cloud", "deepseek-cloud"}: + return "deepseek" + return normalised + + +def get_llm_provider_settings( + requested_model: str | None = None, + requested_provider: str | None = None, +) -> LLMProviderSettings: + """Resolve the configured OpenAI-compatible provider and model. + + DeepSeek remains the default provider. Local VibeThinker is selected when + DEVBOT_PROVIDER names it, when the requested model is a local alias, or + when no DeepSeek key is available but LOCAL_LLM_BASE_URL is. + """ + explicit_provider = _normalise_provider_name( + requested_provider or os.environ.get("DEVBOT_PROVIDER") + ) + requested = requested_model or os.environ.get("DEVBOT_MODEL") + requested_key = (requested or "").strip().lower() + local_base_url = os.environ.get("LOCAL_LLM_BASE_URL") + + provider = explicit_provider + if provider is None and requested_key in LOCAL_MODEL_ALIASES: + provider = LOCAL_PROVIDER_NAME + if provider is None and local_base_url and not os.environ.get("DEEPSEEK_API_KEY"): + provider = LOCAL_PROVIDER_NAME + if provider is None: + provider = "deepseek" + + if provider == LOCAL_PROVIDER_NAME: + local_model = os.environ.get("LOCAL_LLM_MODEL", LOCAL_LLM_DEFAULT_MODEL) + model = local_model if not requested or requested_key in LOCAL_MODEL_ALIASES else requested + return LLMProviderSettings( + provider=LOCAL_PROVIDER_NAME, + display_name="VibeThinker Local", + base_url=local_base_url or LOCAL_LLM_DEFAULT_BASE_URL, + api_key=os.environ.get("LOCAL_LLM_API_KEY", LOCAL_LLM_DEFAULT_API_KEY), + model=model, + ) + + if provider != "deepseek": + raise SystemExit( + f"Unknown LLM provider '{provider}'. Supported providers: " + "deepseek, local-vibethinker." + ) + + return LLMProviderSettings( + provider="deepseek", + display_name="DeepSeek", + base_url=DEEPSEEK_BASE_URL, + api_key=os.environ.get("DEEPSEEK_API_KEY"), + model=requested or DEFAULT_MODEL, + requires_deepseek_key=True, + ) + + +def _clean_local_model_text(text: str) -> tuple[str, bool]: + """Hide VibeThinker text-only reasoning and unsupported tool wrappers.""" + without_thinking = _THINK_BLOCK_RE.sub("", text) + saw_text_tool = bool(_TEXT_TOOL_BLOCK_RE.search(without_thinking)) + cleaned = _TEXT_TOOL_BLOCK_RE.sub("", without_thinking).strip() + return cleaned, saw_text_tool + + +def _show_local_model_text(text: str) -> tuple[str, bool]: + """Show VibeThinker reasoning while still suppressing pseudo tool wrappers.""" + saw_text_tool = bool(_TEXT_TOOL_BLOCK_RE.search(text)) + without_text_tools = _TEXT_TOOL_BLOCK_RE.sub("", text) + + def show_think(match: re.Match) -> str: + thought = (match.group(0) or "") + thought = re.sub(r"^]*>|$", "", thought, + flags=re.IGNORECASE | re.DOTALL).strip() + return f"\n[thinking]\n{thought}\n[/thinking]\n" if thought else "" + + shown = _THINK_BLOCK_RE.sub(show_think, without_text_tools).strip() + return shown, saw_text_tool + + +def _is_trivial_greeting(text: str) -> bool: + return bool(_TRIVIAL_GREETING_RE.match(text)) + + +def _local_user_content(text: str) -> str: + return ( + "/no_think\n" + "Answer directly and briefly. Do not write private reasoning, " + " blocks, or pseudo-tool-call XML/JSON.\n\n" + f"{text}" + ) # Per-1M-token pricing (USD), cache-miss input rates per the DeepSeek pricing # page. deepseek-chat / deepseek-reasoner are legacy aliases for V4-flash and @@ -115,6 +247,16 @@ def _load_dotenv(root: Path): - For long tasks, keep a tiny working summary of decisions and next actions. """ +LOCAL_LLM_ADDENDUM = """\ + +Local VibeThinker mode is active. +- /no_think +- Do not emit chain-of-thought, private reasoning, or blocks. +- For greetings and small talk, answer directly in one short sentence. +- Use tools only through API tool_calls. Never print XML, JSON, or pseudo-tool + call wrappers such as , , or {"name": ...} as normal text. +""" + MANAGER_ADDENDUM = """\ You are running as the MANAGER of an agent swarm. In addition to your own tools, @@ -172,25 +314,31 @@ class Agent: def __init__(self, root: Path, model: str | None = None, auto_approve: bool = False, system_prompt: str | None = None, tool_schemas: list | None = None, swarm: bool = False, megaswarm: bool = False, - label: str | None = None): + label: str | None = None, provider: str | None = None): load_project_config(root) # stores config.toml values for later _load_dotenv(root) # .env overrides config.toml apply_project_config() # apply config.toml (env & .env win) - api_key = os.environ.get("DEEPSEEK_API_KEY") - if not api_key: + provider_settings = get_llm_provider_settings(model, provider) + if provider_settings.requires_deepseek_key and not provider_settings.api_key: raise SystemExit( "DEEPSEEK_API_KEY is not set.\n" "Get a key at https://platform.deepseek.com and set it:\n" ' PowerShell: $env:DEEPSEEK_API_KEY = "sk-..."\n' - " bash/zsh: export DEEPSEEK_API_KEY=sk-..." + " bash/zsh: export DEEPSEEK_API_KEY=sk-...\n" + "Or set DEVBOT_PROVIDER=local-vibethinker to use the local " + "VibeThinker server." ) self.client = OpenAI( - api_key=api_key, base_url=DEEPSEEK_BASE_URL, + api_key=provider_settings.api_key, + base_url=provider_settings.base_url, http_client=httpx.Client( limits=httpx.Limits(max_connections=64, max_keepalive_connections=32)), timeout=httpx.Timeout(120.0, connect=10.0)) - self.model = model or os.environ.get("DEVBOT_MODEL", DEFAULT_MODEL) + self.provider_name = provider_settings.provider + self.provider_display_name = provider_settings.display_name + self.base_url = provider_settings.base_url + self.model = provider_settings.model self.root = root self.auto_approve = auto_approve self.swarm = swarm @@ -214,6 +362,8 @@ def __init__(self, root: Path, model: str | None = None, auto_approve: bool = Fa self._same_call_count = 0 self._last_error = None # last error string for error-loop detection self._same_error_count = 0 + if self.provider_name == LOCAL_PROVIDER_NAME: + self.show_reasoning = False self.tool_schemas = list(tool_schemas) if tool_schemas is not None else list(TOOL_SCHEMAS) if swarm or megaswarm: # the manager gets the delegate + pipeline tools @@ -228,6 +378,8 @@ def __init__(self, root: Path, model: str | None = None, auto_approve: bool = Fa prompt = system_prompt else: prompt = SYSTEM_PROMPT.format(cwd=root, platform=os.name) + if self.provider_name == LOCAL_PROVIDER_NAME: + prompt += LOCAL_LLM_ADDENDUM if megaswarm: prompt += MANAGER_ADDENDUM + MEGASWARM_ADDENDUM elif swarm: @@ -340,6 +492,8 @@ def estimated_cost(self) -> float: tokens are tracked. If the model isn't in the pricing table, defaults to deepseek-v4-flash prices. """ + if getattr(self, "provider_name", "") == LOCAL_PROVIDER_NAME: + return 0.0 pricing = MODEL_PRICING.get(self.model, MODEL_PRICING["deepseek-v4-flash"]) input_price = pricing["input"] / 1_000_000 output_price = pricing["output"] / 1_000_000 @@ -372,7 +526,20 @@ def run(self, user_input: str) -> str: self._last_error = None self._same_error_count = 0 - self.messages.append({"role": "user", "content": user_input}) + if self.provider_name == LOCAL_PROVIDER_NAME and _is_trivial_greeting(user_input): + reply = "Hello! What would you like to work on?" + self.messages.append({"role": "user", "content": user_input}) + self.messages.append({"role": "assistant", "content": reply}) + self.on_text(reply + "\n") + self._auto_save() + return reply + + message_content = ( + _local_user_content(user_input) + if self.provider_name == LOCAL_PROVIDER_NAME and not self.show_reasoning + else user_input + ) + self.messages.append({"role": "user", "content": message_content}) if check_global_budget_exceeded(): self._safe_print( @@ -526,15 +693,21 @@ def run(self, user_input: str) -> str: def _create_stream(self): """Open a streamed completion, retrying transient errors with backoff.""" + kwargs = { + "model": self.model, + "messages": self.messages, + "tools": self.tool_schemas, + "stream": True, + "stream_options": {"include_usage": True}, + } + if self.provider_name == LOCAL_PROVIDER_NAME: + kwargs["max_tokens"] = int( + os.environ.get("LOCAL_LLM_MAX_TOKENS", str(LOCAL_LLM_DEFAULT_MAX_TOKENS)) + or str(LOCAL_LLM_DEFAULT_MAX_TOKENS) + ) for attempt in range(MAX_ATTEMPTS): try: - return self.client.chat.completions.create( - model=self.model, - messages=self.messages, - tools=self.tool_schemas, - stream=True, - stream_options={"include_usage": True}, - ) + return self.client.chat.completions.create(**kwargs) except RETRYABLE as e: if attempt == MAX_ATTEMPTS - 1: raise @@ -548,9 +721,11 @@ def _stream_once(self): assembled from streaming deltas into the standard dict shape.""" stream = self._create_stream() text_parts: list[str] = [] + local_raw_parts: list[str] = [] calls: dict[int, dict] = {} in_reasoning = False finish_reason = None + is_local_provider = getattr(self, "provider_name", "") == LOCAL_PROVIDER_NAME for chunk in stream: # Final chunk (include_usage) carries token counts and no choices. @@ -595,6 +770,24 @@ def _stream_once(self): finish_reason = chunk.choices[0].finish_reason # thinking mode streams chain-of-thought in reasoning_content reasoning = getattr(delta, "reasoning_content", None) + if is_local_provider: + if reasoning: + local_raw_parts.append(reasoning) + if delta.content: + local_raw_parts.append(delta.content) + for tc in delta.tool_calls or []: + slot = calls.setdefault(tc.index, { + "id": "", "type": "function", + "function": {"name": "", "arguments": ""}, + }) + if tc.id: + slot["id"] = tc.id + if tc.function: + if tc.function.name: + slot["function"]["name"] += tc.function.name + if tc.function.arguments: + slot["function"]["arguments"] += tc.function.arguments + continue if reasoning: if not in_reasoning and not self.show_reasoning: # Start of CoT: show compact thinking indicator @@ -646,6 +839,17 @@ def _stream_once(self): if tc.function.arguments: slot["function"]["arguments"] += tc.function.arguments + local_text_tool = False + if is_local_provider and local_raw_parts: + local_text = "".join(local_raw_parts) + if self.show_reasoning: + cleaned, local_text_tool = _show_local_model_text(local_text) + else: + cleaned, local_text_tool = _clean_local_model_text(local_text) + if cleaned: + text_parts.append(cleaned) + self.on_text(cleaned) + if in_reasoning: # stream ended mid-reasoning; clean up display if self.show_reasoning: self.on_text("\x1b[0m\n") @@ -653,8 +857,20 @@ def _stream_once(self): self._transient_line(clear=True) if text_parts: self.on_text("\n") + elif local_text_tool and not calls: + msg = ("[devbot] Local model emitted a text-form tool call instead of " + "a normal response. Try rephrasing, or use /cloud for tool-heavy tasks.") + self.on_text(msg + "\n") + text_parts.append(msg) + elif is_local_provider and local_raw_parts: + msg = ("[devbot] Local model spent the whole response in hidden reasoning. " + "Use /think to show it, try a more direct request, or use /cloud for this prompt.") + self.on_text(msg + "\n") + text_parts.append(msg) # P1-7: warn if the model stopped because it hit the output length limit - if finish_reason == "length": + if finish_reason == "length" and text_parts and not ( + is_local_provider and text_parts[-1].startswith("[devbot] Local model spent") + ): print("\x1b[33m[devbot] Warning: response truncated " "(finish_reason=length).\x1b[0m") tool_calls = [calls[i] for i in sorted(calls)] or None @@ -693,7 +909,11 @@ def _compress_conversation(self) -> bool: ] try: - compress_model = os.environ.get("DEVBOT_COMPRESS_MODEL", "deepseek-v4-flash") + default_compress_model = ( + self.model if getattr(self, "provider_name", "") == LOCAL_PROVIDER_NAME + else "deepseek-v4-flash" + ) + compress_model = os.environ.get("DEVBOT_COMPRESS_MODEL", default_compress_model) resp = self.client.chat.completions.create( model=compress_model, messages=compress_msgs, diff --git a/devbot/autopilot.py b/devbot/autopilot.py index d786341..7376cbd 100644 --- a/devbot/autopilot.py +++ b/devbot/autopilot.py @@ -75,7 +75,7 @@ def _run_tests(root: Path) -> tuple[bool, str]: def run_plan(root: Path, plan_path: str = "plan.md", model: str | None = None, - max_phases: int = 20) -> bool: + max_phases: int = 20, provider: str | None = None) -> bool: """Implement every phase in *plan_path* sequentially. Returns True if all phases completed with green tests, False if it stopped early.""" plan_file = root / plan_path @@ -96,7 +96,8 @@ def run_plan(root: Path, plan_path: str = "plan.md", model: str | None = None, print(f"\x1b[36;1m[autopilot] running {len(phases)} phase(s) from {plan_path}\x1b[0m") def _agent() -> Agent: - return Agent(root=root, model=model, auto_approve=True, megaswarm=True) + return Agent(root=root, model=model, auto_approve=True, megaswarm=True, + provider=provider) for idx, ph in enumerate(phases, 1): if check_global_budget_exceeded(): diff --git a/devbot/cli.py b/devbot/cli.py index 3cddca2..10f6fe1 100644 --- a/devbot/cli.py +++ b/devbot/cli.py @@ -1,6 +1,7 @@ """Interactive REPL entry point: `devbot` or `python -m devbot`.""" import argparse +import os import sys from pathlib import Path @@ -9,7 +10,7 @@ # Either way it auto-hooks input() on import, so we just try and move on. # Module-level list of slash commands used by both the completer and the REPL loop. _SLASH_COMMANDS = [ - "/help", "/clear", "/stats", "/model", "/think", "/swarm", "/megaswarm", + "/help", "/clear", "/stats", "/model", "/local", "/cloud", "/think", "/swarm", "/megaswarm", "/resume", "/sessions", "/exit", "/tools", "/cost", "/export", "/undo", ] @@ -63,7 +64,26 @@ def _setup_readline(history_path=None, marker_path=None): from openai import AuthenticationError, PermissionDeniedError from . import __version__ -from .agent import Agent, DEFAULT_MODEL, KNOWN_MODELS +from .agent import Agent, DEFAULT_MODEL, KNOWN_MODELS, LOCAL_MODEL_ALIASES, LOCAL_PROVIDER_NAME + + +def _make_agent( + root: Path, + *, + model: str | None, + auto_approve: bool, + swarm: bool, + megaswarm: bool, + provider: str | None, +) -> Agent: + return Agent( + root=root, + model=model, + auto_approve=auto_approve, + swarm=swarm, + megaswarm=megaswarm, + provider=provider, + ) def _friendly_api_error(e: Exception) -> str | None: @@ -84,7 +104,9 @@ def _friendly_api_error(e: Exception) -> str | None: /help show this help /clear reset the conversation (keeps system prompt) /stats show token usage and message count - /model switch model (deepseek-v4-flash, deepseek-v4-pro) + /model switch model (deepseek-v4-flash, deepseek-v4-pro, or local model id) + /local switch to VibeThinker Local (resets conversation) + /cloud switch back to DeepSeek (resets conversation) /auto toggle auto-approve of tool calls /think toggle display of full chain-of-thought /swarm toggle swarm mode (delegate to specialists; resets conversation) @@ -102,15 +124,19 @@ def _friendly_api_error(e: Exception) -> str | None: | | | |/ _ \\ \\ / / _ \\ / _ \\| __| | |_| | __/\\ V /| |_) | (_) | |_ |____/ \\___| \\_/ |____/ \\___/ \\__| v{version} -\x1b[0m DeepSeek-powered coding agent · model: {model}{mode} · cwd: {cwd} +\x1b[0m OpenAI-compatible coding agent · provider: {provider} · model: {model}{mode} · cwd: {cwd} Type /help for commands. """ def main(): - parser = argparse.ArgumentParser(prog="devbot", description="DeepSeek-powered CLI coding agent") + parser = argparse.ArgumentParser(prog="devbot", description="OpenAI-compatible CLI coding agent") parser.add_argument("prompt", nargs="*", help="One-shot prompt (omit for interactive mode)") parser.add_argument("-m", "--model", default=None, help=f"Model id (default: {DEFAULT_MODEL})") + parser.add_argument("--provider", choices=["deepseek", LOCAL_PROVIDER_NAME], + default=None, help="LLM provider to use") + parser.add_argument("--local", action="store_true", + help="Use the local VibeThinker llama.cpp server") parser.add_argument("-y", "--yes", action="store_true", help="Auto-approve all tool calls") parser.add_argument("-C", "--cwd", default=".", help="Project root to operate in") parser.add_argument("-s", "--swarm", action="store_true", @@ -132,13 +158,14 @@ def main(): args = parser.parse_args() root = Path(args.cwd).resolve() + provider = LOCAL_PROVIDER_NAME if args.local else args.provider # Autopilot: run a whole plan unattended, phase by phase. if args.run_plan is not None: from .autopilot import run_plan print("\x1b[33m[devbot] Autopilot runs unattended with auto-approve and " "shell access. Ctrl+C to stop.\x1b[0m") - ok = run_plan(root, args.run_plan, model=args.model) + ok = run_plan(root, args.run_plan, model=args.model, provider=provider) sys.exit(0 if ok else 1) # Autopilot: self-evolving loop that plans, critiques, and implements unattended. @@ -148,7 +175,7 @@ def main(): "with auto-approve and shell access. It will plan, critique, and " "implement phases, committing each green one. NEVER runs on main/master. " "Ctrl+C to stop.\x1b[0m") - ok = run_evolve(root, args.model) + ok = run_evolve(root, args.model, provider=provider) sys.exit(0 if ok else 1) # Handle --resume: restore from a saved session. @@ -167,14 +194,17 @@ def main(): print(f"[devbot] Resumed session {agent.session_id} " f"({agent.total_tokens:,} tokens, {len(agent.messages)} messages)") else: - agent = Agent(root=root, model=args.model, auto_approve=args.yes, - swarm=args.swarm, megaswarm=args.megaswarm) + agent = _make_agent(root, model=args.model, auto_approve=args.yes, + swarm=args.swarm, megaswarm=args.megaswarm, + provider=provider) if args.prompt: # one-shot mode: devbot "fix the failing test" agent.run(" ".join(args.prompt)) return - print(BANNER.format(version=__version__, model=agent.model, + print(BANNER.format(version=__version__, + provider=getattr(agent, "provider_display_name", "DeepSeek"), + model=agent.model, mode=" · megaswarm" if agent.megaswarm else (" · swarm" if agent.swarm else ""), cwd=root)) while True: @@ -216,21 +246,47 @@ def main(): if user.startswith("/model"): parts = user.split(maxsplit=1) if len(parts) == 2: - if parts[1] not in KNOWN_MODELS: - print(f"\x1b[33m[warning] '{parts[1]}' is not a known DeepSeek model " - f"({', '.join(sorted(KNOWN_MODELS))}). Setting it anyway.\x1b[0m") - agent.model = parts[1] + requested = parts[1] + if requested.lower() in LOCAL_MODEL_ALIASES: + if getattr(agent, "provider_name", "") != LOCAL_PROVIDER_NAME: + print("\x1b[33m[warning] local-vibethinker is a provider alias. " + "Restart with DEVBOT_PROVIDER=local-vibethinker to switch " + "the client base URL.\x1b[0m") + requested = os.environ.get("LOCAL_LLM_MODEL", "vibethinker-q4-vulkan") + if requested not in KNOWN_MODELS and requested.lower() not in LOCAL_MODEL_ALIASES: + if getattr(agent, "provider_name", "") == LOCAL_PROVIDER_NAME: + print(f"\x1b[33m[warning] '{requested}' is not the configured local " + "model name. Setting it anyway.\x1b[0m") + else: + print(f"\x1b[33m[warning] '{requested}' is not a known DeepSeek model " + f"({', '.join(sorted(KNOWN_MODELS))}). Setting it anyway.\x1b[0m") + agent.model = requested print(f"[model: {agent.model}]") else: print(f"[model: {agent.model}] Usage: /model (current: {agent.model})") continue + if user == "/local": + agent = _make_agent(root, model=None, auto_approve=agent.auto_approve, + swarm=agent.swarm, megaswarm=agent.megaswarm, + provider=LOCAL_PROVIDER_NAME) + print(f"[provider: {agent.provider_display_name} | model: {agent.model} | conversation reset]") + continue + if user == "/cloud": + agent = _make_agent(root, model=None, auto_approve=agent.auto_approve, + swarm=agent.swarm, megaswarm=agent.megaswarm, + provider="deepseek") + print(f"[provider: {agent.provider_display_name} | model: {agent.model} | conversation reset]") + continue if user == "/auto": agent.auto_approve = not agent.auto_approve print(f"[auto-approve: {'on' if agent.auto_approve else 'off'}]") continue if user == "/think": agent.show_reasoning = not agent.show_reasoning - print(f"[show reasoning: {'on' if agent.show_reasoning else 'off'}]") + suffix = " | local VibeThinker may use more tokens" if ( + agent.show_reasoning and getattr(agent, "provider_name", "") == LOCAL_PROVIDER_NAME + ) else "" + print(f"[show reasoning: {'on' if agent.show_reasoning else 'off'}{suffix}]") continue if user == "/swarm": # Recreate the agent with swarm toggled; this resets the conversation @@ -240,15 +296,17 @@ def main(): enable = True else: enable = not agent.swarm - agent = Agent(root=root, model=agent.model, auto_approve=agent.auto_approve, - swarm=enable, megaswarm=False) + agent = _make_agent(root, model=agent.model, auto_approve=agent.auto_approve, + swarm=enable, megaswarm=False, + provider=getattr(agent, "provider_name", None)) print(f"[swarm mode: {'on' if agent.swarm else 'off'} — conversation reset]") continue if user == "/megaswarm": # Toggle megaswarm on/off (megaswarm implies swarm). enable = not agent.megaswarm - agent = Agent(root=root, model=agent.model, auto_approve=agent.auto_approve, - swarm=enable, megaswarm=enable) + agent = _make_agent(root, model=agent.model, auto_approve=agent.auto_approve, + swarm=enable, megaswarm=enable, + provider=getattr(agent, "provider_name", None)) print(f"[megaswarm mode: {'on' if enable else 'off'} — conversation reset]") continue if user == "/sessions": diff --git a/devbot/config.py b/devbot/config.py index fc9a3bd..5ba593a 100644 --- a/devbot/config.py +++ b/devbot/config.py @@ -17,6 +17,11 @@ # Mapping from config.toml keys to their corresponding env var names. _CONFIG_KEY_MAP: dict[str, str] = { "model": "DEVBOT_MODEL", + "provider": "DEVBOT_PROVIDER", + "local_llm_base_url": "LOCAL_LLM_BASE_URL", + "local_llm_model": "LOCAL_LLM_MODEL", + "local_llm_api_key": "LOCAL_LLM_API_KEY", + "local_llm_max_tokens": "LOCAL_LLM_MAX_TOKENS", "max_parallel": "DEVBOT_MAX_PARALLEL", "token_budget": "DEVBOT_TOKEN_BUDGET", "global_budget": "DEVBOT_GLOBAL_BUDGET", diff --git a/devbot/evolve.py b/devbot/evolve.py index 78942ef..e46e767 100644 --- a/devbot/evolve.py +++ b/devbot/evolve.py @@ -72,7 +72,7 @@ def _read_readme(root: Path) -> str: # Main driver # --------------------------------------------------------------------------- -def run_evolve(root: Path, model: str | None = None) -> bool: +def run_evolve(root: Path, model: str | None = None, provider: str | None = None) -> bool: """Run the auto-evolve loop. Parameters @@ -82,6 +82,8 @@ def run_evolve(root: Path, model: str | None = None) -> bool: model : str | None Model id to use for all agents (default: ``DEVBOT_MODEL`` env or the built-in default). + provider : str | None + Provider id to use for all agents, e.g. ``local-vibethinker``. Returns ------- @@ -101,13 +103,16 @@ def run_evolve(root: Path, model: str | None = None) -> bool: from devbot.agent import Agent, get_global_token_count from devbot.autopilot import _run_tests - # Pre-flight: load .env so the API-key check works for users who keep - # their key in the repo's .env file rather than in the shell environment. - from devbot.agent import _load_dotenv + # Pre-flight: load project config/.env so provider/key checks match Agent. + from devbot.agent import _load_dotenv, get_llm_provider_settings + from devbot.config import load_project_config, apply_project_config + load_project_config(root) _load_dotenv(root) - if not os.environ.get("DEEPSEEK_API_KEY"): + apply_project_config() + provider_settings = get_llm_provider_settings(model, provider) + if provider_settings.requires_deepseek_key and not provider_settings.api_key: print("[auto-evolve] DEEPSEEK_API_KEY is not set. Please set it and " - "try again.") + "try again, or set DEVBOT_PROVIDER=local-vibethinker.") return False root_str = str(root) @@ -166,7 +171,7 @@ def run_evolve(root: Path, model: str | None = None) -> bool: # --- Generate plan --- try: manager = Agent(root=root, model=model, auto_approve=True, - megaswarm=True) + megaswarm=True, provider=provider) proposed = generate_plan(manager, context) cost_estimate += manager.estimated_cost() except (Exception, SystemExit) as exc: @@ -182,7 +187,7 @@ def run_evolve(root: Path, model: str | None = None) -> bool: # --- Critique plan --- try: critic = Agent(root=root, model=model, auto_approve=True, - megaswarm=True) + megaswarm=True, provider=provider) surviving = critique_plan(critic, proposed) cost_estimate += critic.estimated_cost() except (Exception, SystemExit) as exc: @@ -208,7 +213,7 @@ def run_evolve(root: Path, model: str | None = None) -> bool: try: impl_agent = Agent(root=root, model=model, auto_approve=True, - megaswarm=True) + megaswarm=True, provider=provider) prompt = ( f"You are implementing one phase of a multi-phase plan.\n\n" f"=== PHASE ===\n# {title}\n{body}\n\n" @@ -238,7 +243,7 @@ def run_evolve(root: Path, model: str | None = None) -> bool: f"one fix round") try: fix_agent = Agent(root=root, model=model, auto_approve=True, - megaswarm=True) + megaswarm=True, provider=provider) fix_agent.run( f"After implementing this phase, the test suite is " f"FAILING:\n\n{body}\n\n=== PYTEST OUTPUT ===\n{out}\n\n" diff --git a/devbot/local_llm_health.py b/devbot/local_llm_health.py new file mode 100644 index 0000000..c68e099 --- /dev/null +++ b/devbot/local_llm_health.py @@ -0,0 +1,59 @@ +"""Health check for the local VibeThinker OpenAI-compatible server.""" + +from __future__ import annotations + +import argparse +import os +import sys + +import httpx + +from .agent import ( + LOCAL_LLM_DEFAULT_API_KEY, + LOCAL_LLM_DEFAULT_BASE_URL, + LOCAL_LLM_DEFAULT_MODEL, +) + + +def main() -> int: + parser = argparse.ArgumentParser( + prog="devbot-local-health", + description="Send a small chat-completions request to the local LLM server.", + ) + parser.add_argument("--base-url", default=os.environ.get( + "LOCAL_LLM_BASE_URL", LOCAL_LLM_DEFAULT_BASE_URL)) + parser.add_argument("--model", default=os.environ.get( + "LOCAL_LLM_MODEL", LOCAL_LLM_DEFAULT_MODEL)) + parser.add_argument("--api-key", default=os.environ.get( + "LOCAL_LLM_API_KEY", LOCAL_LLM_DEFAULT_API_KEY)) + parser.add_argument("--prompt", default="Hello") + args = parser.parse_args() + + endpoint = args.base_url.rstrip("/") + "/chat/completions" + payload = { + "model": args.model, + "messages": [{"role": "user", "content": args.prompt}], + "max_tokens": 64, + "temperature": 0.2, + } + headers = {"Authorization": f"Bearer {args.api_key}"} + + try: + with httpx.Client(timeout=httpx.Timeout(60.0, connect=5.0)) as client: + response = client.post(endpoint, json=payload, headers=headers) + response.raise_for_status() + data = response.json() + except Exception as exc: + print(f"[local-llm] health check failed: {type(exc).__name__}: {exc}", file=sys.stderr) + return 1 + + message = data.get("choices", [{}])[0].get("message", {}) + content = (message.get("content") or "").strip() + print(f"[local-llm] ok: {args.model} at {args.base_url}") + if content: + print(content) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index aacc9d0..aa5b660 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ win = ["pyreadline3>=3.5"] [project.scripts] devbot = "devbot.cli:main" +devbot-local-health = "devbot.local_llm_health:main" [tool.setuptools] packages = ["devbot"] diff --git a/tests/test_cli.py b/tests/test_cli.py index e62be6c..45f14a5 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -12,8 +12,8 @@ def test_auto_evolve_flag_calls_run_evolve(monkeypatch, tmp_path): calls = [] - def fake_run_evolve(root, model): - calls.append((root, model)) + def fake_run_evolve(root, model, provider=None): + calls.append((root, model, provider)) return True monkeypatch.setattr("devbot.evolve.run_evolve", fake_run_evolve) @@ -27,14 +27,83 @@ def fake_run_evolve(root, model): cli.main() assert exc.value.code == 0 - assert calls == [(tmp_path.resolve(), "deepseek-v4-pro")] + assert calls == [(tmp_path.resolve(), "deepseek-v4-pro", None)] def test_auto_evolve_flag_exits_nonzero_on_failure(monkeypatch, tmp_path): - monkeypatch.setattr("devbot.evolve.run_evolve", lambda root, model: False) + monkeypatch.setattr("devbot.evolve.run_evolve", lambda root, model, provider=None: False) monkeypatch.setattr(sys, "argv", ["devbot", "--auto-evolve", "-C", str(tmp_path)]) with pytest.raises(SystemExit) as exc: cli.main() assert exc.value.code == 1 + + +def test_local_flag_passes_provider_to_auto_evolve(monkeypatch, tmp_path): + calls = [] + monkeypatch.setattr( + "devbot.evolve.run_evolve", + lambda root, model, provider=None: calls.append((root, model, provider)) or True, + ) + monkeypatch.setattr( + sys, + "argv", + ["devbot", "--local", "--auto-evolve", "-C", str(tmp_path)], + ) + + with pytest.raises(SystemExit) as exc: + cli.main() + + assert exc.value.code == 0 + assert calls == [(tmp_path.resolve(), None, "local-vibethinker")] + + +def test_local_flag_uses_local_provider(monkeypatch, tmp_path): + calls = [] + + class FakeAgent: + def __init__(self, **kwargs): + calls.append(kwargs) + self.model = "vibethinker-q4-vulkan" + self.provider_display_name = "VibeThinker Local" + self.megaswarm = False + self.swarm = False + + def run(self, prompt): + calls.append({"prompt": prompt}) + + monkeypatch.setattr(cli, "Agent", FakeAgent) + monkeypatch.setattr(sys, "argv", ["devbot", "--local", "-C", str(tmp_path), "hello"]) + + cli.main() + + assert calls[0]["provider"] == "local-vibethinker" + assert calls[0]["root"] == tmp_path.resolve() + assert calls[1] == {"prompt": "hello"} + + +def test_think_toggles_for_local_provider(monkeypatch, tmp_path, capsys): + agents = [] + + class FakeAgent: + def __init__(self, **kwargs): + self.model = "vibethinker-q4-vulkan" + self.provider_name = "local-vibethinker" + self.provider_display_name = "VibeThinker Local" + self.show_reasoning = False + self.auto_approve = False + self.megaswarm = False + self.swarm = False + agents.append(self) + + inputs = iter(["/think", "/exit"]) + monkeypatch.setattr(cli, "Agent", FakeAgent) + monkeypatch.setattr("builtins.input", lambda prompt="": next(inputs)) + monkeypatch.setattr(sys, "argv", ["devbot", "--local", "-C", str(tmp_path)]) + + cli.main() + + out = capsys.readouterr().out + assert "show reasoning: on | local VibeThinker may use more tokens" in out + assert agents[0].show_reasoning is True diff --git a/tests/test_evolve.py b/tests/test_evolve.py index 0b6cb20..b4547d1 100644 --- a/tests/test_evolve.py +++ b/tests/test_evolve.py @@ -24,12 +24,14 @@ class _FakeAgent: """A stand-in for ``devbot.agent.Agent`` that records ``.run()`` calls.""" - def __init__(self, root=None, model=None, auto_approve=None, megaswarm=None): + def __init__(self, root=None, model=None, auto_approve=None, megaswarm=None, + provider=None): # Store constructor args for assertions. self.root = root self.model = model self.auto_approve = auto_approve self.megaswarm = megaswarm + self.provider = provider self.runs: list[str] = [] def run(self, prompt: str) -> str: diff --git a/tests/test_evolve_limits.py b/tests/test_evolve_limits.py index 6c812c6..83d70c9 100644 --- a/tests/test_evolve_limits.py +++ b/tests/test_evolve_limits.py @@ -87,21 +87,24 @@ class TestTimeLimit: def test_before_time_limit_no_stop(self, monkeypatch): monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "10") + # Pin the clock to a small, exact base before start() so elapsed + # arithmetic is free of float cancellation (real monotonic() values + # are large and (base + delta) - base can drift below delta). + monkeypatch.setattr(time, "monotonic", lambda: 1000.0) sc = StopController() sc.start() # Simulate only 1 minute elapsed (limit is 10 min) - fake_now = sc._start_time + 60 # 1 minute - monkeypatch.setattr(time, "monotonic", lambda: fake_now) + monkeypatch.setattr(time, "monotonic", lambda: 1000.0 + 60) stopped, _ = sc.should_stop() assert stopped is False def test_after_time_limit_stops(self, monkeypatch): monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "2") + monkeypatch.setattr(time, "monotonic", lambda: 1000.0) sc = StopController() sc.start() # Simulate 2 minutes elapsed exactly (limit is 2 min = 120 s) - fake_now = sc._start_time + 120 - monkeypatch.setattr(time, "monotonic", lambda: fake_now) + monkeypatch.setattr(time, "monotonic", lambda: 1000.0 + 120) stopped, reason = sc.should_stop() assert stopped is True assert "Time limit exceeded" in reason @@ -109,11 +112,11 @@ def test_after_time_limit_stops(self, monkeypatch): def test_past_time_limit_stops(self, monkeypatch): monkeypatch.setenv("DEVBOT_EVOLVE_TIME_LIMIT", "1") + monkeypatch.setattr(time, "monotonic", lambda: 1000.0) sc = StopController() sc.start() # Way past the limit - fake_now = sc._start_time + 999 - monkeypatch.setattr(time, "monotonic", lambda: fake_now) + monkeypatch.setattr(time, "monotonic", lambda: 1000.0 + 999) stopped, reason = sc.should_stop() assert stopped is True assert "Time limit exceeded" in reason diff --git a/tests/test_local_provider.py b/tests/test_local_provider.py new file mode 100644 index 0000000..6270859 --- /dev/null +++ b/tests/test_local_provider.py @@ -0,0 +1,243 @@ +import os + +import pytest + +from devbot.agent import ( + Agent, + DEEPSEEK_BASE_URL, + LOCAL_LLM_DEFAULT_MAX_TOKENS, + LOCAL_LLM_DEFAULT_MODEL, + LOCAL_PROVIDER_NAME, + _clean_local_model_text, + _is_trivial_greeting, + _local_user_content, + _show_local_model_text, + get_llm_provider_settings, +) + + +_LOCAL_ENV_NAMES = ( + "DEVBOT_PROVIDER", + "DEVBOT_MODEL", + "LOCAL_LLM_BASE_URL", + "LOCAL_LLM_MAX_TOKENS", + "LOCAL_LLM_MODEL", + "LOCAL_LLM_API_KEY", +) + + +@pytest.fixture(autouse=True) +def _restore_local_env(): + original = {name: os.environ.get(name) for name in _LOCAL_ENV_NAMES} + yield + for name, value in original.items(): + if value is None: + os.environ.pop(name, None) + else: + os.environ[name] = value + + +def _clear_local_env(monkeypatch): + for name in _LOCAL_ENV_NAMES: + monkeypatch.delenv(name, raising=False) + + +def test_deepseek_remains_default_provider(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-test") + + agent = Agent(root=tmp_path) + + assert agent.provider_name == "deepseek" + assert agent.base_url == DEEPSEEK_BASE_URL + assert agent.model == "deepseek-v4-flash" + + +def test_local_provider_uses_local_env_without_deepseek_key(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + monkeypatch.setenv("DEVBOT_PROVIDER", "local-vibethinker") + monkeypatch.setenv("LOCAL_LLM_BASE_URL", "http://127.0.0.1:8092/v1") + monkeypatch.setenv("LOCAL_LLM_MODEL", "vibethinker-q4-vulkan") + monkeypatch.setenv("LOCAL_LLM_API_KEY", "local") + + agent = Agent(root=tmp_path) + + assert agent.provider_name == LOCAL_PROVIDER_NAME + assert agent.provider_display_name == "VibeThinker Local" + assert agent.base_url == "http://127.0.0.1:8092/v1" + assert agent.model == "vibethinker-q4-vulkan" + assert agent.estimated_cost() == 0.0 + + +def test_local_model_alias_selects_configured_local_model(monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + monkeypatch.setenv("DEVBOT_MODEL", "local-vibethinker") + monkeypatch.setenv("LOCAL_LLM_MODEL", "custom-local-model") + + settings = get_llm_provider_settings() + + assert settings.provider == LOCAL_PROVIDER_NAME + assert settings.model == "custom-local-model" + + +def test_local_provider_defaults_match_vibethinker_server(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + monkeypatch.setenv("DEVBOT_PROVIDER", "local-vibethinker") + + agent = Agent(root=tmp_path) + + assert agent.base_url == "http://127.0.0.1:8092/v1" + assert agent.model == LOCAL_LLM_DEFAULT_MODEL + + +def test_local_provider_can_be_selected_without_env(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + + agent = Agent(root=tmp_path, provider="local-vibethinker") + + assert agent.provider_name == LOCAL_PROVIDER_NAME + assert agent.base_url == "http://127.0.0.1:8092/v1" + assert agent.model == LOCAL_LLM_DEFAULT_MODEL + + +def test_config_toml_can_select_local_provider(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + devbot_dir = tmp_path / ".devbot" + devbot_dir.mkdir() + devbot_dir.joinpath("config.toml").write_text( + 'provider = "local-vibethinker"\n' + 'local_llm_base_url = "http://127.0.0.1:8092/v1"\n' + 'local_llm_model = "vibethinker-q4-vulkan"\n' + 'local_llm_api_key = "local"\n', + encoding="utf-8", + ) + + agent = Agent(root=tmp_path) + + assert agent.provider_name == LOCAL_PROVIDER_NAME + assert os.environ["LOCAL_LLM_BASE_URL"] == "http://127.0.0.1:8092/v1" + assert agent.model == "vibethinker-q4-vulkan" + + +def test_unknown_provider_is_rejected(monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.setenv("DEVBOT_PROVIDER", "bogus") + + with pytest.raises(SystemExit): + get_llm_provider_settings() + + +def test_clean_local_model_text_strips_think_blocks(): + text = "private reasoningHello there." + + cleaned, saw_text_tool = _clean_local_model_text(text) + + assert cleaned == "Hello there." + assert saw_text_tool is False + + +def test_clean_local_model_text_suppresses_text_tool_wrappers(): + text = ( + "private reasoning" + '{ "name": "run_command", "arguments": { "command": "echo hi" }' + "" + ) + + cleaned, saw_text_tool = _clean_local_model_text(text) + + assert cleaned == "" + assert saw_text_tool is True + + +def test_show_local_model_text_keeps_thinking(): + text = "private reasoningVisible answer." + + shown, saw_text_tool = _show_local_model_text(text) + + assert "[thinking]" in shown + assert "private reasoning" in shown + assert "Visible answer." in shown + assert saw_text_tool is False + + +def test_trivial_greeting_bypasses_local_model(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + agent = Agent(root=tmp_path, provider="local-vibethinker") + + def fail_stream(): + raise AssertionError("trivial greeting should not call the model") + + monkeypatch.setattr(agent, "_stream_once", fail_stream) + seen = [] + monkeypatch.setattr(agent, "on_text", seen.append) + + result = agent.run("hello") + + assert result == "Hello! What would you like to work on?" + assert seen == ["Hello! What would you like to work on?\n"] + + +def test_local_stream_uses_default_max_tokens(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + agent = Agent(root=tmp_path, provider="local-vibethinker") + captured = {} + + def fake_create(**kwargs): + captured.update(kwargs) + return iter(()) + + agent.client.chat.completions.create = fake_create + + agent._create_stream() + + assert captured["max_tokens"] == LOCAL_LLM_DEFAULT_MAX_TOKENS + + +def test_is_trivial_greeting(): + assert _is_trivial_greeting("hello") + assert _is_trivial_greeting("hey!") + assert not _is_trivial_greeting("hello, fix the tests") + + +def test_local_user_content_forces_no_think(): + content = _local_user_content("what do you think of devbot") + + assert content.startswith("/no_think") + assert "Answer directly and briefly" in content + assert "what do you think of devbot" in content + + +def test_local_run_wraps_user_message(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + agent = Agent(root=tmp_path, provider="local-vibethinker") + monkeypatch.setattr(agent, "_stream_once", lambda: ("ok", None)) + monkeypatch.setattr(agent, "on_text", lambda text: None) + + result = agent.run("what do you think of devbot") + + assert result == "ok" + assert agent.messages[1]["role"] == "user" + assert agent.messages[1]["content"].startswith("/no_think") + assert "what do you think of devbot" in agent.messages[1]["content"] + + +def test_local_run_does_not_wrap_when_thinking_enabled(tmp_path, monkeypatch): + _clear_local_env(monkeypatch) + monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False) + agent = Agent(root=tmp_path, provider="local-vibethinker") + agent.show_reasoning = True + monkeypatch.setattr(agent, "_stream_once", lambda: ("ok", None)) + monkeypatch.setattr(agent, "on_text", lambda text: None) + + result = agent.run("what do you think of devbot") + + assert result == "ok" + assert agent.messages[1]["content"] == "what do you think of devbot"