Parallel Claude API workers within heartbeat, with budget reservations #75

Open
teahubxyz wants to merge 9 commits from chatgpt/parallel-api into main
pull from: chatgpt/parallel-api
Showing only changes of commit 15f6e39164 - Show all commits
+57 -29
View File
@@ -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.