Compare commits
11
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
481b38aece | ||
|
|
97ad17704a | ||
|
|
f32c7597ff | ||
|
|
6f34770288 | ||
|
|
7ab34cf6e7 | ||
|
|
4a7fa2045a | ||
|
|
44cfaab718 | ||
|
|
a8f860c559 | ||
|
|
2f637f628d | ||
|
|
a49d96eb13 | ||
|
|
2cb29056f9 |
No files matched your search
@@ -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())
|
||||
@@ -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")
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user