Router v1: the factory ranks its own next work (dry to start) (#169)

Router v1: the factory ranks its own next work, dry to start (#169)

Astra's seat judges every candidate (kind, value, size); router.py ranks them with the owner's weights per kind
(time/router.json), a restless boost and card age. dry posts the table; live queues from the top within the day's
money and the heartbeat promotes queued cards in ranked order. Astra reads the last 14 days of its log. Closes #168.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017K4vyY3yUD3GsvoZaAtKkD

.github/time/astra.py

@@ -11,9 +11,18 @@ TIME_ASTRA_MAX_READY cards queued and TIME_ASTRA_MAX_NEW cards created per run;
 has a `DONE:` line and none of `queued`/`ready`/`needs-human`/`andon`; an open `andon` issue means Astra writes its note
 and queues nothing; the spend cap is checked by the engine like any other call. Everything Astra does is a comment on
 one issue, "Astra log", so a human can read the day's reasoning in one place and object there.
+
+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
+nothing; `live` queues from the top while the day's money left covers each card's estimated cost, creates the seat's
+new cards unqueued (they are candidates the next day), and the heartbeat promotes queued cards in the ranked order.
+`off`, or no router.json, is the planner as before.
 """
 from __future__ import annotations
 
+import base64
+import datetime as dt
 import importlib.util
 import json
 import os
@@ -25,6 +34,10 @@ HERE = Path(__file__).resolve().parent
 _spec = importlib.util.spec_from_file_location("tm", HERE / "time.py")
 tm = importlib.util.module_from_spec(_spec)
 _spec.loader.exec_module(tm)
+_rspec = importlib.util.spec_from_file_location("router", HERE / "router.py")
+router = importlib.util.module_from_spec(_rspec)
+_rspec.loader.exec_module(router)
+ROUTER_FILE = "time/router.json"
 
 SPINE = os.environ.get("TIME_SPINE") or "teahubxyz/teahub"   # where AGENTS.md and docs/decisions.md live
 LOG_TITLE = "Astra log"
@@ -51,6 +64,25 @@ Answer with JSON only, no prose around it:
   "new": [{{"title": "<card title>", "body": "<what to change, as a list, ending with a DONE: line>", "queue": true}}]}}"""
 
 
+ROUTER = """You are Astra, the daily planner of a small, fully AI-run software factory (Teahub). Today you judge the
+candidates below; code ranks them with the owner's weights and queues from the top, so don't choose: judge each one.
+
+For every candidate give:
+- kind: venture (a new venture, or a venture's next version), feature (new functionality), refinement (making what
+  exists better: fixes, polish, a live venture's page or skills) or optimization (cost, speed, reliability)
+- value: 1-5, what finishing it is worth to the factory now (5: it unblocks the factory, or the owner asked for it and
+  it matters today; 1: nice to have, stale, or a duplicate of another card)
+- size: S (well under one worker session), M (one session), L (more than one: it must be split first)
+- why: one short sentence
+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.
+
+Answer with JSON only, no prose around it:
+{{"note": "<3-6 sentences: what happened yesterday and what matters today>",
+  "assess": [{{"key": "<the candidate's key, e.g. #82>", "kind": "feature", "value": 3, "size": "M", "why": "..."}}],
+  "new": [{{"title": "<card title>", "body": "<what to change, as a list, ending with a DONE: line>"}}]}}"""
+
+
 def env_int(name: str, default: int) -> int:
     try:
         return int(os.environ.get(name) or default)
@@ -103,6 +135,94 @@ def decide(state: dict, context: str, seat: str) -> dict:
             "new": [c for c in out.get("new") or [] if isinstance(c, dict) and c.get("title")]}
 
 
+def router_config() -> dict:
+    """time/router.json on the default branch; missing or unreadable, the router is off and Astra plans as before."""
+    try:
+        f = tm.gh("GET", f"/repos/{tm.REPO}/contents/{ROUTER_FILE}")
+        return router.config(base64.b64decode(f.get("content") or "").decode(errors="replace"))
+    except Exception as e:  # noqa: BLE001
+        if "-> 404" not in str(e):
+            print(f"router config unreadable: {str(e)[:120]}")
+        return router.config(None)
+
+
+def recent(log: int, days: int = 14) -> list[dict]:
+    """The log's comments of the last `days` days, oldest first (Gitea pages from the oldest, so a plain limit=20
+    read the thread's first twenty comments forever)."""
+    since = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")
+    out: list[dict] = []
+    for page in range(1, 5):
+        batch = tm.gh("GET", f"/repos/{tm.REPO}/issues/{log}/comments?since={since}&limit=50&page={page}") or []
+        out += batch
+        if len(batch) < 50:
+            break
+    return out
+
+
+def parse(raw: str) -> dict:
+    """The first JSON object in the seat's reply."""
+    dec = json.JSONDecoder()
+    for m in re.finditer(r"\{", raw or ""):
+        try:
+            out, _ = dec.raw_decode(raw, m.start())
+        except ValueError:
+            continue
+        if isinstance(out, dict):
+            return out
+    return {}
+
+
+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]
+    cands = router.candidates(open_issues) + [dict(router.NEW_VENTURE)]
+    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()
+    user = (f"Factory context (the spine's AGENTS.md and the newest decisions):\n{context}\n\nSpend: ${today:.2f} of "
+            f"${day_cap:.2f} today, ${month:.2f} of ${month_cap:.2f} this month.\n\nCandidates ({len(cands)}):\n\n{listing}")
+    try:
+        out = parse(tm.think(seat, None, None, ROUTER.format(max_new=max_new), user))
+    except tm.Budget as e:
+        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())
+    picks = router.pick(ranked, cfg, max(0.0, day_cap - today), max_ready)
+    new = [c for c in out.get("new") or [] if isinstance(c, dict) and c.get("title")][:max_new]
+    live, created = cfg["mode"] == "live", []
+    if live:
+        for r in picks:
+            tm.label(r["n"], "queued")   # the heartbeat promotes it, in ranked order
+        for c in new:
+            body = str(c.get("body") or "")
+            if not router.DONE.search(body):
+                body += "\n\nDONE: PR open that does the above, tests green."
+            created.append(tm.new_issue(str(c["title"])[:200], body + "\n\n<sub>card written by Astra</sub>", [])["number"])
+    unjudged = sum(1 for r in ranked if not r["judged"])
+    w = cfg["weights"]
+    head = (f"🔭 Astra ({seat}), router `{cfg['mode']}`, spend ${today:.2f} of ${day_cap:.2f} today. Weights: "
+            + ", ".join(f"{k} {w[k]:g}" for k in router.KINDS) + f"; restless {cfg['restless']:g}/day.")
+    lines = [head, "", str(out.get("note") or "(no note)"), "", router.table(ranked, picks, cfg["mode"]), ""]
+    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.")
+        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") + ".")
+        if new:
+            lines.append("It would write: " + "; ".join(router._txt(c["title"], 100) for c in new) + ".")
+    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}
+    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)}")
+    return 0
+
+
 def tick() -> int:
     seat = (os.environ.get("TIME_ASTRA_SEAT") or "opus").strip()
     max_ready, max_new = env_int("TIME_ASTRA_MAX_READY", 2), env_int("TIME_ASTRA_MAX_NEW", 2)
@@ -114,10 +234,13 @@ def tick() -> int:
         return 0
     context = spine_file("AGENTS.md")[:6000] + "\n\n" + "\n".join(spine_file("docs/decisions.md").splitlines()[-25:])
     # the log thread is part of the context: humans object there
-    thread = tm.gh("GET", f"/repos/{tm.REPO}/issues/{log}/comments?limit=20")
+    thread = recent(log)
     human = [c["body"][:500] for c in thread if not tm.is_time(c)][-5:]
     if human:
         context += "\n\nRecent human replies on the Astra log (follow these):\n- " + "\n- ".join(human)
+    cfg = router_config()
+    if cfg["mode"] != "off":
+        return router_tick(cfg, seat, log, context, thread, max_ready, max_new)
     try:
         plan = decide(state, context, seat)
     except tm.Budget as e:

.github/time/router.py

@@ -0,0 +1,185 @@
+"""The router: what the factory works on next (time#68, #72).
+
+Astra, the daily planner, gathers the candidates and asks its seat to judge each one; this module turns those
+judgments into one ranked list, in arithmetic anyone can check, and picks from the top. In live mode the heartbeat
+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.
+
+    score = weight[kind] x restless(kind) x age x (value + 0.5 x unblocks) / size
+
+- value (1-5) and size (S, M, L = 1, 2, 4) are the seat's judgment; unblocks counts the open cards that wait on it.
+- weight is the owner's dial per kind, in time/router.json (all 1: balanced).
+- restless: a kind the factory hasn't picked for a while climbs, 1 + rate x days (at most 10 days), so none starves.
+- age: an old card climbs a little, 1 + 0.02 x days open (at most 20 days).
+
+An L is ranked but never queued: a card is what one worker session finishes, so it is split first. The policy is
+`score`; a bandit policy (Beth, Jev, Fable's restless-bandit maths) can replace it without touching how candidates are
+gathered or how picks are queued.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import json
+import re
+
+KINDS = ("venture", "feature", "refinement", "optimization")
+SIZES = {"S": 1, "M": 2, "L": 4}
+MODES = ("off", "dry", "live")
+DEFAULTS = {"mode": "off", "weights": dict.fromkeys(KINDS, 1.0), "restless": 0.05, "cost_usd": {"S": 0.5, "M": 1.5, "L": 4.0}}
+LOOP = {"venture", "venture-batch", "signal", "signup"}       # the venture loop's own cards: it runs them itself
+HELD = {"needs-human", "andon", "astra", "planned"} | LOOP     # a human's, an alarm, Astra's log, parents, the loop's
+LINED_UP = {"queued", "ready"}                                 # already moving: ranked (for promotion), never queued again
+DONE = re.compile(r"^\s*DONE:", re.M | re.I)
+NEW_VENTURE = {"key": "venture:new", "n": None, "title": "Start a new venture: research, pick, build and launch one page",
+               "body": "The venture loop (venture.py): one venture, researched and picked by the lead, built and judged.",
+               "labels": [], "lined_up": False, "actionable": False, "kind": "venture", "unblocks": 0, "created": "",
+               "why_not": "not wired to the router yet; the venture loop is paused (TIME_VENTURE=off)"}
+
+
+def _num(v, default: float, lo: float = 0.0, hi: float = 100.0) -> float:
+    try:
+        return min(hi, max(lo, float(v)))
+    except (TypeError, ValueError):
+        return default
+
+
+def _txt(v, n: int) -> str:
+    return re.sub(r"\s+", " ", str(v or "")).strip()[:n]
+
+
+def config(raw: str | None) -> dict:
+    """time/router.json over the defaults. Unreadable is the defaults, and an unknown mode is off."""
+    try:
+        got = json.loads(raw) if raw else {}
+    except ValueError:
+        got = {}
+    got = got if isinstance(got, dict) else {}
+    w = got.get("weights") if isinstance(got.get("weights"), dict) else {}
+    c = got.get("cost_usd") if isinstance(got.get("cost_usd"), dict) else {}
+    mode = str(got.get("mode", "off")).strip().lower()
+    return {"mode": mode if mode in MODES else "off",
+            "weights": {k: _num(w.get(k), 1.0) for k in KINDS},
+            "restless": _num(got.get("restless"), DEFAULTS["restless"], 0.0, 1.0),
+            "cost_usd": {s: _num(c.get(s), DEFAULTS["cost_usd"][s]) for s in SIZES}}
+
+
+def labels_of(issue: dict) -> set[str]:
+    return {lb["name"] if isinstance(lb, dict) else lb for lb in issue.get("labels") or []}
+
+
+def guess_kind(title: str, body: str, labels: set[str]) -> str:
+    """The kind when the seat gives none: a `kind:<x>` label, the loop's labels, a live venture's folder, then words."""
+    for k in KINDS:
+        if f"kind:{k}" in labels:
+            return k
+    if labels & LOOP:
+        return "venture"
+    if re.search(r"^\s*Base:\s*gh-pages\b", body or "", re.M | re.I):
+        return "refinement"   # work on a launched venture's folder
+    text = f"{title}\n{body}".lower()
+    if re.search(r"\b(cost|spend|cach\w*|cheaper|faster|slow\w*|latency|tokens?|optimi[sz]\w*)\b", text):
+        return "optimization"
+    if re.search(r"\b(fix\w*|polish\w*|refine\w*|tighten\w*|clean ?up|rename\w*|bug\w*)\b", text):
+        return "refinement"
+    return "feature"
+
+
+def candidates(issues: list[dict]) -> list[dict]:
+    """Every open card a worker could take, lined-up ones included (ranked, so the heartbeat promotes in order)."""
+    waits: dict[int, int] = {}
+    for i in issues:
+        for d in set(re.findall(r"Depends on #(\d+)", i.get("body") or "")):
+            waits[int(d)] = waits.get(int(d), 0) + 1
+    out = []
+    for i in issues:
+        labels, body = labels_of(i), i.get("body") or ""
+        if labels & HELD or not DONE.search(body):
+            continue
+        out.append({"key": f"#{i['number']}", "n": i["number"], "title": _txt(i.get("title"), 120), "body": body[:500],
+                    "labels": sorted(labels), "lined_up": bool(labels & LINED_UP), "actionable": True,
+                    "kind": guess_kind(i.get("title") or "", body, labels), "unblocks": waits.get(i["number"], 0),
+                    "created": str(i.get("created_at") or "")})
+    return out
+
+
+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] = {}
+    for b in beats:
+        if b.get("router") != "live":
+            continue
+        day = str(b.get("at") or "")[:10]
+        for p in b.get("picked") or []:
+            k = p.get("kind") if isinstance(p, dict) else None
+            if k in KINDS and day > out.get(k, ""):
+                out[k] = day
+    return out
+
+
+def _days(since: str, today: dt.date, cap: float) -> float:
+    try:
+        return float(min(cap, max(0, (today - dt.date.fromisoformat(since[:10])).days)))
+    except ValueError:
+        return float(cap)   # never picked (or no date): as restless as it gets
+
+
+def score(c: dict, a: dict, cfg: dict, last: dict[str, str], today: dt.date) -> dict:
+    """One candidate, judged (`a`: the seat's kind, value, size, why) and scored, with every factor kept for the table."""
+    kind = a.get("kind") if a.get("kind") in KINDS else c["kind"]
+    size = str(a.get("size") or "M").strip().upper()[:1]
+    size = size if size in SIZES else "M"
+    value = _num(a.get("value"), 3.0, 1.0, 5.0)
+    restless = 1 + cfg["restless"] * _days(last.get(kind, ""), today, 10)
+    age = 1 + 0.02 * (_days(c["created"], today, 20) if c.get("created") else 0)
+    s = cfg["weights"][kind] * restless * age * (value + 0.5 * c["unblocks"]) / SIZES[size]
+    out = {**c, "kind": kind, "value": value, "size": size, "why": _txt(a.get("why"), 200),
+           "restless": round(restless, 2), "age": round(age, 2), "score": round(s, 3), "judged": bool(a)}
+    if size == "L" and out["actionable"]:
+        out["actionable"], out["why_not"] = False, "L: split it into cards a worker finishes in one session first"
+    return out
+
+
+def rank(cands: list[dict], assess: dict[str, dict], cfg: dict, last: dict[str, str], today: dt.date) -> list[dict]:
+    """Best first; ties to the older card."""
+    scored = [score(c, assess.get(c["key"]) or {}, cfg, last, today) for c in cands]
+    return sorted(scored, key=lambda r: (-r["score"], r["n"] if r["n"] is not None else 10**9))
+
+
+def pick(ranked: list[dict], cfg: dict, left_usd: float, max_n: int) -> list[dict]:
+    """From the top: cards not yet lined up, while the day's money left covers their size's estimated cost."""
+    out, spent = [], 0.0
+    for r in ranked:
+        if len(out) >= max_n:
+            break
+        if r["lined_up"] or not r["actionable"]:
+            continue
+        cost = cfg["cost_usd"][r["size"]]
+        if spent + cost > left_usd:
+            continue
+        out.append(r)
+        spent += cost
+    return out
+
+
+def order(ranked: list[dict]) -> list[int]:
+    """Card numbers best first: the heartbeat promotes queued cards in this order."""
+    return [r["n"] for r in ranked if r["n"] is not None]
+
+
+def table(ranked: list[dict], picked: list[dict], mode: str, limit: int = 15) -> str:
+    chosen = {r["key"] for r in picked}
+    mark = "queued today" if mode == "live" else "would queue"
+    rows = []
+    for k, r in enumerate(ranked[:limit], 1):
+        note = (f"**{mark}**" if r["key"] in chosen else "lined up" if r["lined_up"]
+                else r.get("why_not", "") if not r["actionable"] else "")
+        name = f"{r['key']} {r['title']}" if r["n"] is not None 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('|', '/')} |")
+    more = f"\n\n_{len(ranked) - limit} more ranked below these._" if len(ranked) > limit else ""
+    return ("| # | card | kind | value | size | waiting | restless | score | why | status |\n|---|---|---|---|---|---|---|---|---|---|\n"
+            + "\n".join(rows) + more)

.github/time/test_time.py

@@ -1131,6 +1131,106 @@ assert gv.research(repos79, "opus", gv.registry()) is None
 gv.tm.comment, gv.registry, gv.ask_json, gv.readme_of = real79c
 print("ALL OK 79c")
 
+# 80 the router (time#68, #72): the seat judges every candidate, router.py ranks them with the owner's weights per kind,
+# Astra queues from the top in live mode and only shows the ranking in dry mode, and the heartbeat promotes queued
+# cards in the live ranking's order
+rspec = importlib.util.spec_from_file_location("rt", Path(__file__).resolve().parent / "router.py")
+rt = importlib.util.module_from_spec(rspec); rspec.loader.exec_module(rt)
+cfg80 = rt.config(None)
+assert cfg80["mode"] == "off" and cfg80["weights"] == dict.fromkeys(rt.KINDS, 1.0), "no router.json: off, balanced"
+assert rt.config("not json")["mode"] == "off" and rt.config('{"mode": "LIVE"}')["mode"] == "live" and rt.config('{"mode": "x"}')["mode"] == "off"
+c80 = rt.config('{"mode": "dry", "weights": {"venture": 2, "feature": "junk"}, "restless": 9, "cost_usd": {"S": 1}, "_about": "ok"}')
+assert c80["weights"]["venture"] == 2 and c80["weights"]["feature"] == 1 and c80["restless"] == 1, c80
+assert c80["cost_usd"] == {"S": 1, "M": 1.5, "L": 4.0}
+# the kind when the seat gives none
+assert rt.guess_kind("x", "Base: gh-pages\nDONE: y", set()) == "refinement", "a live venture's folder"
+assert rt.guess_kind("Adaptive budget", "pace spend to the pool", set()) == "optimization"
+assert rt.guess_kind("Fix the header", "", set()) == "refinement" and rt.guess_kind("Mentions are handoffs", "", set()) == "feature"
+assert rt.guess_kind("Fix it", "", {"kind:optimization"}) == "optimization" and rt.guess_kind("x", "", {"signal"}) == "venture"
+# candidates: a DONE line and not held; lined-up cards are ranked (for promotion) but never queued again
+iss80 = lambda n, title, body="DONE: x", labels=(), created="2026-10-18T00:00:00Z": {
+    "number": n, "title": title, "body": body, "labels": [{"name": l} for l in labels], "created_at": created}
+open80 = [iss80(1, "Engine feature"), iss80(2, "Fix the docs", labels=["queued"]), iss80(3, "Parent", labels=["planned"]),
+          iss80(4, "No done line", body="just an idea"), iss80(5, "Cheaper reviews", body="Depends on #1\nDONE: z"),
+          iss80(6, "A human's", labels=["needs-human"]), iss80(7, "Huge rewrite", created="2026-10-01T00:00:00Z"),
+          iss80(8, "Huebrain: want the next version?", labels=["signal"])]
+cands80 = rt.candidates(open80)
+assert [c["n"] for c in cands80] == [1, 2, 5, 7] and cands80[0]["unblocks"] == 1 and cands80[1]["lined_up"], cands80
+# scoring: the seat's value and size, the owner's weight per kind, restless per kind, age per card
+today80 = dt.date(2026, 10, 19)
+assess80 = {"#1": {"kind": "feature", "value": 4, "size": "M", "why": "w"}, "#2": {"kind": "refinement", "value": 5, "size": "s"},
+            "#5": {"kind": "optimization", "value": 2, "size": "S"}, "#7": {"kind": "feature", "value": 5, "size": "L"},
+            "venture:new": {"kind": "venture", "value": 3, "size": "M"}}
+live80 = rt.config('{"mode": "live"}')
+r80 = rt.rank(cands80 + [dict(rt.NEW_VENTURE)], assess80, live80, {}, today80)
+s80 = {r["key"]: r for r in r80}
+assert [r["key"] for r in r80] == ["#2", "#1", "#5", "#7", "venture:new"], [(r["key"], r["score"]) for r in r80]
+assert s80["#1"]["restless"] == 1.5 and abs(s80["#1"]["score"] - 1.5 * 1.02 * 4.5 / 2) < 0.002, s80["#1"]   # never picked: 1.5
+assert s80["#7"]["actionable"] is False and "split" in s80["#7"]["why_not"] and s80["#7"]["age"] == 1.36, s80["#7"]
+assert s80["venture:new"]["actionable"] is False and s80["venture:new"]["n"] is None and s80["#2"]["size"] == "S"
+assert rt.order(r80) == [2, 1, 5, 7]
+# a kind picked yesterday is less restless; the owner's weight lifts a kind
+r80b = rt.rank(cands80, assess80, rt.config('{"weights": {"optimization": 4}}'), {"refinement": "2026-10-18"}, today80)
+assert {r["key"]: r for r in r80b}["#2"]["restless"] == 1.05 and r80b[0]["key"] == "#5", [(r["key"], r["score"]) for r in r80b]
+u80 = rt.score(cands80[0], {}, live80, {}, today80)
+assert u80["value"] == 3 and u80["size"] == "M" and u80["kind"] == "feature" and not u80["judged"], "unjudged: 3, M, the guessed kind"
+assert rt.score(cands80[0], {"kind": "nonsense", "value": 99, "size": "huge"}, live80, {}, today80)["value"] == 5
+# picks: from the top, past lined-up and unwired candidates, while the money left covers each size's estimate
+assert [r["key"] for r in rt.pick(r80, live80, 10.0, 2)] == ["#1", "#5"]
+assert [r["key"] for r in rt.pick(r80, live80, 1.0, 2)] == ["#5"], "an M doesn't fit $1; the S behind it does"
+assert rt.pick(r80, live80, 10.0, 0) == []
+beats80 = [{"beat": "astra", "router": "live", "at": "2026-10-17T13:13:00Z", "picked": [{"n": 1, "kind": "feature"}]},
+           {"beat": "astra", "router": "dry", "at": "2026-10-18T13:13:00Z", "picked": [{"n": 2, "kind": "refinement"}]},
+           {"beat": "astra", "router": "live", "at": "2026-10-18T13:13:00Z", "picked": [{"n": 5, "kind": "feature"}, "junk"]}]
+assert rt.last_picked(beats80) == {"feature": "2026-10-18"}, "live picks only"
+t80 = rt.table(r80, rt.pick(r80, live80, 10.0, 2), "dry")
+assert "| 1 | #2 Fix the docs |" in t80 and "lined up" in t80 and "**would queue**" in t80 and "split" in t80 and "not wired" in t80
+assert "**queued today**" in rt.table(r80, rt.pick(r80, live80, 10.0, 2), "live")
+# Astra's router tick: dry shows the ranking and touches nothing; live queues the picks and writes the new cards unqueued
+aspec = importlib.util.spec_from_file_location("astra80", Path(__file__).resolve().parent / "astra.py")
+astra = importlib.util.module_from_spec(aspec); aspec.loader.exec_module(astra)
+said80, labelled80, made80, asked80 = [], [], [], []
+astra.tm.issues = lambda q: open80 + [iss80(57, "Astra log", labels=["astra"])]
+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)}
+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]
+assert astra.router_tick(rt.config('{"mode": "dry"}'), "opus", 57, "ctx", [], 2, 2) == 0
+assert not labelled80 and not made80, "dry: nothing queued, nothing created"
+n80, text80, meta80 = said80[-1]
+assert n80 == 57 and meta80["router"] == "dry" and meta80["would"] == [1, 5] and meta80["queued"] == 0 and meta80["order"] == [2, 1, 5, 7]
+assert "Dry run" in text80 and "Write a thing" in text80 and "| 1 | #2 Fix the docs" in text80 and "the note" in text80
+assert "### #4" not in asked80[0][2] and "Astra log" not in asked80[0][2] and "venture:new" in asked80[0][2] and "ctx" in asked80[0][2]
+assert asked80[0][0] == "opus" and "judge each one" in asked80[0][1]
+assert astra.router_tick(rt.config('{"mode": "live"}'), "opus", 57, "ctx", [], 2, 2) == 0
+assert labelled80 == [(1, "queued"), (5, "queued")] and [m[0] for m in made80] == ["Write a thing"] and made80[0][2] == []
+assert "DONE:" in made80[0][1] and "card written by Astra" in made80[0][1]
+meta80 = said80[-1][2]
+assert meta80["picked"] == [{"n": 1, "kind": "feature"}, {"n": 5, "kind": "optimization"}] and meta80["new"] == [91] and meta80["queued"] == 2
+astra.tm.think = lambda *a: (_ for _ in ()).throw(astra.tm.Budget("today's spend is over the cap"))
+assert astra.router_tick(rt.config('{"mode": "live"}'), "opus", 57, "ctx", [], 2, 2) == 0 and said80[-1][2]["capped"]
+assert astra.parse("x {bad} y {\"a\": 1}") == {"a": 1} and astra.parse("") == {}
+# the heartbeat promotes queued cards in the live ranking's order, oldest first without one
+real80 = (tm.issues, tm.gh, tm.label, tm.unlabel)
+moved80 = []
+q80 = [iss80(10, "a", labels=["queued"]), iss80(11, "b", labels=["queued"]), iss80(12, "c", body="Depends on #99\nDONE: x", labels=["queued"])]
+beat80 = {"user": {"login": tm.TIME_BOT}, "created_at": "2026-10-19T13:13:00Z",
+          "body": "x\n<!-- time " + json.dumps({"beat": "astra", "router": "live", "order": [12, 11, 10]}) + " -->"}
+tm.issues = lambda q: {"state=open&labels=ready": [], "state=open&labels=queued": q80,
+                       "state=open&labels=astra": [iss80(57, "Astra log", labels=["astra"])]}[q]
+tm.gh = lambda method, path, body=None, **k: [beat80] if "/comments" in path else {"state": "open" if path.endswith("/issues/99") else "closed"}
+tm.label = lambda n, name: moved80.append((n, name)); tm.unlabel = lambda n, name: None
+assert tm.rank_order() == [12, 11, 10] and tm.promote_next() == 11 and moved80[-1] == (11, "ready"), "the best whose dependencies are closed"
+beat80["body"] = beat80["body"].replace('"live"', '"dry"')
+assert tm.rank_order() == [] and tm.promote_next() == 10, "a dry ranking moves nothing: oldest first"
+tm.gh = lambda *a, **k: (_ for _ in ()).throw(RuntimeError("GET x -> 500"))
+assert tm.rank_order() == [], "the ranking is a preference: an unreadable log never stops promotion"
+tm.issues, tm.gh, tm.label, tm.unlabel = real80
+print("ALL OK 80")
+
 # ---------- venture loop lifecycle: these still run against the old GitHub-shaped fake (Pages API, author_association,
 # per_page); the Gitea shapes are pinned by 79 above. They run only with TIME_TEST_VENTURE=1 until the fake is ported.
 if not os.environ.get("TIME_TEST_VENTURE"):

.github/time/time.py

@@ -620,13 +620,31 @@ def plan_spec(spec: dict) -> None:
     label(n, "planned")
 
 
+def rank_order() -> list[int]:
+    """The router's latest live ranking (Astra's beat on its log, router.py), card numbers best first; [] when the
+    router is off or dry, or the log can't be read: then cards are promoted oldest first, as before."""
+    try:
+        log = next(iter(issues("state=open&labels=astra")), None)
+        if not log:
+            return []
+        since = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=3)).strftime("%Y-%m-%dT%H:%M:%SZ")
+        beats = [b for b in beats_of(gh("GET", f"/repos/{REPO}/issues/{log['number']}/comments?since={since}&limit=50") or [])
+                 if b.get("beat") == "astra" and b.get("router") == "live" and b.get("order")]
+        return [int(n) for n in beats[-1]["order"]] if beats else []
+    except Exception as e:  # noqa: BLE001  the ranking is a preference; oldest first still works
+        print(f"rank order unavailable: {str(e)[:120]}")
+        return []
+
+
 def promote_next() -> int | None:
-    """Label the oldest `queued` card whose dependencies are closed `ready`; at most TIME_PARALLEL cards in flight."""
+    """Label the best `queued` card whose dependencies are closed `ready`: the router's live ranking first (Astra's
+    latest), then the oldest; at most TIME_PARALLEL cards in flight."""
     ready = issues("state=open&labels=ready")
     if len(ready) >= int(os.environ.get("TIME_PARALLEL") or 1):
         return None
     queued = issues("state=open&labels=queued")
-    for card in sorted(queued, key=lambda i: i["number"]):
+    pos = {n: k for k, n in enumerate(rank_order())}
+    for card in sorted(queued, key=lambda i: (pos.get(i["number"], len(pos)), i["number"])):
         deps = [int(d) for d in re.findall(r"Depends on #(\d+)", card.get("body") or "")]
         if all(gh("GET", f"/repos/{REPO}/issues/{d}")["state"] == "closed" for d in deps):
             try: