← refacto-feature-scout
"""`scout` command: daily / poll / scan / build / status / add / review / watch ..."""
from __future__ import annotations
import argparse
import os
import sys
import threading
import traceback
from datetime import datetime
from pathlib import Path
from typing import Any
from . import build as buildmod
from .config import CLOUD, DB_PATH, LOG_DIR, ensure_dirs
from .locks import Busy, Lock
from .llm import HOW_TO_FIX, NoLLMAccess, check_access, require_llm_access, save_api_key
from .projects import ConfigProblem, Project, load_projects, validate
from . import repos
from .scan import run_scan
from .store import Store, by_value
def log(msg: str) -> None:
print(f"{datetime.now():%Y-%m-%d %H:%M:%S} {msg}", flush=True)
def alert(store: Store | None, subject: str, body: str) -> None:
"""Something went wrong. It is logged, written to a file, and put on the board."""
log(f"ALERT {subject}: {body[:300]}")
try:
LOG_DIR.mkdir(parents=True, exist_ok=True)
path = LOG_DIR / f"problem-{datetime.now():%Y%m%d-%H%M%S}.log"
path.write_text(f"{subject}\n\n{body}")
print(f"Something went wrong: {subject}. The details are in {path}")
except OSError:
pass
if store is not None:
try:
store.notice("problem", f"{subject}: {body[-600:]}")
except Exception as e: # noqa: BLE001
log(f"could not record the problem on the board: {e}")
_project_cache: dict[str, Project] | None = None
def known_projects(refresh: bool = False) -> dict[str, Project]:
"""Every project this computer looks after. A broken settings file must not stop a
command that doesn't need it, so the problem is reported and the rest carries on."""
global _project_cache
if _project_cache is None or refresh:
try:
_project_cache = load_projects()
except ConfigProblem as e:
log(str(e))
_project_cache = {}
return _project_cache
def project_for(idea: dict[str, Any]) -> Project | None:
p = known_projects().get(idea["platform"])
if p is None:
log(f"#{idea['id']} belongs to {idea['platform']}, which isn't set up on this computer — skipping")
return p
def chosen_projects(args: argparse.Namespace) -> list[Project]:
"""The projects a command applies to: the ones named, or all of them."""
known = known_projects()
keys = list(getattr(args, "project", None) or []) + list(getattr(args, "platform", None) or [])
if not keys:
return list(known.values())
out = []
for key in keys:
if key not in known:
raise ConfigProblem(f'I don\'t know a project called "{key}". '
f'Ones I know: {", ".join(sorted(known)) or "none yet"}.')
out.append(known[key])
return out
def spent_today(store: Store, project: Project) -> float:
midnight = int(datetime.now().replace(hour=0, minute=0, second=0, microsecond=0).timestamp())
return store.spend_since(midnight, project.key)
def over_spend_limit(store: Store, project: Project) -> str | None:
"""The reason to stop, or None. A limit of zero means the person set no limit."""
if project.spend_limit_usd <= 0:
return None
spent = spent_today(store, project)
if spent >= project.spend_limit_usd:
return (f"{project.name}: ${spent:.2f} of Claude has been used today, which is the "
f"${project.spend_limit_usd:.2f} limit you set. Nothing more will run until "
"tomorrow — raise the limit in this project's settings to carry on.")
return None
# ---- commands ---------------------------------------------------------------
def cmd_dry_run(store: Store, projects: list[Project]) -> int:
"""Show exactly what would be asked, and spend nothing doing it."""
from .scan import build_prompt, prior_titles, repo_context
for p in projects:
print(f"===== {p.name} ({p.key}) =====")
print(build_prompt(p, repo_context(p),
"(the posts and pages gathered on the day would go here)",
prior_titles(store, p), count=p.ideas_per_scan))
return 0
def cmd_scan(store: Store, projects: list[Project]) -> list[dict[str, Any]]:
results = []
for p in projects:
key = p.key
stop = over_spend_limit(store, p)
if stop:
log(stop)
results.append({"project": key, "ok": False, "error": stop, "found": 0, "added": 0,
"duplicates": 0})
continue
run_id = store.start_run(f"scan:{key}")
if CLOUD:
try:
repos.sync_main(p)
except Exception as e: # noqa: BLE001
r = {"project": key, "ok": False, "error": f"could not fetch the app's code: {e}", "found": 0,
"added": 0, "duplicates": 0}
store.finish_run(run_id, False, r); results.append(r); continue
log(f"scanning for {p.name} ...")
r = run_scan(p, store)
store.finish_run(run_id, r["ok"], r)
log(f"{p.name}: {r}")
results.append(r)
return results
def cmd_scan_report(store: Store, scan_results: list[dict[str, Any]]) -> None:
"""Say how the day's look for ideas went. New ideas show up on the board by themselves;
a scan that found nothing because it broke must not look like a quiet day."""
names = {k: p.name for k, p in known_projects().items()}
notes = []
for r in scan_results:
name = names.get(r["project"], r["project"])
if not r["ok"]:
notes.append(f"{name} scan failed ({r.get('error', '')[:120]}) — nothing examined for it today")
else:
notes.append(f"{name}: {r['found']} found, {r['added']} new, {r['duplicates']} repeats, "
f"{r.get('reddit_posts', 0)} Reddit posts read")
for line in notes:
log(line)
if scan_results and not any(r["ok"] for r in scan_results):
alert(store, "every scan failed", "\n".join(notes))
else:
added = sum(r.get("added", 0) for r in scan_results)
if added:
store.notice("ideas", f"{added} new idea{'s' if added != 1 else ''} to look at: " + "; ".join(notes))
def cmd_build(store: Store, limit: int = 1) -> None:
if os.environ.get("SCOUT_PAUSE_BUILDS"):
log("builds and ships are paused (SCOUT_PAUSE_BUILDS is set)")
return
# 0) anything still 'building' was cut off by a restart (a deploy, say): queue it again.
for iid in store.requeue_interrupted_builds():
log(f"#{iid} was mid-build when the last run stopped; queued again in its old place")
# 1) ships requested (cheap, do first) 2) then build up to `limit` approved ideas
for idea in buildmod.pending_ships(store):
_ship_one(store, idea)
# Build in the order the ideas were approved, so approving in batches sets the queue.
approved = sorted(store.ideas(status="approved"), key=lambda i: (i.get("decided_at") or 0, i["id"]))[:limit]
for idea in approved:
p = project_for(idea)
if p is None:
continue
stop = over_spend_limit(store, p)
if stop:
log(stop)
continue
log(f"building #{idea['id']} {idea['title']} for {p.name} ...")
run_id = store.start_run(f"build:{idea['id']}")
report = buildmod.build_idea(p, idea, store)
store.finish_run(run_id, report.get("status") == "done", report)
log(f"build #{idea['id']}: {report.get('status')} cost=${report.get('cost_usd', 0):.2f}")
if report.get("status") == "paused":
# Claude's usage limit: nothing else will build until it resets, so stop this pass.
# The idea stays approved in its place; the next pass simply tries again.
log(f"builds paused until Claude's usage limit resets ({report.get('error', '')[:120]})")
if not store.last_note_recent("paused", 3 * 3600):
store.notice("paused", f"Paused: Claude's usage limit is reached. Building #{idea['id']} "
f"{idea['title']} could not start ({report.get('error', '')[:200]}). Nothing is "
"lost — the queue picks up again by itself once the limit resets.")
break
fresh = store.idea(idea["id"]) or idea
store.notice(report.get("status", "failed"), build_outcome(fresh, report))
# A step in a hands-off chain ships right away, so the next step can build on it.
if fresh.get("auto_ship") and fresh.get("status") == "built":
store.request_ship(fresh["id"])
_ship_one(store, store.idea(fresh["id"]) or fresh)
def build_outcome(idea: dict[str, Any], report: dict[str, Any]) -> str:
"""One line on how a build went, for the board."""
st = report.get("status", "failed")
if st == "done":
line = f"Built #{idea['id']} {idea['title']} — it is ready to try"
return line + (" and ships by itself." if idea.get("auto_ship") else ". Put it in when you are happy with it.")
why = report.get("blocker") or report.get("summary") or report.get("error") or "unknown"
if st == "blocked":
return f"Couldn't build #{idea['id']} {idea['title']} yet: {why[:300]}"
return f"The build of #{idea['id']} {idea['title']} didn't work: {why[:300]}"
def _ship_one(store: Store, idea: dict[str, Any]) -> None:
p = project_for(idea)
if p is None:
return
log(f"shipping #{idea['id']} {idea['title']} ...")
r = buildmod.ship_idea(p, idea, store)
log(f"ship #{idea['id']}: {r}")
if r["ok"] or not r["retry"]:
store.clear_ship_request(idea["id"]) # done with, one way or the other
released = store.release_dependents(idea["id"]) if r["ok"] else []
for dep in released:
nxt = store.idea(dep) or {"title": ""}
log(f"#{dep} {nxt['title']}: approved automatically — it was waiting on #{idea['id']}")
if r["ok"] or not r["retry"]:
body = r["message"]
if released:
body += " Next in the chain, approved automatically: " + ", ".join(
f"#{d} {(store.idea(d) or {}).get('title', '')}" for d in released)
store.notice("shipped" if r["ok"] else "problem",
body if r["ok"] else f"Could not put #{idea['id']} {idea['title']} in: {body}")
elif r["retry"]:
# saying it every half hour is noisy; only say it the first time
if idea.get("error") != "SHIP_WAITING":
store.notice("waiting", f"Waiting to put #{idea['id']} {idea['title']} in: {r['message']}")
store.conn.execute("UPDATE ideas SET error='SHIP_WAITING' WHERE id=? AND "
"(error IS NULL OR error='SHIP_REQUESTED')", (idea["id"],))
store.conn.commit()
def cmd_doctor(store: Store, projects: list[Project], set_key: str | None = None) -> int:
"""Check, in plain words, that everything this needs is in place."""
ok = True
if set_key:
path = save_api_key(set_key)
print(f"Saved your Claude key on this computer only ({path}).")
print("Claude")
from .claude_runner import claude_bin
where = claude_bin()
print(f" the claude command: {where if Path(where).exists() else where + ' (not found)'}")
access = check_access(force=True)
if access.usable:
print(f" account in use: {access.detail}")
else:
ok = False
print(f" PROBLEM: {access.detail}")
print(" " + HOW_TO_FIX.replace("\n", "\n "))
for project in projects:
print(f"\n{project.name}")
print(f" folder: {project.folder}")
for problem in validate(project):
ok = False
print(f" PROBLEM: {problem}")
if project.test_command:
good, tail = buildmod.run_tests(project, project.folder)
if good:
print(f" checking its work: `{project.test_command}` passes right now")
else:
ok = False
print(f" PROBLEM: `{project.test_command}` does not pass right now, so I have no "
"way to tell whether something I build is good:")
print(" " + tail.strip().splitlines()[-1][:200] if tail.strip() else "")
else:
print(" checking its work: no test command set — I'll build, but I can't check it")
spent = spent_today(store, project)
limit = f"of your ${project.spend_limit_usd:.2f} limit" if project.spend_limit_usd > 0 else "(no limit set)"
print(f" spent today: ${spent:.2f} {limit}")
print("\nEverything Feature Scout does is billed to your own Claude account.")
return 0 if ok else 1
def cmd_score(store: Store, projects: list[Project], *, rescore: bool = False,
dry_run: bool = False) -> int:
"""Value-score the ideas that have none (older ones), so the whole list orders by value."""
from .scan import build_score_prompt, score_ideas
rc = 0
for p in projects:
if dry_run:
open_statuses = ("proposed", "approved", "building", "built", "blocked", "failed")
ideas = [i for i in store.ideas(platform=p.key, statuses=open_statuses)
if rescore or not i.get("value_score")]
print(f"{p.name}: {len(ideas)} idea(s) to score")
if ideas:
print(build_score_prompt(p, ideas))
continue
stop = over_spend_limit(store, p)
if stop:
log(stop); rc = 1; continue
r = score_ideas(p, store, rescore=rescore)
log(f"{p.name}: {r}")
if not r["ok"]:
rc = 1
elif r["asked"] == 0:
print(f"{p.name}: every idea already has a value score")
else:
left = r.get("unscored_ids") or []
print(f"{p.name}: scored {r['scored']} of {r['asked']} idea(s)"
+ (f"; still unscored: {left}" if left else ""))
return rc
def cmd_status(store: Store) -> None:
names = {k: p.name for k, p in known_projects().items()}
counts: dict[str, int] = {}
for i in store.ideas():
counts[i["status"]] = counts.get(i["status"], 0) + 1
print("ideas by status:", counts or "none yet")
# Most valuable first; ideas without a score (from before scoring existed) come last —
# `scout score` gives them one.
for i in by_value(store.ideas(statuses=("proposed", "approved", "building", "built", "blocked", "failed"))):
pname = names.get(i["platform"], i["platform"])
value = f"{int(i['value_score'])}/5" if i.get("value_score") else "-/5"
plan = (i.get("plan") or "-")[:5]
chain = (" [ships itself" + (f", after #{i['after_id']}" if i.get("after_id") else "") + "]") if i.get("auto_ship") else ""
print(f" #{i['id']:<4} {value:<4} {plan:<5} {i['status']:<9} {pname:<16} {i.get('venue','both'):<10} {i['title']}"
+ chain + (f" [{i['error'][:60]}]" if i.get("error") else ""))
for kind in [f"scan:{k}" for k in known_projects()]:
r = store.last_run(kind)
if r:
print(f"last {kind}: {datetime.fromtimestamp(r['started_at']):%Y-%m-%d %H:%M} ok={r['ok']} {r['detail']}")
WELCOME = """Feature Scout finds things people want from your project, asks you, builds the
ones you say yes to, and checks its work before anything goes live.
Two things to know before you start:
- It works on a copy of your project. Your own files are never touched until you say so.
- It runs on your own Claude account, so building an idea costs you a few dollars.
Start here: go to your project's folder and run
feature-scout init
"""
def main(argv: list[str] | None = None) -> int:
if argv is None and len(sys.argv) == 1:
print(WELCOME)
return 0
ap = argparse.ArgumentParser(prog="feature-scout")
sub = ap.add_subparsers(dest="cmd", required=True)
for name in ("daily", "poll", "scan", "build", "status"):
sp = sub.add_parser(name)
sp.add_argument("--project", action="append", help="which project (default: all of them)")
sp.add_argument("--platform", action="append", help=argparse.SUPPRESS) # what it used to be called
sp.add_argument("--limit", type=int, default=1, help="builds per pass")
sp.add_argument("--dry-run", dest="dry_run", action="store_true",
help="show what would be asked, without asking or spending anything")
rev = sub.add_parser("review", help="see the ideas and pick the ones you want")
rev.add_argument("--project", action="append")
rev.add_argument("--port", type=int)
rev.add_argument("--no-open", dest="no_open", action="store_true", help="don't open a browser")
rev.add_argument("--serve", action="store_true",
help=argparse.SUPPRESS) # the hosted machine: stay up and serve the page
und = sub.add_parser("undo", help="take a change back out of your project")
und.add_argument("ids", type=int, nargs="+")
und.add_argument("--project", action="append")
wat = sub.add_parser("watch", help="leave it running: it looks for ideas and builds what you approve")
wat.add_argument("--project", action="append")
wat.add_argument("--port", type=int)
wat.add_argument("--no-open", dest="no_open", action="store_true")
wat.add_argument("--limit", type=int, default=1)
for name, help_text in (("setup", "set up a project, your Claude and how you want to be told (start here)"),
("init", argparse.SUPPRESS)):
ini = sub.add_parser(name, help=help_text)
ini.add_argument("--folder")
ini.add_argument("--no-ai", dest="no_ai", action="store_true",
help="don't offer to read the project and write its description")
doc = sub.add_parser("doctor", help="check that everything this needs is in place")
doc.add_argument("--project", action="append")
doc.add_argument("--set-key", dest="set_key", help="save your own Claude API key on this computer")
sco = sub.add_parser("score", help="give a value score (1-5) to ideas that have none, so the list orders by value")
sco.add_argument("--project", action="append")
sco.add_argument("--all", dest="rescore", action="store_true", help="score every open idea again")
sco.add_argument("--dry-run", dest="dry_run", action="store_true", help="show what would be asked")
chn = sub.add_parser("chain", help="build and ship these ideas one after another without asking between them")
chn.add_argument("ids", type=int, nargs="+", help="in order; the first should be approved, the rest follow by themselves")
dec = sub.add_parser("decide", help="approve / reject / ship ideas by number (the same as pressing the button on the page)")
dec.add_argument("verb", choices=["approve", "reject", "ship"])
dec.add_argument("ids", type=int, nargs="+")
addp = sub.add_parser("add", help="add an idea by hand")
addp.add_argument("--project")
addp.add_argument("--platform", help=argparse.SUPPRESS)
addp.add_argument("--title", required=True)
addp.add_argument("--spec", required=True)
addp.add_argument("--approve", action="store_true")
args = ap.parse_args(argv)
ensure_dirs()
if args.cmd not in ("init", "setup", "doctor", "decide", "chain") and not known_projects():
if CLOUD:
# A server with nothing to look after is broken, not new. Say so loudly rather
# than printing a welcome and exiting quietly, which would look like a fine run.
alert(None, "no projects configured",
"This machine has no projects set up, so there was nothing to do.")
return 1
print(WELCOME)
return 0
store = Store(DB_PATH)
try:
projects = chosen_projects(args) if args.cmd not in ("decide", "init", "setup", "chain") else []
except ConfigProblem as e:
print(e); return 1
if args.cmd in ("scan", "daily", "build", "poll", "score") and not getattr(args, "dry_run", False):
try:
require_llm_access()
except NoLLMAccess as e:
print(e); return 1
try:
if args.cmd == "watch":
from .review import serve
from .watch import watch as watch_loop
if not projects:
print("You haven't set up a project yet. Run `feature-scout setup` first."); return 1
_, url = serve(projects, port=args.port, open_browser=not args.no_open)
names = ", ".join(p.name for p in projects)
print(f"Feature Scout is watching {names}.\n"
f"Your ideas are at {url}\n"
f"Looking for ideas each morning, and building what you've approved.\n"
f"Press Ctrl-C to stop.")
try:
from .config import scan_time
hour, minute = scan_time()
watch_loop(projects, store,
scanner=lambda st, ps: cmd_scan(st, ps),
poller=lambda st: cmd_build(st, limit=args.limit),
hour=hour, minute=minute)
except KeyboardInterrupt:
print("\nStopped.")
return 0
if args.cmd == "chain":
missing = [i for i in args.ids if store.idea(i) is None]
if missing:
print(f"no such idea: {missing}"); return 1
store.chain(args.ids)
names = [f"#{i} {(store.idea(i) or {}).get('title', '')}" for i in args.ids]
print("Hands-off chain set. Each ships as soon as it is built and tested; each one after the "
"first is approved when the one before it has shipped:\n " + "\n ".join(names))
return 0
if args.cmd == "decide":
from .decisions import apply_commands
for line in apply_commands(store, [(args.verb, args.ids)]):
print(line)
return 0
if args.cmd == "add":
key = args.project or args.platform or (projects[0].key if projects else None)
if not key:
print("Say which project with --project."); return 1
iid = store.add_idea(key, args.title, args.spec, spec=args.spec, source="manual",
evidence=[{"url": "", "note": "added by hand"}])
if args.approve:
store.set_status(iid, "approved")
print(f"added #{iid}"); return 0
if args.cmd in ("init", "setup"):
from .init import run_init
try:
run_init(folder=Path(args.folder) if args.folder else None,
use_ai=not args.no_ai)
except (NoLLMAccess, FileNotFoundError) as e:
print(e); return 1
except (KeyboardInterrupt, EOFError):
print("\nStopped. Nothing was saved."); return 1
return 0
if args.cmd == "review":
from .review import hosted_settings, serve
if not projects:
print("You haven't set up a project yet. Run `feature-scout setup` first."); return 1
if args.serve:
# The hosted machine: the same page, on a private link, for as long as it runs.
hosted = hosted_settings()
if not hosted:
print("SCOUT_REVIEW_TOKEN is not set (or is too short), so there is nothing "
"to protect the page with. Refusing to serve it.")
return 1
secret, base = hosted
server, url = serve(projects, port=args.port or 8080, open_browser=False,
host="0.0.0.0", public=True, token=secret, base_url=base)
log(f"the review page is up on port {server.server_address[1]}")
try:
threading.Event().wait()
except KeyboardInterrupt:
server.shutdown()
return 0
server, url = serve(projects, port=args.port, open_browser=not args.no_open)
print(f"Your ideas are at {url}\nPress Ctrl-C when you're done.")
try:
threading.Event().wait()
except KeyboardInterrupt:
server.shutdown()
print("\nStopped.")
return 0
if args.cmd == "undo":
for idea_id in args.ids:
idea = store.idea(idea_id)
if not idea:
print(f"I can't find #{idea_id}."); continue
p = project_for(idea)
if p is None:
continue
print(buildmod.undo_idea(p, idea, store)["message"])
return 0
if args.cmd == "status":
cmd_status(store); return 0
if args.cmd == "score":
return cmd_score(store, projects, rescore=args.rescore, dry_run=args.dry_run)
if args.cmd == "doctor":
return cmd_doctor(store, projects, getattr(args, "set_key", None))
if args.cmd == "scan" and getattr(args, "dry_run", False):
return cmd_dry_run(store, projects)
if args.cmd == "scan":
with Lock(wait_s=0):
cmd_scan(store, projects); return 0
if args.cmd == "build":
with Lock(wait_s=0):
cmd_build(store, limit=args.limit); return 0
if args.cmd == "daily":
with Lock(wait_s=1800):
results = cmd_scan(store, projects)
cmd_scan_report(store, results)
cmd_build(store, limit=args.limit)
return 0
if args.cmd == "poll":
with Lock(wait_s=0):
cmd_build(store, limit=args.limit)
return 0
except Busy as e:
# Something else is mid-job. That is normal and must never look like a failure,
# or every poll while a build is running would report a problem.
log(str(e))
return 0
except SystemExit:
raise
except Exception: # noqa: BLE001
tb = traceback.format_exc()
alert(store, f"{args.cmd} crashed", tb[-3000:])
return 1
return 0
if __name__ == "__main__":
sys.exit(main())