From a6425e0b93182e057556e7aa696117776a644c31 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:52:31 -0700 Subject: [PATCH 01/13] docs: design draft for one job loop shared by generate and judge Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 87 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 docs/design/job-loop.md diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md new file mode 100644 index 00000000..32a6d509 --- /dev/null +++ b/docs/design/job-loop.md @@ -0,0 +1,87 @@ +# One job loop for generation and judging + +**Status:** draft for discussion, not committed. Lands with the first PR that builds it. + +## Problem + +Generation and judging each hand-roll the same asyncio queue-and-workers pool, and the two copies have drifted: + +| | Generate (`generate/runner.py`) | Judge (`judge/runner.py`) | +|---|---|---| +| Job | 4-tuple `(dict, max_turns, idx, run)` | 7-tuple, with output folder, rubric, and params repeated in every job | +| Resume | Parses transcript filenames back into persona/run/session, checked twice (per job and per session) | Checks the TSV basename in the output folder | +| `max_concurrent=0` | No limit | Was zero workers (fixed in #261) | +| Failures | Three layers of `except`; recorded as skipped results | One `except` in the worker; failure not recorded, worked out afterwards by subtraction | +| Summary | One place | Printed in three places | + +On top of that, each stage has 4–5 layers that mostly forward arguments: `vera_cli/._execute` → `run_*` → runner `__init__` / `judge_conversations` → `batch_evaluate…` → `_run_workers_with_queue` → `_worker`. Many of those layers exist for `legacy/`. + +## Proposal + +One function holds all the concurrency code in the repo: + +```python +# workers/loop.py +async def run_jobs(jobs: list[Job], max_concurrent: int | None) -> list[Outcome]: + todo = [job for job in jobs if not job.done()] # resume = skip what's on disk + limit = asyncio.Semaphore(max_concurrent or len(todo) or 1) + + async def run_one(job: Job) -> Outcome: + async with limit: + try: + await job.run() + return Outcome.ok(job) + except Exception as error: + return Outcome.failed(job, error) + + finished = await asyncio.gather(*(run_one(job) for job in todo)) + return [Outcome.skipped(job) for job in jobs if job not in todo] + finished +``` + +A job is anything with a label, a way to tell whether its output already exists, and a way to produce it: + +```python +class Job(Protocol): + label: str + def done(self) -> bool: ... + async def run(self) -> None: ... +``` + +Each stage then reads top to bottom: + +```python +# generate +jobs = [Conversation(persona, run, run_dir) for persona in personas for run in runs] +report(await run_jobs(jobs, max_concurrent)) + +# judge +jobs = [Evaluation(conversation, judge, instance, run_dir) + for conversation in conversations + for judge in judges + for instance in judge.instances] +report(await run_jobs(jobs, max_concurrent)) +write_results_csv(run_dir) +``` + +`Conversation.run()` is today's `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. A semaphore plus `gather` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. + +## What this removes + +Once `legacy/` is gone (it needs a tagged release that announces the removal; roadmap, Deferred): + +- `run_for_user_models` and `_legacy_model_config`; `run_generation`'s always-constant parameters (`persona_names`, `run_id`, `verbose`, the list-of-one `persona_files`) +- The `verbose_workers` branches, `judge_single_conversation`, and the legacy helpers in `judge/utils.py` +- `_run_workers_with_queue`, both `_worker`s, the dead `_create_conversation_jobs`, and the three judge summaries +- The CLI parity tests and most of `test_pipeline.py` + +## Open questions + +1. **Resume granularity.** Multi-session generation skips per session, inside the job. Should a job be one session instead of one persona × run, so `done()` stays a single file check? +2. **`per_judge` pools.** Keep one limit per judge model, or replace it with a per-provider concurrency cap in `llm_clients/` (roadmap, Shared engine)? +3. **Progress.** `report` prints one flushed line as each job starts and finishes. That also fixes the backlog item about runs that look hung under redirection. Enough, or do we want turns-completed while a conversation runs? +4. **Fail fast.** Should `run_jobs` (or the caller) raise when every job failed? This is the backlog item where a generation stage with zero conversations only fails one stage later, in judging. +5. **Retries.** Keep retries per LLM call only (`llm_clients`), or also retry a failed job once? + +## Testing + +The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, concurrency cap holds, failures recorded and not raised, `max_concurrent` of `None` and `0` both unlimited. From 48af215bf0f10bc46ac83df2924105c656f481e0 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:35:45 -0700 Subject: [PATCH 02/13] =?UTF-8?q?docs:=20job-loop=20design=20=E2=80=94=20a?= =?UTF-8?q?nswer=20review:=20layers,=20sessions,=20retries,=20follow-ups?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 90 ++++++++++++++++++++++++++++------------- 1 file changed, 61 insertions(+), 29 deletions(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 32a6d509..adbd05dc 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -1,6 +1,6 @@ # One job loop for generation and judging -**Status:** draft for discussion, not committed. Lands with the first PR that builds it. +**Status:** draft for discussion. Lands with the first PR that builds it. ## Problem @@ -16,54 +16,94 @@ Generation and judging each hand-roll the same asyncio queue-and-workers pool, a On top of that, each stage has 4–5 layers that mostly forward arguments: `vera_cli/._execute` → `run_*` → runner `__init__` / `judge_conversations` → `batch_evaluate…` → `_run_workers_with_queue` → `_worker`. Many of those layers exist for `legacy/`. +## Who does what + +``` +vera_cli/generate.py resolve and validate settings → RunConfig (no jobs, no I/O beyond reading inputs) + │ + ▼ +generate/run.py run_generation(config): list the jobs, hand them to the loop, report +generate/conversation.py the Session job: what generating one session means + │ + ▼ +workers/loop.py run_jobs(...): skip done jobs, run the rest with a concurrency cap, collect outcomes +``` + +Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging`) → `judge/evaluation.py` (the `Evaluation` job) → `workers/loop.py`. The CLI never calls the loop directly; it resolves everything up front and calls one domain entry point, which keeps the layer rule in [architecture.md](../architecture.md) (`vera_cli` → domain → `workers/`). + ## Proposal -One function holds all the concurrency code in the repo: +### The loop + +All the concurrency code in the repo, in one function: ```python # workers/loop.py -async def run_jobs(jobs: list[Job], max_concurrent: int | None) -> list[Outcome]: +async def run_jobs(jobs: list[Job], max_concurrent: int | None, attempts: int = 1) -> list[Outcome]: todo = [job for job in jobs if not job.done()] # resume = skip what's on disk limit = asyncio.Semaphore(max_concurrent or len(todo) or 1) - async def run_one(job: Job) -> Outcome: + async def run_one(job: Job) -> list[Outcome]: async with limit: - try: - await job.run() - return Outcome.ok(job) - except Exception as error: - return Outcome.failed(job, error) + outcome = await run_with_retries(job, attempts) + # A finished job may hand back the work that depended on it. + follow_ups = outcome.follow_ups if outcome.ok else [] + return [outcome, *await gather_flat(run_one(next_job) for next_job in follow_ups)] - finished = await asyncio.gather(*(run_one(job) for job in todo)) + finished = await gather_flat(run_one(job) for job in todo) return [Outcome.skipped(job) for job in jobs if job not in todo] + finished + + +async def run_with_retries(job: Job, attempts: int) -> Outcome: + for attempt in range(1, attempts + 1): + try: + return Outcome.ok(job, follow_ups=await job.run()) + except Exception as error: + if attempt == attempts or not job.retryable(error): + return Outcome.failed(job, error) ``` -A job is anything with a label, a way to tell whether its output already exists, and a way to produce it: +A job is plain data plus two methods: ```python class Job(Protocol): label: str - def done(self) -> bool: ... - async def run(self) -> None: ... + def done(self) -> bool: ... # is my output already on disk? + async def run(self) -> list["Job"]: ... # do the work; return jobs that were waiting on me + def retryable(self, error: Exception) -> bool: ... ``` -Each stage then reads top to bottom: +### Each stage ```python -# generate -jobs = [Conversation(persona, run, run_dir) for persona in personas for run in runs] -report(await run_jobs(jobs, max_concurrent)) +# generate/run.py +jobs = [first_unfinished_session(persona, repeat, run_dir) + for persona in personas for repeat in repeats] +report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) -# judge +# judge/run.py jobs = [Evaluation(conversation, judge, instance, run_dir) for conversation in conversations for judge in judges for instance in judge.instances] -report(await run_jobs(jobs, max_concurrent)) +report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) write_results_csv(run_dir) ``` -`Conversation.run()` is today's `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. A semaphore plus `gather` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. +`Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. A semaphore plus `gather` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. + +## Decisions from review + +1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*: its `run()` returns the next session's job. Generation lists only the first unfinished session of each conversation. `done()` stays a single file check, and resume restarts exactly at the first missing session. +2. **Retries live in the loop, the policy in the stage.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job type says which errors are worth retrying (`retryable`); the stage picks the number of attempts. +3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one global cap, and a per-model (or per-provider) cap comes with the Shared engine phase. +4. **Progress display is a separate PR.** The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. +5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. + +## Open questions + +1. **Stream generation into judging?** Today a pipeline judges only after every conversation is generated. With follow-ups, a finished session could return its `Evaluation` jobs, so judging starts while generation continues. Gains: wall-clock time, and a broken judge setup fails in minutes instead of after generation. Costs: generation and judging compete for the concurrency cap, and the judge needs its run folder and rubric ready before generation starts. Proposal: keep the stage barrier in the first version; the follow-up mechanism makes streaming a later, local change. +2. **External queue.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. Proposal: don't build the interface until there's a concrete need, same rule as the roadmap's Storage phase; just keep jobs serializable. ## What this removes @@ -74,14 +114,6 @@ Once `legacy/` is gone (it needs a tagged release that announces the removal; ro - `_run_workers_with_queue`, both `_worker`s, the dead `_create_conversation_jobs`, and the three judge summaries - The CLI parity tests and most of `test_pipeline.py` -## Open questions - -1. **Resume granularity.** Multi-session generation skips per session, inside the job. Should a job be one session instead of one persona × run, so `done()` stays a single file check? -2. **`per_judge` pools.** Keep one limit per judge model, or replace it with a per-provider concurrency cap in `llm_clients/` (roadmap, Shared engine)? -3. **Progress.** `report` prints one flushed line as each job starts and finishes. That also fixes the backlog item about runs that look hung under redirection. Enough, or do we want turns-completed while a conversation runs? -4. **Fail fast.** Should `run_jobs` (or the caller) raise when every job failed? This is the backlog item where a generation stage with zero conversations only fails one stage later, in judging. -5. **Retries.** Keep retries per LLM call only (`llm_clients`), or also retry a failed job once? - ## Testing -The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, concurrency cap holds, failures recorded and not raised, `max_concurrent` of `None` and `0` both unlimited. +The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, concurrency cap holds, failures recorded and not raised, retries stop at `attempts` and skip non-retryable errors, follow-ups run after their parent and not when it fails, `max_concurrent` of `None` and `0` both unlimited. From 48c189c4139a749f3dc8525bfcbb9310d9b74fbd Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:29:00 -0700 Subject: [PATCH 03/13] =?UTF-8?q?docs:=20job-loop=20=E2=80=94=20why=20jobs?= =?UTF-8?q?=20are=20plain=20data=20rather=20than=20closures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index adbd05dc..bd1da1d4 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -103,7 +103,7 @@ write_results_csv(run_dir) ## Open questions 1. **Stream generation into judging?** Today a pipeline judges only after every conversation is generated. With follow-ups, a finished session could return its `Evaluation` jobs, so judging starts while generation continues. Gains: wall-clock time, and a broken judge setup fails in minutes instead of after generation. Costs: generation and judging compete for the concurrency cap, and the judge needs its run folder and rubric ready before generation starts. Proposal: keep the stage barrier in the first version; the follow-up mechanism makes streaming a later, local change. -2. **External queue.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. Proposal: don't build the interface until there's a concrete need, same rule as the roadmap's Storage phase; just keep jobs serializable. +2. **External queue.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. Proposal: don't build the interface until there's a concrete need, same rule as the roadmap's Storage phase; just keep jobs serializable. ## What this removes From f1687951298de9bf648e5ab82e01a859ad61eadf Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 16 Oct 2026 16:30:21 -0700 Subject: [PATCH 04/13] =?UTF-8?q?docs:=20job-loop=20=E2=80=94=20the=20pipe?= =?UTF-8?q?line=20as=20one=20job=20list,=20judging=20as=20follow-ups?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index bd1da1d4..71f83583 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -92,6 +92,25 @@ write_results_csv(run_dir) `Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. A semaphore plus `gather` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. +### The pipeline: one job list + +Today `vera pipeline` runs a generation pool for one user model, then a judging pool for its run, then the next user model, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one call: + +```python +# vera_cli/pipeline.py, after creating every run root up front (run-lifecycle.md, step 5) +jobs = [first_unfinished_session(persona, repeat, run_dir_for(user_model)) + for user_model in user_models for persona in personas for repeat in repeats] +report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) +for run_dir in run_dirs: + write_results_csv(evaluation_dir_for(run_dir)) + score(evaluation_dir_for(run_dir)) +pool(run_dirs) +``` + +A conversation's last session returns that conversation's `Evaluation` jobs as follow-ups, so it is judged as soon as it exists, while other conversations are still being generated. Every user model shares one concurrency cap, and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. + +What this needs: the judge's run folder and rubric must exist before generation starts, which the run-lifecycle order already gives (all run roots are created before any job runs). + ## Decisions from review 1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*: its `run()` returns the next session's job. Generation lists only the first unfinished session of each conversation. `done()` stays a single file check, and resume restarts exactly at the first missing session. @@ -102,7 +121,7 @@ write_results_csv(run_dir) ## Open questions -1. **Stream generation into judging?** Today a pipeline judges only after every conversation is generated. With follow-ups, a finished session could return its `Evaluation` jobs, so judging starts while generation continues. Gains: wall-clock time, and a broken judge setup fails in minutes instead of after generation. Costs: generation and judging compete for the concurrency cap, and the judge needs its run folder and rubric ready before generation starts. Proposal: keep the stage barrier in the first version; the follow-up mechanism makes streaming a later, local change. +1. **One cap or two?** In the pipeline, generation and judging jobs share one concurrency cap. Judging calls are short and many, generation calls long and few, so one cap can let a burst of evaluations delay generation. Start with one cap and split it per job kind only if runs show the problem. 2. **External queue.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. Proposal: don't build the interface until there's a concrete need, same rule as the roadmap's Storage phase; just keep jobs serializable. ## What this removes From 152b52bca72df8d0c5574ee55a3e3a8fa75c68ce Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:26:44 -0700 Subject: [PATCH 05/13] =?UTF-8?q?docs:=20job-loop=20design=20=E2=80=94=20w?= =?UTF-8?q?hat=20happens=20to=20unfinished=20judging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 71f83583..4b1412d2 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -118,6 +118,7 @@ What this needs: the judge's run folder and rubric must exist before generation 3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one global cap, and a per-model (or per-provider) cap comes with the Shared engine phase. 4. **Progress display is a separate PR.** The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. 5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. + - **Unfinished judging works the same way.** An `Evaluation` is done when its TSV exists, and the TSV is written only after the judge has answered every question it needed. A judging stopped partway through a conversation leaves no TSV, so rerunning with `--into` judges that conversation again from its first question; conversations already judged are skipped. `results.csv` and the scores are rebuilt from the TSVs after the loop, so they always reflect exactly the finished evaluations. Saving progress question by question isn't worth it: one conversation is a few dozen short judge calls. ## Open questions From 8604717f5a2e95ff02b15fca035d322eabb29c84 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:39:14 -0700 Subject: [PATCH 06/13] docs: job-loop design settles the cap and external-queue questions One cap per job kind, from the two max_concurrent settings the config already has; jobs stay serializable, enforced by a JSON round-trip test. Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 4b1412d2..eaa115b5 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -39,12 +39,15 @@ All the concurrency code in the repo, in one function: ```python # workers/loop.py -async def run_jobs(jobs: list[Job], max_concurrent: int | None, attempts: int = 1) -> list[Outcome]: +async def run_jobs( + jobs: list[Job], max_concurrent: dict[str, int | None], attempts: int = 1 +) -> list[Outcome]: todo = [job for job in jobs if not job.done()] # resume = skip what's on disk - limit = asyncio.Semaphore(max_concurrent or len(todo) or 1) + # One cap per job kind (decision 6); None or 0 means no limit. + limits = {kind: asyncio.Semaphore(cap or len(todo) or 1) for kind, cap in max_concurrent.items()} async def run_one(job: Job) -> list[Outcome]: - async with limit: + async with limits[job.kind]: outcome = await run_with_retries(job, attempts) # A finished job may hand back the work that depended on it. follow_ups = outcome.follow_ups if outcome.ok else [] @@ -67,6 +70,7 @@ A job is plain data plus two methods: ```python class Job(Protocol): + kind: str # "session" or "evaluation": picks the cap (and the code a remote worker would run) label: str def done(self) -> bool: ... # is my output already on disk? async def run(self) -> list["Job"]: ... # do the work; return jobs that were waiting on me @@ -107,7 +111,7 @@ for run_dir in run_dirs: pool(run_dirs) ``` -A conversation's last session returns that conversation's `Evaluation` jobs as follow-ups, so it is judged as soon as it exists, while other conversations are still being generated. Every user model shares one concurrency cap, and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. +A conversation's last session returns that conversation's `Evaluation` jobs as follow-ups, so it is judged as soon as it exists, while other conversations are still being generated. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. What this needs: the judge's run folder and rubric must exist before generation starts, which the run-lifecycle order already gives (all run roots are created before any job runs). @@ -115,15 +119,12 @@ What this needs: the judge's run folder and rubric must exist before generation 1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*: its `run()` returns the next session's job. Generation lists only the first unfinished session of each conversation. `done()` stays a single file check, and resume restarts exactly at the first missing session. 2. **Retries live in the loop, the policy in the stage.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job type says which errors are worth retrying (`retryable`); the stage picks the number of attempts. -3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one global cap, and a per-model (or per-provider) cap comes with the Shared engine phase. -4. **Progress display is a separate PR.** The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. +3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one cap per job kind (decision 6), and a per-model (or per-provider) cap comes with the Shared engine phase. +4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. 5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. - **Unfinished judging works the same way.** An `Evaluation` is done when its TSV exists, and the TSV is written only after the judge has answered every question it needed. A judging stopped partway through a conversation leaves no TSV, so rerunning with `--into` judges that conversation again from its first question; conversations already judged are skipped. `results.csv` and the scores are rebuilt from the TSVs after the loop, so they always reflect exactly the finished evaluations. Saving progress question by question isn't worth it: one conversation is a few dozen short judge calls. - -## Open questions - -1. **One cap or two?** In the pipeline, generation and judging jobs share one concurrency cap. Judging calls are short and many, generation calls long and few, so one cap can let a burst of evaluations delay generation. Start with one cap and split it per job kind only if runs show the problem. -2. **External queue.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. Proposal: don't build the interface until there's a concrete need, same rule as the roadmap's Storage phase; just keep jobs serializable. +6. **One cap per job kind.** The config already sets two: `generation.max_concurrent` caps `Session` jobs and `judging.max_concurrent` caps `Evaluation` jobs, so in the pipeline a burst of short judge calls can't crowd out the long generation calls. `run_jobs` keeps one semaphore per job kind instead of one in total. The per-judge-model cap (`per_judge`) still comes later (decision 3). +7. **Jobs stay serializable, so an external queue stays possible.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. The interface isn't built until there's a concrete need, same rule as the roadmap's Storage phase. What keeps it possible is enforced, not assumed: each job type is a dataclass of JSON-safe fields plus `kind`, and a unit test round-trips every job type through JSON and runs the copy. ## What this removes @@ -136,4 +137,4 @@ Once `legacy/` is gone (it needs a tagged release that announces the removal; ro ## Testing -The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, concurrency cap holds, failures recorded and not raised, retries stop at `attempts` and skip non-retryable errors, follow-ups run after their parent and not when it fails, `max_concurrent` of `None` and `0` both unlimited. +The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised, retries stop at `attempts` and skip non-retryable errors, follow-ups run after their parent and not when it fails, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). From 77a7eb8b7f79e0bc1d5f4fba99ea4079a160d469 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:51:12 -0700 Subject: [PATCH 07/13] docs: job-loop design adds vera pipeline --into Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index eaa115b5..3f0ef014 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -115,6 +115,8 @@ A conversation's last session returns that conversation's `Evaluation` jobs as f What this needs: the judge's run folder and rubric must exist before generation starts, which the run-lifecycle order already gives (all run roots are created before any job runs). +It also gives the pipeline the resume that `generate` and `judge` have. `vera pipeline` takes no `--into` today. With one job list fixed by the run's `config.json`, `vera pipeline --into ` reopens the run roots, skips every job that's done (decision 5), and then runs scoring and pooling as usual. The end summary and the Ctrl-C message of the terminal interface ([#271](https://github.com/SpringCare/VERA-MH/pull/271)) print this command. + ## Decisions from review 1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*: its `run()` returns the next session's job. Generation lists only the first unfinished session of each conversation. `done()` stays a single file check, and resume restarts exactly at the first missing session. From 410a58454fce64969584c000aa7a3f91f4927340 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:20:10 -0700 Subject: [PATCH 08/13] docs: job-loop design review fixes, and a CHANGELOG entry - Follow-ups come from follow_ups(), read from a job's fields and disk, so a skipped job hands them on too; each is checked for done(). Resume then reaches evaluations of fully generated conversations, and redoes no completed later session. - Kinds with no cap get no semaphore, so follow-ups aren't throttled. - A session saves the chatbot's conversation_id and the next session creates its client with it, so server-side history survives job boundaries, retries, and resume. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + docs/design/job-loop.md | 43 ++++++++++++++++++++++++++--------------- 2 files changed, 28 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 732a9fd5..155e7476 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Documentation and development workflow +- **Job-loop proposal**—[`docs/design/job-loop.md`](docs/design/job-loop.md) proposes one job loop for generation and judging: jobs as plain data, resume by skipping what's on disk, retries, one concurrency cap per job kind, and the pipeline as a single job list with `vera pipeline --into`. This is a draft design, not an implemented change. - **Release 2.1 plan**—[`docs/roadmap.md`](docs/roadmap.md#release-21-plan) lists the steps for 2.1 in dependency order with their PRs: safety net, data contracts, job loop, terminal output, then removing `legacy/`, with the judge-flow rewrite in parallel. They accumulate on a `release/2.1` branch that merges into `main` once, at the 2.1.0 tag, so `main` stays releasable; CI now also runs for `release/2.1`. ## [v2.0.4](https://github.com/SpringCare/VERA-MH/releases/tag/v2.0.4) \- 2026-10-20 diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 3f0ef014..6183ea6f 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -42,25 +42,33 @@ All the concurrency code in the repo, in one function: async def run_jobs( jobs: list[Job], max_concurrent: dict[str, int | None], attempts: int = 1 ) -> list[Outcome]: - todo = [job for job in jobs if not job.done()] # resume = skip what's on disk - # One cap per job kind (decision 6); None or 0 means no limit. - limits = {kind: asyncio.Semaphore(cap or len(todo) or 1) for kind, cap in max_concurrent.items()} + # One cap per job kind (decision 6). None or 0 means no limit, so no semaphore: + # sizing one from the job list would throttle follow-ups, which can outnumber it. + limits = { + kind: asyncio.Semaphore(cap) if cap else contextlib.nullcontext() + for kind, cap in max_concurrent.items() + } async def run_one(job: Job) -> list[Outcome]: - async with limits[job.kind]: - outcome = await run_with_retries(job, attempts) - # A finished job may hand back the work that depended on it. - follow_ups = outcome.follow_ups if outcome.ok else [] + if job.done(): # resume = skip what's on disk + outcome = Outcome.skipped(job) + else: + async with limits[job.kind]: + outcome = await run_with_retries(job, attempts) + # A finished or skipped job hands on the jobs waiting on it, and each of + # those is checked for done() in turn, so resume reaches every missing job + # however far down the chain it is, and redoes nothing. + follow_ups = job.follow_ups() if outcome.ok or outcome.skipped else [] return [outcome, *await gather_flat(run_one(next_job) for next_job in follow_ups)] - finished = await gather_flat(run_one(job) for job in todo) - return [Outcome.skipped(job) for job in jobs if job not in todo] + finished + return await gather_flat(run_one(job) for job in jobs) async def run_with_retries(job: Job, attempts: int) -> Outcome: for attempt in range(1, attempts + 1): try: - return Outcome.ok(job, follow_ups=await job.run()) + await job.run() + return Outcome.ok(job) except Exception as error: if attempt == attempts or not job.retryable(error): return Outcome.failed(job, error) @@ -73,7 +81,9 @@ class Job(Protocol): kind: str # "session" or "evaluation": picks the cap (and the code a remote worker would run) label: str def done(self) -> bool: ... # is my output already on disk? - async def run(self) -> list["Job"]: ... # do the work; return jobs that were waiting on me + async def run(self) -> None: ... # do the work + def follow_ups(self) -> list["Job"]: ... # the jobs waiting on me, from my fields and disk alone, + # so a skipped job hands them on too def retryable(self, error: Exception) -> bool: ... ``` @@ -81,7 +91,7 @@ class Job(Protocol): ```python # generate/run.py -jobs = [first_unfinished_session(persona, repeat, run_dir) +jobs = [Session(persona, repeat, index=1, run_dir=run_dir) for persona in personas for repeat in repeats] report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) @@ -102,7 +112,7 @@ Today `vera pipeline` runs a generation pool for one user model, then a judging ```python # vera_cli/pipeline.py, after creating every run root up front (run-lifecycle.md, step 5) -jobs = [first_unfinished_session(persona, repeat, run_dir_for(user_model)) +jobs = [Session(persona, repeat, index=1, run_dir=run_dir_for(user_model), judges=judges) for user_model in user_models for persona in personas for repeat in repeats] report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) for run_dir in run_dirs: @@ -111,7 +121,7 @@ for run_dir in run_dirs: pool(run_dirs) ``` -A conversation's last session returns that conversation's `Evaluation` jobs as follow-ups, so it is judged as soon as it exists, while other conversations are still being generated. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. +A conversation's last session hands on that conversation's `Evaluation` jobs as follow-ups (the `judges` field says which; `vera generate` leaves it empty), so it is judged as soon as it exists, while other conversations are still being generated. Resume works through the same chain: if a run stopped after a conversation's last transcript but before its judging finished, every session of it is skipped as done, and its unfinished evaluations are reached and run. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. What this needs: the judge's run folder and rubric must exist before generation starts, which the run-lifecycle order already gives (all run roots are created before any job runs). @@ -119,7 +129,8 @@ It also gives the pipeline the resume that `generate` and `judge` have. `vera pi ## Decisions from review -1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*: its `run()` returns the next session's job. Generation lists only the first unfinished session of each conversation. `done()` stays a single file check, and resume restarts exactly at the first missing session. +1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*. Generation lists every conversation's first session; a done session is skipped but still hands on its successor, so resume restarts exactly at the first missing session and nothing after it is redone. `done()` stays a single file check. + - **Provider state crosses sessions as data.** Some chatbots keep the history server-side, under the client's `conversation_id` (`EndpointLLM`). Today that survives between sessions only because the runner reuses one client for all of a conversation's sessions, and it's lost on resume. Instead, a session saves the chatbot's `conversation_id` with its transcript, and the next session reads it from there and creates its client with it. Continuity then holds across job boundaries, retries (a retried session starts again from the previous session's state), and resume. 2. **Retries live in the loop, the policy in the stage.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job type says which errors are worth retrying (`retryable`); the stage picks the number of attempts. 3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one cap per job kind (decision 6), and a per-model (or per-provider) cap comes with the Shared engine phase. 4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. @@ -139,4 +150,4 @@ Once `legacy/` is gone (it needs a tagged release that announces the removal; ro ## Testing -The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised, retries stop at `attempts` and skip non-retryable errors, follow-ups run after their parent and not when it fails, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). +The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised, retries stop at `attempts` and skip non-retryable errors, follow-ups run after their parent and not when it fails, a skipped job still hands on its follow-ups and each is checked for done in turn, follow-ups aren't throttled when their kind has no cap, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). From e7a749c6e19971d820b229b46fd1d92c25fb73ab Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:48:10 -0800 Subject: [PATCH 09/13] docs: job-loop design runs jobs from a queue Answers the review: one queue and one cap per job kind behind a non-blocking submit(); follow-ups and retries go back on the queue, with the attempt count on the job. Also explains `session` and the pipeline's several run dirs, and notes legacy/ is already removed on release/2.1. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- docs/design/job-loop.md | 146 +++++++++++++++++++++++----------------- docs/roadmap.md | 2 +- 3 files changed, 88 insertions(+), 62 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 155e7476..060adfb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Documentation and development workflow -- **Job-loop proposal**—[`docs/design/job-loop.md`](docs/design/job-loop.md) proposes one job loop for generation and judging: jobs as plain data, resume by skipping what's on disk, retries, one concurrency cap per job kind, and the pipeline as a single job list with `vera pipeline --into`. This is a draft design, not an implemented change. +- **Job-loop proposal**—[`docs/design/job-loop.md`](docs/design/job-loop.md) proposes one job loop for generation and judging: jobs as plain data, resume by skipping what's on disk, one queue and one concurrency cap per job kind with failed jobs re-queued for retry, and the pipeline as a single job list with `vera pipeline --into`. This is a draft design, not an implemented change. - **Release 2.1 plan**—[`docs/roadmap.md`](docs/roadmap.md#release-21-plan) lists the steps for 2.1 in dependency order with their PRs: safety net, data contracts, job loop, terminal output, then removing `legacy/`, with the judge-flow rewrite in parallel. They accumulate on a `release/2.1` branch that merges into `main` once, at the 2.1.0 tag, so `main` stays releasable; CI now also runs for `release/2.1`. ## [v2.0.4](https://github.com/SpringCare/VERA-MH/releases/tag/v2.0.4) \- 2026-10-20 diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 6183ea6f..6d212d51 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -22,11 +22,11 @@ On top of that, each stage has 4–5 layers that mostly forward arguments: `vera vera_cli/generate.py resolve and validate settings → RunConfig (no jobs, no I/O beyond reading inputs) │ ▼ -generate/run.py run_generation(config): list the jobs, hand them to the loop, report +generate/run.py run_generation(config): load the queue with the first jobs, run the loop, report generate/conversation.py the Session job: what generating one session means │ ▼ -workers/loop.py run_jobs(...): skip done jobs, run the rest with a concurrency cap, collect outcomes +workers/loop.py JobLoop: one queue and one cap per job kind; skip done jobs, run the rest, collect outcomes ``` Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging`) → `judge/evaluation.py` (the `Evaluation` job) → `workers/loop.py`. The CLI never calls the loop directly; it resolves everything up front and calls one domain entry point, which keeps the layer rule in [architecture.md](../architecture.md) (`vera_cli` → domain → `workers/`). @@ -35,51 +35,69 @@ Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging` ### The loop -All the concurrency code in the repo, in one function: +All the concurrency code in the repo, in one class. Each job kind (`session`, `evaluation`) has its own queue and its own cap. Everything that adds work goes through `submit`, which never waits: the stage loading its first jobs, a finished job handing on its follow-ups, and a failed job going back for another try. ```python # workers/loop.py -async def run_jobs( - jobs: list[Job], max_concurrent: dict[str, int | None], attempts: int = 1 -) -> list[Outcome]: - # One cap per job kind (decision 6). None or 0 means no limit, so no semaphore: - # sizing one from the job list would throttle follow-ups, which can outnumber it. - limits = { - kind: asyncio.Semaphore(cap) if cap else contextlib.nullcontext() - for kind, cap in max_concurrent.items() - } - - async def run_one(job: Job) -> list[Outcome]: - if job.done(): # resume = skip what's on disk - outcome = Outcome.skipped(job) +class JobLoop: + def __init__(self, max_concurrent: dict[str, int | None]): + self.caps = max_concurrent # per kind; None or 0 means no limit + self.queues = {kind: deque() for kind in max_concurrent} + self.running: dict[asyncio.Task, Job] = {} + self.outcomes: list[Outcome] = [] + + def submit(self, job: Job) -> None: + """Put a job on its kind's queue. Never waits.""" + if job.done(): # resume: its output is already on disk, + self.outcomes.append(Outcome.skipped(job)) + for next_job in job.follow_ups(): # but the jobs after it may not be + self.submit(next_job) else: - async with limits[job.kind]: - outcome = await run_with_retries(job, attempts) - # A finished or skipped job hands on the jobs waiting on it, and each of - # those is checked for done() in turn, so resume reaches every missing job - # however far down the chain it is, and redoes nothing. - follow_ups = job.follow_ups() if outcome.ok or outcome.skipped else [] - return [outcome, *await gather_flat(run_one(next_job) for next_job in follow_ups)] - - return await gather_flat(run_one(job) for job in jobs) - - -async def run_with_retries(job: Job, attempts: int) -> Outcome: - for attempt in range(1, attempts + 1): - try: - await job.run() - return Outcome.ok(job) - except Exception as error: - if attempt == attempts or not job.retryable(error): - return Outcome.failed(job, error) + self.queues[job.kind].append(job) + + async def run(self) -> list[Outcome]: + """Run until every queue is empty and nothing is running.""" + while True: + self.start_jobs_that_fit() + if not self.running: + return self.outcomes + finished, _ = await asyncio.wait(self.running, return_when=asyncio.FIRST_COMPLETED) + for task in finished: + self.record(self.running.pop(task), task.exception()) + + def start_jobs_that_fit(self) -> None: + for kind, queue in self.queues.items(): + while queue and self.has_room(kind): + job = queue.popleft() + self.running[asyncio.create_task(job.run())] = job + + def has_room(self, kind: str) -> bool: + cap = self.caps[kind] + running = sum(1 for job in self.running.values() if job.kind == kind) + return not cap or running < cap + + def record(self, job: Job, error: BaseException | None) -> None: + if error is None: + self.outcomes.append(Outcome.ok(job)) + for next_job in job.follow_ups(): + self.submit(next_job) + elif job.attempt < job.max_attempts and job.retryable(error): + self.submit(dataclasses.replace(job, attempt=job.attempt + 1)) # back of its queue + else: + self.outcomes.append(Outcome.failed(job, error)) ``` -A job is plain data plus two methods: +A done job is skipped in `submit` but still hands on its follow-ups, and each of those is checked in turn, so resume reaches every missing job however far down the chain it is, and redoes nothing. Because the cap is checked when a job starts, not sized from the first jobs, follow-ups are never throttled by how few jobs the stage loaded. + +A job is plain data plus four methods: ```python class Job(Protocol): - kind: str # "session" or "evaluation": picks the cap (and the code a remote worker would run) + kind: str # "session" or "evaluation": picks the queue and the cap + # (and the code a remote worker would run) label: str + attempt: int # 1 on the first try; the loop adds 1 on each retry + max_attempts: int # set by the stage from the config def done(self) -> bool: ... # is my output already on disk? async def run(self) -> None: ... # do the work def follow_ups(self) -> list["Job"]: ... # the jobs waiting on me, from my fields and disk alone, @@ -89,39 +107,48 @@ class Job(Protocol): ### Each stage +A stage only loads the queue and runs the loop: + ```python # generate/run.py -jobs = [Session(persona, repeat, index=1, run_dir=run_dir) - for persona in personas for repeat in repeats] -report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) +loop = JobLoop(max_concurrent) +for persona in personas: + for repeat in repeats: + loop.submit(Session(persona, repeat, session=1, run_dir=run_dir, max_attempts=config.attempts)) +report(await loop.run()) # judge/run.py -jobs = [Evaluation(conversation, judge, instance, run_dir) - for conversation in conversations - for judge in judges - for instance in judge.instances] -report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) +loop = JobLoop(max_concurrent) +for conversation in conversations: + for judge in judges: + for instance in judge.instances: + loop.submit(Evaluation(conversation, judge, instance, run_dir, max_attempts=config.attempts)) +report(await loop.run()) write_results_csv(run_dir) ``` -`Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. A semaphore plus `gather` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. +`session` is the session's number within its conversation, so generation loads only each conversation's first session; session 2 arrives as session 1's follow-up, and so on. `Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. One `JobLoop` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. ### The pipeline: one job list -Today `vera pipeline` runs a generation pool for one user model, then a judging pool for its run, then the next user model, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one call: +`vera pipeline` can make several runs at once: one per user model (`-u gpt-5.4 -u claude-sonnet-5` makes two runs, whose scores are then pooled), and one per target when there are several. Today it runs a generation pool for one run, then a judging pool for it, then the next run, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one loop for every run: ```python # vera_cli/pipeline.py, after creating every run root up front (run-lifecycle.md, step 5) -jobs = [Session(persona, repeat, index=1, run_dir=run_dir_for(user_model), judges=judges) - for user_model in user_models for persona in personas for repeat in repeats] -report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) +loop = JobLoop(max_concurrent) +for user_model in user_models: + for persona in personas: + for repeat in repeats: + loop.submit(Session(persona, repeat, session=1, run_dir=run_dir_for(user_model), + judges=judges, max_attempts=config.attempts)) +report(await loop.run()) for run_dir in run_dirs: write_results_csv(evaluation_dir_for(run_dir)) score(evaluation_dir_for(run_dir)) pool(run_dirs) ``` -A conversation's last session hands on that conversation's `Evaluation` jobs as follow-ups (the `judges` field says which; `vera generate` leaves it empty), so it is judged as soon as it exists, while other conversations are still being generated. Resume works through the same chain: if a run stopped after a conversation's last transcript but before its judging finished, every session of it is skipped as done, and its unfinished evaluations are reached and run. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. +A conversation's last session hands on that conversation's `Evaluation` jobs as follow-ups (the `judges` field says which; `vera generate` leaves it empty). They go on the evaluation queue, so the conversation is judged as soon as it exists, while other conversations are still being generated. Resume works through the same chain: if a run stopped after a conversation's last transcript but before its judging finished, every session of it is skipped as done, and its unfinished evaluations are reached and run. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. What this needs: the judge's run folder and rubric must exist before generation starts, which the run-lifecycle order already gives (all run roots are created before any job runs). @@ -131,23 +158,22 @@ It also gives the pipeline the resume that `generate` and `judge` have. `vera pi 1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*. Generation lists every conversation's first session; a done session is skipped but still hands on its successor, so resume restarts exactly at the first missing session and nothing after it is redone. `done()` stays a single file check. - **Provider state crosses sessions as data.** Some chatbots keep the history server-side, under the client's `conversation_id` (`EndpointLLM`). Today that survives between sessions only because the runner reuses one client for all of a conversation's sessions, and it's lost on resume. Instead, a session saves the chatbot's `conversation_id` with its transcript, and the next session reads it from there and creates its client with it. Continuity then holds across job boundaries, retries (a retried session starts again from the previous session's state), and resume. -2. **Retries live in the loop, the policy in the stage.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job type says which errors are worth retrying (`retryable`); the stage picks the number of attempts. +2. **A failed job goes back on the queue.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job carries its attempt count and its `max_attempts`; when it fails with an error its type calls worth retrying (`retryable`), the loop puts a copy with `attempt + 1` at the back of its queue, so other jobs run in the meantime and a retry doesn't hold a slot while it waits. That's how hosted queues do it too (a message carries its delivery count), so it carries over to an external queue unchanged. The stage sets `max_attempts` from the config. 3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one cap per job kind (decision 6), and a per-model (or per-provider) cap comes with the Shared engine phase. 4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. 5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. - **Unfinished judging works the same way.** An `Evaluation` is done when its TSV exists, and the TSV is written only after the judge has answered every question it needed. A judging stopped partway through a conversation leaves no TSV, so rerunning with `--into` judges that conversation again from its first question; conversations already judged are skipped. `results.csv` and the scores are rebuilt from the TSVs after the loop, so they always reflect exactly the finished evaluations. Saving progress question by question isn't worth it: one conversation is a few dozen short judge calls. -6. **One cap per job kind.** The config already sets two: `generation.max_concurrent` caps `Session` jobs and `judging.max_concurrent` caps `Evaluation` jobs, so in the pipeline a burst of short judge calls can't crowd out the long generation calls. `run_jobs` keeps one semaphore per job kind instead of one in total. The per-judge-model cap (`per_judge`) still comes later (decision 3). -7. **Jobs stay serializable, so an external queue stays possible.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `run_jobs` becomes one implementation of an executor interface, and a remote one would submit serialized jobs instead. The interface isn't built until there's a concrete need, same rule as the roadmap's Storage phase. What keeps it possible is enforced, not assumed: each job type is a dataclass of JSON-safe fields plus `kind`, and a unit test round-trips every job type through JSON and runs the copy. +6. **One cap per job kind.** The config already sets two: `generation.max_concurrent` caps `Session` jobs and `judging.max_concurrent` caps `Evaluation` jobs, so in the pipeline a burst of short judge calls can't crowd out the long generation calls. The loop keeps one queue and one cap per job kind instead of one in total. The per-judge-model cap (`per_judge`) still comes later (decision 3). +7. **Jobs stay serializable, so an external queue stays possible.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: the in-memory queues are the seam. An external version keeps `submit` and `run`, but `submit` sends the job's JSON to the external queue, workers elsewhere pull from it, and the cap becomes how many workers pull. Stages don't change, since they only call `submit` and `run`. The interface isn't built until there's a concrete need, same rule as the roadmap's Storage phase. What keeps it possible is enforced, not assumed: each job type is a dataclass of JSON-safe fields plus `kind`, and a unit test round-trips every job type through JSON and runs the copy. ## What this removes -Once `legacy/` is gone (it needs a tagged release that announces the removal; roadmap, Deferred): +`legacy/` is already gone on `release/2.1` (#287), with its parity tests, `test_pipeline.py`, and the leftovers that had no callers (`judge_single_conversation`, `_create_conversation_jobs`; #303). The forwarding layers go in the two PRs before the loop (roadmap, Release 2.1 plan, step 4): `run_for_user_models`, `_legacy_model_config`, and the always-constant parameters with generation taking `GenerationConfig` (#309), and the `verbose_workers` branches with judging taking `JudgingConfig`. What the loop itself removes: -- `run_for_user_models` and `_legacy_model_config`; `run_generation`'s always-constant parameters (`persona_names`, `run_id`, `verbose`, the list-of-one `persona_files`) -- The `verbose_workers` branches, `judge_single_conversation`, and the legacy helpers in `judge/utils.py` -- `_run_workers_with_queue`, both `_worker`s, the dead `_create_conversation_jobs`, and the three judge summaries -- The CLI parity tests and most of `test_pipeline.py` +- `_run_workers_with_queue` and both `_worker`s +- The three judge summaries, replaced by one `report(outcomes)` +- Failures worked out by subtraction afterwards: every job ends as exactly one outcome ## Testing -The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `run_jobs` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised, retries stop at `attempts` and skip non-retryable errors, follow-ups run after their parent and not when it fails, a skipped job still hands on its follow-ups and each is checked for done in turn, follow-ups aren't throttled when their kind has no cap, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). +The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `JobLoop` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised, a retryable failure goes back on the queue with `attempt + 1` and stops at `max_attempts`, non-retryable errors aren't retried, follow-ups run after their parent and not when it fails, a skipped job still hands on its follow-ups and each is checked for done in turn, follow-ups aren't throttled when their kind has no cap, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). diff --git a/docs/roadmap.md b/docs/roadmap.md index d88f63d5..27828a51 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -43,7 +43,7 @@ Smaller items that belong to no phase are in the [Backlog](#backlog). |---|---|---| | 1 | Phase 1(b): generation takes `GenerationConfig` directly | `release/2.1` | | 2 | Phase 1(c): judging takes `JudgingConfig` directly | `release/2.1` | - | 3 | `workers/loop.py`: `run_jobs`, one cap per job kind, JSON-safe jobs; events, the plain-line reporter, and `events.jsonl`. No callers yet | `release/2.1` | + | 3 | `workers/loop.py`: `JobLoop`, one queue and one cap per job kind, JSON-safe jobs; events, the plain-line reporter, and `events.jsonl`. No callers yet | `release/2.1` | | 4 | Judge on the loop: the `Evaluation` job; deletes `judge/runner.py` | 2, 3 | | 5 | Generate on the loop: the `Session` job; deletes `generate/runner.py` | 1, 3 | | 6 | The pipeline as one job list: judging starts as each conversation finishes, fail-fast when a stage produces nothing, `vera pipeline --into`, two-step Ctrl-C | 4, 5 | From ac2d3fa70b3113b10e821494ad779a97a65b0831 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 4 Nov 2026 09:51:52 -0800 Subject: [PATCH 10/13] docs: job-loop sessions are named types, not numbers Session jobs carry the configured session_types and a position instead of a session number; says why the loop is a class and that the pipeline loads every run's first sessions up front. Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 6d212d51..10706430 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -35,7 +35,7 @@ Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging` ### The loop -All the concurrency code in the repo, in one class. Each job kind (`session`, `evaluation`) has its own queue and its own cap. Everything that adds work goes through `submit`, which never waits: the stage loading its first jobs, a finished job handing on its follow-ups, and a failed job going back for another try. +All the concurrency code in the repo, in one class. It's a class rather than a function because `submit` is called from three places (the stage, a finished job's follow-ups, a retry) and they all need the same queues, so the queues live on an object instead of in a closure. Each job kind (`session`, `evaluation`) has its own queue and its own cap. Everything that adds work goes through `submit`, which never waits: the stage loading its first jobs, a finished job handing on its follow-ups, and a failed job going back for another try. ```python # workers/loop.py @@ -114,7 +114,8 @@ A stage only loads the queue and runs the loop: loop = JobLoop(max_concurrent) for persona in personas: for repeat in repeats: - loop.submit(Session(persona, repeat, session=1, run_dir=run_dir, max_attempts=config.attempts)) + loop.submit(Session(persona, repeat, session_types=config.sessions, position=0, + run_dir=run_dir, max_attempts=config.attempts)) report(await loop.run()) # judge/run.py @@ -127,11 +128,11 @@ report(await loop.run()) write_results_csv(run_dir) ``` -`session` is the session's number within its conversation, so generation loads only each conversation's first session; session 2 arrives as session 1's follow-up, and so on. `Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. One `JobLoop` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. +Sessions are named, not numbered: `session_types` is the configured list (`--sessions intake,coaching`; one unnamed session when unset), and `position` says which entry this job runs. Generation loads only each conversation's first session; `coaching` arrives as `intake`'s follow-up. That's a different thing from `attempt`, which counts tries at the same session. `Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. One `JobLoop` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. ### The pipeline: one job list -`vera pipeline` can make several runs at once: one per user model (`-u gpt-5.4 -u claude-sonnet-5` makes two runs, whose scores are then pooled), and one per target when there are several. Today it runs a generation pool for one run, then a judging pool for it, then the next run, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one loop for every run: +`vera pipeline` can make several runs at once: one per user model (`-u gpt-5.4 -u claude-sonnet-5` makes two runs, whose scores are then pooled), and one per target when there are several. Today it runs a generation pool for one run, then a judging pool for it, then the next run, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one loop for every run: the first session of every conversation of every run goes on the session queue up front, and evaluations join the evaluation queue as conversations finish. ```python # vera_cli/pipeline.py, after creating every run root up front (run-lifecycle.md, step 5) @@ -139,8 +140,9 @@ loop = JobLoop(max_concurrent) for user_model in user_models: for persona in personas: for repeat in repeats: - loop.submit(Session(persona, repeat, session=1, run_dir=run_dir_for(user_model), - judges=judges, max_attempts=config.attempts)) + loop.submit(Session(persona, repeat, session_types=config.sessions, position=0, + run_dir=run_dir_for(user_model), judges=judges, + max_attempts=config.attempts)) report(await loop.run()) for run_dir in run_dirs: write_results_csv(evaluation_dir_for(run_dir)) From d31f0717fe6f9f45b0d4e9992aa7af9b67479253 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:15:16 -0800 Subject: [PATCH 11/13] docs: job-loop retries go back on the queue with no delay, on purpose Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 10706430..941a417e 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -161,6 +161,7 @@ It also gives the pipeline the resume that `generate` and `judge` have. `vera pi 1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*. Generation lists every conversation's first session; a done session is skipped but still hands on its successor, so resume restarts exactly at the first missing session and nothing after it is redone. `done()` stays a single file check. - **Provider state crosses sessions as data.** Some chatbots keep the history server-side, under the client's `conversation_id` (`EndpointLLM`). Today that survives between sessions only because the runner reuses one client for all of a conversation's sessions, and it's lost on resume. Instead, a session saves the chatbot's `conversation_id` with its transcript, and the next session reads it from there and creates its client with it. Continuity then holds across job boundaries, retries (a retried session starts again from the previous session's state), and resume. 2. **A failed job goes back on the queue.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job carries its attempt count and its `max_attempts`; when it fails with an error its type calls worth retrying (`retryable`), the loop puts a copy with `attempt + 1` at the back of its queue, so other jobs run in the meantime and a retry doesn't hold a slot while it waits. That's how hosted queues do it too (a message carries its delivery count), so it carries over to an external queue unchanged. The stage sets `max_attempts` from the config. + - **No delay before a retry, on purpose.** Errors that pass with time (rate limits, timeouts, 5xx) are retried per call in `llm_clients/`, with backoff, and never reach the loop. What reaches it fails the same way however long you wait, like an answer that breaks the schema, so the retry can go straight to the back of the queue. A retried `Session` reruns the whole session, which is expensive, so the stage can give sessions fewer attempts than evaluations. 3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one cap per job kind (decision 6), and a per-model (or per-provider) cap comes with the Shared engine phase. 4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. 5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. From f7a11cd5f7277ec4a1869a3e7fb47d61e887783e Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 4 Nov 2026 11:17:53 -0800 Subject: [PATCH 12/13] docs: job-loop retries happen only per call, in llm_clients The loop no longer resubmits failed jobs: a failed job is recorded and rerun later with --into. attempt, max_attempts, and retryable leave the Job protocol; clients report their retries as call_retrying events. Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 32 ++++++++++++++------------------ 1 file changed, 14 insertions(+), 18 deletions(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index 941a417e..acb9330a 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -35,7 +35,7 @@ Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging` ### The loop -All the concurrency code in the repo, in one class. It's a class rather than a function because `submit` is called from three places (the stage, a finished job's follow-ups, a retry) and they all need the same queues, so the queues live on an object instead of in a closure. Each job kind (`session`, `evaluation`) has its own queue and its own cap. Everything that adds work goes through `submit`, which never waits: the stage loading its first jobs, a finished job handing on its follow-ups, and a failed job going back for another try. +All the concurrency code in the repo, in one class. It's a class rather than a function because `submit` is called from two places (the stage, and a finished job handing on its follow-ups) and both need the same queues, so the queues live on an object instead of in a closure; it also gives an external queue one thing to replace, anything with `submit` and `run` (decision 7). Each job kind (`session`, `evaluation`) has its own queue and its own cap. Everything that adds work goes through `submit`, which never waits: the stage loading its first jobs, and a finished job handing on its follow-ups. The loop never retries a job: retries happen per call, inside it (decision 2). ```python # workers/loop.py @@ -81,28 +81,23 @@ class JobLoop: self.outcomes.append(Outcome.ok(job)) for next_job in job.follow_ups(): self.submit(next_job) - elif job.attempt < job.max_attempts and job.retryable(error): - self.submit(dataclasses.replace(job, attempt=job.attempt + 1)) # back of its queue else: - self.outcomes.append(Outcome.failed(job, error)) + self.outcomes.append(Outcome.failed(job, error)) # rerun later with --into ``` A done job is skipped in `submit` but still hands on its follow-ups, and each of those is checked in turn, so resume reaches every missing job however far down the chain it is, and redoes nothing. Because the cap is checked when a job starts, not sized from the first jobs, follow-ups are never throttled by how few jobs the stage loaded. -A job is plain data plus four methods: +A job is plain data plus three methods: ```python class Job(Protocol): kind: str # "session" or "evaluation": picks the queue and the cap # (and the code a remote worker would run) label: str - attempt: int # 1 on the first try; the loop adds 1 on each retry - max_attempts: int # set by the stage from the config def done(self) -> bool: ... # is my output already on disk? async def run(self) -> None: ... # do the work def follow_ups(self) -> list["Job"]: ... # the jobs waiting on me, from my fields and disk alone, # so a skipped job hands them on too - def retryable(self, error: Exception) -> bool: ... ``` ### Each stage @@ -115,7 +110,7 @@ loop = JobLoop(max_concurrent) for persona in personas: for repeat in repeats: loop.submit(Session(persona, repeat, session_types=config.sessions, position=0, - run_dir=run_dir, max_attempts=config.attempts)) + run_dir=run_dir)) report(await loop.run()) # judge/run.py @@ -123,12 +118,12 @@ loop = JobLoop(max_concurrent) for conversation in conversations: for judge in judges: for instance in judge.instances: - loop.submit(Evaluation(conversation, judge, instance, run_dir, max_attempts=config.attempts)) + loop.submit(Evaluation(conversation, judge, instance, run_dir)) report(await loop.run()) write_results_csv(run_dir) ``` -Sessions are named, not numbered: `session_types` is the configured list (`--sessions intake,coaching`; one unnamed session when unset), and `position` says which entry this job runs. Generation loads only each conversation's first session; `coaching` arrives as `intake`'s follow-up. That's a different thing from `attempt`, which counts tries at the same session. `Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. One `JobLoop` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. +Sessions are named, not numbered: `session_types` is the configured list (`--sessions intake,coaching`; one unnamed session when unset), and `position` says which entry this job runs. Generation loads only each conversation's first session; `coaching` arrives as `intake`'s follow-up. `Session.run()` is today's per-session body of `run_single_conversation`; `Evaluation.run()` is today's `_evaluate_single_conversation_with_judge`. One `JobLoop` replaces both queue-and-worker pools: no worker IDs, no `get_nowait` loop, no shared results list. ### The pipeline: one job list @@ -141,8 +136,7 @@ for user_model in user_models: for persona in personas: for repeat in repeats: loop.submit(Session(persona, repeat, session_types=config.sessions, position=0, - run_dir=run_dir_for(user_model), judges=judges, - max_attempts=config.attempts)) + run_dir=run_dir_for(user_model), judges=judges)) report(await loop.run()) for run_dir in run_dirs: write_results_csv(evaluation_dir_for(run_dir)) @@ -159,11 +153,13 @@ It also gives the pipeline the resume that `generate` and `judge` have. `vera pi ## Decisions from review 1. **A job is one session.** Sessions must run in order, so session *k* is a follow-up of session *k − 1*. Generation lists every conversation's first session; a done session is skipped but still hands on its successor, so resume restarts exactly at the first missing session and nothing after it is redone. `done()` stays a single file check. - - **Provider state crosses sessions as data.** Some chatbots keep the history server-side, under the client's `conversation_id` (`EndpointLLM`). Today that survives between sessions only because the runner reuses one client for all of a conversation's sessions, and it's lost on resume. Instead, a session saves the chatbot's `conversation_id` with its transcript, and the next session reads it from there and creates its client with it. Continuity then holds across job boundaries, retries (a retried session starts again from the previous session's state), and resume. -2. **A failed job goes back on the queue.** LLM calls already retry per call in `llm_clients/` (3 attempts with backoff). The loop adds job-level attempts for what that doesn't cover, such as a judge answer that fails the structured-output schema. Each job carries its attempt count and its `max_attempts`; when it fails with an error its type calls worth retrying (`retryable`), the loop puts a copy with `attempt + 1` at the back of its queue, so other jobs run in the meantime and a retry doesn't hold a slot while it waits. That's how hosted queues do it too (a message carries its delivery count), so it carries over to an external queue unchanged. The stage sets `max_attempts` from the config. - - **No delay before a retry, on purpose.** Errors that pass with time (rate limits, timeouts, 5xx) are retried per call in `llm_clients/`, with backoff, and never reach the loop. What reaches it fails the same way however long you wait, like an answer that breaks the schema, so the retry can go straight to the back of the queue. A retried `Session` reruns the whole session, which is expensive, so the stage can give sessions fewer attempts than evaluations. + - **Provider state crosses sessions as data.** Some chatbots keep the history server-side, under the client's `conversation_id` (`EndpointLLM`). Today that survives between sessions only because the runner reuses one client for all of a conversation's sessions, and it's lost on resume. Instead, a session saves the chatbot's `conversation_id` with its transcript, and the next session reads it from there and creates its client with it. Continuity then holds across job boundaries and resume (a rerun session starts again from the previous session's state). +2. **Retries happen in one place: per call, in `llm_clients/`.** Every LLM call already retries with backoff (`_run_with_retry`, up to `max_llm_retries`), and that covers the whole call, parsing included, so a structured answer that breaks the schema is re-asked too. The loop adds no second mechanism. A failed job is recorded as failed and leaves the queue for good. Its output was never written, so rerunning with `--into` picks it up like any missing job. + - **Why not retry the job as well.** A job fails only after one of its calls has used up its retries, so the cause is persistent (quota, auth, a model that keeps failing) and an immediate rerun rarely helps. Retrying per call also retries the smallest unit: a rate limit on turn 15 re-sends that one call, where rerunning the job would redo the whole session from turn 1. + - **The cost.** While a call backs off, its job keeps its slot, so one fewer job of that kind runs meanwhile. The waits are short and capped (`retry_max_delay_seconds`). + - **The clients still tell the loop.** Each per-call retry emits a `call_retrying` event through the same `report()` jobs use (decision 4), so the display can show "rate limited, retrying 2/4" against the right conversation. That's the only link: no retry logic is shared. 3. **Concurrency per judge model, later.** Today's `per_judge` keeps one cap per judge model. That stays the intended behavior but isn't pressing; the first version has one cap per job kind (decision 6), and a per-model (or per-provider) cap comes with the Shared engine phase. -4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. +4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*, a call retrying) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. 5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. - **Unfinished judging works the same way.** An `Evaluation` is done when its TSV exists, and the TSV is written only after the judge has answered every question it needed. A judging stopped partway through a conversation leaves no TSV, so rerunning with `--into` judges that conversation again from its first question; conversations already judged are skipped. `results.csv` and the scores are rebuilt from the TSVs after the loop, so they always reflect exactly the finished evaluations. Saving progress question by question isn't worth it: one conversation is a few dozen short judge calls. 6. **One cap per job kind.** The config already sets two: `generation.max_concurrent` caps `Session` jobs and `judging.max_concurrent` caps `Evaluation` jobs, so in the pipeline a burst of short judge calls can't crowd out the long generation calls. The loop keeps one queue and one cap per job kind instead of one in total. The per-judge-model cap (`per_judge`) still comes later (decision 3). @@ -179,4 +175,4 @@ It also gives the pipeline the resume that `generate` and `judge` have. `vera pi ## Testing -The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `JobLoop` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised, a retryable failure goes back on the queue with `attempt + 1` and stops at `max_attempts`, non-retryable errors aren't retried, follow-ups run after their parent and not when it fails, a skipped job still hands on its follow-ups and each is checked for done in turn, follow-ups aren't throttled when their kind has no cap, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). +The roadmap's four engine-testing tiers apply (structural parity with the mock harness, live smoke, semantic similarity, manual spot-check). `JobLoop` itself gets plain unit tests: skip-on-done, each kind's concurrency cap holds, failures recorded and not raised and never resubmitted, follow-ups run after their parent and not when it fails, a skipped job still hands on its follow-ups and each is checked for done in turn, follow-ups aren't throttled when their kind has no cap, `max_concurrent` of `None` and `0` both unlimited, and every job type survives a JSON round trip (decision 7). From 728edfb9a369f667a89660e4b0f78012c9f4b7da Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:18:38 -0800 Subject: [PATCH 13/13] docs: job-loop sketch uses a semaphore per kind, as built in #308 Co-Authored-By: Claude Opus 5.5 --- docs/design/job-loop.md | 85 ++++++++++++++++++----------------------- 1 file changed, 37 insertions(+), 48 deletions(-) diff --git a/docs/design/job-loop.md b/docs/design/job-loop.md index acb9330a..3586b0ac 100644 --- a/docs/design/job-loop.md +++ b/docs/design/job-loop.md @@ -22,11 +22,11 @@ On top of that, each stage has 4–5 layers that mostly forward arguments: `vera vera_cli/generate.py resolve and validate settings → RunConfig (no jobs, no I/O beyond reading inputs) │ ▼ -generate/run.py run_generation(config): load the queue with the first jobs, run the loop, report +generate/run.py run_generation(config): submit the first jobs, run the loop, report generate/conversation.py the Session job: what generating one session means │ ▼ -workers/loop.py JobLoop: one queue and one cap per job kind; skip done jobs, run the rest, collect outcomes +workers/loop.py JobLoop: one cap per job kind; skip done jobs, run the rest, collect outcomes ``` Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging`) → `judge/evaluation.py` (the `Evaluation` job) → `workers/loop.py`. The CLI never calls the loop directly; it resolves everything up front and calls one domain entry point, which keeps the layer rule in [architecture.md](../architecture.md) (`vera_cli` → domain → `workers/`). @@ -35,63 +35,52 @@ Judging is the same shape: `vera_cli/judge.py` → `judge/run.py` (`run_judging` ### The loop -All the concurrency code in the repo, in one class. It's a class rather than a function because `submit` is called from two places (the stage, and a finished job handing on its follow-ups) and both need the same queues, so the queues live on an object instead of in a closure; it also gives an external queue one thing to replace, anything with `submit` and `run` (decision 7). Each job kind (`session`, `evaluation`) has its own queue and its own cap. Everything that adds work goes through `submit`, which never waits: the stage loading its first jobs, and a finished job handing on its follow-ups. The loop never retries a job: retries happen per call, inside it (decision 2). +All the concurrency code in the repo, in one class. It's a class rather than a function because `submit` is called from two places (the stage, and a finished job handing on its follow-ups) and both need the same caps and the same set of running jobs; it also gives an external queue one thing to replace, anything with `submit` and `run` (decision 7). Each job kind (`session`, `evaluation`) has its own cap, an `asyncio.Semaphore`. Everything that adds work goes through `submit`, which never waits: it starts the job as a task, and the task waits for a free slot of its kind. The tasks waiting on a kind's semaphore are that kind's queue, served in order. The loop never retries a job: retries happen per call, inside it (decision 2). ```python # workers/loop.py class JobLoop: - def __init__(self, max_concurrent: dict[str, int | None]): - self.caps = max_concurrent # per kind; None or 0 means no limit - self.queues = {kind: deque() for kind in max_concurrent} - self.running: dict[asyncio.Task, Job] = {} + def __init__(self, max_concurrent: dict[str, int | None], reporters=()): + self.limits = {kind: asyncio.Semaphore(cap) if cap else contextlib.nullcontext() + for kind, cap in max_concurrent.items()} # None or 0: no limit + self.tasks: set[asyncio.Task] = set() self.outcomes: list[Outcome] = [] def submit(self, job: Job) -> None: - """Put a job on its kind's queue. Never waits.""" - if job.done(): # resume: its output is already on disk, - self.outcomes.append(Outcome.skipped(job)) - for next_job in job.follow_ups(): # but the jobs after it may not be - self.submit(next_job) - else: - self.queues[job.kind].append(job) + """Start a job; it waits for a free slot of its kind. Never waits itself.""" + self.tasks.add(asyncio.create_task(self.run_one(job))) async def run(self) -> list[Outcome]: - """Run until every queue is empty and nothing is running.""" - while True: - self.start_jobs_that_fit() - if not self.running: - return self.outcomes - finished, _ = await asyncio.wait(self.running, return_when=asyncio.FIRST_COMPLETED) - for task in finished: - self.record(self.running.pop(task), task.exception()) - - def start_jobs_that_fit(self) -> None: - for kind, queue in self.queues.items(): - while queue and self.has_room(kind): - job = queue.popleft() - self.running[asyncio.create_task(job.run())] = job - - def has_room(self, kind: str) -> bool: - cap = self.caps[kind] - running = sum(1 for job in self.running.values() if job.kind == kind) - return not cap or running < cap - - def record(self, job: Job, error: BaseException | None) -> None: - if error is None: - self.outcomes.append(Outcome.ok(job)) - for next_job in job.follow_ups(): - self.submit(next_job) - else: - self.outcomes.append(Outcome.failed(job, error)) # rerun later with --into + """Wait until every job has finished, follow-ups included.""" + while self.tasks: # follow-ups add tasks while we wait + finished, _ = await asyncio.wait(self.tasks) + self.tasks -= finished + return self.outcomes + + async def run_one(self, job: Job) -> None: + try: # any error is this job's outcome + if job.done(): # resume: its output is already on disk, + status = "skipped" + else: + async with self.limits[job.kind]: + await job.run() + status = "finished" + follow_ups = job.follow_ups() # but the jobs after it may not be + except Exception as error: + self.outcomes.append(Outcome(job, "failed", error)) # rerun later with --into + return + self.outcomes.append(Outcome(job, status)) + for next_job in follow_ups: + self.submit(next_job) ``` -A done job is skipped in `submit` but still hands on its follow-ups, and each of those is checked in turn, so resume reaches every missing job however far down the chain it is, and redoes nothing. Because the cap is checked when a job starts, not sized from the first jobs, follow-ups are never throttled by how few jobs the stage loaded. +A done job is skipped but still hands on its follow-ups, and each of those is checked in turn, so resume reaches every missing job however far down the chain it is, and redoes nothing. Because a job takes a slot when it starts, follow-ups are never throttled by how few jobs the stage loaded. A job is plain data plus three methods: ```python class Job(Protocol): - kind: str # "session" or "evaluation": picks the queue and the cap + kind: str # "session" or "evaluation": picks the cap # (and the code a remote worker would run) label: str def done(self) -> bool: ... # is my output already on disk? @@ -102,7 +91,7 @@ class Job(Protocol): ### Each stage -A stage only loads the queue and runs the loop: +A stage only submits its first jobs and runs the loop: ```python # generate/run.py @@ -127,7 +116,7 @@ Sessions are named, not numbered: `session_types` is the configured list (`--ses ### The pipeline: one job list -`vera pipeline` can make several runs at once: one per user model (`-u gpt-5.4 -u claude-sonnet-5` makes two runs, whose scores are then pooled), and one per target when there are several. Today it runs a generation pool for one run, then a judging pool for it, then the next run, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one loop for every run: the first session of every conversation of every run goes on the session queue up front, and evaluations join the evaluation queue as conversations finish. +`vera pipeline` can make several runs at once: one per user model (`-u gpt-5.4 -u claude-sonnet-5` makes two runs, whose scores are then pooled), and one per target when there are several. Today it runs a generation pool for one run, then a judging pool for it, then the next run, then pools the results: one pool after another, with idle capacity in between. With the loop it becomes one loop for every run: the first session of every conversation of every run is submitted up front, and evaluations are submitted as conversations finish, each kind under its own cap. ```python # vera_cli/pipeline.py, after creating every run root up front (run-lifecycle.md, step 5) @@ -144,7 +133,7 @@ for run_dir in run_dirs: pool(run_dirs) ``` -A conversation's last session hands on that conversation's `Evaluation` jobs as follow-ups (the `judges` field says which; `vera generate` leaves it empty). They go on the evaluation queue, so the conversation is judged as soon as it exists, while other conversations are still being generated. Resume works through the same chain: if a run stopped after a conversation's last transcript but before its judging finished, every session of it is skipped as done, and its unfinished evaluations are reached and run. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. +A conversation's last session hands on that conversation's `Evaluation` jobs as follow-ups (the `judges` field says which; `vera generate` leaves it empty). They wait for an evaluation slot, so the conversation is judged as soon as it exists, while other conversations are still being generated. Resume works through the same chain: if a run stopped after a conversation's last transcript but before its judging finished, every session of it is skipped as done, and its unfinished evaluations are reached and run. Every user model shares the same caps (decision 6), and the slowest persona no longer holds up the next model's start. Run-level steps (`results.csv`, scores, pooling) still wait for every job, because they need all the rows. What this needs: the judge's run folder and rubric must exist before generation starts, which the run-lifecycle order already gives (all run roots are created before any job runs). @@ -162,8 +151,8 @@ It also gives the pipeline the resume that `generate` and `judge` have. `vera pi 4. **Progress display is a separate PR** ([terminal-interface.md](./terminal-interface.md), #271). The loop emits events (job started, job finished, and from inside a job: user responding, chatbot responding, judge on question *n*, a call retrying) to a reporter. The first reporter prints one flushed line per event, which fixes the "looks hung under redirection" backlog item. A later PR adds a live display: a progress bar per stage and a box per model saying what it's doing. 5. **Missing work is re-done, not fatal.** The run's `config.json` (see [run-lifecycle.md](./run-lifecycle.md)) fixes the job list, so `done()` over that list is the manifest: rerunning with `--into` does exactly the missing jobs. A stage still stops the pipeline when it produced nothing (nothing to judge), with a message naming the run folder and the failure count; partial failures warn and continue. - **Unfinished judging works the same way.** An `Evaluation` is done when its TSV exists, and the TSV is written only after the judge has answered every question it needed. A judging stopped partway through a conversation leaves no TSV, so rerunning with `--into` judges that conversation again from its first question; conversations already judged are skipped. `results.csv` and the scores are rebuilt from the TSVs after the loop, so they always reflect exactly the finished evaluations. Saving progress question by question isn't worth it: one conversation is a few dozen short judge calls. -6. **One cap per job kind.** The config already sets two: `generation.max_concurrent` caps `Session` jobs and `judging.max_concurrent` caps `Evaluation` jobs, so in the pipeline a burst of short judge calls can't crowd out the long generation calls. The loop keeps one queue and one cap per job kind instead of one in total. The per-judge-model cap (`per_judge`) still comes later (decision 3). -7. **Jobs stay serializable, so an external queue stays possible.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: the in-memory queues are the seam. An external version keeps `submit` and `run`, but `submit` sends the job's JSON to the external queue, workers elsewhere pull from it, and the cap becomes how many workers pull. Stages don't change, since they only call `submit` and `run`. The interface isn't built until there's a concrete need, same rule as the roadmap's Storage phase. What keeps it possible is enforced, not assumed: each job type is a dataclass of JSON-safe fields plus `kind`, and a unit test round-trips every job type through JSON and runs the copy. +6. **One cap per job kind.** The config already sets two: `generation.max_concurrent` caps `Session` jobs and `judging.max_concurrent` caps `Evaluation` jobs, so in the pipeline a burst of short judge calls can't crowd out the long generation calls. The loop keeps one cap per job kind instead of one in total. The per-judge-model cap (`per_judge`) still comes later (decision 3). +7. **Jobs stay serializable, so an external queue stays possible.** Jobs are plain data (paths, model names, indices), not closures, so they can be serialized as JSON. A closure would carry live state with it (an LLM client's HTTP session, its API key), which can't or shouldn't travel; instead each job names its `kind`, and the worker looks up the code to run and builds its own clients. That keeps an external queue (a cloud task queue, a cluster scheduler) possible: `submit` is the seam. An external version keeps `submit` and `run`, but `submit` sends the job's JSON to the external queue, workers elsewhere pull from it, and the cap becomes how many workers pull. Stages don't change, since they only call `submit` and `run`. The interface isn't built until there's a concrete need, same rule as the roadmap's Storage phase. What keeps it possible is enforced, not assumed: each job type is a dataclass of JSON-safe fields plus `kind`, and a unit test round-trips every job type through JSON and runs the copy. ## What this removes