From 8c27c537c1a6af7900339d4c4acef1503410bd6b Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sat, 17 Oct 2026 18:39:54 +0000 Subject: [PATCH 1/9] time: prepare in-process concurrent model calls with bounded reservations --- .github/time/time.py | 40 +++++++++++++++++++++++++++++++++++----- 1 file changed, 35 insertions(+), 5 deletions(-) diff --git a/.github/time/time.py b/.github/time/time.py index cf826d9..a500953 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,17 +235,45 @@ 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 {}) - return out + """Reserve worst-case in-flight API spend before concurrent network requests. + + This is a *process-local* reservation: it protects concurrent API calls + within one heartbeat. Independent Actions processes require shared atomic + reservations before cross-run worker fan-out can be enabled. + """ + 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") + # Input token count cannot exceed the number of UTF-8 bytes; quoting + # every byte as a token plus maximum output is deliberately cautious. + max_output = max(0, int(body.get("max_tokens") or 4000)) + quote = (len(payload) * price[0] + max_output * price[1]) / 1e6 + with _LEDGER_LOCK: + budget_check() + day, day_cap, month, month_cap = budget() + if day + _RESERVED_USD + quote > day_cap or month + _RESERVED_USD + quote > month_cap: + raise Budget("not enough unreserved spend for another in-flight model call") + _RESERVED_USD += quote + try: + out = (anthropic_chat if provider_of(model) == "anthropic" else venice_chat)(body, timeout) + with _LEDGER_LOCK: + book(model, out.get("usage") or {}) + return out + finally: + with _LEDGER_LOCK: + _RESERVED_USD -= quote # ---------- spend: a ledger in the repo, and a governor that reads it ---------- # Deterministic on purpose: arithmetic over usage numbers decides, never a model. LEDGER: list[dict] = [] +_LEDGER_LOCK = threading.RLock() # never hold during HTTP or subprocess work +_RESERVED_USD = 0.0 # in-flight quote, replaced by measured spend on completion _ledger_files: dict[str, tuple[list[dict], str | None]] = {} # month -> (lines, sha) as read this process -- 2.54.0 From d16c6acab66fef57a00db55041b3acf9372223ed Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sat, 17 Oct 2026 18:40:15 +0000 Subject: [PATCH 2/9] time: fill API worker slots before waiting for model completion --- .github/time/time.py | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/.github/time/time.py b/.github/time/time.py index a500953..ae1232e 100644 --- a/.github/time/time.py +++ b/.github/time/time.py @@ -1070,10 +1070,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 -- 2.54.0 From 3ac25db0ce715d095111e77239aec87bf18252c2 Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sat, 17 Oct 2026 18:40:48 +0000 Subject: [PATCH 3/9] time: overlap Claude API card workers while preserving receipts and failures --- .github/time/time.py | 41 +++++++++++++++++++++++++++++++++++++---- 1 file changed, 37 insertions(+), 4 deletions(-) diff --git a/.github/time/time.py b/.github/time/time.py index ae1232e..a264301 100644 --- a/.github/time/time.py +++ b/.github/time/time.py @@ -1127,6 +1127,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") @@ -1159,10 +1165,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}) @@ -1215,6 +1225,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) -- 2.54.0 From 9eb0c833a6acf046b53c1de00755be2495702783 Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sat, 17 Oct 2026 18:41:30 +0000 Subject: [PATCH 4/9] tests: prove concurrent Claude calls and atomic in-flight spend reservations --- .github/time/test_time.py | 82 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/.github/time/test_time.py b/.github/time/test_time.py index 75fa50f..145b9c6 100644 --- a/.github/time/test_time.py +++ b/.github/time/test_time.py @@ -1517,3 +1517,85 @@ 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 + + +# 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): + 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 A concurrent call cannot pass a spend check while another reserves the +# remaining budget. The lock MUST be released while Anthropic is waiting. +_original_api42 = tm.anthropic_chat +_original_ledger42 = list(tm.LEDGER) +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}] +_entered42, _release42 = _th41.Event(), _th41.Event() +_api_calls42 = [] +def _wait_for_api42(body, timeout): + _api_calls42.append(body["model"]) + _entered42.set() + assert _release42.wait(5), "other budget claimant should resolve without waiting on provider" + return {"usage": {"prompt_tokens": 10, "completion_tokens": 10}, "choices": []} +tm.anthropic_chat = _wait_for_api42 +tm.http = World([]).http +os.environ.update(TIME_DAILY_USD="5", TIME_MONTHLY_USD="100") +_body42 = {"model": "claude-sonnet-5-5", "max_tokens": 16000, + "messages": [{"role": "user", "content": "bounded task"}]} +with cf.ThreadPoolExecutor(2) as _pool42: + _futures42 = [_pool42.submit(tm.chat, _body42, 3) for _ in range(2)] + assert _entered42.wait(5), "one API call started" + _t41.sleep(0.15) # second task must reach the budget check while first is in flight + _release42.set() + _out42 = [] + for _f42 in _futures42: + try: + _out42.append(_f42.result()) + except tm.Budget: + _out42.append("capped") +assert len(_api_calls42) == 1 and _out42.count("capped") == 1, (_api_calls42, _out42) +assert abs(tm._RESERVED_USD) < 1e-8, tm._RESERVED_USD +tm.anthropic_chat = _original_api42 +tm.LEDGER[:] = _original_ledger42 +tm._ledger_files.clear() +os.environ.pop("TIME_PARALLEL", None) +os.environ.pop("TIME_DAILY_CAP", None) +os.environ.pop("TIME_DAILY_USD", None) +os.environ.pop("TIME_MONTHLY_USD", None) +print("ALL OK 41-42 (overlapping Claude calls, bounded spend)") -- 2.54.0 From 38a7006bea85358f50f1ffd6edb29c6c49f2c41a Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sat, 17 Oct 2026 18:42:52 +0000 Subject: [PATCH 5/9] ci: run API concurrency and spend reservation tests before venture opt-in exit --- .github/time/test_time.py | 161 +++++++++++++++++++------------------- 1 file changed, 81 insertions(+), 80 deletions(-) diff --git a/.github/time/test_time.py b/.github/time/test_time.py index 145b9c6..9a60262 100644 --- a/.github/time/test_time.py +++ b/.github/time/test_time.py @@ -560,6 +560,87 @@ 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): + 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 A concurrent call cannot pass a spend check while another reserves the +# remaining budget. The lock MUST be released while Anthropic is waiting. +_original_api42 = tm.anthropic_chat +_original_ledger42 = list(tm.LEDGER) +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}] +_entered42, _release42 = _th41.Event(), _th41.Event() +_api_calls42 = [] +def _wait_for_api42(body, timeout): + _api_calls42.append(body["model"]) + _entered42.set() + assert _release42.wait(5), "other budget claimant should resolve without waiting on provider" + return {"usage": {"prompt_tokens": 10, "completion_tokens": 10}, "choices": []} +tm.anthropic_chat = _wait_for_api42 +tm.http = World([]).http +os.environ.update(TIME_DAILY_USD="5", TIME_MONTHLY_USD="100") +_body42 = {"model": "claude-sonnet-5-5", "max_tokens": 16000, + "messages": [{"role": "user", "content": "bounded task"}]} +with cf.ThreadPoolExecutor(2) as _pool42: + _futures42 = [_pool42.submit(tm.chat, _body42, 3) for _ in range(2)] + assert _entered42.wait(5), "one API call started" + _t41.sleep(0.15) # second task must reach the budget check while first is in flight + _release42.set() + _out42 = [] + for _f42 in _futures42: + try: + _out42.append(_f42.result()) + except tm.Budget: + _out42.append("capped") +assert len(_api_calls42) == 1 and _out42.count("capped") == 1, (_api_calls42, _out42) +assert abs(tm._RESERVED_USD) < 1e-8, tm._RESERVED_USD +tm.anthropic_chat = _original_api42 +tm.LEDGER[:] = _original_ledger42 +tm._ledger_files.clear() +os.environ.pop("TIME_PARALLEL", None) +os.environ.pop("TIME_DAILY_CAP", None) +os.environ.pop("TIME_DAILY_USD", None) +os.environ.pop("TIME_MONTHLY_USD", None) +print("ALL OK 41-42 (overlapping Claude calls, bounded spend)") + # ---------- 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"): @@ -1519,83 +1600,3 @@ 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 -# 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): - 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 A concurrent call cannot pass a spend check while another reserves the -# remaining budget. The lock MUST be released while Anthropic is waiting. -_original_api42 = tm.anthropic_chat -_original_ledger42 = list(tm.LEDGER) -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}] -_entered42, _release42 = _th41.Event(), _th41.Event() -_api_calls42 = [] -def _wait_for_api42(body, timeout): - _api_calls42.append(body["model"]) - _entered42.set() - assert _release42.wait(5), "other budget claimant should resolve without waiting on provider" - return {"usage": {"prompt_tokens": 10, "completion_tokens": 10}, "choices": []} -tm.anthropic_chat = _wait_for_api42 -tm.http = World([]).http -os.environ.update(TIME_DAILY_USD="5", TIME_MONTHLY_USD="100") -_body42 = {"model": "claude-sonnet-5-5", "max_tokens": 16000, - "messages": [{"role": "user", "content": "bounded task"}]} -with cf.ThreadPoolExecutor(2) as _pool42: - _futures42 = [_pool42.submit(tm.chat, _body42, 3) for _ in range(2)] - assert _entered42.wait(5), "one API call started" - _t41.sleep(0.15) # second task must reach the budget check while first is in flight - _release42.set() - _out42 = [] - for _f42 in _futures42: - try: - _out42.append(_f42.result()) - except tm.Budget: - _out42.append("capped") -assert len(_api_calls42) == 1 and _out42.count("capped") == 1, (_api_calls42, _out42) -assert abs(tm._RESERVED_USD) < 1e-8, tm._RESERVED_USD -tm.anthropic_chat = _original_api42 -tm.LEDGER[:] = _original_ledger42 -tm._ledger_files.clear() -os.environ.pop("TIME_PARALLEL", None) -os.environ.pop("TIME_DAILY_CAP", None) -os.environ.pop("TIME_DAILY_USD", None) -os.environ.pop("TIME_MONTHLY_USD", None) -print("ALL OK 41-42 (overlapping Claude calls, bounded spend)") -- 2.54.0 From 943e084482b229e714ed38567f1b2b0438af55bd Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sun, 18 Oct 2026 00:57:50 +0000 Subject: [PATCH 6/9] PR75: reconcile landed engine changes and clarify reservation semantics --- .github/time/time.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/.github/time/time.py b/.github/time/time.py index a264301..99ce1f9 100644 --- a/.github/time/time.py +++ b/.github/time/time.py @@ -235,7 +235,7 @@ def anthropic_chat(body: dict, timeout: float) -> dict: def chat(body: dict, timeout: float) -> dict: - """Reserve worst-case in-flight API spend before concurrent network requests. + """Reserve a conservative in-flight API spend estimate before concurrent requests. This is a *process-local* reservation: it protects concurrent API calls within one heartbeat. Independent Actions processes require shared atomic @@ -248,8 +248,9 @@ def chat(body: dict, timeout: float) -> dict: if price: payload = json.dumps({"messages": body.get("messages"), "tools": body.get("tools")}, ensure_ascii=False, default=str).encode("utf-8") - # Input token count cannot exceed the number of UTF-8 bytes; quoting - # every byte as a token plus maximum output is deliberately cautious. + # Quote UTF-8 bytes as text tokens plus the requested output limit. + # Deliberately pessimistic for this text-only lane; not a provider + # billing guarantee (hidden framing, retries and unknown usage vary). max_output = max(0, int(body.get("max_tokens") or 4000)) quote = (len(payload) * price[0] + max_output * price[1]) / 1e6 with _LEDGER_LOCK: @@ -272,7 +273,7 @@ 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() # never hold during HTTP or subprocess work +_LEDGER_LOCK = threading.RLock() # cached ledger read may do HTTP; never hold across model/tool work _RESERVED_USD = 0.0 # in-flight quote, replaced by measured spend on completion _ledger_files: dict[str, tuple[list[dict], str | None]] = {} # month -> (lines, sha) as read this process @@ -572,7 +573,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}") @@ -905,7 +908,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"] @@ -1056,7 +1060,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"}) -- 2.54.0 From 8edea322793bed92f761e7b7b3df6eb104d75605 Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sun, 18 Oct 2026 00:58:03 +0000 Subject: [PATCH 7/9] PR75: retain main regression tests and forward-compatible worker stub --- .github/time/test_time.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/.github/time/test_time.py b/.github/time/test_time.py index 9a60262..4efbe39 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"}) @@ -567,7 +582,7 @@ _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): +def _overlap41(card, done, spec, hands_cls=None, review=None, **kwargs): global _peak41, _active41 with _lock41: _active41 += 1 -- 2.54.0 From 175049768c203e59de6e0f2a5204c1ddfe3e2de5 Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sun, 18 Oct 2026 01:01:03 +0000 Subject: [PATCH 8/9] PR75: wait for temporary budget contention and settle reservations atomically --- .github/time/time.py | 53 ++++++++++++++++++++++++++++---------------- 1 file changed, 34 insertions(+), 19 deletions(-) diff --git a/.github/time/time.py b/.github/time/time.py index 99ce1f9..240861c 100644 --- a/.github/time/time.py +++ b/.github/time/time.py @@ -235,11 +235,12 @@ def anthropic_chat(body: dict, timeout: float) -> dict: def chat(body: dict, timeout: float) -> dict: - """Reserve a conservative in-flight API spend estimate before concurrent requests. + """Reserve a conservative cost estimate; overlap requests that fit. - This is a *process-local* reservation: it protects concurrent API calls - within one heartbeat. Independent Actions processes require shared atomic - reservations before cross-run worker fan-out can be enabled. + 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"] @@ -248,25 +249,38 @@ def chat(body: dict, timeout: float) -> dict: if price: payload = json.dumps({"messages": body.get("messages"), "tools": body.get("tools")}, ensure_ascii=False, default=str).encode("utf-8") - # Quote UTF-8 bytes as text tokens plus the requested output limit. - # Deliberately pessimistic for this text-only lane; not a provider - # billing guarantee (hidden framing, retries and unknown usage vary). + # 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 _LEDGER_LOCK: - budget_check() - day, day_cap, month, month_cap = budget() - if day + _RESERVED_USD + quote > day_cap or month + _RESERVED_USD + quote > month_cap: - raise Budget("not enough unreserved spend for another in-flight model call") - _RESERVED_USD += quote + 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) - with _LEDGER_LOCK: + 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 {}) - return out - finally: - with _LEDGER_LOCK: - _RESERVED_USD -= quote + finally: + # Replace reservation with measured spend in ONE locked transition. + _RESERVED_USD = max(0.0, _RESERVED_USD - quote) + _BUDGET_CHANGED.notify_all() + return out # ---------- spend: a ledger in the repo, and a governor that reads it ---------- @@ -274,7 +288,8 @@ def chat(body: dict, timeout: float) -> dict: LEDGER: list[dict] = [] _LEDGER_LOCK = threading.RLock() # cached ledger read may do HTTP; never hold across model/tool work -_RESERVED_USD = 0.0 # in-flight quote, replaced by measured spend on completion +_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 -- 2.54.0 From 15f6e3916432bfbee747e8e01914e858aec1fc5d Mon Sep 17 00:00:00 2001 From: teahubxyz Date: Sun, 18 Oct 2026 01:01:22 +0000 Subject: [PATCH 9/9] PR75: test automatic reservation recovery deterministically through heartbeat --- .github/time/test_time.py | 86 ++++++++++++++++++++++++++------------- 1 file changed, 57 insertions(+), 29 deletions(-) diff --git a/.github/time/test_time.py b/.github/time/test_time.py index 4efbe39..1e957d3 100644 --- a/.github/time/test_time.py +++ b/.github/time/test_time.py @@ -615,46 +615,74 @@ w41 = run(World([card(n, D) for n in (1, 2, 3)]), { 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 A concurrent call cannot pass a spend check while another reserves the -# remaining budget. The lock MUST be released while Anthropic is waiting. -_original_api42 = tm.anthropic_chat +# 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) -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}] -_entered42, _release42 = _th41.Event(), _th41.Event() +_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), "other budget claimant should resolve without waiting on provider" + assert _release42.wait(5), "provider release timed out" return {"usage": {"prompt_tokens": 10, "completion_tokens": 10}, "choices": []} -tm.anthropic_chat = _wait_for_api42 -tm.http = World([]).http -os.environ.update(TIME_DAILY_USD="5", TIME_MONTHLY_USD="100") _body42 = {"model": "claude-sonnet-5-5", "max_tokens": 16000, "messages": [{"role": "user", "content": "bounded task"}]} -with cf.ThreadPoolExecutor(2) as _pool42: - _futures42 = [_pool42.submit(tm.chat, _body42, 3) for _ in range(2)] - assert _entered42.wait(5), "one API call started" - _t41.sleep(0.15) # second task must reach the budget check while first is in flight - _release42.set() - _out42 = [] - for _f42 in _futures42: - try: - _out42.append(_f42.result()) - except tm.Budget: - _out42.append("capped") -assert len(_api_calls42) == 1 and _out42.count("capped") == 1, (_api_calls42, _out42) +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 -tm.anthropic_chat = _original_api42 +# 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() -os.environ.pop("TIME_PARALLEL", None) -os.environ.pop("TIME_DAILY_CAP", None) -os.environ.pop("TIME_DAILY_USD", None) -os.environ.pop("TIME_MONTHLY_USD", None) -print("ALL OK 41-42 (overlapping Claude calls, bounded spend)") +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. -- 2.54.0