← refacto-feature-scout
import json
import subprocess
from pathlib import Path
import pytest
from scout import build as b
from scout.claude_runner import ClaudeResult
from scout.projects import Project
def make_repo(tmp_path: Path) -> Path:
repo = tmp_path / "repo"
repo.mkdir()
subprocess.run(["git", "init", "-q", "-b", "main", str(repo)], check=True)
subprocess.run(["git", "-C", str(repo), "config", "user.email", "t@t"], check=True)
subprocess.run(["git", "-C", str(repo), "config", "user.name", "t"], check=True)
(repo / "README.md").write_text("hello\n")
(repo / "node_modules").mkdir()
subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
subprocess.run(["git", "-C", str(repo), "commit", "-q", "-m", "init"], check=True)
return repo
def project(repo: Path) -> Project:
return Project(key="terminal", name="PolyTerm", areas=["Polymarket"], folder=repo, url=None,
what_it_is="d", test_command="true", deploy_command=None,
link_dirs=("node_modules",))
@pytest.fixture(autouse=True)
def worktree_dir(tmp_path, monkeypatch):
monkeypatch.setattr(b, "WORKTREE_DIR", tmp_path / "wt")
monkeypatch.setattr(b, "LOG_DIR", tmp_path / "logs")
def committing_runner(text: str):
"""A fake builder that commits a file in the worktree and returns `text`."""
def runner(prompt, *, cwd, **kw):
(Path(cwd) / "feature.txt").write_text("built\n")
subprocess.run(["git", "-C", str(cwd), "add", "."], check=True)
subprocess.run(["git", "-C", str(cwd), "commit", "-q", "-m", "feat"], check=True)
return ClaudeResult(ok=True, text=text, cost_usd=1.0)
return runner
def test_build_success_uses_worktree_and_verifies(store, tmp_path):
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
store.set_status(iid, "approved")
report = b.build_idea(p, store.idea(iid), store, runner=committing_runner(json.dumps(
{"status": "done", "summary": "ok", "tests": {"passed": True}})), tester=lambda p, cwd: (True, "ok"))
idea = store.idea(iid)
assert report["status"] == "done" and idea["status"] == "built"
assert idea["build_branch"] == f"scout/{iid}-whale-feed"
wt = Path(idea["build_dir"])
assert (wt / "feature.txt").exists()
deps = wt / "node_modules" # its own hard-linked copy, never a symlink (see _link)
assert deps.is_dir() and not deps.is_symlink()
# the owner's checkout is untouched
assert not (repo / "feature.txt").exists()
assert subprocess.run(["git", "-C", str(repo), "rev-parse", "--abbrev-ref", "HEAD"], capture_output=True,
text=True).stdout.strip() == "main"
def test_build_rejects_builder_claim_when_tests_fail(store, tmp_path):
repo = make_repo(tmp_path)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
report = b.build_idea(project(repo), store.idea(iid), store,
runner=committing_runner('{"status":"done","summary":"trust me"}'),
tester=lambda p, cwd: (False, "1 failed"))
assert report["status"] == "failed" and "test suite fails" in report["blocker"]
assert store.idea(iid)["status"] == "failed"
def test_build_without_commit_is_failed(store, tmp_path):
repo = make_repo(tmp_path)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
report = b.build_idea(project(repo), store.idea(iid), store,
runner=lambda *a, **k: ClaudeResult(ok=True, text='{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (True, ""))
assert report["status"] == "failed" and "no commit" in report["blocker"]
def test_blocked_and_runner_error(store, tmp_path):
repo = make_repo(tmp_path)
iid = store.add_idea("terminal", "Kalshi thing", "s", evidence=[{"url": "https://e", "note": ""}])
report = b.build_idea(project(repo), store.idea(iid), store,
runner=lambda *a, **k: ClaudeResult(ok=True, text='{"status":"blocked","blocker":"no adapter"}'))
assert report["status"] == "blocked" and store.idea(iid)["status"] == "blocked"
assert store.idea(iid)["error"] == "no adapter"
iid2 = store.add_idea("terminal", "Other", "s", evidence=[{"url": "https://e", "note": ""}])
report = b.build_idea(project(repo), store.idea(iid2), store, runner=lambda *a, **k: ClaudeResult(ok=False, error="timed out"))
assert report["status"] == "failed" and store.idea(iid2)["status"] == "failed"
def test_ship_merges_then_refuses_dirty_tree(store, tmp_path):
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
b.build_idea(p, store.idea(iid), store, runner=committing_runner('{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (True, ""))
# dirty checkout → wait
(repo / "README.md").write_text("edited\n")
r = b.ship_idea(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""))
assert r["ok"] is False and r["retry"] is True and "unsaved changes" in r["message"]
subprocess.run(["git", "-C", str(repo), "checkout", "--", "README.md"], check=True)
# clean → merged, tests run in the main checkout, worktree removed
r = b.ship_idea(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""))
assert r["ok"] and (repo / "feature.txt").exists()
assert store.idea(iid)["status"] == "shipped"
assert not Path(store.idea(iid)["build_dir"]).exists()
def test_ship_undoes_merge_when_tests_fail(store, tmp_path):
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
b.build_idea(p, store.idea(iid), store, runner=committing_runner('{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (True, ""))
head = subprocess.run(["git", "-C", str(repo), "rev-parse", "HEAD"], capture_output=True, text=True).stdout
r = b.ship_idea(p, store.idea(iid), store, tester=lambda p, cwd: (False, "boom"))
assert r["ok"] is False and "undid the merge" in r["message"]
assert subprocess.run(["git", "-C", str(repo), "rev-parse", "HEAD"], capture_output=True, text=True).stdout == head
assert not (repo / "feature.txt").exists()
def test_ship_deploy_failure_keeps_merge_and_retries(store, tmp_path):
repo = make_repo(tmp_path)
p = Project(key="terminal", name="PolyTerm", areas=["Polymarket"], folder=repo, url="https://x",
what_it_is="d", test_command="true", deploy_command="false")
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
b.build_idea(p, store.idea(iid), store, runner=committing_runner('{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (True, ""))
r = b.ship_idea(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""), deployer=lambda p: (False, "fly down"))
assert r["retry"] and "deploy failed" in r["message"] and (repo / "feature.txt").exists()
assert store.idea(iid)["status"] == "built" and store.idea(iid)["error"].startswith("deploy failed")
assert [i["id"] for i in b.pending_ships(store)] == [iid]
r = b.ship_idea(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""), deployer=lambda p: (True, "ok"))
# already merged: a second merge is a no-op, deploy succeeds
assert r["ok"] and store.idea(iid)["status"] == "shipped"
def test_cloud_ship_rebases_pushes_and_retries_once(store, tmp_path, monkeypatch):
"""Cloud mode: Library Mode moved on after the build; the ship must land on top of the new tip."""
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
b.build_idea(p, store.idea(iid), store, runner=committing_runner('{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (True, ""))
# meanwhile Ken saved a new version: main gains a commit the branch doesn't have
(repo / "other.txt").write_text("ken\n")
subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
subprocess.run(["git", "-C", str(repo), "commit", "-q", "-m", "Saved"], check=True)
monkeypatch.setattr(b, "CLOUD", True)
pushes = []
def pusher(platform):
pushes.append(1)
return (len(pushes) == 2), "rejected: fetch first" if len(pushes) == 1 else "ok"
r = b.ship_idea(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""), deployer=lambda p: (True, "ok"))
# `_ship_cloud` takes syncer/pusher via ship_idea's defaults only in production; call it directly here
assert r["ok"] is False # default syncer refuses outside the cloud → retry later, nothing broken
r = b._ship_cloud(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""), deployer=lambda p: (True, "ok"),
syncer=lambda p: "tip", pusher=pusher)
assert r["ok"] and len(pushes) == 2 and "new version" in r["message"]
log = subprocess.run(["git", "-C", str(repo), "log", "--oneline", "--no-merges", "-3"], capture_output=True, text=True).stdout
assert "feat" in log and "Saved" in log and "Merge" not in log # one straight line
assert (repo / "feature.txt").exists() and (repo / "other.txt").exists()
assert store.idea(iid)["status"] == "shipped"
def test_cloud_ship_conflict_is_reported_not_forced(store, tmp_path, monkeypatch):
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
b.build_idea(p, store.idea(iid), store, runner=committing_runner('{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (True, ""))
(repo / "feature.txt").write_text("conflicting\n")
subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
subprocess.run(["git", "-C", str(repo), "commit", "-q", "-m", "Saved"], check=True)
monkeypatch.setattr(b, "CLOUD", True)
r = b._ship_cloud(p, store.idea(iid), store, tester=lambda p, cwd: (True, ""), deployer=None,
syncer=lambda p: "tip", pusher=lambda p: (True, "ok"))
assert r["ok"] is False and "conflict" in r["message"] and r.get("rebuild") is True
assert (repo / "feature.txt").read_text() == "conflicting\n"
# 2026-09-07: #31 and #32 hit exactly this and sat stuck as 'built'. Now the feature goes
# back to the builder, keeping its branch and its place, with the conflict noted.
idea = store.idea(iid)
assert idea["status"] == "approved" and idea["error"].startswith(b.REBASE_MARK)
assert idea["build_branch"] and idea["ship_requested_at"] is None
prompt = b.build_prompt(p, idea, idea["build_branch"])
assert "fit this work onto the latest version" in prompt and "git rebase main" in prompt
assert prompt.index("fit this work onto") < prompt.index("## How to work")
clean = b.build_prompt(p, {**idea, "error": None}, idea["build_branch"])
assert "rebase" not in clean.lower() and "$" not in clean
def test_build_nudges_a_session_that_stopped_early(store, tmp_path):
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
calls = []
def runner(prompt, *, cwd, resume=None, **kw):
calls.append(resume)
if resume is None:
return ClaudeResult(ok=True, text="I'll wait for the background task notification.", cost_usd=1.0, session_id="sess-1")
(Path(cwd) / "feature.txt").write_text("built\n")
subprocess.run(["git", "-C", str(cwd), "add", "."], check=True)
subprocess.run(["git", "-C", str(cwd), "commit", "-q", "-m", "feat"], check=True)
return ClaudeResult(ok=True, text='{"status":"done","summary":"finished after nudge"}', cost_usd=0.5, session_id="sess-1")
report = b.build_idea(p, store.idea(iid), store, runner=runner, tester=lambda p, cwd: (True, ""))
assert calls == [None, "sess-1"] # resumed the same session once
assert report["status"] == "done" and report["nudges"] == 1 and report["cost_usd"] == 1.5
assert store.idea(iid)["status"] == "built"
def test_build_gives_up_after_two_nudges(store, tmp_path):
repo = make_repo(tmp_path)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
runner = lambda *a, **k: ClaudeResult(ok=True, text="still waiting...", cost_usd=1.0, session_id="s")
report = b.build_idea(project(repo), store.idea(iid), store, runner=runner)
assert report["status"] == "failed" and report["nudges"] == 2 and "no final report" in report["error"]
def test_a_build_cut_off_by_a_restart_goes_back_in_the_queue_in_its_old_place(store, monkeypatch):
"""2026-09-07: #14 and #15 sat in 'building' for hours after deploys restarted the machine
mid-build; the queue only ever picks 'approved', so they were stranded."""
import time as _time
from scout import cli
early = store.add_idea("terminal", "Older approval", "s", evidence=[{"url": "https://e", "note": ""}])
later = store.add_idea("terminal", "Newer approval", "s", evidence=[{"url": "https://e", "note": ""}])
store.set_status(early, "approved")
store.conn.execute("UPDATE ideas SET decided_at=? WHERE id=?", (int(_time.time()) - 3600, early))
store.set_status(later, "approved")
store.set_status(early, "building", build_branch="scout/1-older")
run_id = store.start_run(f"build:{early}")
monkeypatch.delenv("SCOUT_PAUSE_BUILDS", raising=False)
cli.cmd_build(store, limit=0) # a pass that builds nothing, just tidies up
rows = {i["id"]: i for i in store.ideas()}
assert rows[early]["status"] == "approved" and "interrupted" in rows[early]["error"]
assert rows[early]["build_branch"] == "scout/1-older", "the branch is kept so the retry continues its work"
assert rows[early]["decided_at"] < rows[later]["decided_at"], "it keeps its turn, not sent to the back"
run = store.conn.execute("SELECT finished_at, ok FROM runs WHERE id=?", (run_id,)).fetchone()
assert run[0] is not None and run[1] == 0
assert store.requeue_interrupted_builds() == [], "nothing left to tidy"
def test_the_builder_is_forbidden_the_projects_never_run_commands_and_nudges_are_short(store, tmp_path):
"""2026-09-07: four builds in a row sat for the full 2.5 h inside a nudge, waiting on a Playwright
run that can never finish on the cloud machine. The block is enforced by Claude's permission
system on every call, and a nudge only gets 20 minutes."""
from dataclasses import replace
from scout.config import BUILD_TIMEOUT_S, NUDGE_TIMEOUT_S
repo = make_repo(tmp_path)
p = replace(project(repo), never_run=["Bash(npx playwright*)", "Bash(npm run verify*)"])
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
seen = []
def runner(prompt, *, cwd, resume=None, **kw):
seen.append((resume, kw.get("disallowed_tools"), kw.get("timeout_s")))
if resume is None:
assert "Never run `npx playwright`, `npm run verify`" in prompt
return ClaudeResult(ok=True, text="Waiting for the Playwright run in the background.", cost_usd=1.0, session_id="s")
return ClaudeResult(ok=True, text='{"status":"blocked","summary":"x","blocker":"y"}', cost_usd=0.1, session_id="s")
b.build_idea(p, store.idea(iid), store, runner=runner, tester=lambda p, cwd: (True, ""))
assert seen == [(None, ["Bash(npx playwright*)", "Bash(npm run verify*)"], BUILD_TIMEOUT_S),
("s", ["Bash(npx playwright*)", "Bash(npm run verify*)"], NUDGE_TIMEOUT_S)]
assert NUDGE_TIMEOUT_S <= 30 * 60 < BUILD_TIMEOUT_S
def test_a_project_with_nothing_blocked_gets_no_such_rule(store, tmp_path):
repo = make_repo(tmp_path)
p = project(repo)
prompt = b.build_prompt(p, {"id": 1, "title": "t", "summary": "s", "user_value": "u", "spec": "x"}, "scout/1-t")
assert "Never run" not in prompt and "$" not in prompt
def test_a_work_area_gets_its_own_copy_of_the_dependencies_not_a_link(tmp_path):
"""2026-09-07: a builder's `npm ci` followed the symlink and emptied the main mirror's
node_modules mid-install; the ship then found no test runner. A hard-linked copy keeps
the builder's changes in the work area."""
src = tmp_path / "node_modules"; (src / ".bin").mkdir(parents=True)
(src / ".bin" / "vitest").write_text("#!/bin/sh\necho ok\n")
(src / "pkg.txt").write_text("v1")
dst = tmp_path / "wt" / "node_modules"; dst.parent.mkdir()
b._link(src, dst)
assert not dst.is_symlink() and (dst / ".bin" / "vitest").read_text().startswith("#!")
assert (dst / "pkg.txt").stat().st_ino == (src / "pkg.txt").stat().st_ino, "shares content, no copy cost"
import shutil
shutil.rmtree(dst) # what `npm ci` does first
assert (src / ".bin" / "vitest").exists(), "the shared copy is untouched"
def test_missing_dependencies_are_installed_once_and_tests_rerun(tmp_path):
from dataclasses import replace
repo = make_repo(tmp_path)
marker = tmp_path / "installed"
p = replace(project(repo), test_command=f"test -f {marker} && echo ok || (echo 'sh: 1: vitest: not found'; exit 127)",
install_hint=f"touch {marker}")
calls = []
def installer(cmd, cwd, timeout):
calls.append(cmd)
return b._run(cmd, cwd, timeout)
ok, tail = b.run_tests(p, repo, installer=installer)
assert ok and calls == [f"touch {marker}"]
marker.unlink()
p2 = replace(p, install_hint=f"false")
ok, tail = b.run_tests(p2, repo)
assert not ok and "could not be installed" in tail
p3 = replace(p, test_command="echo 'a real failure'; exit 1")
calls.clear()
ok, tail = b.run_tests(p3, repo, installer=installer)
assert not ok and calls == [], "a genuine test failure does not trigger an install"
def test_a_test_that_merely_timed_out_gets_one_more_run(tmp_path):
"""2026-09-07: #36 failed to build because two password-hashing tests took over 5 s on the
busy cloud machine; the feature was fine (637 tests passed for the builder)."""
from dataclasses import replace
repo = make_repo(tmp_path)
counter = tmp_path / "runs"
counter.write_text("0")
cmd = (f"n=$(cat {counter}); n=$((n+1)); echo $n > {counter}; "
f"if [ $n -eq 1 ]; then echo 'Error: Test timed out in 5000ms.'; exit 1; fi; echo fine")
ok, tail = b.run_tests(replace(project(repo), test_command=cmd), repo)
assert ok and "second run" in tail and counter.read_text().strip() == "2"
counter.write_text("0")
always = f"n=$(cat {counter}); echo $((n+1)) > {counter}; echo 'Error: Test timed out in 5000ms.'; exit 1"
ok, tail = b.run_tests(replace(project(repo), test_command=always), repo)
assert not ok and counter.read_text().strip() == "2", "a repeat timeout is a real failure, tried exactly twice"
def test_a_retry_whose_work_was_committed_last_time_still_counts(store, tmp_path):
"""2026-09-08: #36's retry only re-ran the tests (the feature was already committed by the
first attempt), moved no commit, and was called 'no commit was made'."""
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
first = b.build_idea(p, store.idea(iid), store, runner=committing_runner('{"status":"done","summary":"x"}'),
tester=lambda p, cwd: (False, "flaky"))
assert first["status"] == "failed" and "test suite" in first["blocker"]
store.set_status(iid, "approved")
again = b.build_idea(p, store.idea(iid), store,
runner=lambda *a, **k: ClaudeResult(ok=True, text='{"status":"done","summary":"re-tested"}'),
tester=lambda p, cwd: (True, ""))
assert again["status"] == "done" and store.idea(iid)["status"] == "built"
def test_changes_the_builder_forgot_to_commit_are_committed_for_it(store, tmp_path):
"""2026-09-07: #33 was built twice and failed twice for 'uncommitted changes were left'
(a file-mode flip on scripts/deploy.sh). Those changes would vanish at ship time, so the
scout commits tracked leftovers itself — never untracked folders like node_modules."""
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
def runner(prompt, *, cwd, **kw):
(Path(cwd) / "feature.txt").write_text("built\n")
subprocess.run(["git", "-C", str(cwd), "add", "feature.txt"], check=True)
subprocess.run(["git", "-C", str(cwd), "commit", "-q", "-m", "feat"], check=True)
(Path(cwd) / "feature.txt").write_text("built, then tweaked\n") # left uncommitted
(Path(cwd) / "scratch").mkdir(); (Path(cwd) / "scratch" / "junk").write_text("x")
return ClaudeResult(ok=True, text='{"status":"done","summary":"x"}', session_id="s")
report = b.build_idea(p, store.idea(iid), store, runner=runner, tester=lambda p, cwd: (True, ""))
assert report["status"] == "done" and "committed for it" in report["note"]
wt = Path(store.idea(iid)["build_dir"])
log = subprocess.run(["git", "-C", str(wt), "log", "--oneline", "-3"], capture_output=True, text=True).stdout
assert "remaining changes" in log
status = subprocess.run(["git", "-C", str(wt), "status", "--short"], capture_output=True, text=True).stdout
assert " M " not in status and "?? scratch/" in status, "tracked leftovers committed; untracked junk left alone"
def test_a_retry_that_built_nothing_starts_from_todays_main(store, tmp_path):
"""2026-09-08: #10 and #11 got work areas during the usage-limit pause, hours before
several ships; their real build then collided with everything since. An untouched work
area moves to the latest main; one with work on it is left alone."""
repo = make_repo(tmp_path)
p = project(repo)
iid = store.add_idea("terminal", "Whale feed", "s", evidence=[{"url": "https://e", "note": ""}])
wt, branch = b.prepare_worktree(p, store.idea(iid))
old_tip = subprocess.run(["git", "-C", str(wt), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip()
(repo / "later.txt").write_text("shipped meanwhile\n")
subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
subprocess.run(["git", "-C", str(repo), "commit", "-q", "-m", "Something shipped"], check=True)
new_tip = subprocess.run(["git", "-C", str(repo), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip()
wt2, _ = b.prepare_worktree(p, store.idea(iid))
assert wt2 == wt
assert subprocess.run(["git", "-C", str(wt), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip() == new_tip
assert (wt / "later.txt").exists()
# With work on the branch, nothing moves (the rebase path handles that case).
(wt / "feature.txt").write_text("built\n")
subprocess.run(["git", "-C", str(wt), "add", "."], check=True)
subprocess.run(["git", "-C", str(wt), "commit", "-q", "-m", "feat"], check=True)
built_tip = subprocess.run(["git", "-C", str(wt), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip()
(repo / "later2.txt").write_text("x\n")
subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
subprocess.run(["git", "-C", str(repo), "commit", "-q", "-m", "More shipped"], check=True)
b.prepare_worktree(p, store.idea(iid))
assert subprocess.run(["git", "-C", str(wt), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip() == built_tip
assert old_tip != new_tip