diff --git a/.github/time/test_time.py b/.github/time/test_time.py
index 75fa50f..1e957d3 100644
--- a/.github/time/test_time.py
+++ b/.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
+
+
diff --git a/.github/time/time.py b/.github/time/time.py
index cf826d9..240861c 100644
--- a/.github/time/time.py
+++ b/.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\nworker: {self.label} via Time"
+ 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\nworker: {self.label} via Time"
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\nreview by {seat} · {model or SEATS[seat]['default']} · via Time"})
@@ -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)