← refacto-feature-scout
"""Build approved ideas in an isolated git worktree; merge (and deploy) on `ship`."""
from __future__ import annotations
import json
import os
import re
import shutil
import subprocess
import time
from pathlib import Path
from string import Template
from typing import Any, Callable
from .claude_runner import ClaudeResult, extract_json, run_claude
from .config import BUILD_TIMEOUT_S, NUDGE_TIMEOUT_S, CLOUD, LOG_DIR, TEST_TIMEOUT_S, WORKTREE_DIR, load_env
from .projects import Project
from . import repos
from .store import Store
PROMPT_PATH = Path(__file__).parent / "prompts" / "build.md"
BUILD_TOOLS = ["Read", "Edit", "Write", "Glob", "Grep", "MultiEdit", "TodoWrite",
"Bash(npm *)", "Bash(npx *)", "Bash(node *)", "Bash(git *)", "Bash(ls *)", "Bash(cat *)",
"Bash(mkdir *)", "Bash(cp *)", "Bash(mv *)", "Bash(rm *)", "Bash(sed *)", "Bash(grep *)",
"Bash(find *)", "Bash(wc *)", "Bash(head *)", "Bash(tail *)", "Bash(echo *)", "Bash(cd *)",
"Bash(pwd)", "Bash(tsx *)", "Bash(sqlite3 *)", "Bash(diff *)", "Bash(true)"]
MAX_NUDGES = 2
NUDGE_PROMPT = (
"You stopped before finishing. You are running headless: no background task will ever notify you, so run "
"every remaining command in the foreground now and wait for it. If you started anything in the background, "
"do NOT wait for it or check on it again — treat it as something that will never finish, and do not start it "
"again. Continue the work in this worktree, make sure everything is committed on this branch, run the test "
"suite and typecheck, and then end with ONLY the final JSON report described in your original instructions."
)
def _has_report(text: str) -> bool:
try:
parsed = extract_json(text)
except ValueError:
return False
return isinstance(parsed, dict) and "status" in parsed
def slugify(title: str) -> str:
s = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
return s[:40] or "feature"
def git(repo: Path | str, *args: str, check: bool = True, timeout: int = 300) -> subprocess.CompletedProcess:
return subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True,
check=check, timeout=timeout)
def tracked_dirty(repo: Path) -> list[str]:
"""Modified/staged tracked files (untracked files are fine to leave alone)."""
out = git(repo, "status", "--porcelain", "--untracked-files=no").stdout
return [line for line in out.splitlines() if line.strip()]
def main_branch(repo: Path) -> str:
for cand in ("main", "master"):
if git(repo, "rev-parse", "--verify", "--quiet", cand, check=False).returncode == 0:
return cand
raise RuntimeError(f"{repo}: no main/master branch")
def prepare_worktree(project: Project, idea: dict[str, Any]) -> tuple[Path, str]:
branch = f"scout/{idea['id']}-{slugify(idea['title'])}"
wt = WORKTREE_DIR / project.key / f"{idea['id']}-{slugify(idea['title'])}"
base = main_branch(project.folder)
if wt.exists():
# A retry: keep the branch, refresh the worktree registration.
git(project.folder, "worktree", "prune", check=False)
if not (wt / ".git").exists():
shutil.rmtree(wt, ignore_errors=True)
if wt.exists() and (wt / ".git").exists():
# A retry whose earlier attempt built NOTHING (it never ran — a usage-limit pause, a
# crash before the first commit) still sits on the main of that earlier time. Move it
# to today's main so it does not collide with everything shipped since (#10, #11 on
# 2026-09-08 each lost a pass to that). Anything already built or in progress is kept.
ahead = git(wt, "rev-list", "--count", f"{base}..HEAD", check=False).stdout.strip()
if ahead == "0" and not tracked_dirty(wt):
git(wt, "reset", "-q", "--hard", base, check=False)
if not wt.exists():
wt.parent.mkdir(parents=True, exist_ok=True)
exists = git(project.folder, "rev-parse", "--verify", "--quiet", branch, check=False).returncode == 0
if exists:
git(project.folder, "worktree", "add", str(wt), branch)
else:
git(project.folder, "worktree", "add", "-b", branch, str(wt), base)
for rel in project.link_dirs:
src, dst = project.folder / rel, wt / rel
if src.exists() and not dst.exists():
dst.parent.mkdir(parents=True, exist_ok=True)
_link(src, dst)
return wt, branch
def _hardlink_tree(src: Path, dst: Path) -> None:
"""A copy of ``src`` at ``dst`` that shares file contents (hard links) but has its own
directory entries — instant and nearly free, and anything the builder does to it
(``npm ci`` empties the folder before reinstalling) stays in the work area."""
for root, dirs, files in os.walk(src):
rel = Path(root).relative_to(src)
(dst / rel).mkdir(parents=True, exist_ok=True)
for d in dirs:
sp = Path(root) / d
if sp.is_symlink():
os.symlink(os.readlink(sp), dst / rel / d)
dirs.remove(d)
for f in files:
sp, dp = Path(root) / f, dst / rel / f
if sp.is_symlink():
os.symlink(os.readlink(sp), dp)
else:
try:
os.link(sp, dp)
except OSError:
shutil.copy2(sp, dp)
def _link(src: Path, dst: Path) -> None:
"""Give the scratch copy the real dependencies without installing them again.
A hard-linked copy, not a symlink: on 2026-09-07 a builder ran ``npm ci`` in its work
area, which followed the symlink and emptied the shared node_modules of the main mirror
mid-install, so the ship's own test run found no test runner. Windows falls back to a
junction, and failing that to nothing at all — the builder is told how to install them.
"""
try:
_hardlink_tree(src, dst)
return
except (OSError, NotImplementedError):
shutil.rmtree(dst, ignore_errors=True)
try:
os.symlink(src, dst)
return
except (OSError, NotImplementedError):
pass
if os.name == "nt": # pragma: no cover - Windows only
try:
import _winapi
_winapi.CreateJunction(str(src), str(dst))
except Exception: # noqa: BLE001
pass
def _context_files(project: Project) -> str:
extra = [f"`{f}`" for f in project.context_files if f.lower() != "readme.md"]
if not extra:
return ""
if len(extra) == 1:
return f" and {extra[0]}"
return " and " + ", ".join(extra[:-1]) + f" and {extra[-1]}"
def _install_step(project: Project) -> str:
if project.install_hint:
return f"If this project's dependencies are missing or broken, run `{project.install_hint}`."
return ("If this project needs its dependencies installed before it will run, install them the "
"way this project already does it.")
def _test_step(project: Project) -> str:
if not project.test_command:
return ("This project has no test command yet. Add at least one test for what you built, "
"run it, and put the command you used in your report so it can be used from now on.")
checks = "".join(f" and `{c}`" for c in project.extra_checks)
return (f"Write tests in the same change. Run the full test suite with `{project.test_command}`"
f"{checks} and make them pass.")
def _never_run(project: Project) -> str:
if not project.never_run:
return ""
names = ", ".join(f"`{pat[len('Bash('):-1].rstrip('*').strip()}`" if pat.startswith("Bash(") else f"`{pat}`"
for pat in project.never_run)
return (f"\n- Never run {names}: it is blocked here and cannot work where you are running. Do not try to "
"install what it needs, and do not look for another way to run it. The owner's own checks run after you finish.")
REBASE_MARK = "needs rebasing onto the latest version"
def _rebase_note(idea: dict[str, Any]) -> str:
if not str(idea.get("error") or "").startswith(REBASE_MARK):
return ""
return ("\n## First, fit this work onto the latest version\n"
"This feature was built and passed its tests, but the project has moved on since and it no "
"longer applies cleanly (a rebase conflict). Before anything else: run `git rebase main` in this "
"worktree, resolve EVERY conflict so that both this feature and what main now has survive (never "
"throw away either side), `git add` the resolved files and `git rebase --continue` until the rebase "
"finishes, then run the test suite and typecheck and fix what the merge broke. The feature itself is "
"already built — do not redesign it. Then continue below, which may mean only the final report.\n")
def build_prompt(project: Project, idea: dict[str, Any], branch: str) -> str:
return Template(PROMPT_PATH.read_text()).substitute(
name=project.name, topic=project.topic, branch=branch, id=idea["id"], title=idea["title"],
summary=idea["summary"], user_value=idea.get("user_value") or "(see summary)",
spec=idea.get("spec") or "(no notes — use your judgement)",
context_files=_context_files(project), install_step=_install_step(project),
test_step=_test_step(project), test_command=project.test_command or "(none yet)",
never_run=_never_run(project), rebase_note=_rebase_note(idea),
house_rules=(f"\n## House rules for this project\n{project.house_rules.strip()}\n"
if project.house_rules.strip() else ""))
def _path(env: dict[str, str] | None = None) -> str:
"""Homebrew's folder isn't on the path of a program started outside a terminal."""
current = (env or os.environ).get("PATH", "")
if os.name == "nt": # pragma: no cover - Windows only
return current
return "/opt/homebrew/bin:/usr/local/bin:" + current
_MISSING_DEPS = ("not found", "Cannot find module", "command not found", "ENOENT", "MODULE_NOT_FOUND")
_TIMEOUT_FLAKE = ("Test timed out in", "Exceeded timeout of", "Timeout - Async callback")
def _run(cmd: str, cwd: Path, timeout: int) -> tuple[bool, str]:
try:
proc = subprocess.run(cmd, shell=True, cwd=str(cwd), capture_output=True, text=True,
timeout=timeout, env={**os.environ, "CI": "1", "PATH": _path()})
except subprocess.TimeoutExpired:
return False, f"`{cmd}` timed out"
return proc.returncode == 0, (proc.stdout + proc.stderr)[-3000:]
def run_tests(project: Project, cwd: Path, *, installer: Callable[[str, Path, int], tuple[bool, str]] = _run
) -> tuple[bool, str]:
"""The project's own check. If it fails because the dependencies are missing (a wiped
or never-installed node_modules), install them once with the project's own install
command and try again — a missing test runner is not a failing test."""
ok, tail = _run(project.test_command, cwd, TEST_TIMEOUT_S)
if not ok and project.install_hint and any(m in tail for m in _MISSING_DEPS):
iok, itail = installer(project.install_hint, cwd, TEST_TIMEOUT_S)
if not iok:
return False, f"dependencies could not be installed (`{project.install_hint}`): {itail[-1500:]}"
ok, tail = _run(project.test_command, cwd, TEST_TIMEOUT_S)
if not ok and any(m in tail for m in _TIMEOUT_FLAKE):
# A single test hitting its own time limit on a small, busy machine (scrypt password
# hashing took >5 s on the cloud box on 2026-09-07 while a deploy ran next door) is
# not the feature failing. One more run decides; a real failure fails twice.
ok2, tail2 = _run(project.test_command, cwd, TEST_TIMEOUT_S)
if ok2:
return True, "passed on a second run (the first run had a test time out on this machine):\n" + tail2[-1500:]
return False, tail2
return ok, tail
_LIMIT_SIGNS = ("session limit", "usage limit", "rate limit", "rate_limit", "429", "resets ", "overloaded",
"try again later", "capacity")
def is_usage_limit(error: str) -> bool:
"""Claude saying 'not now' (a Max plan session limit, a 429, an overloaded API)."""
e = (error or "").lower()
return any(sign in e for sign in _LIMIT_SIGNS)
def build_idea(project: Project, idea: dict[str, Any], store: Store, *,
runner: Callable[..., ClaudeResult] = run_claude,
tester: Callable[[Project, Path], tuple[bool, str]] = run_tests) -> dict[str, Any]:
"""Build one idea. Returns the report dict that was stored (status done/blocked/failed)."""
store.set_status(idea["id"], "building")
try:
if CLOUD:
repos.sync_main(project)
wt, branch = prepare_worktree(project, idea)
except Exception as e: # noqa: BLE001
report = {"status": "failed", "error": f"could not prepare a work area: {e}"}
store.set_status(idea["id"], "failed", build_report=json.dumps(report), error=report["error"])
return report
store.set_status(idea["id"], "building", build_branch=branch, build_dir=str(wt))
blocked = list(project.never_run)
from .config import build_model
res = runner(build_prompt(project, idea, branch), cwd=wt, allowed_tools=BUILD_TOOLS, max_turns=250,
timeout_s=BUILD_TIMEOUT_S, permission_mode="acceptEdits", disallowed_tools=blocked,
model=build_model(),
log_path=LOG_DIR / f"build-{project.key}-{idea['id']}.json")
cost = res.cost_usd
# A headless session that ends its turn early (e.g. "I'll wait for the notification") never
# wakes up on its own. Resume it and ask it to finish, up to twice, before calling it failed.
nudges = 0
while res.ok and not _has_report(res.text) and res.session_id and nudges < MAX_NUDGES:
nudges += 1
res = runner(NUDGE_PROMPT, cwd=wt, allowed_tools=BUILD_TOOLS, max_turns=120, timeout_s=NUDGE_TIMEOUT_S,
permission_mode="acceptEdits", resume=res.session_id, disallowed_tools=blocked,
model=build_model(),
log_path=LOG_DIR / f"build-{project.key}-{idea['id']}-nudge{nudges}.json")
cost += res.cost_usd
report: dict[str, Any]
if not res.ok and is_usage_limit(res.error):
# Claude's own usage limit (a Max plan resets every few hours): the idea did not fail,
# the account was busy. Leave it approved in its place and tell the caller to stop
# this pass rather than march down the queue marking everything failed.
report = {"status": "paused", "error": res.error, "cost_usd": cost, "project": project.key}
store.keep_approved(idea["id"], f"waiting for Claude's usage limit to reset: {res.error[:200]}")
return report
if not res.ok:
report = {"status": "failed", "error": res.error, "cost_usd": cost}
else:
try:
parsed = extract_json(res.text)
report = parsed if isinstance(parsed, dict) else {"status": "failed", "error": "no report"}
except ValueError:
report = {"status": "failed", "error": "the builder gave no final report", "tail": res.text[-800:]}
report["cost_usd"] = cost
report["nudges"] = nudges
# Independent verification — never trust the builder's own word.
if report.get("status") == "done":
problems = []
if tracked_dirty(wt):
# Finished work the builder forgot to commit would be LOST at ship time (only the
# branch's commits merge). Tracked files only — never sweep in node_modules and co.
git(wt, "add", "-u", check=False)
c = git(wt, "commit", "-q", "-m", f"Feature Scout #{idea['id']}: remaining changes", check=False)
if c.returncode != 0 or tracked_dirty(wt):
problems.append("uncommitted changes were left in the work area and could not be committed")
else:
report["note"] = "the builder left some changes uncommitted; they were committed for it"
# Work counts if the branch carries commits beyond where it started from — including
# commits made by an EARLIER attempt (a retry that only re-tests moves nothing).
base = main_branch(project.folder)
ahead = git(wt, "rev-list", "--count", f"{base}..HEAD", check=False).stdout.strip()
if not ahead.isdigit() or int(ahead) == 0:
problems.append("no commit was made")
ok, tail = tester(project, wt)
if not ok:
problems.append("the test suite fails: " + tail[-600:])
if problems:
report["status"] = "failed"
report["blocker"] = "; ".join(problems)
report["error"] = report["blocker"]
report["project"] = project.key # so what it cost can be attributed
status = {"done": "built", "blocked": "blocked"}.get(report.get("status", "failed"), "failed")
store.set_status(idea["id"], status, build_report=json.dumps(report),
error=None if status == "built" else str(report.get("blocker") or report.get("error") or ""))
return report
def ship_idea(project: Project, idea: dict[str, Any], store: Store, *,
tester: Callable[[Project, Path], tuple[bool, str]] = run_tests,
deployer: Callable[[Project], tuple[bool, str]] | None = None) -> dict[str, Any]:
"""Merge a built branch into the owner's main checkout, run tests, deploy if configured."""
repo, branch = project.folder, idea.get("build_branch")
if not branch:
return {"ok": False, "retry": False, "message": "no build branch recorded"}
if CLOUD:
return _ship_cloud(project, idea, store, tester=tester, deployer=deployer)
dirty = tracked_dirty(repo)
if dirty:
return {"ok": False, "retry": True,
"message": f"your {project.name} working copy has unsaved changes ({len(dirty)} files); "
"I'll try again on the next pass"}
base = main_branch(repo)
current = git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
if current != base:
return {"ok": False, "retry": True,
"message": f"your {project.name} checkout is on branch {current}, not {base}; I'll retry later"}
before = git(repo, "rev-parse", "HEAD").stdout.strip()
store.set_status(idea["id"], "built", merge_base_sha=before) # so `undo` can put it back
merge = git(repo, "merge", "--no-ff", "--no-edit", "-m", f"Merge {branch} (Feature Scout #{idea['id']})",
branch, check=False)
if merge.returncode != 0:
git(repo, "merge", "--abort", check=False)
return {"ok": False, "retry": False, "message": f"merge conflict: {merge.stdout[-500:]}{merge.stderr[-500:]}"}
ok, tail = tester(project, repo)
if not ok:
git(repo, "reset", "--hard", before)
store.set_status(idea["id"], "failed", error="tests failed after merge: " + tail[-600:])
return {"ok": False, "retry": False, "message": "tests failed after merging, so I undid the merge: " + tail[-600:]}
deploy_note = ""
if project.deploy_command:
deployer = deployer or run_deploy
dok, dmsg = deployer(project)
if not dok:
store.set_status(idea["id"], "built", error="deploy failed: " + dmsg[-600:])
return {"ok": False, "retry": True,
"message": f"merged into {base}, but the deploy failed; I'll retry the deploy next pass: {dmsg[-600:]}"}
deploy_note = f" and deployed to {project.url}"
store.set_status(idea["id"], "shipped", error=None)
wt = idea.get("build_dir")
if wt and Path(wt).exists():
git(repo, "worktree", "remove", "--force", wt, check=False)
return {"ok": True, "retry": False, "message": f"#{idea['id']} {idea['title']} is merged into {project.name}{deploy_note}."}
def _ship_cloud(project: Project, idea: dict[str, Any], store: Store, *, tester, deployer,
syncer: Callable[[Project], str] = repos.sync_main,
pusher: Callable[[Project], tuple[bool, str]] = repos.push_main) -> dict[str, Any]:
"""The cloud mirror is disposable: take Library Mode's latest, put the feature on top as one
straight line, test, push it back as a new version, then deploy."""
repo, branch = project.folder, idea["build_branch"]
wt = idea.get("build_dir")
for attempt in (1, 2):
try:
syncer(project)
except Exception as e: # noqa: BLE001
return {"ok": False, "retry": True, "message": f"could not fetch the latest version from Library Mode: {e}"}
if wt and Path(wt).exists():
rb = git(wt, "rebase", "main", check=False)
if rb.returncode != 0:
git(wt, "rebase", "--abort", check=False)
detail = (rb.stdout + rb.stderr)[-500:]
# Not a dead end: the builder can fit it onto the new version. Back to the
# queue (same place, same branch) with the conflict noted for its next attempt.
store.send_back_for_rebase(idea["id"], detail)
return {"ok": False, "retry": False, "rebuild": True,
"message": "the feature no longer fits on top of the latest version (conflict), so I have "
"handed it back to the builder to fit it onto the newest version; it will ship "
"again by itself once that is done. The conflict: " + detail}
before = git(repo, "rev-parse", "HEAD").stdout.strip()
ff = git(repo, "merge", "--ff-only", branch, check=False)
if ff.returncode != 0:
return {"ok": False, "retry": False, "message": "could not add the feature on top of the latest version: "
+ (ff.stdout + ff.stderr)[-500:]}
ok, tail = tester(project, repo)
if not ok:
git(repo, "reset", "--hard", before)
store.set_status(idea["id"], "failed", error="tests failed on top of the latest version: " + tail[-600:])
return {"ok": False, "retry": False,
"message": "tests failed once the feature sat on the latest version, so it was not shipped: " + tail[-600:]}
pushed, msg = pusher(project)
if pushed:
break
git(repo, "reset", "--hard", before)
if attempt == 2 or "history_protected" in msg:
return {"ok": False, "retry": True, "message": f"Library Mode refused the new version: {msg[-400:]}"}
# someone saved a newer version in between: loop once more from the fresh tip
deploy_note = ""
if project.deploy_command:
deployer = deployer or run_deploy
dok, dmsg = deployer(project)
if not dok:
store.set_status(idea["id"], "built", error="deploy failed: " + dmsg[-600:])
return {"ok": False, "retry": True,
"message": f"saved to Library Mode as a new version, but the deploy failed; I'll retry it next pass: {dmsg[-600:]}"}
deploy_note = f" and deployed to {project.url}"
store.set_status(idea["id"], "shipped", error=None)
if wt and Path(wt).exists():
git(repo, "worktree", "remove", "--force", wt, check=False)
git(repo, "branch", "-D", branch, check=False)
return {"ok": True, "retry": False,
"message": f"#{idea['id']} {idea['title']} is now a new version of {project.name} on Library Mode{deploy_note}. "
f"Your Mac copy is behind by one version: run `librarymode latest` (or let the next Claude Code session pull it) before saving new work there."}
def run_deploy(project: Project) -> tuple[bool, str]:
env = dict(os.environ)
env["PATH"] = _path(env)
cfg = Path.home() / ".fly" / "config.yml"
if env.get("FLY_API_TOKEN"):
env["FLY_ACCESS_TOKEN"] = env["FLY_API_TOKEN"]
elif cfg.exists():
m = re.search(r"^access_token:\s*(\S+)", cfg.read_text(), re.M)
if m:
env["FLY_ACCESS_TOKEN"] = m.group(1)
try:
proc = subprocess.run(project.deploy_command, shell=True, cwd=str(project.folder), capture_output=True,
text=True, timeout=30 * 60, env=env)
except subprocess.TimeoutExpired:
return False, "deploy timed out"
return proc.returncode == 0, (proc.stdout + proc.stderr)[-2000:]
def undo_idea(project: Project, idea: dict[str, Any], store: Store) -> dict[str, Any]:
"""Put the project back the way it was before a change was added.
If nothing has happened since, the change is lifted straight off. If the owner has done
work on top of it, that work is kept and only this change is reversed.
"""
repo = project.folder
before = idea.get("merge_base_sha")
if idea.get("status") != "shipped" or not before:
return {"ok": False, "message": f"#{idea['id']} isn't something I've added to your project, "
"so there's nothing to take back."}
dirty = tracked_dirty(repo)
if dirty:
return {"ok": False, "message": f"You have unsaved changes in {project.name} ({len(dirty)} "
"files). Save or put those aside first, then try again."}
base = main_branch(repo)
current = git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
if current != base:
return {"ok": False, "message": f"Your {project.name} folder is on {current}, not {base}."}
# Everything added on the main line since the project was last untouched by this change.
since = git(repo, "rev-list", "--first-parent", f"{before}..HEAD", check=False)
added = since.stdout.split() if since.returncode == 0 else []
if not added:
return {"ok": False, "message": f"#{idea['id']} isn't in your project any more."}
if len(added) == 1:
git(repo, "reset", "--hard", before)
how = "Your project is back exactly as it was."
else:
commit = added[-1] # the one that put this change in
parents = git(repo, "rev-list", "--parents", "-n", "1", commit).stdout.split()
args = ["revert", "--no-edit"] + (["-m", "1"] if len(parents) > 2 else []) + [commit]
undone = git(repo, *args, check=False)
if undone.returncode != 0:
git(repo, "revert", "--abort", check=False)
return {"ok": False, "message": "I couldn't take it back cleanly because of work done "
"since. Nothing was changed."}
how = "Everything you've done since is untouched; just this change was reversed."
store.set_status(idea["id"], "rejected", error="taken back by you")
return {"ok": True, "message": f"#{idea['id']} {idea['title']} has been taken back out. {how}"}
def machine_started_at() -> float | None:
"""When this machine last booted, where the system will say."""
try:
with open("/proc/uptime") as f: # Linux, which is where it runs
return time.time() - float(f.read().split()[0])
except (OSError, ValueError, IndexError):
return None
def was_interrupted(store: Store, idea: dict[str, Any], now: float | None = None,
booted: float | None = None) -> bool:
"""True when an idea says it is building but nothing is actually building it.
A build killed with the machine never gets to finish, so its run is left open too — which
makes "did the run finish?" exactly the wrong question. What gives it away is that the work
began before this machine did, or longer ago than a build is ever allowed to take.
"""
if idea.get("status") != "building":
return False
now = time.time() if now is None else now
booted = machine_started_at() if booted is None else booted
run = store.last_run(f"build:{idea['id']}")
started = float((run or {}).get("started_at") or 0)
return (run is None
or bool(run.get("finished_at"))
or (booted is not None and started < booted)
or (now - started) > BUILD_TIMEOUT_S + 600)
def pending_ships(store: Store) -> list[dict[str, Any]]:
"""Built ideas the owner has asked for. The old marker is still read so that anything
already waiting when this version arrives is not stranded."""
return [i for i in store.ideas(status="built")
if i.get("ship_requested_at") or i.get("error") == "SHIP_REQUESTED"
or (i.get("error") or "").startswith("deploy failed")]