Router: optimization candidates from the ledger, CI and parked cards
The router only ranked work someone had written a card for, so optimization showed up only when a human noticed it. It now reads three of the factory's own records over the last seven days and proposes the rest itself, deterministically. router.py, one synthetic candidate per finding (`opt:<source>:<id>`, kind optimization, no card yet, a drafted title and body that carries its own key and a `DONE:` line): - `ledger_candidates`: the week's spend grouped by model and by entry point (a line's `entry` where it has one, else `pool`, the route it booked to). A group that took over half the bill is a candidate, where there is a second group in that dimension to compare it with; so is a model whose spend per call is double its own median, which is a few heavy calls behind a cheap average. Pennies (under $0.25 a week) are left alone and one group yields at most one candidate. - `run_candidates`: Actions runs as `/actions/runs` gives them (the shape `tea_actions` reads: the list, or `workflow_runs`). A workflow with two or more failed or cancelled runs in the window is a candidate, with the failure count and the last run's id. - `parked_candidates`: a card parked by an `error` or `stuck` beat older than a day, with no open PR and #78's automatic retries spent ("retry or split #N"). One Time will still retry on its own is left to it. - `named` / `opt_candidates`: a source an open card already names (`Router: opt:<source>:<id>` in the body) is not proposed again. astra.py gathers the three sources (`ledger_week`, `action_runs`, `parked_cards`; each is context, so unreadable means no candidates from it, never a failed run) and the seat judges them like any other candidate. Live, a picked one is written as a card, unqueued and a candidate from tomorrow, and is never worked on in place; dry, the ranked table shows it as "would be written as a card". The log line and the beat's `opt` say which source became which card. Tests (test_time.py section 171): a ledger with one dominant route yields one candidate, and so does the entry-point dimension, the median rule, pennies, a single group and a stale window; two failed runs of one workflow yield one candidate with its count and last run id; a parked card yields one unless it has an open PR, retries left or was parked today; a source an open card names yields none; Astra live writes the card unqueued while dry writes nothing. Closes #171 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EzRXxLDJF4YDoZgzSsbW24
This commit is contained in:
4 files changed
+356
-14
No files matched your search
+74
-9
@@ -14,7 +14,10 @@ one issue, "Astra log", so a human can read the day's reasoning in one place and
|
||||
|
||||
The router (time/router.json, `mode`: off | dry | live; router.py): instead of picking cards in prose, the seat judges
|
||||
every candidate (its kind, its value 1-5, its size) and code ranks them with the owner's weights per kind, a restless
|
||||
boost for kinds not picked lately, and the card's age. The log shows the ranked table. `dry` writes it and queues
|
||||
boost for kinds not picked lately, and the card's age. The candidates include the optimizations the factory's own
|
||||
records show (#171): the ledger's spend by model and entry point, workflows whose runs keep failing, and cards parked
|
||||
with their retries spent. Those have no card yet, so a picked one is written as one (unqueued) instead of being queued,
|
||||
and a source an open card already names is not proposed again. The log shows the ranked table. `dry` writes it and queues
|
||||
nothing; `live` queues from the top while the day's money left covers each card's estimated cost (the allowance left
|
||||
in the pool that takes the worker's route, else the daily cap's remainder, never more than the month has left),
|
||||
creates the seat's new cards unqueued (they are candidates the next day; never a repeat of an open or lately closed
|
||||
@@ -79,6 +82,10 @@ For every candidate give:
|
||||
- blocked: only when it can't move today, what it waits on (a PR that can't merge, a human, something outside the
|
||||
repo); leave it out otherwise. A blocked card is never queued, however it scores. A card whose `Depends on #N` card
|
||||
is still open is already held back by code.
|
||||
A candidate whose key starts `opt:` is not a card yet: it is something the factory's own records show (a route that took
|
||||
over half the week's spend, a workflow that keeps failing, a card parked with its retries spent), with a title and body
|
||||
already drafted. Judge it like the rest; if it is picked it is written as a card for a worker to take later, never
|
||||
worked on today, so judge whether that work is worth a card at all.
|
||||
You may also write up to {max_new} new cards for work the board is missing; they are candidates from tomorrow on.
|
||||
Never write one that posts outside the repo or touches workflows, secrets or variables, and never one for work that is
|
||||
already a candidate or is in the list of cards closed lately: that work is done or under way.
|
||||
@@ -201,6 +208,51 @@ def closed_lately(days: int = 14) -> list[dict]:
|
||||
return []
|
||||
|
||||
|
||||
def ledger_week(days: int = router.OPT_DAYS) -> list[dict]:
|
||||
"""The ledger lines the router reads (#171): this month's file, and the previous one when the window straddles it.
|
||||
router.ledger_candidates keeps the last `days` days of them."""
|
||||
now = dt.datetime.now(dt.timezone.utc)
|
||||
months = sorted({now.strftime("%Y-%m"), (now - dt.timedelta(days=days)).strftime("%Y-%m")})
|
||||
try:
|
||||
return [r for m in months for r in tm.ledger_lines(m) if isinstance(r, dict)]
|
||||
except Exception as e: # noqa: BLE001 a source, not a dependency: unreadable is no candidates from it
|
||||
print(f"ledger unreadable: {str(e)[:120]}")
|
||||
return []
|
||||
|
||||
|
||||
def action_runs(limit: int = 50) -> list[dict]:
|
||||
"""The newest workflow runs, from the endpoint `tea_actions` reads (`/actions/runs`; Gitea wraps them in
|
||||
`workflow_runs`)."""
|
||||
try:
|
||||
got = tm.gh("GET", f"/repos/{tm.REPO}/actions/runs?limit={limit}")
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"actions runs unreadable: {str(e)[:120]}")
|
||||
return []
|
||||
return got if isinstance(got, list) else (got or {}).get("workflow_runs") or []
|
||||
|
||||
|
||||
def parked_cards() -> list[dict]:
|
||||
"""The `ready` cards whose last real beat parked them, with what the router needs of each: the beat and when it was
|
||||
written, whether a PR of theirs is open, and how many of #78's automatic retries are left."""
|
||||
auto_max = env_int("TIME_AUTO_RETRY", 1)
|
||||
out: list[dict] = []
|
||||
try:
|
||||
prs = tm.gh("GET", f"/repos/{tm.REPO}/pulls?state=open&limit=50") or []
|
||||
for card in tm.issues("state=open&labels=ready"):
|
||||
beats = [b for b in tm.beats_of(tm.gh("GET", f"/repos/{tm.REPO}/issues/{card['number']}/comments?limit=100") or [])
|
||||
if b.get("beat") not in ("blocked", "capped")]
|
||||
if not beats or beats[-1].get("beat") not in ("error", "stuck"):
|
||||
continue
|
||||
# an automatic retry is one run of consecutive `auto_retry` beats, counted as the heartbeat counts it
|
||||
used = sum(1 for k, b in enumerate(beats) if b.get("auto_retry") and not (k and beats[k - 1].get("auto_retry")))
|
||||
out.append({"n": card["number"], "title": card.get("title") or "", "beat": beats[-1]["beat"],
|
||||
"at": beats[-1]["at"], "retries_left": max(0, auto_max - used),
|
||||
"open_pr": any(p.get("state", "open") == "open" for p in tm.linked_prs(prs, card["number"]))})
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"parked cards unreadable: {str(e)[:120]}")
|
||||
return out
|
||||
|
||||
|
||||
def _norm(title: str) -> str:
|
||||
return re.sub(r"[^a-z0-9]+", " ", (title or "").lower()).strip()
|
||||
|
||||
@@ -221,7 +273,10 @@ def repeat_of(title: str, known: list[dict]) -> int | None:
|
||||
def router_tick(cfg: dict, seat: str, log: int, context: str, thread: list[dict], max_ready: int, max_new: int) -> int:
|
||||
"""The seat judges every candidate; router.py ranks them; live queues from the top, dry only shows the ranking."""
|
||||
open_issues = [i for i in tm.issues("state=open") if i.get("title") != LOG_TITLE]
|
||||
today_d = dt.datetime.now(dt.timezone.utc).date()
|
||||
cands = router.candidates(open_issues) + [dict(router.NEW_VENTURE)]
|
||||
# what no one wrote a card for (#171): the ledger, the Actions runs and the parked cards, each proposed once
|
||||
cands += router.opt_candidates(ledger_week(), action_runs(), parked_cards(), today_d, router.named(open_issues))
|
||||
listing = "\n\n".join(f"### {c['key']}: {c['title']}\nlabels: {', '.join(c['labels']) or '-'}; opened "
|
||||
f"{c['created'][:10] or '-'}; cards waiting on it: {c['unblocks']}\n{c['body'][:400]}" for c in cands)
|
||||
today, day_cap, month, month_cap = tm.budget()
|
||||
@@ -239,7 +294,7 @@ def router_tick(cfg: dict, seat: str, log: int, context: str, thread: list[dict]
|
||||
tm.comment(log, f"🔭 Astra: {e}. No ranking today.", {"beat": "astra", "router": cfg["mode"], "queued": 0, "capped": True})
|
||||
return 0
|
||||
assess = {str(a.get("key") or "").strip(): a for a in out.get("assess") or [] if isinstance(a, dict)}
|
||||
ranked = router.rank(cands, assess, cfg, router.last_picked(tm.beats_of(thread)), dt.datetime.now(dt.timezone.utc).date())
|
||||
ranked = router.rank(cands, assess, cfg, router.last_picked(tm.beats_of(thread)), today_d)
|
||||
in_lane = sum(1 for i in open_issues if router.CLOUD in router.labels_of(i))
|
||||
to_cloud = router.pick_cloud(ranked, cfg, in_lane) # the cloud lane takes its cards first: its credit is separate
|
||||
sent = {r["key"] for r in to_cloud}
|
||||
@@ -249,10 +304,14 @@ def router_tick(cfg: dict, seat: str, log: int, context: str, thread: list[dict]
|
||||
n = repeat_of(str(c["title"]), open_issues + closed)
|
||||
(repeats.append((c, n)) if n else new.append(c))
|
||||
new = new[:max_new]
|
||||
queued, found = [r for r in picks if r["n"] is not None], [r for r in picks if r["n"] is None]
|
||||
live, created = cfg["mode"] == "live", []
|
||||
if live:
|
||||
for r in picks:
|
||||
for r in queued:
|
||||
tm.label(r["n"], "queued") # the heartbeat promotes it, in ranked order
|
||||
for r in found:
|
||||
# a candidate with no card (an optimization the records showed): it becomes a card, never worked on in place
|
||||
r["written"] = tm.new_issue(r["title"][:200], r["body"] + "\n\n<sub>card written by Astra</sub>", [])["number"]
|
||||
for r in to_cloud:
|
||||
tm.label(r["n"], router.CLOUD) # the Mac's launcher starts a cloud session for it
|
||||
for c in new:
|
||||
@@ -270,26 +329,32 @@ def router_tick(cfg: dict, seat: str, log: int, context: str, thread: list[dict]
|
||||
if unjudged:
|
||||
lines.append(f"_{unjudged} candidate(s) the seat didn't judge are scored at value 3, size M._")
|
||||
if live:
|
||||
lines.append(("Queued: " + ", ".join(f"#{r['n']}" for r in picks)) if picks else "Queued nothing today.")
|
||||
lines.append(("Queued: " + ", ".join(f"#{r['n']}" for r in queued)) if queued else "Queued nothing today.")
|
||||
if found:
|
||||
lines.append("Written from the factory's own records (candidates from tomorrow): "
|
||||
+ ", ".join(f"#{r['written']} ({r['key']})" for r in found))
|
||||
if to_cloud:
|
||||
lines.append("To the cloud lane: " + ", ".join(f"#{r['n']}" for r in to_cloud))
|
||||
if created:
|
||||
lines.append("New cards (candidates from tomorrow): " + ", ".join(f"#{n}" for n in created))
|
||||
else:
|
||||
lines.append("Dry run: nothing queued or created. Live, it would queue: "
|
||||
+ (", ".join(f"#{r['n']}" for r in picks) or "nothing") + "."
|
||||
+ (", ".join(f"#{r['n']}" for r in queued) or "nothing") + "."
|
||||
+ (" And write a card for: " + ", ".join(r["key"] for r in found) + "." if found else "")
|
||||
+ (" To the cloud lane: " + ", ".join(f"#{r['n']}" for r in to_cloud) + "." if to_cloud else ""))
|
||||
if new:
|
||||
lines.append("It would write: " + "; ".join(router._txt(c["title"], 100) for c in new) + ".")
|
||||
if repeats:
|
||||
lines.append("Not written, repeats of existing cards: "
|
||||
+ "; ".join(f"{router._txt(c['title'], 80)} (#{n})" for c, n in repeats) + ".")
|
||||
meta = {"beat": "astra", "router": cfg["mode"], "queued": len(picks) if live else 0, "order": router.order(ranked),
|
||||
"picked": [{"n": r["n"], "kind": r["kind"]} for r in picks] if live else [],
|
||||
"would": [r["n"] for r in picks] if not live else [], "new": created,
|
||||
meta = {"beat": "astra", "router": cfg["mode"], "queued": len(queued) if live else 0, "order": router.order(ranked),
|
||||
"picked": [{"n": r.get("written", r["n"]), "kind": r["kind"]} for r in picks] if live else [],
|
||||
"would": [r["n"] for r in queued] if not live else [], "new": created,
|
||||
"opt": [{"key": r["key"], "card": r.get("written")} for r in found],
|
||||
"cloud": [r["n"] for r in to_cloud] if live else [], "would_cloud": [r["n"] for r in to_cloud] if not live else []}
|
||||
tm.comment(log, "\n".join(lines), meta)
|
||||
print(f"astra router {cfg['mode']}: picks={[r['n'] for r in picks]} new={created} ranked={len(ranked)}")
|
||||
print(f"astra router {cfg['mode']}: picks={[r['n'] for r in queued]} opt={[r['key'] for r in found]} "
|
||||
f"new={created} ranked={len(ranked)}")
|
||||
return 0
|
||||
|
||||
|
||||
|
||||
+166
-5
@@ -5,8 +5,8 @@ judgments into one ranked list, in arithmetic anyone can check, and picks from t
|
||||
promotes queued cards in the same order instead of oldest first.
|
||||
|
||||
A candidate is an open card a worker can take (a `DONE:` line; not a parent, not a human's, not the venture loop's
|
||||
own), or work with no card yet that the factory could start (a new venture). Each has a kind: venture, feature,
|
||||
refinement or optimization.
|
||||
own), or work with no card yet that the factory could start (a new venture, or an optimization the factory's own
|
||||
records show: see below). Each has a kind: venture, feature, refinement or optimization.
|
||||
|
||||
score = weight[kind] x restless(kind) x age x (value + 0.5 x unblocks) / size
|
||||
|
||||
@@ -21,6 +21,12 @@ worker. A launcher signed in to the owner's claude.ai account starts a Claude Co
|
||||
session credit pays); the session's `claude/card-N-*` branch comes home through sync.py as a PR, sync relabels the card
|
||||
`ready`, and the heartbeat adopts and reviews it like any card PR. `per_day: 0` (the default) is no cloud lane.
|
||||
|
||||
Optimization candidates (#171): the router reads three of the factory's own records over the last week and proposes the
|
||||
work no one wrote a card for — the ledger (a route or model that takes over half the week's spend, a model whose spend
|
||||
per call is double its own median), the Actions runs (a workflow that keeps failing or being cancelled) and the parked
|
||||
cards (an `error` or `stuck` beat Time will not retry again). Each finding is a synthetic candidate the seat judges like
|
||||
any other; live, a picked one is written as a card, never worked on in place. See the section below.
|
||||
|
||||
An L is ranked but never queued: a card is what one worker session finishes, so it is split first. Nor is a card that
|
||||
waits: on a card still open (its `Depends on #N`), or on whatever the seat says blocks it (a PR that can't merge, a
|
||||
human). Cards that can move come first in the promotion order, the ones that wait after them. The policy is
|
||||
@@ -121,6 +127,156 @@ def candidates(issues: list[dict]) -> list[dict]:
|
||||
return out
|
||||
|
||||
|
||||
# ---------- optimization candidates from the factory's own records (#171) ----------
|
||||
# Three sources, each read by arithmetic over the last week, so optimization work no longer waits for a human to notice
|
||||
# it. Every finding is a synthetic candidate: key `opt:<source>:<id>`, kind optimization, a drafted title and body, and
|
||||
# no card (`n` is None). The seat judges it like any other candidate; live, a picked one is written as a card (unqueued,
|
||||
# a candidate from tomorrow) and never worked on in place. A source an open card already names is not proposed again.
|
||||
|
||||
OPT = "opt:"
|
||||
OPT_DAYS = 7 # the window the ledger and the runs are read over
|
||||
OPT_MIN_USD = 0.25 # a week's spend under this is pennies: nothing to optimize
|
||||
OPT_SHARE = 0.5 # a group over this share of the week's spend is a candidate
|
||||
OPT_SKEW = 2.0 # a model whose spend per call is this many times its own median is a candidate
|
||||
OPT_CALLS = 4 # fewer calls than this say nothing about a model's median
|
||||
OPT_FAILS = 2 # failed or cancelled runs of one workflow that make a candidate
|
||||
OPT_PARKED_DAYS = 1 # a parked card is a candidate once its beat is older than this
|
||||
BAD_RUN = ("failure", "cancelled", "canceled")
|
||||
NAMES = re.compile(rf"^\s*Router:\s*({OPT}\S+)", re.M)
|
||||
|
||||
|
||||
def _recent(at, today: dt.date, days: int) -> bool:
|
||||
"""Is this timestamp inside the last `days` days? By the day, like the rest of the router's arithmetic."""
|
||||
return str(at or "")[:10] >= (today - dt.timedelta(days=days)).isoformat()
|
||||
|
||||
|
||||
def _median(xs: list[float]) -> float:
|
||||
s, n = sorted(xs), len(xs)
|
||||
return 0.0 if not n else s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2
|
||||
|
||||
|
||||
def _opt(source: str, ident: str, title: str, body: list[str], created: str,
|
||||
done: str = "PR open that does the above, tests green.") -> dict:
|
||||
"""One synthetic candidate. The body names its own key, which is how a source is proposed once and not every day,
|
||||
and carries the `DONE:` line the card is written with. A key holds no spaces: `named` reads it back as one word."""
|
||||
word = re.sub(r"\s+", "-", str(ident).strip()) or "unknown"
|
||||
key = f"{OPT}{source}:{word}"
|
||||
return {"key": key, "n": None, "title": _txt(title, 120),
|
||||
"body": "\n".join(body) + f"\n\nRouter: {key}\n\nDONE: {done}",
|
||||
"labels": [], "lined_up": False, "actionable": True, "kind": "optimization", "unblocks": 0,
|
||||
"created": created or ""}
|
||||
|
||||
|
||||
def ledger_candidates(rows: list[dict], today: dt.date, days: int = OPT_DAYS) -> list[dict]:
|
||||
"""The week's spend, grouped by model and by entry point (a line's `entry` where it has one, else `pool`, the route
|
||||
it booked to). Two questions are worth a card: a group that took over half of what the factory spent, and a model
|
||||
whose spend per call is double its own median (a few heavy calls hiding behind a cheap average). A dimension with
|
||||
one group says nothing — everything went through it by definition — pennies are left alone (OPT_MIN_USD), and one
|
||||
group yields at most one candidate."""
|
||||
week = [r for r in rows or [] if isinstance(r, dict) and _recent(r.get("at"), today, days)]
|
||||
total = sum(_num(r.get("usd"), 0.0, 0.0, 1e6) for r in week)
|
||||
calls: dict[tuple[str, str], list[float]] = {}
|
||||
for r in week:
|
||||
for dim, name in (("model", r.get("model")), ("entry", r.get("entry") or r.get("pool"))):
|
||||
if name:
|
||||
calls.setdefault((dim, _txt(name, 60)), []).append(_num(r.get("usd"), 0.0, 0.0, 1e6))
|
||||
alone = {dim for dim in ("model", "entry") if len([k for k in calls if k[0] == dim]) < 2}
|
||||
out = []
|
||||
for (dim, name), usd in sorted(calls.items()):
|
||||
spent, n = sum(usd), len(usd)
|
||||
what = "the model" if dim == "model" else "the entry point"
|
||||
if spent < OPT_MIN_USD:
|
||||
continue
|
||||
if total and spent > OPT_SHARE * total and dim not in alone:
|
||||
out.append(_opt("ledger", f"{dim}:{name}", f"Why did {name} cost ${spent:.2f} of the week's ${total:.2f}?", [
|
||||
f"- Over the last {days} days {name} ({what}) booked ${spent:.2f} of the ${total:.2f} the factory spent "
|
||||
f"on models, over {n} call(s): more than half of the bill.",
|
||||
"- Read what those calls are and make them cheaper: a shorter system prompt, the prompt cache, a smaller "
|
||||
"seat for the easy half of the work, or simply fewer calls.",
|
||||
"- Change no behaviour the factory depends on, and say in the PR what the same week would have cost."],
|
||||
today.isoformat()))
|
||||
continue
|
||||
per, mid = spent / n, _median(usd)
|
||||
if dim == "model" and n >= OPT_CALLS and mid > 0 and per >= OPT_SKEW * mid:
|
||||
out.append(_opt("ledger", f"{dim}:{name}", f"Why does {name} cost ${per:.4f} a call against a ${mid:.4f} median?", [
|
||||
f"- {name} made {n} call(s) in the last {days} days for ${spent:.2f}: ${per:.4f} a call against a median "
|
||||
f"of ${mid:.4f} a call, so a few heavy calls carry the bill.",
|
||||
"- Find the heavy ones (the longest prompts, the longest answers) and cut what they send or ask for; the "
|
||||
"cheap calls are already cheap.",
|
||||
"- Say in the PR which calls they were and what they cost before and after."],
|
||||
today.isoformat()))
|
||||
return out
|
||||
|
||||
|
||||
def _workflow(run: dict) -> str:
|
||||
"""Which workflow a run belongs to: its file's name, else the workflow's name, else its id."""
|
||||
path = str(run.get("path") or "").strip().rsplit("/", 1)[-1]
|
||||
return _txt(path or run.get("name") or run.get("workflow_id") or "unknown", 60)
|
||||
|
||||
|
||||
def run_candidates(runs, today: dt.date, days: int = OPT_DAYS) -> list[dict]:
|
||||
"""Actions runs as `/actions/runs` gives them (the shape `tea_actions` reads: the list, or `{"workflow_runs": [...]}`).
|
||||
A workflow with OPT_FAILS or more failed or cancelled runs in the window is one candidate, with the failure count
|
||||
and the last run's id, which is where to start reading."""
|
||||
rows = runs.get("workflow_runs") if isinstance(runs, dict) else runs
|
||||
bad: dict[str, list[dict]] = {}
|
||||
for r in rows or []:
|
||||
if not isinstance(r, dict) or str(r.get("conclusion") or "").strip().lower() not in BAD_RUN:
|
||||
continue
|
||||
if _recent(r.get("created_at") or r.get("run_started_at") or r.get("started_at") or r.get("updated_at"), today, days):
|
||||
bad.setdefault(_workflow(r), []).append(r)
|
||||
out = []
|
||||
for wf, rs in sorted(bad.items()):
|
||||
if len(rs) < OPT_FAILS:
|
||||
continue
|
||||
last = max(rs, key=lambda r: (str(r.get("created_at") or ""), str(r.get("id") or "")))
|
||||
out.append(_opt("ci", wf, f"Why does {wf} keep failing? {len(rs)} runs in {days} days", [
|
||||
f"- `{wf}` has {len(rs)} failed or cancelled run(s) in the last {days} days; the last one is run "
|
||||
f"`{last.get('id')}`. A workflow that fails this often is either broken or flaky, and either way the factory "
|
||||
"reads its red as noise.",
|
||||
"- Read that run, find the one cause, and fix it in the code the workflow runs (a test, a script, a missing "
|
||||
"guard), not by making the check weaker.",
|
||||
"- Workflow files are a human's: if the change belongs in `.github/workflows/`, write the exact change in the "
|
||||
"PR and the commit message and stop at the code."],
|
||||
today.isoformat()))
|
||||
return out
|
||||
|
||||
|
||||
def parked_candidates(parked: list[dict], today: dt.date, days: int = OPT_PARKED_DAYS) -> list[dict]:
|
||||
"""Cards parked by an `error` or `stuck` beat (as astra.parked_cards reads them): one candidate each, to retry it by
|
||||
hand or split it. A card with an open PR belongs to the review path, and one whose automatic retries (#78) are not
|
||||
spent is Time's own to retry; neither is proposed, nor is one parked only today."""
|
||||
out = []
|
||||
for p in parked or []:
|
||||
if not isinstance(p, dict):
|
||||
continue
|
||||
n, at, beat = p.get("n"), str(p.get("at") or ""), str(p.get("beat") or "")
|
||||
if n is None or beat not in ("error", "stuck") or p.get("open_pr"):
|
||||
continue
|
||||
if _num(p.get("retries_left"), 0.0) > 0 or _recent(at, today, days):
|
||||
continue
|
||||
out.append(_opt("parked", str(n), f"Retry or split #{n}: {_txt(p.get('title'), 70)}", [
|
||||
f"- #{n} has been parked since {at[:10] or 'a while ago'}: its last beat was `{beat}` and Time's automatic "
|
||||
"retries are spent, so it sits still with no PR of its own and no one waiting on it.",
|
||||
f"- Read the beats on #{n}, find what stopped the worker, and either fix that so a retry goes through, or "
|
||||
"split the card into ones a worker finishes in one session.",
|
||||
f"- Leave #{n} open: the work here is getting it moving again."],
|
||||
at or today.isoformat(),
|
||||
done=f"#{n} is moving again (a retry that goes through, or the cards that replace it), tests green."))
|
||||
return out
|
||||
|
||||
|
||||
def named(issues: list[dict]) -> set[str]:
|
||||
"""The sources open cards already name (`Router: opt:<source>:<id>` in the body): proposed once, not every day."""
|
||||
return {m for i in issues for m in NAMES.findall(i.get("body") or "")}
|
||||
|
||||
|
||||
def opt_candidates(ledger: list[dict], runs, parked: list[dict], today: dt.date, known=frozenset()) -> list[dict]:
|
||||
"""Every optimization the three sources show today, minus the ones an open card already names."""
|
||||
found = ledger_candidates(ledger, today) + run_candidates(runs, today) + parked_candidates(parked, today)
|
||||
return [c for c in found if c["key"] not in known]
|
||||
|
||||
|
||||
def last_picked(beats: list[dict]) -> dict[str, str]:
|
||||
"""The day each kind was last picked, from Astra's live beats."""
|
||||
out: dict[str, str] = {}
|
||||
@@ -208,9 +364,14 @@ def table(ranked: list[dict], picked: list[dict], mode: str, limit: int = 15, cl
|
||||
rows = []
|
||||
for k, r in enumerate(ranked[:limit], 1):
|
||||
lane = "cloud lane" if CLOUD in r.get("labels", ()) else "lined up" if r["lined_up"] else ""
|
||||
note = (f"**{mark}**" if r["key"] in chosen else ("**to the cloud today**" if mode == "live" else "**would go to the cloud**")
|
||||
if r["key"] in sent else "; ".join(x for x in (lane, r.get("why_not", "") if not r["actionable"] else "") if x))
|
||||
name = f"{r['key']} {r['title']}" if r["n"] is not None else r["title"]
|
||||
if r["key"] in chosen: # a candidate with no card is written as one instead of being queued
|
||||
note = f"**written as #{r['written']}**" if r.get("written") else \
|
||||
f"**{mark}**" if r["n"] is not None else "**would be written as a card**"
|
||||
elif r["key"] in sent:
|
||||
note = "**to the cloud today**" if mode == "live" else "**would go to the cloud**"
|
||||
else:
|
||||
note = "; ".join(x for x in (lane, r.get("why_not", "") if not r["actionable"] else "") if x)
|
||||
name = f"{r['key']} {r['title']}" if r["n"] is not None or r["key"].startswith(OPT) else r["title"]
|
||||
rows.append(f"| {k} | {_txt(name, 80).replace('|', '/')} | {r['kind']} | {r['value']:g} | {r['size']} | "
|
||||
f"{r['unblocks']} | {r['restless']:g} | {r['score']:g} | {_txt(r['why'], 140).replace('|', '/')} | "
|
||||
f"{_txt(note, 90).replace('|', '/')} |")
|
||||
|
||||
@@ -1226,6 +1226,8 @@ astra.tm.budget = lambda: (1.0, 15.0, 10.0, 100.0)
|
||||
astra.tm.comment = lambda n, text, meta: said80.append((n, text, meta))
|
||||
astra.tm.label = lambda n, name: labelled80.append((n, name))
|
||||
astra.tm.new_issue = lambda title, body, labels: made80.append((title, body, labels)) or {"number": 90 + len(made80)}
|
||||
astra.tm.ledger_lines = lambda month: [] # the optimization sources (#171) are the next section's; none of them here
|
||||
astra.tm.gh = lambda method, path, body=None, **k: [] if "/comments" in path or "/pulls" in path else {"workflow_runs": []}
|
||||
reply80 = {"note": "the note", "assess": [dict(assess80[k], key=k) for k in assess80] + ["junk"],
|
||||
"new": [{"title": "Write a thing", "body": "do it"}, {"no": "title"}]}
|
||||
astra.tm.think = lambda seat, a, b, system, user: (asked80.append((seat, system, user)), "Sure:\n" + json.dumps(reply80))[1]
|
||||
@@ -1331,6 +1333,119 @@ else: os.environ["TIME_PARALLEL"] = par135
|
||||
tm.issues, tm.gh, tm.label, tm.unlabel = real80
|
||||
print("ALL OK 80")
|
||||
|
||||
# 171 optimization candidates the factory finds itself: the ledger, the Actions runs and the parked cards. Each source
|
||||
# is a synthetic candidate the seat judges like any other; live, a picked one is written as a card, never worked on in place.
|
||||
today171 = dt.date(2026, 10, 19)
|
||||
led171 = [{"at": "2026-10-16T09:00:00Z", "model": "sonnet", "usd": 2.0}] + \
|
||||
[{"at": "2026-10-17T09:00:00Z", "model": "haiku", "usd": 0.1}] * 3
|
||||
one171 = rt.ledger_candidates(led171, today171)
|
||||
assert [c["key"] for c in one171] == ["opt:ledger:model:sonnet"], one171
|
||||
assert one171[0]["title"] == "Why did sonnet cost $2.00 of the week's $2.30?" and one171[0]["kind"] == "optimization"
|
||||
assert one171[0]["n"] is None and one171[0]["actionable"] and not one171[0]["lined_up"] and one171[0]["created"] == "2026-10-19"
|
||||
assert "Router: opt:ledger:model:sonnet" in one171[0]["body"] and rt.DONE.search(one171[0]["body"]), one171[0]["body"]
|
||||
assert "$2.00 of the $2.30" in one171[0]["body"] and "3 call(s)" not in one171[0]["body"]
|
||||
# by entry point too (`entry`, else the `pool` the call booked to); a dimension with one group says nothing
|
||||
pooled171 = [dict(r, pool="anthropic" if r["model"] == "sonnet" else "venice") for r in led171]
|
||||
assert [c["key"] for c in rt.ledger_candidates(pooled171, today171)] == ["opt:ledger:entry:anthropic", "opt:ledger:model:sonnet"]
|
||||
assert [c["key"] for c in rt.ledger_candidates([dict(r, entry="heartbeat") for r in led171], today171)] == ["opt:ledger:model:sonnet"]
|
||||
assert rt.ledger_candidates(led171[:1], today171) == [], "one model, every dollar: nothing to compare it with"
|
||||
assert rt.ledger_candidates([{"at": "2026-10-16T09:00:00Z", "model": m, "usd": 0.01} for m in ("a", "b", "b")], today171) == [], "pennies"
|
||||
assert rt.ledger_candidates([dict(r, at="2026-10-01T09:00:00Z") for r in led171], today171) == [], "older than the window"
|
||||
assert rt.ledger_candidates([], today171) == [] and rt.ledger_candidates([{"at": "2026-10-16T09:00:00Z"}], today171) == []
|
||||
# a model whose spend per call is double its own median: a few heavy calls behind a cheap average
|
||||
skew171 = [{"at": "2026-10-16T09:00:00Z", "model": "sonnet", "usd": u} for u in (0.01, 0.01, 0.01, 0.01, 1.0)]
|
||||
assert [c["key"] for c in rt.ledger_candidates(skew171, today171)] == ["opt:ledger:model:sonnet"]
|
||||
assert "a $0.0100 median" in rt.ledger_candidates(skew171, today171)[0]["title"]
|
||||
assert rt.ledger_candidates(skew171[:4], today171) == [], "four cheap calls: no median worth a card"
|
||||
assert rt._median([3.0, 1.0]) == 2.0 and rt._median([3.0, 1.0, 2.0]) == 2.0 and rt._median([]) == 0.0
|
||||
# the Actions runs: two or more failed or cancelled runs of one workflow, with the count and the last run's id
|
||||
run171 = lambda i, wf, c, at: {"id": i, "path": f".gitea/workflows/{wf}", "conclusion": c, "created_at": at}
|
||||
runs171 = {"workflow_runs": [run171(51, "ci.yml", "failure", "2026-10-18T10:00:00Z"), run171(52, "ci.yml", "cancelled", "2026-10-19T10:00:00Z"),
|
||||
run171(53, "heartbeat.yml", "failure", "2026-10-19T10:00:00Z"), run171(54, "ci.yml", "success", "2026-10-19T11:00:00Z"),
|
||||
run171(55, "old.yml", "failure", "2026-10-01T10:00:00Z"), run171(56, "old.yml", "failure", "2026-10-02T10:00:00Z")]}
|
||||
ci171 = rt.run_candidates(runs171, today171)
|
||||
assert [c["key"] for c in ci171] == ["opt:ci:ci.yml"], [c["key"] for c in ci171]
|
||||
assert ci171[0]["title"] == "Why does ci.yml keep failing? 2 runs in 7 days" and "run `52`" in ci171[0]["body"], ci171[0]
|
||||
assert "2 failed or cancelled run(s)" in ci171[0]["body"] and "Router: opt:ci:ci.yml" in ci171[0]["body"]
|
||||
assert rt.run_candidates(runs171["workflow_runs"], today171) == ci171, "the list and the wrapped shape read the same"
|
||||
assert [c["key"] for c in rt.run_candidates([{"id": 60, "name": "sync", "conclusion": "failure", "created_at": "2026-10-18T10:00:00Z"}] * 2,
|
||||
today171)] == ["opt:ci:sync"], "no path: the workflow's name"
|
||||
spaced171 = rt.run_candidates([{"id": 61, "name": "CI build", "conclusion": "failure", "updated_at": "2026-10-18T10:00:00Z"}] * 2, today171)
|
||||
assert [c["key"] for c in spaced171] == ["opt:ci:CI-build"] and rt.named([iss80(82, "x", body=spaced171[0]["body"])]) == {"opt:ci:CI-build"}
|
||||
assert rt.run_candidates(None, today171) == [] and rt.run_candidates([{"junk": 1}, "junk"], today171) == []
|
||||
# parked cards: an `error` or `stuck` beat older than a day, no open PR, and #78's automatic retries spent
|
||||
park171 = lambda n, title, beat="error", at="2026-10-16T08:00:00Z", left=0, pr=False: {
|
||||
"n": n, "title": title, "beat": beat, "at": at, "retries_left": left, "open_pr": pr}
|
||||
parked171 = [park171(11, "A stuck card", beat="stuck"), park171(12, "Retries left", left=1), park171(13, "Has a PR", pr=True),
|
||||
park171(14, "Parked today", at="2026-10-19T08:00:00Z"), park171(15, "Moving", beat="dispatched")]
|
||||
p171 = rt.parked_candidates(parked171, today171)
|
||||
assert [c["key"] for c in p171] == ["opt:parked:11"], [c["key"] for c in p171]
|
||||
assert p171[0]["title"] == "Retry or split #11: A stuck card" and p171[0]["created"] == "2026-10-16T08:00:00Z"
|
||||
assert "last beat was `stuck`" in p171[0]["body"] and "DONE: #11 is moving again" in p171[0]["body"]
|
||||
assert rt.parked_candidates([], today171) == [] and rt.parked_candidates(["junk", {"n": None}], today171) == []
|
||||
# no duplicates: a source an open card already names is not proposed again
|
||||
seen171 = [iss80(80, "Why did sonnet cost so much?", body="- look at it\n\nRouter: opt:ledger:model:sonnet\n\nDONE: cheaper")]
|
||||
assert rt.named(seen171) == {"opt:ledger:model:sonnet"} and rt.named([iss80(81, "x")]) == set()
|
||||
assert [c["key"] for c in rt.opt_candidates(led171, runs171, parked171, today171)] == \
|
||||
["opt:ledger:model:sonnet", "opt:ci:ci.yml", "opt:parked:11"]
|
||||
assert [c["key"] for c in rt.opt_candidates(led171, runs171, parked171, today171, rt.named(seen171))] == ["opt:ci:ci.yml", "opt:parked:11"]
|
||||
assert rt.opt_candidates(led171, runs171, parked171, today171,
|
||||
{"opt:ledger:model:sonnet", "opt:ci:ci.yml", "opt:parked:11"}) == [], "all three already have cards"
|
||||
# the table shows a synthetic candidate by its key, and what would become of it
|
||||
r171 = rt.rank(rt.opt_candidates(led171, runs171, parked171, today171), {}, live80, {}, today171)
|
||||
assert "| opt:ci:ci.yml Why does ci.yml keep failing" in rt.table(r171, [], "dry")
|
||||
assert "**would be written as a card**" in rt.table(r171, r171[:1], "dry")
|
||||
r171[0]["written"] = 99
|
||||
assert "**written as #99**" in rt.table(r171, r171[:1], "live")
|
||||
|
||||
# Astra gathers the three sources and, live, writes the picked one as an unqueued card; dry writes nothing
|
||||
said171, labelled171, made171, asked171 = [], [], [], []
|
||||
beat171 = lambda b, h, **extra: {"user": {"login": astra.tm.TIME_BOT}, "created_at": iso(h),
|
||||
"body": "x\n<!-- time " + json.dumps({"beat": b, **extra}) + " -->"}
|
||||
beats171 = [beat171("dispatched", 96, launched=True), beat171("error", 90),
|
||||
beat171("dispatched", 84, launched=True, auto_retry=True), beat171("error", 72, auto_retry=True)]
|
||||
led171live = [{"at": iso(48), "model": "sonnet", "usd": 2.0}] + [{"at": iso(48), "model": "haiku", "usd": 0.1}] * 3
|
||||
runs171live = [run171(51, "ci.yml", "failure", iso(48)), run171(52, "ci.yml", "cancelled", iso(24))]
|
||||
card171 = [iss80(30, "A parked card", labels=["ready"])]
|
||||
astra.tm.issues = lambda q: [] if q.startswith("state=closed") else card171
|
||||
astra.tm.gh = lambda method, path, body=None, **k: beats171 if "/comments" in path else [] if "/pulls" in path \
|
||||
else {"workflow_runs": runs171live}
|
||||
astra.tm.ledger_lines = lambda month: led171live if month == NOW.strftime("%Y-%m") else []
|
||||
astra.tm.pools = lambda: []
|
||||
astra.tm.budget = lambda: (1.0, 15.0, 10.0, 100.0)
|
||||
astra.tm.comment = lambda n, text, meta: said171.append((n, text, meta))
|
||||
astra.tm.label = lambda n, name: labelled171.append((n, name))
|
||||
astra.tm.new_issue = lambda title, body, labels: made171.append((title, body, labels)) or {"number": 200 + len(made171)}
|
||||
assert [p["n"] for p in astra.parked_cards()] == [30] and astra.parked_cards()[0]["retries_left"] == 0, astra.parked_cards()
|
||||
assert astra.parked_cards()[0]["beat"] == "error" and not astra.parked_cards()[0]["open_pr"]
|
||||
assert len(astra.ledger_week()) == 4 and [r["id"] for r in astra.action_runs()] == [51, 52]
|
||||
keys171 = ["opt:ledger:model:sonnet", "opt:ci:ci.yml", "opt:parked:30"]
|
||||
reply171 = {"note": "the records", "assess": [{"key": k, "kind": "optimization", "value": 5, "size": "S"} for k in keys171]}
|
||||
astra.tm.think = lambda seat, a, b, system, user: (asked171.append(user), json.dumps(reply171))[1]
|
||||
assert astra.router_tick(rt.config('{"mode": "dry"}'), "opus", 57, "ctx", [], 2, 2) == 0
|
||||
assert not made171 and not labelled171, "dry: no card written, nothing queued"
|
||||
text171, meta171 = said171[-1][1], said171[-1][2]
|
||||
assert "And write a card for: opt:parked:30, opt:ledger:model:sonnet." in text171, text171
|
||||
assert "**would be written as a card**" in text171 and meta171["queued"] == 0 and meta171["would"] == []
|
||||
assert meta171["opt"] == [{"key": "opt:parked:30", "card": None}, {"key": "opt:ledger:model:sonnet", "card": None}]
|
||||
assert all(k in asked171[-1] for k in keys171) and "Retry or split #30" in asked171[-1], "the seat judges them like any candidate"
|
||||
assert astra.router_tick(rt.config('{"mode": "live"}'), "opus", 57, "ctx", [], 2, 2) == 0
|
||||
assert [m[0] for m in made171] == ["Retry or split #30: A parked card", "Why did sonnet cost $2.00 of the week's $2.30?"], made171
|
||||
assert made171[0][2] == [] and labelled171 == [], "written unqueued; #30 itself is never worked on in place"
|
||||
assert "Router: opt:parked:30" in made171[0][1] and "card written by Astra" in made171[0][1] and rt.DONE.search(made171[0][1])
|
||||
text171, meta171 = said171[-1][1], said171[-1][2]
|
||||
assert "Written from the factory's own records (candidates from tomorrow): #201 (opt:parked:30), #202 (opt:ledger:model:sonnet)" in text171
|
||||
assert "**written as #201**" in text171 and "Queued nothing today." in text171
|
||||
assert meta171["opt"] == [{"key": "opt:parked:30", "card": 201}, {"key": "opt:ledger:model:sonnet", "card": 202}]
|
||||
assert meta171["picked"] == [{"n": 201, "kind": "optimization"}, {"n": 202, "kind": "optimization"}] and meta171["queued"] == 0
|
||||
# the card Astra wrote names its source, so the same source is not proposed again
|
||||
card171.append(iss80(31, "Retry or split #30: A parked card", body=made171[0][1], labels=["queued"]))
|
||||
made171.clear()
|
||||
assert astra.router_tick(rt.config('{"mode": "live"}'), "opus", 57, "ctx", [], 2, 2) == 0
|
||||
assert [m[0] for m in made171] == ["Why did sonnet cost $2.00 of the week's $2.30?", "Why does ci.yml keep failing? 2 runs in 7 days"]
|
||||
assert said171[-1][2]["opt"] == [{"key": "opt:ledger:model:sonnet", "card": 201}, {"key": "opt:ci:ci.yml", "card": 202}]
|
||||
print("ALL OK 171")
|
||||
|
||||
# 82 adaptive budget: each route is paced to its capacity pool; no pool, no change
|
||||
import io, contextlib
|
||||
real_pools = tm.pools
|
||||
|
||||
@@ -28,6 +28,7 @@
|
||||
- The second fix round escalates to a different model from the first round: `claude:opus` with the Claude seats (it was `venice:kimi code`, and grok 4.7 before that). `TIME_VENTURE_ESCALATE` overrides it.
|
||||
|
||||
### Added
|
||||
- The router proposes optimization work the factory's own records show, instead of waiting for a human to notice it (#171; `router.py`, `astra.py`). Three sources, read by arithmetic over the last seven days: the ledger (a model or an entry point that booked over half the week's spend, where there is a second one to compare it with; a model whose spend per call is double its own median, so a few heavy calls carry the bill; pennies are left alone), the Actions runs (a workflow with two or more failed or cancelled runs, with the count and the last run's id) and the parked cards (an `error` or `stuck` beat older than a day, with no open PR and #78's automatic retries spent: "retry or split #N"). Each finding is a candidate with no card yet, keyed `opt:<source>:<id>` and kind optimization, with a title and body already drafted; Astra's seat judges it like any other candidate. Live, a picked one is written as a card (unqueued, a candidate from tomorrow) and never worked on in place; dry, the ranked table shows it. A source an open card already names (`Router: opt:<source>:<id>` in the body) is not proposed again.
|
||||
- The cloud lane: Claude Code cloud sessions build cards on the owner's cloud-session credit, beside the API worker. With `cloud: {"per_day": N, "sizes": ["M"]}` in `time/router.json`, Astra's live run labels the best-ranked actionable cards of those sizes `cloud`, until N cards are in the lane, before the API worker's picks; a `cloud` card is lined up, so it is never queued for the API worker. A launcher on hb-5 (a user agent in the desktop session, where the claude.ai sign-in lives; it reads the public issue list and holds no Teahub credentials) starts one `claude --cloud` session per card, with the card as the prompt, on the GitHub copy. The session pushes a `claude/card-N-*` branch; sync.py brings it home as a PR that closes the card (adding `Closes #N` when the commits don't), moves the card from `cloud` to `ready` with a note that carries no beat, and the heartbeat adopts the PR, so the council reviews and lands it like any card PR. A card the commits close leaves the lane too, whatever the branch is called. `per_day: 0`, the default, is no lane.
|
||||
- Time brings a PR branch up to date with its base (#123). An approved PR Gitea reports as not `mergeable` (or whose merge call answers 409) gets `git merge origin/<base>` (the PR's own base) pushed as one commit by Time, `card #N: merge <base>`; CI runs on the new head and landing follows on a later beat (a `ci-dispatched` or `updated` beat now reaches `land()` again). A conflict is not guessed at: the merge is aborted, the branch left as it was, and the conflicting files and `<<<<<<<` blocks (trimmed) are posted on the card and the PR with the `needs-human` label; `/time retry` hands them to the worker as the reviewer's text. Known limit: the worker's `write_file` of a resolved file is a plain commit, not the merge, so Time's next try can conflict again.
|
||||
- A card waiting on a person holds no `ready` slot (#135). `promote_next` counts toward `TIME_PARALLEL` only the `ready` cards that are moving; one whose last real beat is `needs-human`, `error` or `stuck` (an objection, or a failure Time won't retry again) keeps its label, so `/time retry` still finds it, but frees its slot. Three such cards held every slot twice, the second time for ten hours on 2026-10-19 while 22 cards sat `queued`. The run log prints `promote: ready=N moving=M`.
|
||||
|
||||
Reference in new issue
Block a user