card #80: kept .github/time/test_steward.py (generated)

.github/time/steward.py

@@ -0,0 +1,403 @@
+"""Steward: the factory's repair tick, inside Time. A block that is only reported is a status page; this one repairs.
+
+    python3 .github/time/steward.py      # every 6 hours (steward.yml) and on workflow_dispatch
+
+Two parts, in order.
+
+1. Deterministic, free, no model: merge card PRs that carry a trusted Approve and green checks and that `touches_protected`
+   does not hold (the same tests `land()` applies, TIME_AUTOMERGE included); close an `andon` whose cause has healed (a daily
+   spend cap once midnight UTC has passed, a key that is now set); re-dispatch the heartbeat once if its latest run failed.
+2. Only for what is left and is a real block: one call to the steward seat (TIME_STEWARD, default `claude:opus`; `off` disables
+   the whole tick) with the repair charter. It books to the ledger like every other call. It repairs with the tools below
+   (labels, the few variables it may touch, a heartbeat retry, a comment that hands a card back to its worker) and may
+   escalate to the operator only for money beyond the caps or a human-only step. Code checks that, not the prompt.
+
+One entry per run goes to the Astra log, in Astra's thread and style: Merged / Fixed / Left, with the reason for anything left.
+A comment the steward leaves on a card carries `{"steward": true}`, never a `beat`, so it cannot change the card's state.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import importlib.util
+import json
+import os
+import re
+import sys
+from pathlib import Path
+
+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)
+
+LOG_TITLE = "Astra log"   # astra.py's thread: one place a human reads the factory's day
+DEFAULT_SEAT = "claude:opus"
+HEARTBEAT = "heartbeat.yml"
+STEWARD_LABELS = {"ready", "queued", "needs-human"}
+# the only variables the steward may set, and the values it may set them to; never money (the caps), merge policy or trust
+STEWARD_VARS = {"TIME_MODE": r"live|dry-run", "TIME_PARALLEL": r"[1-9]", "TIME_STUCK_HOURS": r"\d{1,3}", "TIME_DAILY_CAP": r"\d{1,3}"}
+PARKED = ("error", "stuck", "blocked")   # a card whose last beat says its worker stopped
+
+CHARTER = """You are the steward of a small, fully AI-run software factory (Time). The deterministic part of this run has
+already merged what could be merged and closed what had healed. What follows is what is left, as JSON.
+
+Your job, in order, for each block: name it; repair it with the tools you have; escalate only if it cannot be repaired.
+- Tools: `label` (ready / queued / needs-human), `comment` (on a card or PR: hand a card back to its worker with the reason,
+  or leave a note the next run reads), `set_variable` (TIME_MODE, TIME_PARALLEL, TIME_STUCK_HOURS, TIME_DAILY_CAP only),
+  `dispatch_heartbeat` (a retry of the clock), `escalate`.
+- Escalate to the operator ONLY for money beyond the caps (the caps are never yours to raise) or a human-only step: a login, a
+  payment, a legal signature. A failing check, an objection, a parked card, a conflict, a wrong label is yours: fix it or hand
+  the card back to its worker. "Needs a human" about something you could have repaired is the failure this seat exists to end.
+- Never touch workflows, secrets, the spend caps, TIME_AUTOMERGE, TIME_PROTECTED or TIME_TRUSTED, and never merge anything.
+- Be brief. When you are done, reply with 2-4 sentences: what you found, what you did, what you left and why."""
+
+TOOLS = [
+    {"type": "function", "function": {"name": "label", "description": "Add or remove one label on a card or PR.",
+     "parameters": {"type": "object", "properties": {"number": {"type": "integer"}, "name": {"type": "string", "enum": sorted(STEWARD_LABELS)},
+                    "action": {"type": "string", "enum": ["add", "remove"]}}, "required": ["number", "name", "action"]}}},
+    {"type": "function", "function": {"name": "comment", "description": "Comment on a card or PR as Time. Use it to hand a card back to its worker.",
+     "parameters": {"type": "object", "properties": {"number": {"type": "integer"}, "text": {"type": "string"}}, "required": ["number", "text"]}}},
+    {"type": "function", "function": {"name": "set_variable", "description": "Set one Actions variable: " + ", ".join(sorted(STEWARD_VARS)) + ".",
+     "parameters": {"type": "object", "properties": {"name": {"type": "string", "enum": sorted(STEWARD_VARS)}, "value": {"type": "string"}},
+                    "required": ["name", "value"]}}},
+    {"type": "function", "function": {"name": "dispatch_heartbeat", "description": "Run the heartbeat workflow once, now.",
+     "parameters": {"type": "object", "properties": {}, "required": []}}},
+    {"type": "function", "function": {"name": "escalate", "description": "Open one issue for the operator. Only for money beyond the caps or a human-only step.",
+     "parameters": {"type": "object", "properties": {"kind": {"type": "string", "enum": ["money", "human-only"]},
+                    "title": {"type": "string"}, "reason": {"type": "string", "description": "What is blocked, what you tried, what exactly the human must do."}},
+                    "required": ["kind", "title", "reason"]}}},
+]
+
+
+class Report:
+    """What this run did: the three sections of the log entry."""
+
+    def __init__(self):
+        self.merged: list[str] = []
+        self.fixed: list[str] = []
+        self.left: list[dict] = []        # {"what", "why", "ask"}: ask = a real block the model may repair
+        self.escalated: list[str] = []
+
+    def leave(self, what: str, why: str, ask: bool = False, card: int | None = None, since: str = "") -> None:
+        self.left.append({"what": what, "why": why, "ask": ask, "card": card, "since": since})
+
+
+def log_entry(seat: str, spend: tuple[float, float], rep: Report, note: str) -> tuple[str, dict]:
+    """The log comment, in Astra's style: a header with the spend, a note, then Merged / Fixed / Left."""
+    def section(name: str, items: list[str]) -> list[str]:
+        return [f"{name}: none"] if not items else [f"{name}:"] + [f"- {i}" for i in items]
+    lines = [f"🧭 Steward ({seat}), spend ${spend[0]:.2f} of ${spend[1]:.2f} today:", "", note, ""]
+    lines += section("Merged", rep.merged) + section("Fixed", rep.fixed)
+    lines += section("Left", [f"{l['what']}: {l['why']}" for l in rep.left])
+    if rep.escalated:
+        lines += section("Escalated to the operator", rep.escalated)
+    meta = {"beat": "steward", "merged": len(rep.merged), "fixed": len(rep.fixed), "left": len(rep.left), "escalated": len(rep.escalated)}
+    return "\n".join(lines), meta
+
+
+def log_issue() -> int:
+    for i in tm.issues("state=open"):
+        if i["title"] == LOG_TITLE:
+            return i["number"]
+    return tm.new_issue(LOG_TITLE, "Astra, the daily planner, writes one comment here per day: what it read, what it "
+                        "queued, and why. Object in a reply; Astra reads this thread too.", ["astra"])["number"]
+
+
+def say(number: int, text: str, **meta) -> None:
+    tm.comment(number, f"🧭 Steward: {text}", {"steward": True, **meta})
+
+
+# ---------- 1. deterministic ----------
+
+def card_of(pr: dict) -> int | None:
+    """The card a PR answers: `Closes #N` in its body, or its branch `time/card-N`."""
+    m = re.search(r"\b(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)\s+#(\d+)", pr.get("body") or "", re.I) \
+        or re.match(r"time/card-(\d+)", (pr.get("head") or {}).get("ref") or "")
+    return int(m.group(1)) if m else None
+
+
+def merge_ready(rep: Report) -> None:
+    """Merge every card PR that land() would merge. What it would not merge is left, with the reason."""
+    automerge = (os.environ.get("TIME_AUTOMERGE") or "off").strip() == "on"
+    for listed in tm.gh("GET", f"/repos/{tm.REPO}/pulls?state=open&limit=50"):
+        n = card_of(listed)
+        if n is None or listed.get("state", "open") != "open":
+            continue
+        what = f"PR #{listed['number']} (card #{n})"
+        try:
+            comments = tm.gh("GET", f"/repos/{tm.REPO}/issues/{n}/comments?limit=100")
+            beats = tm.beats_of(comments)
+            last = beats[-1]["at"] if beats else ""
+            pr = tm.gh("GET", f"/repos/{tm.REPO}/pulls/{listed['number']}")
+            if pr.get("merged_at") or pr.get("state", "open") != "open":
+                continue
+            revised = max((b["at"] for b in beats if b.get("revised")), default="")
+            cast = tm.votes(pr, max(pr["created_at"], revised))
+            v = tm.decide(cast)
+            if v == "object":
+                rep.leave(what, "a reviewer objected", ask=True, card=n, since=last)
+            elif v is None:
+                rep.leave(what, "no trusted verdict yet" if not cast else "the council is split or short of the quorum", ask=bool(cast), card=n, since=last)
+            elif (hit := tm.touches_protected(pr["number"])):
+                rep.leave(what, f"approved, but it touches protected paths ({', '.join(hit[:5])}): a human merges this one")
+            elif not automerge:
+                rep.leave(what, "approved, but TIME_AUTOMERGE is not on")
+            else:
+                _merge(pr, n, what, beats, rep)
+        except Exception as e:  # noqa: BLE001  one PR must not stop the others
+            rep.leave(what, f"could not be checked: {type(e).__name__}: {str(e)[:160]}", ask=True)
+
+
+def _merge(pr: dict, n: int, what: str, beats: list[dict], rep: Report) -> None:
+    c = tm.checks(pr["head"]["sha"])
+    need = (os.environ.get("TIME_REQUIRE_CHECKS") or "on").strip() != "off"
+    if c == "pending":
+        rep.leave(what, "approved; checks are pending")
+    elif c == "none" and need:
+        if beats and beats[-1]["beat"] in ("no-checks", "ci-dispatched"):
+            rep.leave(what, "approved, but there are still no checks and I only merge on green", ask=True, card=n, since=beats[-1]["at"])
+        elif tm.dispatch_ci(pr):
+            say(n, f"approved; running CI on PR #{pr['number']} before merging.", beat="ci-dispatched", pr=pr["number"])
+            rep.fixed.append(f"dispatched CI on {what}, which had none")
+        else:
+            rep.leave(what, "approved, but it has no checks and there is no CI workflow to dispatch", ask=True)
+    elif c == "fail":
+        rep.leave(what, "approved, but checks are failing", ask=True, card=n, since=beats[-1]["at"] if beats else "")
+    elif pr.get("mergeable") is False:
+        rep.leave(what, "approved and green, but it has merge conflicts", ask=True, card=n, since=beats[-1]["at"] if beats else "")
+    else:
+        tm.gh("POST", f"/repos/{tm.REPO}/pulls/{pr['number']}/merge", {"Do": "squash", "delete_branch_after_merge": True}, as_time=True)
+        if tm.gh("GET", f"/repos/{tm.REPO}/issues/{n}")["state"] == "open":
+            tm.gh("PATCH", f"/repos/{tm.REPO}/issues/{n}", {"state": "closed"})
+        # the one steward comment that keeps its beat: it is the same "merged" beat land() writes, so the card reads as done
+        tm.comment(n, f"⏱️ Time: approved and green, merged PR #{pr['number']}. Card done.", {"beat": "merged", "pr": pr["number"], "steward": True})
+        rep.merged.append(what)
+
+
+def healed(andon: dict) -> str | None:
+    """Why this andon's cause is gone, or None when it is not (or cannot be told to be)."""
+    text = f"{andon.get('title') or ''}\n{andon.get('body') or ''}"
+    today, day_cap, month, month_cap = tm.budget()
+    now = dt.datetime.now(dt.timezone.utc)
+    if "today's spend is" in text:   # budget_check's daily reason: it heals itself at midnight UTC
+        if (andon.get("created_at") or "9")[:10] < now.strftime("%Y-%m-%d") and today < day_cap:
+            return f"it was a daily cap, and today's spend is ${today:.2f} of ${day_cap:.2f} after midnight UTC"
+        return None
+    if "this month's spend is" in text:   # monthly: a human raises the cap and sets TIME_MODE back; not mine to close
+        return None
+    keys = re.findall(r"\bno ([A-Z][A-Z0-9_]+) secret\b", text)
+    if keys and all(os.environ.get(k) for k in keys):
+        return f"{', '.join(keys)} {'is' if len(keys) == 1 else 'are'} set now"
+    return None
+
+
+def close_healed(rep: Report) -> list[dict]:
+    """Close each open andon whose cause has healed. Returns the ones still open."""
+    still = []
+    for a in tm.issues("state=open&labels=andon"):
+        why = healed(a)
+        if not why:
+            still.append(a)
+            continue
+        try:
+            say(a["number"], f"closing this andon: {why}.", healed=True)
+            tm.gh("PATCH", f"/repos/{tm.REPO}/issues/{a['number']}", {"state": "closed"})
+            rep.fixed.append(f"closed andon #{a['number']}: {why}")
+        except Exception as e:  # noqa: BLE001
+            still.append(a)
+            rep.leave(f"andon #{a['number']}", f"healed ({why}) but I could not close it: {str(e)[:160]}", ask=True)
+    return still
+
+
+def heartbeat_failed() -> dict | None:
+    """The heartbeat's latest run if it failed, else None. Raises when the forge cannot list runs."""
+    out = tm.gh("GET", f"/repos/{tm.REPO}/actions/runs?limit=50")
+    runs = out.get("workflow_runs") if isinstance(out, dict) else out
+    mine = [r for r in runs or [] if str(r.get("path") or "").startswith(HEARTBEAT) or r.get("name") == "heartbeat"]
+    if not mine:
+        return None
+    latest = max(mine, key=lambda r: r.get("id") or 0)
+    return latest if str(latest.get("conclusion") or latest.get("status") or "").lower() == "failure" else None
+
+
+def dispatch_heartbeat() -> None:
+    tm.gh("POST", f"/repos/{tm.REPO}/actions/workflows/{HEARTBEAT}/dispatches", {"ref": tm.default_branch()})
+
+
+def retry_heartbeat(rep: Report) -> None:
+    try:
+        run = heartbeat_failed()
+    except Exception as e:  # noqa: BLE001
+        rep.leave("heartbeat", f"I could not read its runs: {str(e)[:160]}")
+        return
+    if not run:
+        return
+    try:
+        dispatch_heartbeat()
+        rep.fixed.append(f"the heartbeat's latest run (#{run.get('id')}) failed; dispatched it once")
+    except Exception as e:  # noqa: BLE001
+        rep.leave("heartbeat", f"its latest run (#{run.get('id')}) failed and I could not dispatch it: {str(e)[:160]}", ask=True)
+
+
+def parked_cards(rep: Report) -> None:
+    """A `ready` card whose last beat says its worker stopped is a block, unless the steward has already spoken on it."""
+    for card in tm.issues("state=open&labels=ready"):
+        try:
+            cs = tm.gh("GET", f"/repos/{tm.REPO}/issues/{card['number']}/comments?limit=100")
+            real = [b for b in tm.beats_of(cs) if b["beat"] not in ("capped",)]
+            if real and real[-1]["beat"] in PARKED:
+                rep.leave(f"card #{card['number']}", f"parked on `{real[-1]['beat']}`", ask=True, card=card["number"], since=real[-1]["at"])
+        except Exception as e:  # noqa: BLE001
+            print(f"card #{card['number']}: {e}")
+
+
+def already_handled(rep: Report) -> None:
+    """Blocks the steward has already spoken on since the card's last beat go back to waiting: no second call for the same thing."""
+    for item in rep.left:
+        if not (item["ask"] and item["card"]):
+            continue
+        cs = tm.gh("GET", f"/repos/{tm.REPO}/issues/{item['card']}/comments?limit=100")
+        if any(tm.is_time(c) and '"steward":true' in (c.get("body") or "") and c["created_at"] > item["since"] for c in cs):
+            item["ask"], item["why"] = False, item["why"] + " (the steward has handed this back already; it waits for the card to move)"
+
+
+# ---------- 2. the model, for what is left ----------
+
+class Tools:
+    def __init__(self, rep: Report):
+        self.rep = rep
+
+    def call(self, name: str, a: dict) -> str:
+        try:
+            return getattr(self, f"t_{name}")(a)
+        except AttributeError:
+            return f"unknown tool {name}"
+        except Exception as e:  # noqa: BLE001  the model reads the error and tries something else
+            return f"error: {type(e).__name__}: {str(e)[:300]}"
+
+    def t_label(self, a: dict) -> str:
+        n, name = int(a["number"]), str(a["name"])
+        if name not in STEWARD_LABELS:
+            return f"refused: the steward may only use {', '.join(sorted(STEWARD_LABELS))}"
+        (tm.label if a.get("action") == "add" else tm.unlabel)(n, name)
+        self.rep.fixed.append(f"{a.get('action')} label `{name}` on #{n}")
+        return "done"
+
+    def t_comment(self, a: dict) -> str:
+        say(int(a["number"]), str(a["text"])[:3000])
+        self.rep.fixed.append(f"commented on #{a['number']}")
+        return "done"
+
+    def t_set_variable(self, a: dict) -> str:
+        name, value = str(a["name"]), str(a["value"])
+        if name not in STEWARD_VARS or not re.fullmatch(STEWARD_VARS[name], value):
+            return f"refused: allowed variables are {', '.join(f'{k} ({v})' for k, v in STEWARD_VARS.items())}"
+        if name == "TIME_MODE" and value == "live" and tm.budget()[2] >= tm.budget()[3]:
+            return "refused: the monthly cap is reached; raising it is the operator's"
+        path = f"/repos/{tm.REPO}/actions/variables/{name}"
+        try:
+            tm.gh("PUT", path, {"value": value})
+        except RuntimeError as e:
+            if "-> 404" not in str(e):
+                raise
+            tm.gh("POST", path, {"value": value})
+        self.rep.fixed.append(f"set variable {name}={value}")
+        return "done"
+
+    def t_dispatch_heartbeat(self, a: dict) -> str:
+        dispatch_heartbeat()
+        self.rep.fixed.append("dispatched the heartbeat")
+        return "dispatched"
+
+    def t_escalate(self, a: dict) -> str:
+        kind = str(a.get("kind"))
+        if kind not in ("money", "human-only"):
+            return "refused: escalate only for `money` beyond the caps or a `human-only` step (a login, a payment, a legal signature)"
+        title = f"Steward: {str(a['title'])[:150]}"
+        if any(i["title"] == title for i in tm.issues("state=open&labels=needs-human")):
+            return "already open"
+        tm.new_issue(title, f"🧭 The steward cannot repair this and hands it to the operator ({kind}).\n\n{str(a['reason'])[:3000]}\n\n"
+                     "Close this once it's handled.", ["needs-human"])
+        self.rep.escalated.append(f"{kind}: {title}")
+        return "opened"
+
+
+def ask_steward(spec: str, blocks: list[dict], context: dict, rep: Report) -> str:
+    """One conversation with the steward seat. Books to the ledger through tm.chat; a cap or a missing key ends it politely."""
+    seat, model, effort = tm.worker_seat_model(spec)
+    user = "Blocks:\n" + json.dumps(blocks, indent=1) + "\n\nContext:\n" + json.dumps(context, indent=1)
+    msgs = [{"role": "system", "content": CHARTER}, {"role": "user", "content": user}]
+    tools, last = Tools(rep), ""
+    for _ in range(int(os.environ.get("TIME_STEWARD_STEPS") or 8)):
+        body = {"model": model, "temperature": 0.2, "max_tokens": 4000, "messages": msgs, "tools": TOOLS,
+                "venice_parameters": {"include_venice_system_prompt": False}}
+        if effort:
+            body["reasoning_effort"] = effort
+        try:
+            msg = tm.chat(body, timeout=240)["choices"][0]["message"]
+        except tm.Budget as e:
+            return f"No model call: {e}."
+        except LookupError as e:
+            return f"No model call: {e}."
+        msgs.append({k: v for k, v in msg.items() if k in ("role", "content", "tool_calls")})
+        last = msg.get("content") or last
+        calls = msg.get("tool_calls") or []
+        if not calls:
+            break
+        for c in calls:
+            try:
+                args = json.loads(c["function"].get("arguments") or "{}")
+            except ValueError:
+                args = {}
+            msgs.append({"role": "tool", "tool_call_id": c["id"], "content": tools.call(c["function"]["name"], args)})
+    return re.sub(r"<think>.*?</think>", "", last, flags=re.S).strip()[:1200] or "(the steward gave no summary)"
+
+
+def context() -> dict:
+    today, day_cap, month, month_cap = tm.budget()
+    return {"mode": os.environ.get("TIME_MODE") or "dry-run", "automerge": os.environ.get("TIME_AUTOMERGE") or "off",
+            "spend": {"today": round(today, 2), "daily_cap": day_cap, "month": round(month, 2), "monthly_cap": month_cap},
+            "open_andons": [{"n": i["number"], "title": i["title"], "body": (i.get("body") or "")[:500]} for i in tm.issues("state=open&labels=andon")],
+            "needs_human": [{"n": i["number"], "title": i["title"]} for i in tm.issues("state=open&labels=needs-human")]}
+
+
+# ---------- the tick ----------
+
+def tick() -> int:
+    spec = (os.environ.get("TIME_STEWARD") or DEFAULT_SEAT).strip()
+    if spec.lower() == "off":
+        print("steward: off")
+        return 0
+    rep = Report()
+    for step in (merge_ready, close_healed, retry_heartbeat, parked_cards):
+        try:
+            step(rep)
+        except Exception as e:  # noqa: BLE001  a forge hiccup in one step must not skip the rest or the log
+            rep.leave(step.__name__, f"{type(e).__name__}: {str(e)[:160]}")
+    for a in tm.issues("state=open&labels=andon"):
+        if healed(a) or any(l["what"] == f"andon #{a['number']}" for l in rep.left):
+            continue
+        text = f"{a.get('title')}\n{a.get('body')}"
+        if "this month's spend is" in text:
+            rep.leave(f"andon #{a['number']}", "the monthly cap is money: the operator raises it, and sets TIME_MODE back to live")
+        elif "today's spend is" in text:
+            rep.leave(f"andon #{a['number']}", "a daily cap, which heals after midnight UTC")
+        else:
+            rep.leave(f"andon #{a['number']}", "its cause has not healed", ask=True)
+    try:
+        already_handled(rep)
+    except Exception as e:  # noqa: BLE001
+        print(f"steward: {e}")
+    blocks = [l for l in rep.left if l["ask"]]
+    note = "Nothing needed a model call."
+    if blocks:
+        note = ask_steward(spec, [{"what": b["what"], "why": b["why"]} for b in blocks], context(), rep)
+    spend = tm.budget()
+    text, meta = log_entry(spec, (spend[0], spend[1]), rep, note)
+    tm.comment(log_issue(), text, meta)
+    print(f"steward: merged={len(rep.merged)} fixed={len(rep.fixed)} left={len(rep.left)} asked={bool(blocks)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(tick())

.github/time/test_steward.py

@@ -0,0 +1,250 @@
+"""Offline tests for steward.py: a fake forge, no network, no model. Run: python3 .github/time/test_steward.py
+(test_time.py runs this file too, so the one CI command covers it.)"""
+import datetime as dt, importlib.util, json, os, urllib.parse
+from pathlib import Path
+
+for _k in [k for k in os.environ if k.startswith("TIME_")] + ["ANTHROPIC_API_KEY", "VENICE_API_KEY", "CURSOR_API_KEY"]:
+    os.environ.pop(_k, None)
+os.environ.update(GITHUB_REPOSITORY="o/r", GITHUB_TOKEN="gh", GITHUB_SERVER_URL="https://forge.test", GITHUB_API_URL="https://forge.test/api/v1")
+spec = importlib.util.spec_from_file_location("steward", Path(__file__).resolve().parent / "steward.py")
+st = importlib.util.module_from_spec(spec); spec.loader.exec_module(st)
+tm = st.tm
+
+NOW = dt.datetime.now(dt.timezone.utc)
+iso = lambda h=0: (NOW - dt.timedelta(hours=h)).strftime("%Y-%m-%dT%H:%M:%SZ")
+BOT = {"login": "gitea-actions"}
+
+
+class Forge:
+    def __init__(self):
+        self.prs, self.comments, self.reviews, self.files, self.statuses = {}, {}, {}, {}, {}
+        self.issues = {57: {"number": 57, "title": "Astra log", "body": "", "state": "open", "labels": ["astra"], "created_at": iso(100)}}
+        self.merged, self.dispatched, self.runs, self.ci, self.vars, self.chats = [], [], [], [], {}, 0
+        self.trusted = {"rev"}
+
+    def pr(self, n, card, files=("app.py",), status="success", approve=True, mergeable=True, body=None):
+        self.prs[n] = {"number": n, "state": "open", "merged_at": None, "created_at": iso(3), "mergeable": mergeable, "body": body or f"Closes #{card}",
+                       "head": {"ref": f"time/card-{card}", "sha": f"sha{n}"}}
+        self.files[n] = list(files)
+        self.statuses[f"sha{n}"] = [{"context": "ci / test", "status": status}] if status else []
+        self.issues.setdefault(card, {"number": card, "title": f"card {card}", "body": "", "state": "open", "labels": [], "created_at": iso(5)})
+        self.comments.setdefault(card, [])
+        if approve:
+            self.comments[n] = [{"body": "**Approve**\nlooks right", "user": {"login": "rev"}, "created_at": iso(1)}]
+
+    def add_issue(self, n, title, body, labels, created=0):
+        self.issues[n] = {"number": n, "title": title, "body": body, "state": "open", "labels": labels, "created_at": iso(created)}
+
+    def gh(self, method, path, body=None, *, as_time=None):
+        base = "/repos/o/r"
+        p, _, q = path.partition("?")
+        q = urllib.parse.parse_qs(q)
+        seg = p[len(base):].strip("/").split("/")
+        if method == "GET" and seg[0] == "pulls" and len(seg) == 1:
+            return [dict(x) for x in self.prs.values() if x["state"] == "open"]
+        if seg[0] == "pulls" and len(seg) == 2 and method == "GET":
+            return dict(self.prs[int(seg[1])])
+        if seg[0] == "pulls" and seg[-1] == "files":
+            return [{"filename": f} for f in self.files[int(seg[1])]]
+        if seg[0] == "pulls" and seg[-1] == "reviews":
+            return self.reviews.get(int(seg[1]), [])
+        if seg[0] == "pulls" and seg[-1] == "merge":
+            n = int(seg[1]); assert body["Do"] == "squash"
+            self.prs[n].update(merged_at=iso(), state="closed"); self.merged.append(n); return {}
+        if seg[0] == "commits":
+            return {"statuses": self.statuses[seg[1]]}
+        if seg[0] == "collaborators":
+            if seg[1] in self.trusted: return {}
+            raise RuntimeError(f"GET {path} -> 404: no")
+        if seg[0] == "issues" and len(seg) == 1 and method == "GET":
+            out = [i for i in self.issues.values() if i["state"] == q.get("state", ["open"])[0]
+                   and all(l in i["labels"] for l in q.get("labels", [""])[0].split(",") if l)]
+            return [dict(i, labels=[{"name": l} for l in i["labels"]]) for i in out]
+        if seg[0] == "issues" and len(seg) == 1 and method == "POST":
+            n = max(self.issues) + 1
+            self.add_issue(n, body["title"], body["body"], [self.lname(l) for l in body["labels"]])
+            return {"number": n}
+        if seg[0] == "issues" and seg[-1] == "comments":
+            n = int(seg[1])
+            if method == "POST":
+                self.comments.setdefault(n, []).append({"body": body["body"], "user": BOT, "created_at": iso()}); return {}
+            return self.comments.get(n, [])
+        if seg[0] == "issues" and len(seg) == 2:
+            n = int(seg[1])
+            if method == "PATCH": self.issues[n].update(body); return {}
+            return dict(self.issues[n])
+        if seg[0] == "issues" and seg[2] == "labels":
+            n = int(seg[1])
+            if method == "POST": self.issues[n]["labels"] += [self.lname(l) for l in body["labels"]]
+            else: self.issues[n]["labels"].remove(self.lname(int(seg[3])))
+            return {}
+        if seg[0] == "labels":
+            if method == "POST": return {"id": 99}
+            return [{"name": n, "id": i} for i, n in enumerate(["ready", "queued", "needs-human"], 1)]
+        if seg[0] == "actions" and seg[1] == "runs":
+            return {"workflow_runs": self.runs}
+        if seg[0] == "actions" and seg[1] == "workflows":
+            if seg[2] == "ci.yml": self.ci.append(body["ref"]); return {}
+            self.dispatched.append(seg[2]); return {}
+        if seg[0] == "actions" and seg[1] == "variables":
+            if method == "PUT" and seg[2] not in self.vars: raise RuntimeError(f"PUT {p} -> 404: not found")  # Gitea: PUT updates, POST creates
+            self.vars[seg[2]] = body["value"]; return {}
+        if p == base:
+            return {"default_branch": "main"}
+        raise AssertionError(f"unmocked {method} {path}")
+
+    def lname(self, x):
+        return x if isinstance(x, str) else {1: "ready", 2: "queued", 3: "needs-human"}.get(x, f"label{x}")
+
+    def log(self):
+        return [c["body"] for c in self.comments.get(57, [])]
+
+
+def world(**env):
+    f = Forge()
+    tm.gh = f.gh
+    tm._trust.clear(); tm._label_ids = None
+    for k in [k for k in os.environ if k.startswith("TIME_")] + ["VENICE_API_KEY", "ANTHROPIC_API_KEY"]: os.environ.pop(k, None)
+    os.environ.update({"TIME_AUTOMERGE": "on", **env})
+    tm.budget = lambda: (1.0, 5.0, 10.0, 100.0)
+    def no_model(body, timeout): f.chats += 1; raise AssertionError("the deterministic part must not call a model")
+    tm.chat = no_model
+    return f
+
+
+def run():
+    assert st.tick() == 0
+
+
+def beat_of(comment):
+    return json.loads(comment.split(tm.MARK, 1)[1].rsplit("-->", 1)[0])
+
+
+# 1. a green, approved, unprotected card PR is merged without a model call; the log says so
+f = world()
+f.pr(59, 12)
+run()
+assert f.merged == [59] and f.chats == 0, (f.merged, f.chats)
+assert f.issues[12]["state"] == "closed", "the card closes with its PR"
+assert any(beat_of(c["body"])["beat"] == "merged" for c in f.comments[12]), "the card gets its merged beat, as land() writes it"
+entry = f.log()
+assert len(entry) == 1, "one entry per run"
+assert "Merged:\n- PR #59 (card #12)" in entry[0] and "Fixed: none" in entry[0] and "Left: none" in entry[0], entry[0]
+assert beat_of(entry[0]) == {"beat": "steward", "merged": 1, "fixed": 0, "left": 0, "escalated": 0}
+
+# ... but not on a stranger's Approve, without an approval, on red checks, with TIME_AUTOMERGE off, or on a PR that closes no card
+f = world(); f.pr(60, 13); f.comments[60] = [{"body": "Approve", "user": {"login": "stranger"}, "created_at": iso(1)}]; run()
+assert f.merged == [] and "no trusted verdict yet" in f.log()[0]
+f = world(); f.pr(60, 13, approve=False); run()
+assert f.merged == [] and "no trusted verdict yet" in f.log()[0]
+f = world(); f.pr(60, 13, status="failure"); f.chats = 0
+tm.chat = lambda body, timeout: {"choices": [{"message": {"role": "assistant", "content": "Checks fail on #13; handed back."}}], "usage": {}}
+run()
+assert f.merged == [] and "checks are failing" in f.log()[0] and "handed back" in f.log()[0]
+f = world(TIME_AUTOMERGE="off"); f.pr(60, 13); run()
+assert f.merged == [] and "TIME_AUTOMERGE is not on" in f.log()[0]
+f = world(); f.pr(61, 14, body="no link"); f.prs[61]["head"]["ref"] = "feature/x"; run()
+assert f.merged == [] and "PR #61" not in f.log()[0], "only card PRs are the steward's"
+f = world(); f.pr(60, 13, status="pending"); run()
+assert f.merged == [] and "checks are pending" in f.log()[0] and f.chats == 0, "waiting on CI is not a block"
+
+# no checks yet: one CI dispatch, then it waits (no second dispatch, no merge)
+f = world(); f.pr(60, 13, status=None); run()
+assert f.merged == [] and f.ci == ["time/card-13"] and "dispatched CI" in f.log()[0]
+
+# 2. a protected PR is left, with the reason, and no model is asked
+f = world(); f.pr(62, 15, files=("app.py", ".github/workflows/ci.yml")); run()
+assert f.merged == [] and f.chats == 0
+assert "Left:\n- PR #62 (card #15): approved, but it touches protected paths (.github/workflows/ci.yml): a human merges this one" in f.log()[0], f.log()[0]
+
+# 3. andons: a healed daily cap closes, an unhealed monthly one does not
+DAILY = "⏱️ Time stopped calling models: today's spend is $5.20 of the $5 cap; carrying on tomorrow.\n\n- Daily cap (`TIME_DAILY_USD`): $5.00\n- Monthly cap (`TIME_MONTHLY_USD`): $100.00"
+MONTHLY = "⏱️ Time stopped calling models: this month's spend is $100.10 of the $100.00 cap; Time is back in dry-run until a human raises it.\n\n- Daily cap (`TIME_DAILY_USD`): $5.00\n- Monthly cap (`TIME_MONTHLY_USD`): $100.00"
+f = world()
+f.add_issue(70, "Spend cap reached: $5.20 today / $9.00 this month", DAILY, ["andon"], created=30)
+f.add_issue(71, "Spend cap reached: $5.20 today / $100.10 this month", MONTHLY, ["andon"], created=30)
+f.add_issue(72, "Spend cap reached: $5.20 today / $9.00 this month", DAILY, ["andon"], created=0)   # tripped today: not healed yet
+run()
+assert f.issues[70]["state"] == "closed" and f.issues[71]["state"] == "open" and f.issues[72]["state"] == "open", [f.issues[n]["state"] for n in (70, 71, 72)]
+assert f.chats == 0
+assert "closed andon #70" in f.log()[0] and "andon #71: the monthly cap is money" in f.log()[0] and "andon #72: a daily cap" in f.log()[0], f.log()[0]
+# a daily andon from yesterday is not healed while today is over the cap again
+f = world(); tm.budget = lambda: (5.5, 5.0, 10.0, 100.0)
+f.add_issue(70, "Spend cap reached", DAILY, ["andon"], created=30); run()
+assert f.issues[70]["state"] == "open"
+# a missing key that is set now
+f = world(VENICE_API_KEY="k")
+f.add_issue(73, "Venice key missing", "Time can't go further: no VENICE_API_KEY secret on this repo yet.", ["andon"]); run()
+assert f.issues[73]["state"] == "closed" and "VENICE_API_KEY is set now" in f.log()[0]
+f = world(); f.add_issue(73, "Venice key missing", "no VENICE_API_KEY secret on this repo yet", ["andon"])
+tm.chat = lambda body, timeout: {"choices": [{"message": {"role": "assistant", "content": "A key is a human-only step."}}], "usage": {}}
+run(); assert f.issues[73]["state"] == "open"
+
+# 4. the heartbeat: a failed latest run gets one dispatch; a green one, or another workflow's failure, none
+HB = lambda i, c: {"id": i, "path": "heartbeat.yml@refs/heads/main", "name": "heartbeat", "status": c, "conclusion": c}
+f = world(); f.runs = [HB(1, "success"), HB(2, "failure")]; run()
+assert f.dispatched == ["heartbeat.yml"] and "dispatched it once" in f.log()[0] and f.chats == 0, f.dispatched
+f = world(); f.runs = [HB(1, "failure"), HB(2, "success")]; run()
+assert f.dispatched == [], "only the latest run counts"
+f = world(); f.runs = [HB(1, "success"), {"id": 5, "path": "ci.yml@refs/heads/main", "status": "failure"}]; run()
+assert f.dispatched == []
+f = world(); f.runs = [HB(1, "cancelled")]; run()
+assert f.dispatched == [], "a run the concurrency group cancelled is not a failure"
+
+# 5. the log entry format
+rep = st.Report(); rep.merged.append("PR #1 (card #2)"); rep.fixed.append("closed andon #3"); rep.leave("PR #4 (card #5)", "a reviewer objected")
+text, meta = st.log_entry("claude:opus", (1.234, 5), rep, "Note.")
+assert text.splitlines()[0] == "🧭 Steward (claude:opus), spend $1.23 of $5.00 today:"
+assert text.splitlines()[2:] == ["Note.", "", "Merged:", "- PR #1 (card #2)", "Fixed:", "- closed andon #3", "Left:", "- PR #4 (card #5): a reviewer objected"], text.splitlines()
+assert meta == {"beat": "steward", "merged": 1, "fixed": 1, "left": 1, "escalated": 0}
+
+# 6. what is left and blocking: one model call, booked through tm.chat, with the charter; its tools are fenced
+f = world(); f.pr(80, 20, approve=False); f.comments[80] = [{"body": "**Object**\nno test", "user": {"login": "rev"}, "created_at": iso(1)}]
+f.add_issue(20, "card 20", "", ["ready"])
+calls = []
+def steward_model(body, timeout):
+    calls.append(body)
+    if len(calls) == 1:
+        tc = lambda i, name, args: {"id": f"c{i}", "type": "function", "function": {"name": name, "arguments": json.dumps(args)}}
+        return {"choices": [{"message": {"role": "assistant", "content": None, "tool_calls": [
+            tc(1, "comment", {"number": 20, "text": "The reviewer wants a test; please add one."}),
+            tc(2, "set_variable", {"name": "TIME_AUTOMERGE", "value": "on"}),
+            tc(3, "set_variable", {"name": "TIME_MODE", "value": "live"}),
+            tc(4, "escalate", {"kind": "boredom", "title": "x", "reason": "y"}),
+            tc(5, "label", {"number": 80, "name": "andon", "action": "add"})]}}], "usage": {}}
+    return {"choices": [{"message": {"role": "assistant", "content": "PR #80 was objected to; I handed card #20 back with the reviewer's point."}}], "usage": {}}
+tm.chat = steward_model
+run()
+assert len(calls) == 2 and calls[0]["model"] == "claude-opus-5-5"
+assert "ONLY for money" in calls[0]["messages"][0]["content"] and "a reviewer objected" in calls[0]["messages"][1]["content"]
+tool_results = [m["content"] for m in calls[1]["messages"] if m["role"] == "tool"]
+assert tool_results[0] == "done" and tool_results[1].startswith("refused") and tool_results[2] == "done" and tool_results[3].startswith("refused") and tool_results[4].startswith("refused"), tool_results
+assert f.vars == {"TIME_MODE": "live"}, f.vars
+assert "-> 404" not in "".join(tool_results), "the first set_variable on a fresh repo creates the variable"
+st.Tools(st.Report()).t_set_variable({"name": "TIME_MODE", "value": "dry-run"}); assert f.vars == {"TIME_MODE": "dry-run"}, "an existing one is updated"
+assert any('"steward":true' in c["body"] and "The reviewer wants a test" in c["body"] for c in f.comments[20])
+assert "handed card #20 back" in f.log()[0] and "Left:\n- PR #80 (card #20): a reviewer objected" in f.log()[0]
+# the steward has spoken on card #20 since its last beat: the next run does not ask again
+tm.chat = lambda body, timeout: (_ for _ in ()).throw(AssertionError("asked twice"))
+run(); assert "the steward has handed this back already" in f.log()[1]
+# real money or a human-only step: one issue for the operator, once
+f = world(); f.add_issue(75, "odd", "not a cap or a key", ["andon"])
+def esc(body, timeout):
+    if not any(m["role"] == "tool" for m in body["messages"]):
+        return {"choices": [{"message": {"role": "assistant", "content": None, "tool_calls": [{"id": "e", "type": "function", "function": {"name": "escalate",
+                "arguments": json.dumps({"kind": "human-only", "title": "Log in to the registrar", "reason": "Domain renewal needs a login."})}}]}}], "usage": {}}
+    return {"choices": [{"message": {"role": "assistant", "content": "Needs a login."}}], "usage": {}}
+tm.chat = esc; run()
+assert any(i["title"] == "Steward: Log in to the registrar" and "needs-human" in i["labels"] for i in f.issues.values()) and "Escalated to the operator:" in f.log()[0]
+# over the cap: the call is refused by the ledger's governor and the log says so
+f = world(); f.add_issue(75, "odd", "not a cap or a key", ["andon"])
+def capped(body, timeout): raise tm.Budget("today's spend is $5.00 of the $5.00 cap")
+tm.chat = capped; run()
+assert "No model call: today's spend" in f.log()[0]
+
+# TIME_STEWARD=off disables the whole tick; a seat spec is read from it
+f = world(TIME_STEWARD="off"); f.pr(59, 12); run()
+assert f.merged == [] and f.log() == []
+assert tm.worker_seat_model("claude:opus")[:2] == ("opus", "claude-opus-5-5")
+
+print("ALL OK steward")

.github/time/test_time.py

@@ -889,6 +889,12 @@ tm.venice_work = real_vw78
 for k in ("TIME_WORKER", "TIME_WORKER_STEPS", "TIME_AUTO_RETRY"): os.environ.pop(k, None)
 print("ALL OK 78")
 
+# ---------- 80 the steward tick: its own file (fake forge, no model), run here so the one CI command covers it
+import subprocess  # noqa: E402
+_p = subprocess.run([sys.executable, str(Path(__file__).resolve().parent / "test_steward.py")], capture_output=True, text=True)
+assert _p.returncode == 0 and "ALL OK steward" in _p.stdout, _p.stdout[-1500:] + _p.stderr[-1500:]
+print("ALL OK 80")
+
 # ---------- venture loop: still GitHub-shaped (Pages, author_association, per_page). Ported in its own card;
 # until then these run only with TIME_TEST_VENTURE=1 against the old fake.
 if not os.environ.get("TIME_TEST_VENTURE"):
@@ -1846,4 +1852,3 @@ with cf.ThreadPoolExecutor(2) as ex:
 vt2.tm.Hands, vt2.tm.venice_work = real40
 got = dict((n, (x, y)) for n, x, y in seen40)
 assert got[1][0] == "wrote" and got[1][1].startswith("refused") and got[2][0].startswith("refused") and got[2][1] == "wrote", got
-