PR75: test automatic reservation recovery deterministically through heartbeat

.github/time/test_time.py

@@ -222,6 +222,7 @@ prh = {**pr, "head": {"sha": "abc"}}
 chk = {7: {"pr": {**prh, "state": "open"}, "statuses": [{"status": "pending", "context": "ci"}, {"status": "success", "context": "lint"}]}}
 run(World([card(1, D)], {1: [beat("dispatched", launched=True)]}, [prh], None, chk), {"TIME_MODE": "live", "CURSOR_API_KEY": "ck", "TIME_REVIEW": "venice:kimi"})
 assert "# Checks\n- ci: pending\n- lint: success\n\n# Diff" in seen["user"], seen["user"]
+assert "means that context reported success" in seen["system"] and "tests or builds passed" not in seen["system"]
 assert "evidence you can rely on" in seen["system"] and "pending check is not yet evidence" in seen["system"]
 run(World([card(1, D)], {1: [beat("dispatched", launched=True)]}, [prh]), {"TIME_MODE": "live", "CURSOR_API_KEY": "ck", "TIME_REVIEW": "venice:kimi"})
 assert "# Checks\n(checks unavailable)\n\n# Diff" in seen["user"], seen["user"]
@@ -280,6 +281,14 @@ assert not merged(w) and "objected" in w.posted[0][1]
 # protected paths -> human merges
 w = run(pr_world(files=[".github/workflows/ci.yml"]), {**K, "TIME_AUTOMERGE": "on"})
 assert not merged(w) and "protected paths" in w.posted[0][1]
+# the default protects all of .github/, the engine included; TIME_PROTECTED can narrow that to the workflow dirs, which stay
+# protected whatever it says
+w = run(pr_world(files=[".github/time/time.py", "CHANGELOG.md"]), {**K, "TIME_AUTOMERGE": "on"}); assert not merged(w)
+NARROW = ".github/workflows/,.gitea/workflows/,deploy,infra/,.env,secret"
+w = run(pr_world(files=[".github/time/time.py", "CHANGELOG.md"]), {**K, "TIME_AUTOMERGE": "on", "TIME_PROTECTED": NARROW}); assert merged(w)
+w = run(pr_world(files=[".gitea/workflows/ci.yml"]), {**K, "TIME_AUTOMERGE": "on", "TIME_PROTECTED": "deploy"})
+assert not merged(w) and "protected paths" in w.posted[0][1]
+os.environ.pop("TIME_PROTECTED", None)
 # failing or missing checks block the merge even when approved
 w = run(pr_world(checks=[{"status": "failure", "context": "ci"}]), {**K, "TIME_AUTOMERGE": "on"})
 assert not merged(w) and "checks are failing" in w.posted[0][1]
@@ -395,6 +404,12 @@ assert h.call("open_pr", {"title": "T", "summary": "round two"}) == "updated PR
 assert w.patched[0][0] == 9 and "round two" in w.patched[0][1]["body"] and "Closes #1" in w.patched[0][1]["body"]
 h2 = tm.Hands(2, "time/card-2"); h2.label = "kimi"
 assert h2.call("open_pr", {"title": "T", "summary": "s"}) == "opened PR #42" and len(w.opened) == 1, "a closed PR is not reopened"
+# no duplicate "Closes #N": added when the summary lacks it, left alone when it has it (#1 is not satisfied by #12)
+for summ, n in (("s", 1), ("s\n\nCloses #1", 1), ("s\n\ncloses #1.", 1), ("s\n\nCloses #12", 2)):
+    w = World([], prs=[{"number": 9, "state": "open", "head": {"ref": "time/card-1"}}]); tm.http = w.http
+    h = tm.Hands(1, "time/card-1"); h.label = "kimi"; h.call("open_pr", {"title": "T", "summary": summ})
+    body = w.patched[0][1]["body"]
+    assert body.lower().count("closes #1") == n and body.count("Closes #1\n") + body.count("Closes #1.") <= 1, (summ, body)
 # the Cursor worker answers an objection too: on the PR's own branch, no new PR, the review text in the prompt
 w = objected_world()
 w = run(w, {**K, "TIME_WORKER": "cursor"})
@@ -560,6 +575,115 @@ if _sh.which("node"):
         assert res.returncode == 0 and "aicharts: all good" in res.stdout, res.stdout[-2000:] + res.stderr[-2000:]
         print("ALL OK 41")
 
+# 41 A heartbeat overlaps priced API workers, instead of running them one by one.
+# Barrier fails in a sequential implementation. The per-run cap remains independent.
+import threading as _th41, time as _t41  # noqa: E402
+_original_worker41 = tm.venice_work
+_gate41 = _th41.Barrier(3, timeout=6)
+_seen41, _peak41, _active41 = [], 0, 0
+_lock41 = _th41.Lock()
+def _overlap41(card, done, spec, hands_cls=None, review=None, **kwargs):
+    global _peak41, _active41
+    with _lock41:
+        _active41 += 1
+        _peak41 = max(_peak41, _active41)
+        _seen41.append(card["number"])
+    try:
+        _gate41.wait()
+        _t41.sleep(0.03)
+        return {"pr": 200 + card["number"], "steps": 1, "seat": "sonnet", "model": "claude-sonnet-5-5"}
+    finally:
+        with _lock41:
+            _active41 -= 1
+
+tm.venice_work = _overlap41
+tm.LEDGER.clear()
+tm._ledger_files.clear()
+w41 = run(World([card(n, D) for n in (1, 2, 3)]), {
+    "TIME_MODE": "live", "TIME_WORKER": "claude:sonnet",
+    "ANTHROPIC_API_KEY": "test", "TIME_PARALLEL": "3",
+    "TIME_DAILY_USD": "5", "TIME_MONTHLY_USD": "100"})
+assert _peak41 == 3 and set(_seen41) == {1, 2, 3}, (_peak41, _seen41)
+assert len([body for _, body in w41.posted if "opened PR #" in body]) == 3, w41.posted
+
+_gate41 = _th41.Barrier(2, timeout=6)
+_seen41.clear()
+w41 = run(World([card(n, D) for n in (1, 2, 3)]), {
+    "TIME_MODE": "live", "TIME_WORKER": "claude:sonnet",
+    "ANTHROPIC_API_KEY": "test", "TIME_PARALLEL": "3", "TIME_DAILY_CAP": "2",
+    "TIME_DAILY_USD": "5", "TIME_MONTHLY_USD": "100"})
+assert set(_seen41) == {1, 2} and len([x for _, x in w41.posted if "opened PR #" in x]) == 2, (w41.posted, _seen41)
+tm.venice_work = _original_worker41
+
+# 42 Reservation contention waits and resumes automatically within the heartbeat.
+# Events establish ordering; no sleeps or hoped-for scheduling interleavings.
+_original_api42, _original_worker42 = tm.anthropic_chat, tm.venice_work
+_original_condition42 = tm._BUDGET_CHANGED
+_original_ledger42 = list(tm.LEDGER)
+_entered42, _release42, _waiting42 = _th41.Event(), _th41.Event(), _th41.Event()
+_api_calls42 = []
+class _ObservedBudgetCondition42(_th41.Condition):
+    def wait(self, timeout=None):
+        _waiting42.set()
+        return super().wait(timeout)
+tm._BUDGET_CHANGED = _ObservedBudgetCondition42(tm._LEDGER_LOCK)
+def _wait_for_api42(body, timeout):
+    _api_calls42.append(body["model"])
+    _entered42.set()
+    assert _release42.wait(5), "provider release timed out"
+    return {"usage": {"prompt_tokens": 10, "completion_tokens": 10}, "choices": []}
+_body42 = {"model": "claude-sonnet-5-5", "max_tokens": 16000,
+           "messages": [{"role": "user", "content": "bounded task"}]}
+def _budgeted_worker42(card, done, spec, **kwargs):
+    tm.chat(_body42, 3)
+    return {"pr": 200 + card["number"], "steps": 1, "seat": "sonnet", "model": "claude-sonnet-5-5"}
+tm.anthropic_chat, tm.venice_work = _wait_for_api42, _budgeted_worker42
+tm._ledger_files.clear()
+tm.LEDGER[:] = [{"at": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+                "model": "claude-sonnet-5-5", "in": 0, "out": 0, "usd": 4.75}]
+with cf.ThreadPoolExecutor(1) as _outer42:
+    _run42 = _outer42.submit(run, World([card(n, D) for n in (1, 2)]), {
+        "TIME_MODE": "live", "TIME_WORKER": "claude:sonnet", "ANTHROPIC_API_KEY": "test",
+        "TIME_PARALLEL": "2", "TIME_DAILY_USD": "5", "TIME_MONTHLY_USD": "100"})
+    try:
+        assert _entered42.wait(5) and _waiting42.wait(5), "first request in flight and second waiting"
+        assert len(_api_calls42) == 1, _api_calls42
+    finally:
+        _release42.set()
+    w42 = _run42.result(timeout=5)
+assert len(_api_calls42) == 2, _api_calls42
+assert len([b for _, b in w42.posted if "opened PR #" in b]) == 2, w42.posted
+assert not any(tm.beats_of(w42.comments[n])[-1]["beat"] in ("error", "capped", "stuck") for n in (1, 2))
+assert abs(tm._RESERVED_USD) < 1e-8, tm._RESERVED_USD
+# A call that cannot fit EVEN WITHOUT reservations fails before contacting API.
+tm.LEDGER[-1]["usd"] = 0.24
+_before42 = len(_api_calls42)
+try:
+    tm.chat(_body42, 3)
+    raise AssertionError("hard allowance must reject")
+except tm.Budget:
+    pass
+assert len(_api_calls42) == _before42 and tm._RESERVED_USD == 0
+# An API exception releases its reservation, without swallowing the exception.
+tm.LEDGER.clear()
+def _failed_api42(body, timeout):
+    raise RuntimeError("provider failed")
+tm.anthropic_chat = _failed_api42
+try:
+    tm.chat(_body42, 3)
+    raise AssertionError("provider error must propagate")
+except RuntimeError as e:
+    assert str(e) == "provider failed"
+assert tm._RESERVED_USD == 0
+
+tm.anthropic_chat, tm.venice_work = _original_api42, _original_worker42
+tm._BUDGET_CHANGED = _original_condition42
+tm.LEDGER[:] = _original_ledger42
+tm._ledger_files.clear()
+for _key42 in ("TIME_PARALLEL", "TIME_DAILY_CAP", "TIME_DAILY_USD", "TIME_MONTHLY_USD"):
+    os.environ.pop(_key42, None)
+print("ALL OK 41-42 (API overlap, automatic contention recovery, hard cap, failure cleanup)")
+
 # ---------- 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"):
@@ -1517,3 +1641,5 @@ 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
+
+

.github/time/time.py

@@ -19,6 +19,8 @@ from __future__ import annotations
 import datetime as dt
 import importlib.util
 import atexit
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import threading
 import base64
 import json
 import os
@@ -233,10 +235,51 @@ def anthropic_chat(body: dict, timeout: float) -> dict:
 
 
 def chat(body: dict, timeout: float) -> dict:
-    """Every model call goes through here: the budget is checked first and the spend booked after."""
-    budget_check()
-    out = (anthropic_chat if provider_of(body["model"]) == "anthropic" else venice_chat)(body, timeout)
-    book(body["model"], out.get("usage") or {})
+    """Reserve a conservative cost estimate; overlap requests that fit.
+
+    Temporary reservation contention waits for a completing call, rather than
+    stranding a dispatched card as capped. Only actual allowance exhaustion
+    rejects the call. This protects ONE process, not other Actions runs.
+    Provider requests/retries are timeout-bounded; wait releases the lock.
+    """
+    global _RESERVED_USD
+    model = body["model"]
+    price = price_of(model)
+    quote = 0.0
+    if price:
+        payload = json.dumps({"messages": body.get("messages"), "tools": body.get("tools")},
+                             ensure_ascii=False, default=str).encode("utf-8")
+        # Text-only estimate, deliberately pessimistic; not a billing guarantee
+        # for hidden framing, provider retries, or calls with unknown usage.
+        max_output = max(0, int(body.get("max_tokens") or 4000))
+        quote = (len(payload) * price[0] + max_output * price[1]) / 1e6
+    with _BUDGET_CHANGED:
+        while True:
+            budget_check()
+            day, day_cap, month, month_cap = budget()
+            if day + quote > day_cap or month + quote > month_cap:
+                raise Budget("remaining booked-spend allowance cannot cover this model call")
+            if (day + _RESERVED_USD + quote <= day_cap and
+                    month + _RESERVED_USD + quote <= month_cap):
+                _RESERVED_USD += quote
+                break
+            # Other requests reserve capacity, but this one fits after they
+            # settle. A spurious wakeup simply rechecks the current allowance.
+            _BUDGET_CHANGED.wait()
+    try:
+        out = (anthropic_chat if provider_of(model) == "anthropic" else venice_chat)(body, timeout)
+    except BaseException:
+        with _BUDGET_CHANGED:
+            _RESERVED_USD = max(0.0, _RESERVED_USD - quote)
+            _BUDGET_CHANGED.notify_all()
+        raise
+    with _BUDGET_CHANGED:
+        try:
+            book(model, out.get("usage") or {})
+        finally:
+            # Replace reservation with measured spend in ONE locked transition.
+            _RESERVED_USD = max(0.0, _RESERVED_USD - quote)
+            _BUDGET_CHANGED.notify_all()
     return out
 
 
@@ -244,6 +287,9 @@ def chat(body: dict, timeout: float) -> dict:
 # Deterministic on purpose: arithmetic over usage numbers decides, never a model.
 
 LEDGER: list[dict] = []
+_LEDGER_LOCK = threading.RLock()  # cached ledger read may do HTTP; never hold across model/tool work
+_BUDGET_CHANGED = threading.Condition(_LEDGER_LOCK)
+_RESERVED_USD = 0.0  # in-flight quote, replaced atomically by measured spend
 _ledger_files: dict[str, tuple[list[dict], str | None]] = {}  # month -> (lines, sha) as read this process
 
 
@@ -542,7 +588,9 @@ def verdict(pr: dict, since: str) -> str | None:
 
 def touches_protected(pr_number: int) -> list[str]:
     paths = [p.strip() for p in (os.environ.get("TIME_PROTECTED") or DEFAULT_PROTECTED).split(",") if p.strip()]
-    paths = list(dict.fromkeys([".github/", ".gitea/", *paths]))  # workflows stay a human's merge whatever TIME_PROTECTED says
+    # workflows stay a human's merge whatever TIME_PROTECTED says; the rest is TIME_PROTECTED's call: the default keeps all
+    # of .github/, and a repo that trusts its reviewer can narrow it so the engine under .github/time/ lands like other code
+    paths = list(dict.fromkeys([".github/workflows/", ".gitea/workflows/", *paths]))
     files, page = [], 1
     while True:  # every file, not the first 100; a rename counts under both names
         batch = gh("GET", f"/repos/{REPO}/pulls/{pr_number}/files?limit=100&page={page}")
@@ -875,7 +923,8 @@ class Hands:
             note = ("\n" + "; ".join(kept)) if kept else ""
             return f"exit {p.returncode}\n{(p.stdout + p.stderr)[-6000:]}{note}"
         if name == "open_pr":
-            body = f"{a['summary']}\n\nCloses #{self.n}\n\n<sub>worker: {self.label} via Time</sub>"
+            closes = "" if re.search(rf"\bcloses #{self.n}(?!\d)", a["summary"], re.I) else f"\n\nCloses #{self.n}"  # once only
+            body = f"{a['summary']}{closes}\n\n<sub>worker: {self.label} via Time</sub>"
             if (old := open_pr_for(self.branch)):  # a revision: the PR is open already, so say what changed there
                 gh("PATCH", f"/repos/{REPO}/pulls/{old['number']}", {"title": a["title"][:200], "body": body}, as_time=True)
                 self.pr = old["number"]
@@ -1026,7 +1075,7 @@ def review(card: dict, pr: dict, done: str) -> dict:
                   + REVIEW_PROMPT.format(pr=pr["number"], repo=REPO, n=card["number"], title=card["title"], done=done)
                   .replace("Leave one PR comment that", "Your reply is posted as the PR comment. It")
                   + "\nThe checks listed under # Checks are the head commit's real statuses: evidence you can rely on "
-                    "(a success there means those tests or builds passed). A pending check is not yet evidence either way.")
+                    "(a success there means that context reported success). A pending check is not yet evidence either way.")
         text = think(seat, model, effort, system, user)
         gh("POST", f"/repos/{REPO}/issues/{pr['number']}/comments",
            {"body": f"{text}\n\n<sub>review by {seat} · {model or SEATS[seat]['default']} · via Time</sub>"})
@@ -1040,10 +1089,24 @@ def cmd_heartbeat() -> int:
             plan_spec(spec)
         except Exception as e:  # noqa: BLE001
             print(f"spec #{spec['number']}: {type(e).__name__}: {e}")
-    for _ in range(3):  # a merge can free the slot for the next card; dispatch it in the same run
+    worker = (os.environ.get("TIME_WORKER") or "cursor").strip()
+    parallel = max(1, min(16, int(os.environ.get("TIME_PARALLEL") or 1)))
+    if seat_worker(worker) and price_of(worker_seat_model(worker)[1]) and parallel > 1:
+        # Fill slots before invoking the workers, rather than wait for the
+        # first API call to complete before promoting the second card.
+        for _ in range(parallel):
+            if promote_next() is None:
+                break
         tick()
-        if promote_next() is None:
-            break
+        for _ in range(parallel):
+            if promote_next() is None:
+                break
+            tick()
+    else:
+        for _ in range(3):  # Cursor/unpriced/one-worker preserve the historical path
+            tick()
+            if promote_next() is None:
+                break
     return 0
 
 
@@ -1083,6 +1146,12 @@ def tick() -> None:
         plan.append((card, last, m.group(1).strip() if m else None, blocked, capped, revision, revised_at))
     print(f"tick {now:%Y-%m-%dT%H:%MZ} mode={mode} cards={len(cards)} launches_today={launches_today}/{cap}")
 
+    # API workers have isolated branches; only budget and receipt accounting
+    # need a lock. Submit all eligible cards before awaiting any provider.
+    parallel = max(1, min(16, int(os.environ.get("TIME_PARALLEL") or 1)))
+    pool = (ThreadPoolExecutor(max_workers=parallel)
+            if seat_worker(worker) and price_of(worker_seat_model(worker)[1]) and parallel > 1 else None)
+    pending = {}  # future -> (card, capped, provider, revision)
     tripped = None
     for card, last, done, blocked, capped, revision, revised_at in plan:
         n, state = card["number"], (last or {}).get("beat")
@@ -1115,10 +1184,14 @@ def tick() -> None:
                             unlabel(rev["number"], "needs-human")  # the objection is being answered
                         except RuntimeError as e:
                             print(f"#{n}: needs-human label not removed: {str(e)[:100]}")
-                        r = venice_work(card, done, worker.split(":", 1)[1], review=revision["feedback"])
-                    else:
-                        r = venice_work(card, done, worker.split(":", 1)[1])
-                    # back to `dispatched`: the next beat reviews the PR again and the new verdict is the one that counts
+                    args = (card, done, worker.split(":", 1)[1])
+                    kwargs = {"review": revision["feedback"]} if rev else {}
+                    if pool is not None:
+                        pending[pool.submit(venice_work, *args, **kwargs)] = (card, capped, prov, bool(rev))
+                        continue
+                    r = venice_work(*args, **kwargs)
+                    # The next beat reviews the PR. The existing sequential path
+                    # is unchanged for Cursor and one-worker API configurations.
                     comment(n, f"⏱️ Time: `{r['seat']}` " + ("pushed its answer to" if rev else "opened") + f" PR #{r['pr']} "
                             f"({r['steps']} steps, {r['model']}). Review and landing follow on the next beat.",
                             {"beat": "dispatched", "worker": prov, **({"revised": True} if rev else {}), **r})
@@ -1171,6 +1244,29 @@ def tick() -> None:
             print(f"#{n}: {type(e).__name__}: {e}")
             comment(n, f"⏱️ Time: the worker call failed: `{str(e)[:300]}`. Comment `/time retry` to try again.",
                     {"beat": "error"})
+    # Gather by completion, not submission order. A slow seat does not block
+    # starting or finishing a different card.
+    for future in as_completed(pending):
+        card, capped, prov, revised = pending[future]
+        n = card["number"]
+        try:
+            r = future.result()
+            comment(n, f"⏱️ Time: `{r['seat']}` " + ("pushed its answer to" if revised else "opened") + f" PR #{r['pr']} "
+                    f"({r['steps']} steps, {r['model']}). Review and landing follow on the next beat.",
+                    {"beat": "dispatched", "worker": prov, **({"revised": True} if revised else {}), **r})
+        except Budget as e:
+            print(f"#{n}: spend cap reached: {e}")
+            tripped = tripped or str(e)
+            if not capped:
+                comment(n, f"⏱️ Time: {e}. Nothing is lost; this card resumes once there is budget.", {"beat": "capped"})
+        except LookupError as e:
+            comment(n, f"⏱️ Time: can't go further on this card: {e}. I'll carry on once it's added.", {"beat": "blocked"})
+        except Exception as e:  # noqa: BLE001
+            print(f"#{n}: {type(e).__name__}: {e}")
+            comment(n, f"⏱️ Time: the worker call failed: `{str(e)[:300]}`. Comment `/time retry` to try again.",
+                    {"beat": "error"})
+    if pool is not None:
+        pool.shutdown(wait=True)
     if tripped:
         andon(tripped)