Router: record each pick's outcome, so a learning policy has data (#203)

Co-authored-by: Time (CEO lane) <time@users.noreply.teahub.xyz>

.agents/skills/housekeeping/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: housekeeping
-description: The daily sweep of this repo: prunes merged branches, closes spent signal issues and empty milestones, and lists what is stale. Use when asked what housekeeping did, why a branch or issue was closed, how to run the sweep by hand, or what to do with the stale list.
+description: The daily sweep of this repo: prunes merged branches, closes spent signal issues and empty milestones, records what the router's picks turned into, and lists what is stale. Use when asked what housekeeping did, why a branch or issue was closed, where the router's outcomes come from, how to run the sweep by hand, or what to do with the stale list.
 ---
 
 # Housekeeping
@@ -24,11 +24,17 @@ Deterministic: `.github/time/housekeeping.py` decides, never a model. Every step
    whose venture has no signups: no 👍 from a person and no early-access issue for its slug. One closing line names the count.
    Issues with signups are left to the venture loop.
 3. **Milestones.** Closes milestones with no open issues and at least one closed one. An empty, never-used milestone stays.
-4. **Stale list.** No action, one comment on the Astra log (#57): `needs-human` untouched for over 3 days, open PRs idle
+4. **Router outcomes.** For each card the router picked in live mode (Astra's `picked` beats), one line in
+   `time/ledger/router-outcomes.jsonl` once the card closes: merged or not, its PR, days from pick to close, the spend
+   booked while its worker ran, and, for a venture pick, the venture's signups at day 7. A pick whose card is still open
+   waits for a later sweep, and one already in the file is never written twice. The file's fields are documented in
+   `time/ledger/README.md`; Astra's log summarises it per kind over the ranked table. The data is for a learning policy
+   (a restless bandit) to read later; nothing uses it to pick today.
+5. **Stale list.** No action, one comment on the Astra log (#57): `needs-human` untouched for over 3 days, open PRs idle
    7 days or more, cards parked after their automatic retries, `error` cards older than 2 days. An empty list posts no comment.
 
-Each run ends with one line on the Astra log: branches deleted, issues closed, milestones closed, stale count. A step that
-fails is logged and the others still run; the job exits 1.
+Each run ends with one line on the Astra log: branches deleted, issues closed, milestones closed, picks settled, stale
+count. A step that fails is logged and the others still run; the job exits 1.
 
 ## What the steward does with the stale list
 

.github/time/astra.py

@@ -18,7 +18,8 @@ boost for kinds not picked lately, and the card's age. The log shows the ranked
 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
-card), and the heartbeat promotes queued cards in the ranked order.
+card), and the heartbeat promotes queued cards in the ranked order. Over the table the log shows what each kind's past
+picks returned (picks, merged share, median days, median $), from the outcomes housekeeping records (#172).
 `off`, or no router.json, is the planner as before.
 """
 from __future__ import annotations
@@ -153,6 +154,18 @@ def router_config() -> dict:
         return router.config(None)
 
 
+def outcomes() -> list[dict]:
+    """The settled picks housekeeping has recorded (#172), for the summary per kind over the table. Context, not a
+    dependency: unreadable, the table goes up without it."""
+    try:
+        f = tm.gh("GET", f"/repos/{tm.REPO}/contents/{router.OUTCOMES}")
+        return router.read_outcomes(base64.b64decode(f.get("content") or "").decode(errors="replace"))
+    except Exception as e:  # noqa: BLE001
+        if "-> 404" not in str(e):
+            print(f"router outcomes unreadable: {str(e)[:120]}")
+        return []
+
+
 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)."""
@@ -266,7 +279,9 @@ def router_tick(cfg: dict, seat: str, log: int, context: str, thread: list[dict]
             f"queue against ({source}). Weights: "
             + ", ".join(f"{k} {w[k]:g}" for k in router.KINDS) + f"; restless {cfg['restless']:g}/day."
             + (f" Cloud lane: {in_lane} in it, up to {cfg['cloud']['per_day']}." if cfg["cloud"]["per_day"] else ""))
-    lines = [head, "", str(out.get("note") or "(no note)"), "", router.table(ranked, picks, cfg["mode"], cloud=to_cloud), ""]
+    settled = router.outcome_lines(outcomes())   # what each kind has returned, over the table (#172)
+    lines = [head, "", str(out.get("note") or "(no note)"), ""] + (settled + [""] if settled else []) \
+        + [router.table(ranked, picks, cfg["mode"], cloud=to_cloud), ""]
     if unjudged:
         lines.append(f"_{unjudged} candidate(s) the seat didn't judge are scored at value 3, size M._")
     if live:

.github/time/housekeeping.py

@@ -2,21 +2,26 @@
 
     python3 .github/time/housekeeping.py      # daily (housekeeping.yml), or by hand
 
-Four steps, each idempotent (a second run finds nothing to do) and each logged:
+Five steps, each idempotent (a second run finds nothing to do) and each logged:
   1. branches   delete every branch whose PR merged into the default branch. Never the default branch, `gh-pages`, a
                 protected branch, a branch with an open PR, or a branch that has commits newer than its merge.
   2. signals    close `signal` issues whose venture's measuring window (TIME_VENTURE_MEASURE_DAYS, default 7) has passed
                 with no signups (no 👍 on the issue, no early-access issue for the venture), with one closing line.
   3. milestones close milestones with no open issues and at least one closed one.
-  4. stale      list, never act: `needs-human` untouched for 3 days, open PRs idle 7 days, cards parked after their
+  4. outcomes   append one line to `time/ledger/router-outcomes.jsonl` per router pick that has settled (#172): merged
+                or not, its PR, days from pick to close, the spend booked while its worker ran, and a venture's signups
+                at day 7. A pick whose card is still open is in flight and waits for a later sweep.
+  5. stale      list, never act: `needs-human` untouched for 3 days, open PRs idle 7 days, cards parked after their
                 automatic retries, cards whose last beat is `error` for 2 days. One comment on the Astra log; none if empty.
 Then one summary line on the Astra log. Only TIME_MODE=live changes anything; anything else (the default) prints what it
 would do and writes nothing, comments included. A step that fails is logged and the others still run (exit code 1).
 """
 from __future__ import annotations
 
+import base64
 import datetime as dt
 import importlib.util
+import json
 import os
 import re
 import sys
@@ -27,11 +32,16 @@ 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)
 
 NEVER = {"gh-pages"}          # besides the default branch and protected ones
 LOG_TITLE = "Astra log"
 LOG_FALLBACK = 57             # the Astra log's number here, if it cannot be found by title
 NEEDS_HUMAN_DAYS, PR_IDLE_DAYS, ERROR_DAYS = 3, 7, 2
+VENTURES = "ventures/ventures.json"   # the venture registry (venture.py writes it), on the Pages branch
+PAGES_BRANCH = "gh-pages"
 
 
 def log(msg: str) -> None:
@@ -68,6 +78,29 @@ def age_days(ts: str | None, now: dt.datetime) -> float:
     return (now - d).total_seconds() / 86400 if d else 0.0
 
 
+def stamp(ts: str | dt.datetime | None) -> str:
+    """A timestamp written the way the ledger writes them, so one string comparison orders beats, ledger lines and
+    windows alike whatever offset the API sent."""
+    d = ts if isinstance(ts, dt.datetime) else parse(ts)
+    return d.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") if d else ""
+
+
+def measure_days() -> int:
+    """The venture measuring window: how long a signal issue is given, and the day a pick's signups are counted on."""
+    return int(float(os.environ.get("TIME_VENTURE_MEASURE_DAYS") or 7))
+
+
+def repo_file(path: str, ref: str = "") -> str | None:
+    """A file in the repo, read through the API (the workflow checks out only `.github/time`). None if there is none."""
+    try:
+        f = tm.gh("GET", f"{repo()}/contents/{urllib.parse.quote(path)}" + (f"?ref={ref}" if ref else ""))
+    except RuntimeError as e:
+        if "-> 404" in str(e):
+            return None
+        raise
+    return base64.b64decode(f.get("content") or "").decode(errors="replace")
+
+
 def labelled(name: str, query: str = "state=open") -> list[dict]:
     """Issues (and PRs) that carry `name`. Each row is checked too: Gitea ignores a filter on a label the repo does not
     have and returns everything, which on the first live run listed every open issue as a parked card."""
@@ -124,19 +157,19 @@ def person(user: dict | None) -> str | None:
     return login
 
 
-def signup_people() -> dict[str, set[str]]:
-    """venture slug -> distinct people who opened an early-access issue for it."""
+def signup_people(until: str = "9999") -> dict[str, set[str]]:
+    """venture slug -> distinct people who opened an early-access issue for it before `until` (all of them by default)."""
     out: dict[str, set[str]] = {}
     for i in labelled("signup", "state=all&type=issues"):
         m = re.search(r"^###\s*Venture\s*$\n+(.*?)(?=^###\s|\Z)", i.get("body") or "", re.M | re.S)
         who = person(i.get("user"))
-        if m and who and m.group(1).strip():
+        if m and who and m.group(1).strip() and stamp(i.get("created_at")) < until:
             out.setdefault(m.group(1).strip().lower(), set()).add(who)
     return out
 
 
 def step_signals(live: bool, now: dt.datetime) -> int:
-    days = int(float(os.environ.get("TIME_VENTURE_MEASURE_DAYS") or 7))
+    days = measure_days()
     due = [i for i in labelled("signal", "state=open&type=issues")
            if not i.get("pull_request") and age_days(i.get("created_at"), now) >= days]
     if not due:
@@ -181,7 +214,114 @@ def step_milestones(live: bool) -> int:
     return closed
 
 
-# ---------- 4. the stale list ----------
+# ---------- 4. the router's outcomes ----------
+# What a pick turned into, so a learning policy (a restless bandit) has rewards to read. The ledger has no card field,
+# so a card's spend is the dollars booked while its worker ran: each `dispatched` beat until the next beat on the card
+# (the chosen proxy of the two the card offered; review and planning calls on the card are not counted).
+
+def run_windows(beats: list[dict], settled: str) -> list[tuple[str, str]]:
+    """The card's worker runs as [start, end) in UTC: a `dispatched` beat until the next beat on the card, the last one
+    until the card settled."""
+    at = [stamp(b.get("at")) for b in beats]
+    out = [(at[k], next((s for s in at[k + 1:] if s), settled)) for k, b in enumerate(beats)
+           if b.get("beat") == "dispatched" and at[k]]
+    return [(a, z) for a, z in out if a < z]
+
+
+def months(since: str, until: str) -> list[str]:
+    """The ledger months a span touches, `2026-10` first."""
+    out, m = [], since[:7]
+    while m <= until[:7] and len(out) < 24:
+        out.append(m)
+        m = f"{int(m[:4]) + (m[5:] == '12')}-{int(m[5:]) % 12 + 1:02d}"
+    return out
+
+
+def window_usd(windows: list[tuple[str, str]]) -> float:
+    """Dollars the ledger booked inside any of these windows."""
+    if not windows:
+        return 0.0
+    rows = [r for m in months(min(a for a, _ in windows), max(z for _, z in windows)) for r in tm.ledger_lines(m)]
+    return sum(r.get("usd") or 0 for r in rows if any(a <= str(r.get("at") or "") < z for a, z in windows))
+
+
+def ventures() -> list[dict]:
+    """The venture registry on the Pages branch. Read-only here; unreadable is an empty studio."""
+    try:
+        reg = json.loads(repo_file(VENTURES, PAGES_BRANCH) or "{}")
+    except (ValueError, RuntimeError) as e:
+        log(f"outcomes: venture registry unreadable: {str(e)[:120]}")
+        return []
+    return [v for v in (reg.get("ventures") if isinstance(reg, dict) else None) or [] if isinstance(v, dict)]
+
+
+def signups_by(v: dict, until: str) -> int:
+    """Distinct people (not bots, not the owner) who asked for early access for this venture or 👍'd its signal issue
+    before `until`. The slug match is the signal step's, so a signup that named the venture some other way is missed."""
+    people = set(signup_people(until).get(str(v.get("slug") or "").lower(), set()))
+    if v.get("signal"):
+        people |= {w for r in paged(f"{repo()}/issues/{v['signal']}/reactions") if r.get("content") in ("+1", "👍")
+                   and stamp(r.get("created_at")) < until and (w := person(r.get("user")))}
+    return len(people)
+
+
+def day_after(day: str, days: int) -> str:
+    """Midnight `days` days after a date, as a ledger timestamp; "" when the date is not one."""
+    try:
+        return (dt.date.fromisoformat(day[:10]) + dt.timedelta(days=days)).strftime("%Y-%m-%dT00:00:00Z")
+    except ValueError:
+        return ""
+
+
+def step_outcomes(live: bool, now: dt.datetime) -> int:
+    """One line per settled pick in `router.OUTCOMES`, appended in one commit. A pick already in the file is never
+    written twice; a pick whose card is still open, or whose venture has not reached day 7, waits for a later sweep."""
+    picks = router.picks(tm.beats_of(paged(f"{repo()}/issues/{log_issue()}/comments")))
+    if not picks:
+        return 0
+    have = {r["n"] for r in router.read_outcomes(repo_file(router.OUTCOMES))}
+    rows, registry = [], None
+    for n, p in sorted(picks.items()):
+        if n in have:
+            continue
+        try:
+            card = tm.gh("GET", f"{repo()}/issues/{n}")
+        except RuntimeError as e:
+            if "-> 404" not in str(e):
+                raise
+            log(f"outcomes: skip #{n} (no such card)")
+            continue
+        if (card.get("state") or "open") != "closed":
+            log(f"outcomes: keep #{n} (still in flight)")
+            continue
+        settled = stamp(card.get("closed_at")) or stamp(card.get("updated_at")) or stamp(now)
+        beats = tm.beats_of(paged(f"{repo()}/issues/{n}/comments"))
+        signups = None
+        if p["kind"] == "venture":
+            registry = ventures() if registry is None else registry
+            v = next((x for x in registry if x.get("card") == n), None)
+            day7 = day_after(str(v.get("launched") or v.get("planned") or p["picked"]), measure_days()) if v else ""
+            if day7 and day7 > stamp(now):
+                log(f"outcomes: keep #{n} (its venture's day {measure_days()} is {day7[:10]})")
+                continue
+            signups = signups_by(v, day7 or "9999") if v else None
+        row = router.outcome(p, card, beats, window_usd(run_windows(beats, settled)), signups)
+        if row:
+            rows.append(row)
+            log(f"outcomes: #{n} settled ({'merged' if row['merged'] else 'not merged'}, {row['days']} day(s), "
+                f"${row['usd'] or 0:.2f}" + (f", {row['signups']} signup(s)" if row["signups"] is not None else "") + ")")
+    if not rows:
+        return 0
+    if not live:
+        log(f"outcomes: would record {len(rows)} settled pick(s)")
+        return 0
+    tm.git_commit(tm.default_branch(), {}, f"router: {len(rows)} settled pick(s)",
+                  append={router.OUTCOMES: "".join(json.dumps(r, separators=(",", ":")) + "\n" for r in rows).encode()})
+    log(f"outcomes: recorded {len(rows)} settled pick(s) in {router.OUTCOMES}")
+    return len(rows)
+
+
+# ---------- 5. the stale list ----------
 
 def stale(now: dt.datetime) -> dict[str, list[tuple[int, float, str]]]:
     """category -> [(number, age in days, title)]. Read-only."""
@@ -235,10 +375,10 @@ def main() -> int:
     now = dt.datetime.now(dt.timezone.utc)
     log(f"housekeeping {now:%Y-%m-%dT%H:%MZ} mode={'live' if live else 'dry-run'}")
     failed = 0
-    done = {"branches": 0, "signals": 0, "milestones": 0}
+    done = {"branches": 0, "signals": 0, "milestones": 0, "outcomes": 0}
     found: dict = {}
     for name, step in (("branches", lambda: step_branches(live)), ("signals", lambda: step_signals(live, now)),
-                       ("milestones", lambda: step_milestones(live))):
+                       ("milestones", lambda: step_milestones(live)), ("outcomes", lambda: step_outcomes(live, now))):
         try:
             done[name] = step()
         except Exception as e:  # noqa: BLE001  one step must not stop the others
@@ -253,7 +393,8 @@ def main() -> int:
     if found:
         log(stale_text(found))
     line = (f"🧹 Housekeeping: {done['branches']} branch(es) deleted, {done['signals']} signal issue(s) closed, "
-            f"{done['milestones']} milestone(s) closed, {count} stale." + (f" {failed} step(s) failed; see the run log." if failed else ""))
+            f"{done['milestones']} milestone(s) closed, {done['outcomes']} router pick(s) settled, {count} stale."
+            + (f" {failed} step(s) failed; see the run log." if failed else ""))
     log(line)
     if live:
         try:

.github/time/router.py

@@ -26,6 +26,10 @@ waits: on a card still open (its `Depends on #N`), or on whatever the seat says
 human). Cards that can move come first in the promotion order, the ones that wait after them. 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.
+
+Outcomes (#172): a bandit learns from rewards, so every live pick's outcome is recorded once the card settles — one
+line in `time/ledger/router-outcomes.jsonl`, written by the daily housekeeping sweep, and one summary line per kind
+over Astra's table. Nothing here reads them back into a pick yet; `score` is still the whole policy.
 """
 from __future__ import annotations
 
@@ -202,6 +206,99 @@ def order(ranked: list[dict]) -> list[int]:
     return [r["n"] for r in cards if r["actionable"]] + [r["n"] for r in cards if not r["actionable"]]
 
 
+# ---------- outcomes: what a pick turned into (#172) ----------
+# A learning policy needs rewards. Housekeeping settles each live pick into one line here; these functions are the
+# shape of a line, the arithmetic over the lines, and nothing else. No pick reads them yet.
+
+OUTCOMES = "time/ledger/router-outcomes.jsonl"   # under time/ledger/, the one path CI ignores: the daily append is not code
+FIELDS = ("n", "kind", "picked", "settled", "merged", "pr", "days", "usd", "signups")
+
+
+def picks(beats: list[dict]) -> dict[int, dict]:
+    """Every card the router picked in live mode, from Astra's beats: card number -> {n, kind, picked (the day)}. A
+    card picked twice (it came back after a park) keeps its first pick; that is when the factory spent the choice."""
+    out: dict[int, dict] = {}
+    for b in beats:
+        if b.get("router") != "live":
+            continue
+        day = str(b.get("at") or "")[:10]
+        for p in b.get("picked") or []:
+            n = p.get("n") if isinstance(p, dict) else None
+            if isinstance(n, int) and n not in out:
+                out[n] = {"n": n, "kind": p.get("kind") if p.get("kind") in KINDS else "feature", "picked": day}
+    return out
+
+
+def span(since: str, until: str) -> int | None:
+    """Whole days from one date to another, None when either is not a date."""
+    try:
+        return (dt.date.fromisoformat(until[:10]) - dt.date.fromisoformat(since[:10])).days
+    except ValueError:
+        return None
+
+
+def outcome(p: dict, card: dict, beats: list[dict], usd: float | None, signups: int | None) -> dict | None:
+    """One line for a pick, or None while its card is still open: a pick in flight has no outcome yet. Merged or not
+    and the PR number come from the card's own beats (a `merged` beat is a landing; any beat naming a `pr` is the PR
+    the card had), `days` counts the pick's day to the card's close, `usd` and `signups` are the caller's."""
+    if (card.get("state") or "open") != "closed":
+        return None
+    settled = str(card.get("closed_at") or card.get("updated_at") or "")[:10]
+    merged = next((b for b in beats if b.get("beat") == "merged"), None)
+    pr = (merged or {}).get("pr") or next((b.get("pr") for b in reversed(beats) if b.get("pr")), None)
+    return {"n": p["n"], "kind": p["kind"], "picked": p["picked"], "settled": settled, "merged": bool(merged),
+            "pr": int(pr) if pr else None, "days": span(p["picked"], settled),
+            "usd": round(usd, 4) if usd is not None else None, "signups": signups}
+
+
+def read_outcomes(raw: str | None) -> list[dict]:
+    """The outcomes log as rows, oldest first. A line that is not an outcome is skipped: the file is append-only and
+    one bad line must not hide the rest."""
+    out = []
+    for line in (raw or "").splitlines():
+        try:
+            row = json.loads(line)
+        except ValueError:
+            continue
+        if isinstance(row, dict) and isinstance(row.get("n"), int):
+            out.append(row)
+    return out
+
+
+def median(xs: list[float]) -> float | None:
+    s = sorted(xs)
+    if not s:
+        return None
+    return float(s[len(s) // 2]) if len(s) % 2 else (s[len(s) // 2 - 1] + s[len(s) // 2]) / 2
+
+
+def summary(rows: list[dict]) -> dict[str, dict]:
+    """Per kind, what its settled picks returned: how many, the merged share, the median days from pick to close and
+    the median dollars. A kind with no settled pick is left out."""
+    out = {}
+    for k in KINDS:
+        mine = [r for r in rows if r.get("kind") == k]
+        if mine:
+            out[k] = {"picks": len(mine), "merged": sum(1 for r in mine if r.get("merged")) / len(mine),
+                      "days": median([r["days"] for r in mine if isinstance(r.get("days"), (int, float))]),
+                      "usd": median([r["usd"] for r in mine if isinstance(r.get("usd"), (int, float))])}
+    return out
+
+
+def outcome_lines(rows: list[dict]) -> list[str]:
+    """The header over Astra's table: one line per kind, so the owner sees what a kind has returned before changing
+    its weight. No settled pick yet, no header."""
+    got = summary(rows)
+    if not got:
+        return []
+    out = [f"Settled picks so far ({len(rows)} in `{OUTCOMES}`):"]
+    for k, s in got.items():
+        days = f"{s['days']:g} day(s) median" if s["days"] is not None else "days unknown"
+        usd = f"${s['usd']:.2f} median" if s["usd"] is not None else "$ unknown"
+        out.append(f"- {k}: {s['picks']} pick(s), {s['merged'] * 100:.0f}% merged, {days}, {usd}")
+    return out
+
+
 def table(ranked: list[dict], picked: list[dict], mode: str, limit: int = 15, cloud: list[dict] = ()) -> str:
     chosen, sent = {r["key"] for r in picked}, {r["key"] for r in cloud}
     mark = "queued today" if mode == "live" else "would queue"

.github/time/test_housekeeping.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 """Offline tests for housekeeping.py: a fake forge (like test_time.py's World) that records every write."""
+import base64
 import datetime as dt
 import importlib.util
 import json
@@ -23,13 +24,15 @@ LOG = 57
 
 
 class Forge:
-    def __init__(self, branches=(), prs=(), issues=(), milestones=(), reactions=None, comments=None, repo_labels=None):
+    def __init__(self, branches=(), prs=(), issues=(), milestones=(), reactions=None, comments=None, repo_labels=None, files=None):
         self.branches = [{"name": b} if isinstance(b, str) else b for b in branches]
         self.prs = list(prs)
         self.issues = {i["number"]: {"state": "open", "labels": [], "created_at": iso(0), "updated_at": iso(0), "body": "", **i} for i in issues}
         self.milestones = [dict(m) for m in milestones]
         self.reactions = reactions or {}
         self.cmts = comments or {}
+        self.files = dict(files or {})   # "<path>" (or "<path>@<ref>") -> text, as the contents API serves it
+        self.commits = []                # (branch, message, {path: what was appended}): Time's writes are pushes
         # the repo's labels; by default the ones its issues carry. Gitea ignores a filter on any other label.
         self.repo_labels = set(repo_labels) if repo_labels is not None else {l for i in self.issues.values() for l in i["labels"]}
         self.writes = []   # every non-GET call
@@ -72,6 +75,17 @@ class Forge:
         if route.endswith("/comments") and method == "POST":
             self.posted.append((int(route.split("/")[-2]), body["body"]))
             return {}
+        if route.startswith("/repos/o/r/contents/") and method == "GET":
+            path, ref = urllib.parse.unquote(route.split("/contents/", 1)[1]), q.get("ref", [""])[0]
+            text = self.files.get(f"{path}@{ref}" if ref else path)
+            if text is None:
+                raise RuntimeError(f"GET {route} -> 404: not found")
+            return {"content": base64.b64encode(text.encode()).decode(), "sha": "sha"}
+        if route.startswith("/repos/o/r/issues/") and route.rsplit("/", 1)[1].isdigit() and method == "GET":
+            i = self.issues.get(int(route.rsplit("/", 1)[1]))
+            if i is None:
+                raise RuntimeError(f"GET {route} -> 404: not found")
+            return {**i, "labels": [{"name": l} for l in i["labels"]]}
         if route == "/repos/o/r/issues" and method == "GET":
             rows = [i for i in self.issues.values()
                     if (q.get("state", ["open"])[0] in ("all", i["state"]))
@@ -82,6 +96,10 @@ class Forge:
             return {}
         raise AssertionError(f"unexpected {method} {path}")
 
+    def push(self, branch, files, message, *, base=None, append=None):
+        self.commits.append((branch, message, {p: d.decode() for p, d in (append or {}).items()}))
+        return "sha"
+
     def run(self, mode="live", env=None):
         os.environ.pop("TIME_MODE", None)
         os.environ.pop("TIME_VENTURE_MEASURE_DAYS", None)
@@ -89,6 +107,8 @@ class Forge:
             os.environ["TIME_MODE"] = mode
         os.environ.update(env or {})
         tm.http = self.http
+        tm.git_commit = self.push
+        tm._ledger_files.clear()   # the ledger is read through the API and cached per process, not per world
         return hk.main()
 
 
@@ -184,18 +204,95 @@ def test_unknown_label():
     assert [n for n, _ in f.posted] == [LOG], "the summary goes to the Astra log, found by its own label"
 
 
+def beat(meta, days=0):
+    return {"created_at": iso(days), "user": {"login": "gitea-actions"}, "body": f"x\n\n{tm.MARK}{json.dumps(meta)} -->"}
+
+
+def ledger(*rows):
+    """The month files the ledger lines below fall in (a window can straddle a month end)."""
+    out = {}
+    for at, usd in rows:
+        out.setdefault(f"time/ledger/{at[:7]}.jsonl", "")
+        out[f"time/ledger/{at[:7]}.jsonl"] += json.dumps({"at": at, "model": "m", "usd": usd}) + "\n"
+    return out
+
+
+ASTRA_LOG = {"number": LOG, "title": "Astra log", "labels": ["astra"]}
+
+
+def test_outcomes():
+    """A merged pick and an unmerged one each settle into one line; a pick still open produces none."""
+    picked = beat({"beat": "astra", "router": "live", "picked": [{"n": 1, "kind": "feature"}, {"n": 2, "kind": "refinement"},
+                                                                 {"n": 3, "kind": "feature"}]}, 4)
+    f = Forge(issues=[ASTRA_LOG,
+                      {"number": 1, "title": "merged card", "state": "closed", "closed_at": iso(1)},
+                      {"number": 2, "title": "dropped card", "state": "closed", "closed_at": iso(2)},
+                      {"number": 3, "title": "still going"}],
+              comments={LOG: [picked],
+                        1: [beat({"beat": "dispatched"}, 4), beat({"beat": "pr", "pr": 90}, 3), beat({"beat": "merged", "pr": 90}, 1)],
+                        2: [beat({"beat": "dispatched"}, 2.5), beat({"beat": "stuck", "pr": 91}, 2.2)]},
+              files=ledger((iso(3.5), 0.2), (iso(3.2), 0.05), (iso(2.4), 0.4), (iso(0.5), 9.0)))   # the last: no run window holds it
+    assert f.run() == 0
+    assert len(f.commits) == 1 and f.commits[0][0] == "main" and "2 settled pick(s)" in f.commits[0][1], f.commits
+    rows = [json.loads(l) for l in f.commits[0][2][hk.router.OUTCOMES].splitlines()]
+    assert [r["n"] for r in rows] == [1, 2], rows
+    assert rows[0] == {"n": 1, "kind": "feature", "picked": iso(4)[:10], "settled": iso(1)[:10], "merged": True,
+                       "pr": 90, "days": 3, "usd": 0.25, "signups": None}, rows[0]
+    assert rows[1]["merged"] is False and rows[1]["pr"] == 91 and rows[1]["days"] == 2, rows[1]
+    assert rows[1]["usd"] == 0.4, "only what was booked while its own worker ran"
+    assert any("2 router pick(s) settled" in b for _, b in f.posted), f.posted
+    # idempotent: a pick already in the file is never written twice, and #3 settles once it closes
+    f.files[hk.router.OUTCOMES] = f.commits[0][2][hk.router.OUTCOMES]
+    f.commits.clear()
+    assert f.run() == 0 and f.commits == [], "nothing new to settle"
+    f.issues[3].update(state="closed", closed_at=iso(0))
+    assert f.run() == 0 and [json.loads(l)["n"] for l in f.commits[0][2][hk.router.OUTCOMES].splitlines()] == [3]
+
+
+def test_outcomes_venture():
+    """A venture pick carries the venture's signups at day 7; before day 7 the pick keeps waiting."""
+    picked = beat({"beat": "astra", "router": "live", "picked": [{"n": 5, "kind": "venture"}, {"n": 6, "kind": "venture"}]}, 10)
+    reg = {"ventures": [{"slug": "huebrain", "name": "Huebrain", "card": 5, "signal": 8, "launched": iso(9)[:10]},
+                        {"slug": "later", "name": "Later", "card": 6, "signal": 9, "launched": iso(2)[:10]}]}
+    f = Forge(issues=[ASTRA_LOG,
+                      {"number": 5, "title": "launch huebrain", "state": "closed", "closed_at": iso(8)},
+                      {"number": 6, "title": "launch later", "state": "closed", "closed_at": iso(1)},
+                      {"number": 20, "labels": ["signup"], "user": {"login": "ann"}, "created_at": iso(8),
+                       "body": "### Venture\n\nhuebrain\n\n### Plan\n\nFree"},
+                      {"number": 21, "labels": ["signup"], "user": {"login": "bob"}, "created_at": iso(1),
+                       "body": "### Venture\n\nhuebrain\n\n### Plan\n\nFree"}],
+              comments={LOG: [picked], 5: [beat({"beat": "pr", "pr": 70}, 9), beat({"beat": "merged", "pr": 70}, 8)]},
+              reactions={8: [{"content": "+1", "user": {"login": "cid"}, "created_at": iso(8)},
+                             {"content": "+1", "user": {"login": "dee"}, "created_at": iso(0)}]},
+              files={f"ventures/ventures.json@{hk.PAGES_BRANCH}": json.dumps(reg)})
+    assert f.run() == 0
+    rows = [json.loads(l) for l in f.commits[0][2][hk.router.OUTCOMES].splitlines()]
+    assert [r["n"] for r in rows] == [5], "#6's venture is 2 days old: its day 7 is still ahead"
+    assert rows[0]["signups"] == 2, "ann and cid came in by day 7; bob and dee came later"
+    assert rows[0]["merged"] is True and rows[0]["usd"] == 0.0, "no worker run on the card: no spend of its own"
+    # no registry entry for the card: the line goes in with no count rather than waiting forever
+    f = Forge(issues=[ASTRA_LOG, {"number": 5, "title": "launch huebrain", "state": "closed", "closed_at": iso(8)}],
+              comments={LOG: [beat({"beat": "astra", "router": "live", "picked": [{"n": 5, "kind": "venture"}]}, 10)]})
+    assert f.run() == 0
+    assert json.loads(f.commits[0][2][hk.router.OUTCOMES])["signups"] is None
+
+
 def test_dry_run():
     f = Forge(branches=["main", "time/card-53"], prs=[pr(1, "time/card-53", "closed", True)],
-              issues=[sig(10, "spent", 8), {"number": 30, "labels": ["needs-human"], "updated_at": iso(9)}],
+              issues=[ASTRA_LOG, sig(10, "spent", 8), {"number": 30, "labels": ["needs-human"], "updated_at": iso(9)},
+                      {"number": 1, "title": "merged card", "state": "closed", "closed_at": iso(1)}],
+              comments={LOG: [beat({"beat": "astra", "router": "live", "picked": [{"n": 1, "kind": "feature"}]}, 4)],
+                        1: [beat({"beat": "merged", "pr": 90}, 1)]},
               milestones=[{"id": 1, "title": "done batch", "open_issues": 0, "closed_issues": 3}])
     for mode in ("dry-run", None):   # explicit and unset
         assert f.run(mode) == 0
-        assert f.writes == [] and f.posted == [], (f.writes, f.posted)
+        assert f.writes == [] and f.posted == [] and f.commits == [], (f.writes, f.posted, f.commits)
     assert {b["name"] for b in f.branches} == {"main", "time/card-53"} and f.issues[10]["state"] == "open"
 
 
 if __name__ == "__main__":
-    tests = [test_branches, test_signals, test_milestones, test_stale, test_unknown_label, test_dry_run]
+    tests = [test_branches, test_signals, test_milestones, test_outcomes, test_outcomes_venture, test_stale,
+             test_unknown_label, test_dry_run]
     for t in tests:
         t()
         print("ok", t.__name__)

.github/time/test_time.py

@@ -1226,6 +1226,14 @@ 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)}
+import base64 as _b8172  # noqa: E402
+settled80 = []   # the outcomes log (#172); empty, so nothing is summarised over the table
+def gh80(method, path, body=None, **k):
+    """The outcomes log is the only file this world serves; anything else is a fresh repo's 404."""
+    if rt.OUTCOMES not in path:
+        raise RuntimeError(f"{method} {path} -> 404: not found")
+    return {"content": _b8172.b64encode("".join(json.dumps(r) + "\n" for r in settled80).encode()).decode()}
+astra.tm.gh = gh80
 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 +1339,56 @@ else: os.environ["TIME_PARALLEL"] = par135
 tm.issues, tm.gh, tm.label, tm.unlabel = real80
 print("ALL OK 80")
 
+# 172 the router's outcomes: a live pick settles into one line (housekeeping writes it; test_housekeeping.py checks the
+# sweep), and Astra's log summarises what each kind has returned over the table
+beats172 = [{"beat": "astra", "router": "live", "at": "2026-10-17T13:13:00Z", "picked": [{"n": 1, "kind": "feature"}, {"n": 2, "kind": "venture"}]},
+            {"beat": "astra", "router": "dry", "at": "2026-10-18T13:13:00Z", "picked": [{"n": 3, "kind": "feature"}]},
+            {"beat": "astra", "router": "live", "at": "2026-10-18T13:13:00Z",
+             "picked": [{"n": 1, "kind": "refinement"}, {"n": 4, "kind": "nonsense"}, "junk"]}]
+assert rt.picks(beats172) == {1: {"n": 1, "kind": "feature", "picked": "2026-10-17"},
+                              2: {"n": 2, "kind": "venture", "picked": "2026-10-17"},
+                              4: {"n": 4, "kind": "feature", "picked": "2026-10-18"}}, rt.picks(beats172)
+assert rt.picks([{"beat": "astra", "router": "dry", "picked": [{"n": 9, "kind": "feature"}]}]) == {}, "a dry run picked nothing"
+p172 = rt.picks(beats172)[1]
+beats1 = [{"beat": "dispatched", "at": "2026-10-17T14:00:00Z"}, {"beat": "pr", "pr": 90, "at": "2026-10-17T15:00:00Z"},
+          {"beat": "merged", "pr": 90, "at": "2026-10-19T09:00:00Z"}]
+row172 = rt.outcome(p172, {"state": "closed", "closed_at": "2026-10-19T09:00:05Z"}, beats1, 1.25, None)
+assert row172 == {"n": 1, "kind": "feature", "picked": "2026-10-17", "settled": "2026-10-19", "merged": True,
+                  "pr": 90, "days": 2, "usd": 1.25, "signups": None}, row172
+assert tuple(row172) == rt.FIELDS, "the line's shape"
+assert rt.outcome(p172, {"state": "open"}, beats1, 1.0, None) is None, "a pick in flight has no outcome yet"
+drop172 = rt.outcome(p172, {"state": "closed", "closed_at": "2026-10-18T09:00:00Z"}, [{"beat": "stuck", "pr": 91, "at": "x"}], None, None)
+assert drop172["merged"] is False and drop172["pr"] == 91 and drop172["days"] == 1 and drop172["usd"] is None
+assert rt.outcome(p172, {"state": "closed", "closed_at": "x"}, [], 0.0, None)["days"] is None, "no close date, no span"
+# the summary per kind: picks, merged share, median days, median $
+rows172 = [row172, dict(drop172, n=5), {"n": 6, "kind": "feature", "merged": True, "days": 4, "usd": 3.0},
+           {"n": 7, "kind": "venture", "merged": False, "days": 9, "usd": 0.5, "signups": 2}]
+got172 = rt.summary(rows172)
+assert got172["feature"] == {"picks": 3, "merged": 2 / 3, "days": 2.0, "usd": 2.125}, got172["feature"]
+assert got172["venture"] == {"picks": 1, "merged": 0.0, "days": 9.0, "usd": 0.5} and "refinement" not in got172, got172
+assert rt.median([]) is None and rt.median([3, 1]) == 2 and rt.median([5, 1, 3]) == 3
+assert rt.summary([{"n": 8, "kind": "feature"}])["feature"] == {"picks": 1, "merged": 0.0, "days": None, "usd": None}
+lines172 = rt.outcome_lines(rows172)
+assert lines172[0] == "Settled picks so far (4 in `time/ledger/router-outcomes.jsonl`):", lines172
+assert lines172[1] == "- venture: 1 pick(s), 0% merged, 9 day(s) median, $0.50 median", lines172
+assert lines172[2] == "- feature: 3 pick(s), 67% merged, 2 day(s) median, $2.12 median", lines172   # KINDS order, as the weights
+assert rt.outcome_lines([]) == [] and rt.read_outcomes(None) == [] and rt.read_outcomes("") == []
+assert rt.read_outcomes("not json\n" + "".join(json.dumps(r) + "\n" for r in rows172) + '{"no": "number"}\n') == rows172
+# Astra puts the summary over the table, and goes without it when nothing has settled
+astra.tm.issues = lambda q: closed80 if q.startswith("state=closed") else open80 + [iss80(57, "Astra log", labels=["astra"])]
+astra.tm.think = lambda seat, a, b, system, user: "Sure:\n" + json.dumps(reply80)
+settled80[:] = rows172
+assert astra.router_tick(rt.config('{"mode": "dry"}'), "opus", 57, "ctx", [], 2, 2) == 0
+assert "Settled picks so far (4 in" in said80[-1][1] and "- feature: 3 pick(s)" in said80[-1][1], said80[-1][1]
+assert "| 1 | #2 Fix the docs |" in said80[-1][1], "the table still follows"
+settled80.clear()
+assert astra.router_tick(rt.config('{"mode": "dry"}'), "opus", 57, "ctx", [], 2, 2) == 0
+assert "Settled picks" not in said80[-1][1], "nothing settled yet: no header"
+astra.tm.gh = lambda *a, **k: (_ for _ in ()).throw(RuntimeError("GET x -> 500"))
+assert astra.outcomes() == [], "the outcomes log is context: unreadable, the table goes up without it"
+astra.tm.gh = gh80
+print("ALL OK 172")
+
 # 82 adaptive budget: each route is paced to its capacity pool; no pool, no change
 import io, contextlib
 real_pools = tm.pools

CHANGELOG.md

@@ -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 records what each pick turned into (#172), so a learning policy (a restless bandit) has rewards to read. Housekeeping's daily sweep settles every card the router picked in live mode into one line of `time/ledger/router-outcomes.jsonl` — `{"n", "kind", "picked", "settled", "merged", "pr", "days", "usd", "signups"}` — once the card closes: merged or not and its PR from the card's own beats, the days from pick to close, the dollars the ledger booked while the card's worker ran (each `dispatched` beat until the next beat on the card: the ledger has no card field, so this is the runs'-window proxy, not the card's exact bill), and, for a venture pick, the venture's signups at day 7 (that pick waits until day 7 has passed). A pick whose card is still open is in flight and produces nothing; one already in the file is never written twice. Over the ranked table Astra's log now shows one line per kind — picks, merged share, median days, median $ — so the owner can see what a kind returns before changing its weight. The file sits under `time/ledger/`, the one path CI ignores, since the sweep appends to it daily. No policy change: `router.score` still makes every pick.
 - A cloud PR's revision comes from the cloud, and Time reviews it again. The branch of a cloud-lane PR is the copy's to write, so `/time retry` (the API worker pushing to it) would be overwritten by the next sync. When the reviewer objects instead, the launcher on hb-5 starts one more cloud session on the same branch with the review in its prompt (once; then a human). When the copy's branch moves under a PR that is open and already reviewed, sync.py posts a `dispatched` beat marked `revised` on each card the PR closes and takes `needs-human` off the PR, so the heartbeat reviews the new head and a verdict from before the push no longer counts. A PR not reviewed yet gets nothing; its first review sees the new head.
 - 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.

time/ledger/README.md

@@ -30,6 +30,29 @@ Time stops calling models. The next time the heartbeat has a card to run, it ope
 
 Humans should never edit these files. The caps rely on them being exact, and Time only ever appends. A hand edit can hide spend or break the next append.
 
+## The router's outcomes (`router-outcomes.jsonl`)
+
+Not spend, but it lives here for the same reason: Time appends to it daily and CI ignores this folder. One JSON line per
+router pick that has settled, written by housekeeping (`.github/time/housekeeping.py`, step 4), so a learning policy has
+rewards to read. Fields:
+
+- `n`, `kind`: the card and the kind it was picked as.
+- `picked`, `settled`: the day Astra picked it, and the day the card closed.
+- `merged`, `pr`: whether its PR landed, and which PR.
+- `days`: whole days from `picked` to `settled`.
+- `usd`: the dollars this ledger booked while the card's worker ran (each `dispatched` beat until the next beat on the
+  card). The ledger has no card field, so this is a proxy: review and planning calls on the card are not in it, and two
+  cards whose runs overlap are each charged what was booked in their own window.
+- `signups`: for a venture pick, the venture's signups at day 7 (`TIME_VENTURE_MEASURE_DAYS`); `null` for any other kind.
+
+```json
+{"n":168,"kind":"feature","picked":"2026-10-17","settled":"2026-10-19","merged":true,"pr":190,"days":2,"usd":1.25,"signups":null}
+```
+
+A pick is written once, when its card closes, and never rewritten. Astra's daily log shows one summary line per kind over
+the ranked table (picks, merged share, median days, median $). No pick reads the file yet: `router.score` is still the
+whole policy.
+
 ## Pools (`time/pools.json`)
 
 The operator keeps one entry per capacity pool; a route with no entry is paced by `TIME_DAILY_USD`. Time reads the file through the API once per run. Fields: `name`, `kind`, `routes`, and by kind: