diff --git a/.github/time/test_time.py b/.github/time/test_time.py index 91efef4..e145824 100644 --- a/.github/time/test_time.py +++ b/.github/time/test_time.py @@ -1,5 +1,6 @@ """Offline tests for time.py: fake GitHub + fake Cursor, whole card lifecycle. Run: python3 .github/time/test_time.py""" import importlib.util, json, os, sys, datetime as dt +import concurrent.futures as cf os.environ.update(GITHUB_REPOSITORY="o/r", GITHUB_TOKEN="gh") os.environ.pop("GITHUB_STEP_SUMMARY", None) # keep fake doctor output out of the run summary @@ -799,8 +800,8 @@ def http34(method, url, token, body=None, timeout=0): return W.http(method, url, token, body, timeout) vt.tm.http = http34 seen34 = [] -def work34(card, done, spec): - h = vt.tm.Hands(card["number"], "card-7") +def work34(card, done, spec, hands_cls=None): + h = (hands_cls or vt.tm.Hands)(card["number"], "card-7") seen34.append(h.branch) for path in ("index.html", "ventures/mdecks/../../index.html", "/ventures/mdecks/index.html", "ventures/other/index.html", "ventures/mdecks-2/index.html", "ventures/mdecks/index.html"): @@ -1007,7 +1008,7 @@ class FakeHands: def call(self, name, a): calls.append(a.get("path")); return "wrote" real_hands, real_work = tm.Hands, tm.venice_work tm.Hands = FakeHands -tm.venice_work = lambda card, done, spec: {"refusals": [tm.Hands(1, "x").call("write_file", {"path": p, "content": "x"}) +tm.venice_work = lambda card, done, spec, hands_cls=None: {"refusals": [(hands_cls or tm.Hands)(1, "x").call("write_file", {"path": p, "content": "x"}) for p in ("ventures/s/time.js", "ventures/s/./time.js", "ventures/s/app.js")]} # vt._build_on_pages is a scripted Worker here: load a fresh venture.py for the real one vspec2 = importlib.util.spec_from_file_location("vt2", Path(__file__).resolve().parent / "venture.py") @@ -1111,12 +1112,12 @@ assert all(f"time-batch:{b}:" in W.issues[k["card"]]["body"] for k in kids) before = len(W.issues) # a tick that dies after spawning reruns the spawn: no card is made twice picks = [p for p in field if p["slug"] in {k["slug"] for k in kids}] assert vt.spawn(b, picks, 1, []) == kids and len(W.issues) == before -# one worker run per tick: the first child built and passed, and holds at ready; its siblings are planned -stages = [W.beats(k["card"])[-1] for k in kids] -assert sorted(stages) == ["planned", "planned", "ready"], stages -assert json.loads(W.files["ventures/ventures.json"])["ventures"][0]["status"] == "building", "ready is not live yet" -btick(); btick() -assert [W.beats(k["card"])[-1] for k in kids] == ["ready"] * 3 and W.beats(b)[-1] == "spawned" +# three worker runs at once (TIME_VENTURE_HEAVY, 3): every child built and passed in the first tick, and holds at ready +assert [W.beats(k["card"])[-1] for k in kids] == ["ready"] * 3, [W.beats(k["card"]) for k in kids] +assert len(vt._build_on_pages.briefs) == 3 and W.beats(b)[-1] == "spawned" +reg39 = json.loads(W.files["ventures/ventures.json"])["ventures"] +assert sorted(v["slug"] for v in reg39) == ["groka", "kimia", "qwena"], "parallel cards keep each other's registry entries" +assert {v["status"] for v in reg39} == {"building"}, "ready is not live yet" btick() # every child settled: the batch releases and they launch in the same tick assert W.beats(b)[-1] == "launched" and [W.beats(k["card"])[-1] for k in kids] == ["launched"] * 3 assert {v["status"] for v in json.loads(W.files["ventures/ventures.json"])["ventures"]} == {"live"} @@ -1133,16 +1134,16 @@ kids = {k["slug"]: k["card"] for k in kids_of(b)} # the hold is 0h: the first ready child released the batch; stragglers launched alone once they passed assert W.beats(b)[-1] == "launched" and "launch alone" in W.text(b) assert W.beats(kids["qwena"])[-1] == "launched" and W.beats(kids["kimia"])[-1] == "launched" -for _ in range(6): - if "held" in W.beats(kids["groka"]) and W.beats(kids["groka"]).count("held") >= 1: +for _ in range(12): # the andon holds it, the batch (not a human) gives it one more go, the andon holds it again + if W.beats(kids["groka"]).count("held") == 2: break btick({"TIME_VENTURE_BATCH_HOLD_HOURS": "0"}) -assert W.beats(kids["groka"])[-1] == "held", W.beats(kids["groka"]) -btick({"TIME_VENTURE_BATCH_HOLD_HOURS": "0"}) # the batch, not a human, gives it one more go -assert "Batch #1 retry: no child left behind" in W.text(kids["groka"]) and "needs-human" not in W.issues[kids["groka"]]["labels"] -for _ in range(6): +gb = W.beats(kids["groka"]) +assert gb.count("held") == 2 and gb[-1] == "held" and gb[gb.index("held") + 1] == "built", gb +assert "Batch #1 retry: no child left behind" in W.text(kids["groka"]) +for _ in range(2): btick({"TIME_VENTURE_BATCH_HOLD_HOURS": "0"}) -assert W.beats(kids["groka"]).count("held") == 2 and W.beats(kids["groka"])[-1] == "held" +assert W.beats(kids["groka"])[-1] == "held", "dead stays dead: one batch retry, not one per tick" # day 7 (window 0 here): the launched children are measured; the dead one is not waited on; the retro closes the batch btick({"TIME_VENTURE_BATCH_HOLD_HOURS": "0", "TIME_VENTURE_MEASURE_DAYS": "0"}) assert W.beats(kids["qwena"])[-1] == "done" and W.beats(kids["kimia"])[-1] == "done" @@ -1153,6 +1154,78 @@ assert "2 of 3 launched" in retro and "#4" in retro and "after its batch retry" assert W.issues[kids["groka"]]["state"] == "open", "a dead child stays open: a human can still /time retry it" print("ALL OK 39") + +# 40 parallel work: builds overlap in time, each build's folder lock holds under threads, and a contents PUT that loses +# the race for the branch head (409) is retried +import threading as _th, time as _t # noqa: E402 + + +class SlowWorker(Worker): + def __init__(self): + super().__init__() + self.live, self.peak, self.lock = 0, 0, _th.Lock() + + def __call__(self, card, done, spec, root=""): + with self.lock: + self.live += 1 + self.peak = max(self.peak, self.live) + _t.sleep(0.4) + try: + return super().__call__(card, done, spec, root) + finally: + with self.lock: + self.live -= 1 + + +W = BWorld() +ROUNDS.clear() +vt._build_on_pages = SlowWorker() +t0 = _t.monotonic() +btick() +assert vt._build_on_pages.peak == 3, f"three builds at once, saw at most {vt._build_on_pages.peak}" +assert _t.monotonic() - t0 < 1.1, "the three 0.4s builds overlapped" +W = BWorld() +ROUNDS.clear() +vt._build_on_pages = SlowWorker() +btick({"TIME_VENTURE_HEAVY": "1"}) +assert vt._build_on_pages.peak == 1 and len(vt._build_on_pages.briefs) == 1, "TIME_VENTURE_HEAVY=1 is the old one-at-a-time" + +# the folder lock under threads: two builds at once, each refused outside its own folder +seen40 = [] +class FakeHands40: + def __init__(self, n, branch): self.steps = 0 + def call(self, name, a): return "wrote" +def work40(card, done, spec, hands_cls=None): + h = hands_cls(card["number"], "x") + _t.sleep(0.2) # both builds hold their hands at once + seen40.append((card["number"], h.call("write_file", {"path": "ventures/a/i.html"}), h.call("write_file", {"path": "ventures/b/i.html"}))) + return {} +real40 = (vt2.tm.Hands, vt2.tm.venice_work) +vt2.tm.Hands, vt2.tm.venice_work = FakeHands40, work40 +with cf.ThreadPoolExecutor(2) as ex: + list(ex.map(lambda nr: vt2._build_on_pages({"number": nr[0]}, "d", "venice:qwen", root=nr[1]), [(1, "ventures/a/"), (2, "ventures/b/")])) +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 + +# 409 on a contents PUT is retried; other errors are not +tries = [] +def flaky(method, url, token, body=None, timeout=0): + tries.append(url) + if len(tries) < 3: + raise RuntimeError(f"{method} {url} -> 409: is at abc but expected def") + return {"ok": True} +real_http, real_sleep = tm.http, tm.time.sleep +tm.http, tm.time.sleep = flaky, lambda s: None +assert tm.gh("PUT", "/repos/o/r/contents/ventures/a/i.html", {}) == {"ok": True} and len(tries) == 3 +tries.clear() +try: + tm.gh("POST", "/repos/o/r/issues", {}); raise AssertionError("a POST 409 must not be retried") +except RuntimeError: + assert len(tries) == 1 +tm.http, tm.time.sleep = real_http, real_sleep +print("ALL OK 40") + # the worker itself (worker.js), against a fake D1, Workers AI and Pages import shutil as _sh, subprocess as _sp # noqa: E402 if _sh.which("node"): diff --git a/.github/time/time.py b/.github/time/time.py index b8ae849..5925172 100644 --- a/.github/time/time.py +++ b/.github/time/time.py @@ -42,7 +42,15 @@ def http(method: str, url: str, token: str, body=None, timeout: float = 120.0): def gh(method: str, path: str, body=None): - return http(method, f"{GH}{path}", os.environ["GITHUB_TOKEN"], body) + for attempt in range(5): + try: + return http(method, f"{GH}{path}", os.environ["GITHUB_TOKEN"], body) + except RuntimeError as e: + # parallel writers on one branch: a contents PUT loses the race for the branch head with a 409; the file's own + # sha is still right, so the same write goes through on a retry + if method != "PUT" or "/contents/" not in path or "-> 409" not in str(e) or attempt == 4: + raise + time.sleep(0.5 * (attempt + 1)) def comment(issue: int, text: str, meta: dict) -> None: @@ -523,7 +531,7 @@ def worker_seat_model(spec: str) -> tuple[str, str, str | None]: return seat, model or SEATS[seat].get("aliases", {}).get("code") or SEATS[seat]["default"], effort -def venice_work(card: dict, done: str, spec: str) -> dict: +def venice_work(card: dict, done: str, spec: str, hands_cls=None) -> dict: """Run one card with a Venice seat as the worker. Returns {pr, steps, seat, model}. TIME_WORKER_FALLBACK (default `grok build`, `off` to disable) takes over if the worker's model stays overloaded.""" seat, model, effort = worker_seat_model(spec) @@ -532,7 +540,7 @@ def venice_work(card: dict, done: str, spec: str) -> dict: key = os.environ.get("VENICE_API_KEY") if not key: raise LookupError("no VENICE_API_KEY secret on this repo yet") - hands = Hands(card["number"], f"time/card-{card['number']}") + hands = (hands_cls or Hands)(card["number"], f"time/card-{card['number']}") msgs = [{"role": "system", "content": WORKER_RULES.format(n=card["number"], repo=REPO, title=card["title"], body=(card.get("body") or "")[:20000], branch=hands.branch, done=done)}, {"role": "user", "content": "Begin. Look at the repository first."}] diff --git a/.github/time/venture.py b/.github/time/venture.py index a9ffb70..5f84e14 100644 --- a/.github/time/venture.py +++ b/.github/time/venture.py @@ -639,6 +639,18 @@ def upsert(reg: dict, entry: dict) -> None: reg["ventures"].append(entry) +REGISTRY_LOCK = threading.RLock() + + +def record(entry: dict, message: str) -> None: + """Upsert one venture into ventures.json, read fresh under the lock: cards advance in parallel, and a registry read + before a slow step (a plan round, a build) would otherwise write back over a sibling's entry.""" + with REGISTRY_LOCK: + reg = registry() + upsert(reg, entry) + save_registry(reg, message) + + def save_registry(reg: dict, message: str) -> None: put_pages("ventures/ventures.json", json.dumps(reg, indent=1, ensure_ascii=False), message) put_pages("ventures/index.html", index_html(reg), message) @@ -1297,8 +1309,7 @@ def advance(card: dict, comments: list[dict], beats: list[dict], heavy_ok: bool) if w.get("backend"): entry["backend"] = w["backend"] entry.update({"planned": today(), "status": "building", "signups": 0} if version == 1 else {"status": "iterating"}) - upsert(reg, entry) - save_registry(reg, f"ventures: plan {w['slug']}") + record(entry, f"ventures: plan {w['slug']}") say(plan_summary(w, plan, plans, chair) + backend_note, {"beat": "planned", "winner": w, "plan": plan}) elif stage == "planned": if not heavy_ok: @@ -1387,11 +1398,9 @@ def advance(card: dict, comments: list[dict], beats: list[dict], heavy_ok: bool) def launch(n: int, w: dict, plan: dict, keep: dict, shot: str | None, note: str, say) -> None: - reg = registry() since = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - upsert(reg, {"slug": w["slug"], "status": "live", "launched": today(), "card": n, - "signups": count_signups(w, since if int(w.get("version", 1)) > 1 else None)[0]}) - save_registry(reg, f"ventures: launch {w['slug']}") + record({"slug": w["slug"], "status": "live", "launched": today(), "card": n, + "signups": count_signups(w, since if int(w.get("version", 1)) > 1 else None)[0]}, f"ventures: launch {w['slug']}") say(launch_post(w, plan, shot, note), {"beat": "launched", **keep, "shot": shot, "since": since}) @@ -1405,7 +1414,6 @@ def measure(n: int, st: dict, say) -> None: plan = st.get("plan") or normalize_plan({}, w) count, asks = count_signups(w, st.get("since") if int(w.get("version", 1)) > 1 else None) line = f"{count} signup(s) (success at ≥{plan['success_signups']}, kill at ≤{plan['kill_signups']})" - reg = registry() if count >= plan["success_signups"]: nxt = next_version(w, asks, n) status, text = "iterating", (f"📈 {line}. Success: **{w['name']} v{int(w.get('version', 1)) + 1}** is planned on #{nxt}, " @@ -1416,15 +1424,13 @@ def measure(n: int, st: dict, say) -> None: tm.comment(w["signal"], f"Closed: {line} after the measuring window. Thank you to everyone who weighed in.", {"venture": True}) tm.gh("PATCH", f"/repos/{REPO}/issues/{w['signal']}", {"state": "closed"}) elif not st.get("extended"): - upsert(reg, {"slug": w["slug"], "signups": count}) - save_registry(reg, f"ventures: {w['slug']} measured") + record({"slug": w["slug"], "signups": count}, f"ventures: {w['slug']} measured") say(f"⏳ {line}. Between the lines: watching one more week.", {"beat": "launched", "winner": w, "plan": plan, "extended": True, "since": st.get("since")}) return else: status, text = "parked", f"🅿️ {line}. Still between the lines after another week: parked. The signal issue stays open." - upsert(reg, {"slug": w["slug"], "status": status, "signups": count}) - save_registry(reg, f"ventures: {w['slug']} {status}") + record({"slug": w["slug"], "status": status, "signups": count}, f"ventures: {w['slug']} {status}") say(text, {"beat": "done", "winner": w, "signups": count, "outcome": status}) tm.gh("PATCH", f"/repos/{REPO}/issues/{n}", {"state": "closed"}) @@ -1727,7 +1733,7 @@ def tick() -> int: cm = comments_of(b["number"]) if advance_batch(b, cm, tm.beats_of(cm)) != "ok" or stage_of(tm.beats_of(comments_of(b["number"]))) in ("spawned", "done"): break - heavy_left, skip = 1, set() + heavy_left, skip, ran = int(float(env("TIME_VENTURE_HEAVY", "3"))), set(), set() while time.monotonic() - t0 < budget: due = [] cards = venture_cards() @@ -1743,12 +1749,23 @@ def tick() -> int: if not due: break due.sort(key=lambda x: -priority(x[2])) # the card furthest along first; fix rounds yield to first builds - c, cm, b = due[0] - out = advance(c, cm, b, heavy_ok=heavy_left > 0 and time.monotonic() - t0 < heavy_until) - if out == "heavy": - heavy_left -= 1 - elif out in ("wait", "error"): - skip.add(c["number"]) + # one wave: the top cards at once. Worker runs (a build, a fix round) are handed out in priority order, at most + # heavy_left of them; a card that needs one and didn't get one waits for the next tick. + wave = due[:max(1, heavy_left)] + early = time.monotonic() - t0 < heavy_until + grant, given = [], 0 + for c, cm, b in wave: + ok = early and given < heavy_left and stage_of(b) in ("planned", "built") and c["number"] not in ran + given += ok + grant.append(ok) + with cf.ThreadPoolExecutor(max_workers=len(wave)) as ex: + outs = list(ex.map(lambda x: advance(x[0][0], x[0][1], x[0][2], heavy_ok=x[1]), zip(wave, grant))) + for (c, _, _), out in zip(wave, outs): + if out == "heavy": + heavy_left -= 1 + ran.add(c["number"]) # one worker run per card per tick; its cheap stages (the checks) go on + elif out in ("wait", "error"): + skip.add(c["number"]) print(f"venture tick done in {time.monotonic() - t0:.0f}s") return 0 @@ -1762,34 +1779,27 @@ def _build_on_pages(card: dict, done: str, spec: str, root: str = "") -> dict: """venice_work, but the hands work on gh-pages and never open a PR: ventures ship by landing on the Pages branch. With a root (ventures//) the hands write only inside it: the Pages branch also carries the studio's own site and every other venture, and a fix round once wrote its page over the site's front page.""" - hands_cls = tm.Hands - orig_init, orig_call = hands_cls.__init__, hands_cls.call - - def init(self, n, branch): - orig_init(self, n, PAGES_BRANCH) - - def call(self, name, a): - if name == "open_pr": - self.pr = -1 # truthy sentinel: nothing to open, the page is live on the branch - return "done: the page is on the Pages branch, no PR needed" - if root and name == "write_file" and os.path.normpath(str(a.get("path") or "")) == os.path.normpath(root + "time.js"): - self.steps += 1 - return "refused: time.js is Time's backend client and stays as it is; load it, don't write it" - if root and name == "write_file" and not _inside(a.get("path"), root): - self.steps += 1 - return f"refused: this build writes only under {root} (not {a.get('path')!r}); write the file there" - if root and name == "run" and a.get("keep"): - out_of = [k for k in a["keep"] if not _inside(k, root)] - if out_of: - res = orig_call(self, name, {**a, "keep": [k for k in a["keep"] if _inside(k, root)]}) - return f"{res}\nnot kept, outside {root}: {', '.join(map(str, out_of))}" - return orig_call(self, name, a) - hands_cls.__init__, hands_cls.call = init, call - try: - return tm.venice_work(card, done, spec) - finally: - hands_cls.__init__, hands_cls.call = orig_init, orig_call + class PagesHands(tm.Hands): # one class per build: parallel builds each keep their own folder + def __init__(self, n, branch): + super().__init__(n, PAGES_BRANCH) + def call(self, name, a): + if name == "open_pr": + self.pr = -1 # truthy sentinel: nothing to open, the page is live on the branch + return "done: the page is on the Pages branch, no PR needed" + if root and name == "write_file" and os.path.normpath(str(a.get("path") or "")) == os.path.normpath(root + "time.js"): + self.steps += 1 + return "refused: time.js is Time's backend client and stays as it is; load it, don't write it" + if root and name == "write_file" and not _inside(a.get("path"), root): + self.steps += 1 + return f"refused: this build writes only under {root} (not {a.get('path')!r}); write the file there" + if root and name == "run" and a.get("keep"): + out_of = [k for k in a["keep"] if not _inside(k, root)] + if out_of: + res = super().call(name, {**a, "keep": [k for k in a["keep"] if _inside(k, root)]}) + return f"{res}\nnot kept, outside {root}: {', '.join(map(str, out_of))}" + return super().call(name, a) + return tm.venice_work(card, done, spec, hands_cls=PagesHands) if __name__ == "__main__": sys.exit({"tick": tick}[sys.argv[1] if len(sys.argv) > 1 else "tick"]()) diff --git a/.github/workflows/venture.yml b/.github/workflows/venture.yml index 24f0791..869a1df 100644 --- a/.github/workflows/venture.yml +++ b/.github/workflows/venture.yml @@ -47,6 +47,7 @@ jobs: TIME_VENTURE_BATCH_START: ${{ inputs.batch && '1' || '' }} TIME_VENTURE_BATCH: ${{ vars.TIME_VENTURE_BATCH }} # k; set, a batch replaces the daily single card TIME_VENTURE_BATCH_HOLD_HOURS: ${{ vars.TIME_VENTURE_BATCH_HOLD_HOURS }} # launch-together hold (default 6) + TIME_VENTURE_HEAVY: ${{ vars.TIME_VENTURE_HEAVY }} # worker runs at once per tick (default 3) TIME_VENTURE_PITCH: ${{ vars.TIME_VENTURE_PITCH }} TIME_VENTURE_BOARD: ${{ vars.TIME_VENTURE_BOARD }} TIME_VENTURE_CHAIR: ${{ vars.TIME_VENTURE_CHAIR }} diff --git a/CHANGELOG.md b/CHANGELOG.md index e5a5e1a..d32f9fe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,8 @@ ### Added - Venture batches (multi-ceremony): one batch card scouts once, the seats pitch a field (rounds until there are 2k pitches), and the board ranks the whole field; the top k, each wanted by two seats, become venture cards in the batch's own milestone. They run the ordinary loop side by side and launch together: a child that passes its checks waits at `ready` until every sibling is settled or the hold (`TIME_VENTURE_BATCH_HOLD_HOURS`, 6) runs out; stragglers then launch alone. No child left behind: a held child gets one batch retry, and held again it is counted dead with its reason. A retro on the batch card compares the cohort's signups against each plan's lines, then the card and milestone close. Set `TIME_VENTURE_BATCH=3` to make batches the daily default, or Actions → venture → Run workflow → "batch". -- Fix rounds yield the one worker run per tick to cards still due their first build, so a page that keeps failing can't starve the others. +- Parallel builds: a tick runs up to `TIME_VENTURE_HEAVY` (default 3) worker runs at once, one per card, so a batch builds in one tick instead of three. Each build keeps its own folder lock, `ventures.json` is updated under a lock with a fresh read, and a Pages write that loses the race for the branch head (409) is retried. `TIME_VENTURE_HEAVY=1` is the old one-at-a-time. +- Fix rounds yield the worker runs to cards still due their first build, so a page that keeps failing can't starve the others. - Accounts for ventures, with passkeys: each venture's backend can sign people up and in (Face ID, Touch ID, Windows Hello, a phone), keep a 30-day session, and list what each person saved. No passwords, no email, nothing to configure; accounts live in the venture's own database on its own domain. `time.js` gains `Time.signUp`, `Time.signIn`, `Time.signOut`, `Time.user` and `Time.mine`, and builds are told to use them for the paid tier's promise without ever putting sign-in in front of the free product. - Venture backends use Gemma 4 (`@cf/google/gemma-4-26b-a4b-it`) by default and read both the older `{response}` and the OpenAI-style answer from Workers AI (the first default had been retired). - Ventures can be backed by Cloudflare: with `CLOUDFLARE_API_TOKEN` and `CLOUDFLARE_ACCOUNT_ID`, each venture gets its own Worker, D1 database and Workers AI, and lives at `.` (or workers.dev) while its page stays on GitHub Pages. The page's `time.js` gives share links and a small model, and degrades to null so products still work without it. Existing ventures are backfilled a couple per tick; a new worker redeploys every venture; the doctor checks the token.