Repository navigation
docs: design draft — one job loop for generate and judge - #266
luca-belli wants to merge 14 commits into
Conversation
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
| One function holds all the concurrency code in the repo: | ||
|
|
||
| ```python | ||
| # workers/loop.py |
There was a problem hiding this comment.
does it mean that the vera cli resolves all the dependencies, then invokes the loop?
There was a problem hiding this comment.
Almost: vera_cli resolves and validates, then calls one domain entry point (run_generation(config)). The domain lists the jobs and calls the loop; the CLI never calls workers/ directly. See the new "Who does what" section.
| ```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 |
There was a problem hiding this comment.
could jobs be an abstraction so we can use an external queue?
There was a problem hiding this comment.
Yes, as long as jobs stay plain data (paths, model names, indices), not closures, so they can be serialized. run_jobs then becomes one executor among others. I'd only build that interface once there's a concrete need (same rule as Storage). Open question 2.
There was a problem hiding this comment.
explain teh part about not being closures. can't function be serialized?
There was a problem hiding this comment.
Only partly, and not in a way that suits a job queue.
- Top-level functions serialize by name.
picklestoresgenerate.conversation.run_session, and the worker imports it. That only works if the worker has the same code. - Closures and bound methods don't, with standard
pickle. Today's jobs would carry things like a live LLM client, which holds an HTTP session, the API key, and an asyncio lock.cloudpicklecan serialize closures by value, but then:- the captured state travels with the job, API keys included;
- live objects like HTTP connections don't survive the trip anyway;
- both sides need the same Python and library versions;
- the result is an opaque blob that a queue message, a manifest entry, or a "which jobs are done?" check can't read.
So jobs stay plain data, and the code that runs them is looked up on the worker side:
{"kind": "session", "persona": "Brian", "repeat": 1, "session": 2, "run_dir": "…"} # the job: JSON
RUNNERS = {"session": run_session, "evaluation": run_evaluation} # the worker
await RUNNERS[job["kind"]](**job)The worker builds its own LLM client from model names and its own environment, so keys never travel with the job. The same JSON is what the manifest in #268 lists, which is how --into knows what's missing. Added a sentence to open question 2.
| 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: |
There was a problem hiding this comment.
we could have retries here? or maybe define a retry function to delegate to generate and judge?
There was a problem hiding this comment.
Added: the loop owns retries (attempts), and each job type says which errors are worth retrying (retryable). That covers what the per-call retries in llm_clients/ miss, like a judge answer failing the schema. Decision 2.
| Each stage then reads top to bottom: | ||
|
|
||
| ```python | ||
| # generate |
There was a problem hiding this comment.
where would that code live?
There was a problem hiding this comment.
generate/run.py lists the jobs, generate/conversation.py defines the Session job; judge/run.py and judge/evaluation.py on the judge side; the loop is workers/loop.py. See "Who does what".
| report(await run_jobs(jobs, max_concurrent)) | ||
|
|
||
| # judge | ||
| jobs = [Evaluation(conversation, judge, instance, run_dir) |
There was a problem hiding this comment.
if a conversation is generated, it can't be judged until the previous stage ends, correct? should it be the case? how would that work?
There was a problem hiding this comment.
It doesn't have to be. A job's run() now returns follow-up jobs, so a finished session could hand back its evaluations and judging would overlap generation. I'd keep the stage barrier first (it avoids the two stages competing for the cap and needing the judge folder up front); follow-ups make streaming a small change later. Open question 1.
There was a problem hiding this comment.
i guess the finished generation job can pt in into the judge queue?
There was a problem hiding this comment.
Yes, that's what follow-ups do: a conversation's last session returns that conversation's Evaluation jobs, so it's judged as soon as it exists while other conversations are still generating. The one requirement is that the judge's run folder and rubric exist before generation starts, which the run-lifecycle order (#268) gives. This replaces the old "keep the stage barrier first" proposal. The remaining open question is whether generation and judging should share one cap or get one each.
|
|
||
| ## 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? |
There was a problem hiding this comment.
i think so, assuming all the sessions are required in order
There was a problem hiding this comment.
Agreed. Session k is a follow-up of session k−1, which keeps them in order, and resume restarts at the first missing session. Decision 1.
| ## 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)? |
There was a problem hiding this comment.
should be limit per judge type, but i think it's not a pressing issue
There was a problem hiding this comment.
Noted: per judge model is the intended behavior. First version has one global cap; per-model caps come with the Shared engine phase. Decision 3.
|
|
||
| 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? |
There was a problem hiding this comment.
we definitely want better viz, but this is for a different PR. at the very least a progress bar per stage, and ideally a box per model that says what is happening (e..g user responding, chatbot responding etc)
There was a problem hiding this comment.
Agreed, separate PR. The loop emits events (job started/finished, user/chatbot responding, judge on question n); the first reporter prints one flushed line per event, and the progress bar per stage plus a box per model come later. Decision 4.
| 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. |
There was a problem hiding this comment.
if we have a manifest, we could just re-do the missing ones correct?
There was a problem hiding this comment.
Yes. config.json (#268) fixes the job list, so done() over it is the manifest: rerunning with --into does exactly the missing jobs. A stage still stops the pipeline if it produced nothing; partial failures warn and continue. Decision 5.
…low-ups Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
|
||
| ```python | ||
| # generate/run.py | ||
| jobs = [first_unfinished_session(persona, repeat, run_dir) |
There was a problem hiding this comment.
consider the current vera pipeline; we run on two sepearate genaration+judging pool
this new loop would make that not needed correct? we could just list all the needed jobs here?
There was a problem hiding this comment.
Yes. The pipeline becomes one run_jobs call with every session job for every user model, instead of one generation pool, then one judging pool, per model in turn. All user models share one concurrency cap, so there's no idle time between models. Run-level steps (results.csv, scores, pooling) still run after all jobs finish, since they need every row. Added a "The pipeline: one job list" section.
| report(await run_jobs(jobs, max_concurrent)) | ||
|
|
||
| # judge | ||
| jobs = [Evaluation(conversation, judge, instance, run_dir) |
There was a problem hiding this comment.
i guess the finished generation job can pt in into the judge queue?
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
| report(await run_jobs(jobs, max_concurrent, attempts=config.attempts)) | ||
|
|
||
| # judge/run.py | ||
| jobs = [Evaluation(conversation, judge, instance, run_dir) |
There was a problem hiding this comment.
unfinished judging?
There was a problem hiding this comment.
Added under decision 5. An Evaluation is done when its TSV exists, and the TSV is written only after the judge has answered every question it needed. So a judging stopped partway through a conversation leaves no TSV, and rerunning with --into judges that conversation again from its first question, while conversations already judged are skipped. results.csv and the scores are rebuilt from the TSVs after the loop, so they only ever reflect finished evaluations. I didn't propose saving progress question by question: one conversation is a few dozen short judge calls, so restarting it is cheap.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
|
||
| ## 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. |
There was a problem hiding this comment.
the config file can differntite between the two correct?
There was a problem hiding this comment.
Yes, it already does: generation.max_concurrent and judging.max_concurrent. So the loop keeps one cap per job kind instead of one in total (decision 6 in 8604717), and a burst of short judge calls can't crowd out generation. The per-judge-model cap (per_judge) still comes later (decision 3).
| ## 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. |
There was a problem hiding this comment.
ok but make sure that it would be possible to di it
There was a problem hiding this comment.
Made it a decision (7 in 8604717), enforced rather than assumed: every job type is a dataclass of JSON-safe fields plus a kind, and a unit test round-trips each job type through JSON and runs the copy. The executor interface itself still waits for a concrete need.
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 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
🟡 Changes recommended
The design can leave unfinished evaluations undiscovered on resume and break stateful-provider conversation continuity by rebuilding clients without restoring provider state.
4 open issues
What changed in this PR
This docs-only PR proposes a shared asynchronous job loop for generation and judging as part of the roadmap’s Shared engine phase. Runtime implementation is deferred to later PRs.
Changes:
- Defines session and evaluation jobs with concurrency caps, retries, and outcomes.
- Proposes pipeline scheduling through dependent jobs, including resume behavior.
- Records design decisions, planned cleanup, and testing requirements.
| File | Description |
|---|---|
docs/design/job-loop.md |
Drafts the shared job-loop design, stage responsibilities, pipeline integration, and validation plan. |
Review effort: Balanced
- 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 <noreply@anthropic.com>
| for kind, cap in max_concurrent.items() | ||
| } | ||
|
|
||
| async def run_one(job: Job) -> list[Outcome]: |
There was a problem hiding this comment.
why are we defining a function within a function? that seems uncnencessary hard to read
There was a problem hiding this comment.
Agreed, it is gone in e7a749c. The loop is now a JobLoop class with small named methods: submit, run, start_jobs_that_fit, has_room, record.
| ```python | ||
| # workers/loop.py | ||
| async def run_jobs( | ||
| jobs: list[Job], max_concurrent: dict[str, int | None], attempts: int = 1 |
There was a problem hiding this comment.
should job be a more generic iterable? so we can have a queue later on
There was a problem hiding this comment.
Yes, in effect. The loop no longer takes a list: jobs go in one at a time through submit(), so a stage can feed it from a list, a generator, or anything else, and follow-ups and retries go in the same way. The per-kind queues are the abstraction an external queue would replace (decision 7). e7a749c.
| # 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)] |
There was a problem hiding this comment.
should those jobs be pushed to the queue/ instead? can we do in a unblocking way?
There was a problem hiding this comment.
and i guess there should be a way of putting evaluation jobs in the eval queue as well
There was a problem hiding this comment.
Yes to both, in e7a749c. Follow-ups go back through submit(), which never waits: it puts the job on its kind's queue and returns, and run() starts it when that kind has room. Each kind has its own queue, so a conversation's last session puts its Evaluation jobs on the evaluation queue, and they run under the judging cap while other conversations are still generating.
| return await gather_flat(run_one(job) for job in jobs) | ||
|
|
||
|
|
||
| async def run_with_retries(job: Job, attempts: int) -> Outcome: |
There was a problem hiding this comment.
what if every job had a # of retires associated with it? and if it fails, the job is pushed back on the queue with an extra counter for the the attempt? would thata be a weird design?
There was a problem hiding this comment.
Not weird, it's the design now (decision 2, e7a749c). A retryable failure puts a copy of the job 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. Hosted queues do the same (a message carries its delivery count), so it carries over to an external queue unchanged.
| A job is plain data plus two methods: | ||
|
|
||
| ```python | ||
| class Job(Protocol): |
There was a problem hiding this comment.
my above suggestion implies having # of failures as an item too
There was a problem hiding this comment.
Yes: every job now has attempt (1 on the first try) and max_attempts (set by the stage from the config) in the Job protocol.
|
|
||
| ```python | ||
| # generate/run.py | ||
| jobs = [Session(persona, repeat, index=1, run_dir=run_dir) |
There was a problem hiding this comment.
this would just be loading the queue
There was a problem hiding this comment.
Right. A stage now just loads the queue with submit() and calls run(); see the updated "Each stage" section.
|
|
||
| ```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) |
There was a problem hiding this comment.
what's index here?
There was a problem hiding this comment.
The session's number within its conversation. Renamed to session and explained in the doc: generation loads only each conversation's first session, and session 2 arrives as session 1's follow-up, and so on.
| 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: |
There was a problem hiding this comment.
in what context do we have more run dirs?
There was a problem hiding this comment.
In vera pipeline: one run 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. Added to the pipeline section.
There was a problem hiding this comment.
and now everything would be loaded in the same queue at once, correct?
There was a problem hiding this comment.
Yes, one queue per kind, shared by every run. The first session of every conversation of every run goes on the session queue up front. Later sessions join it as follow-ups, and evaluations join the evaluation queue as conversations finish. So a pipeline with two user models has one session queue holding both models' conversations, under one cap, instead of one pool per model in turn.
|
|
||
| ## What this removes | ||
|
|
||
| Once `legacy/` is gone (it needs a tagged release that announces the removal; roadmap, Deferred): |
There was a problem hiding this comment.
legacy is removed
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 <noreply@anthropic.com>
| async def run_one(job: Job) -> list[Outcome]: | ||
| if job.done(): # resume = skip what's on disk | ||
| outcome = Outcome.skipped(job) | ||
| class JobLoop: |
There was a problem hiding this comment.
what's the advantage of a class over a function here?
There was a problem hiding this comment.
Shared state. submit is called from three places: the stage loading its first jobs, a finished job handing on its follow-ups, and a failed job going back for a retry. All three need the same queues. With a function, those queues would live in a closure, which means the function-inside-a-function you flagged, or they would have to be passed around. On an object they are just self.queues, and each step is a small named method. It also gives an external queue one thing to replace: anything with submit and run. The doc says this since ac2d3fa.
| 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)) |
There was a problem hiding this comment.
is session the same as attempt?
There was a problem hiding this comment.
No. A session is one stage of a multi-session conversation, from --sessions intake,coaching; each is its own job, run in order. An attempt is one try at the same job: if intake fails with a retryable error, its copy with attempt=2 goes back on the queue. ac2d3fa spells out the difference.
| ``` | ||
|
|
||
| `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. |
There was a problem hiding this comment.
but sessions are not necessary indexby numbers
There was a problem hiding this comment.
Right. Since ac2d3fa a session job carries the configured names (session_types, e.g. ["intake", "coaching"], or one unnamed session when unset) plus position, which entry it runs. The position only keeps them in order; the name is what the job runs and what its transcript is called.
| 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: |
There was a problem hiding this comment.
and now everything would be loaded in the same queue at once, correct?
| 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. |
There was a problem hiding this comment.
is that the right approach?
There was a problem hiding this comment.
I think so, with one clarification, added to decision 2 in d31f071: there is no delay before a retry, on purpose. Errors that pass with time (rate limits, timeouts, 5xx) are already 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 a schema-breaking answer, so the back of the queue is the right place, and the job does not hold a slot meanwhile. The cost to watch is that a retried Session reruns the whole session, so the stage can give sessions fewer attempts than evaluations.
There was a problem hiding this comment.
should the retrie be unified? namely should llm_clients be aware of this mechanistm and communicate? so that we don't have retry code in 2 places
There was a problem hiding this comment.
Yes, and I think the simplest way to unify is to keep retries only in llm_clients/ and drop them from the loop.
I checked: _run_with_retry wraps the whole call, parsing included. So a structured answer that breaks the schema, the main case the loop's retries were meant for, is already re-asked per call with backoff, up to max_llm_retries. A job therefore fails only after its call has used up those retries, meaning the failure is persistent: quota, auth, a broken model. Rerunning the whole job then rarely helps, and for a Session it reruns every turn.
So the proposal:
- One retry mechanism, per call, in
llm_clients/. It retries the smallest unit, so a rate limit on turn 15 re-sends one call, not the session. - The loop doesn't retry. A failed job is recorded as failed; rerunning with
--intopicks it up later, like any missing job.attempt,max_attempts, andretryableleave theJobprotocol, andrecord()loses its retry branch. - The clients communicate through events, not shared retry logic. A client's retry emits a
call_retryingevent through the samereport()jobs use, so the display can show "rate limited, retrying 2/4" against the right conversation.
If you agree, I'll update decision 2 and the loop sketch, and #308 follows the simpler loop.
There was a problem hiding this comment.
Done in f7a11cd. Decision 2 now keeps retries only in llm_clients/, per call. A failed job is recorded and leaves the queue; --into reruns it later. attempt, max_attempts, and retryable are out of the Job protocol and the loop sketch, and per-call retries surface as call_retrying events.
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 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
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 <noreply@anthropic.com>
There was a problem hiding this comment.
🟡 Changes recommended
The proposed loop can skip incomplete evaluations, pool runs across unrelated targets, and omit provider-required sessions, leaving substantive correctness issues unresolved.
5 open issues
Use normalized prepared sessions when scheduling and resuming · New Pooling must retain separate groups for each target · New TSV existence can incorrectly mark an incomplete evaluation as done · New Job scheduling crosses the documented architecture boundary · New Retry policy does not account for exhausted provider failures · New
4 resolved since last review
Review effort: Balanced
| 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. |
| for run_dir in run_dirs: | ||
| write_results_csv(evaluation_dir_for(run_dir)) | ||
| score(evaluation_dir_for(run_dir)) | ||
| pool(run_dirs) |
| 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. |
| # vera_cli/pipeline.py, after creating every run root up front (run-lifecycle.md, step 5) | ||
| loop = JobLoop(max_concurrent) |
| 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. |
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>



Summary
Design draft for discussion, docs only:
docs/design/job-loop.md.Generation and judging each hand-roll the same asyncio queue-and-workers pool, and the copies have drifted: different job shapes, different resume checks, different failure handling (see the table in the doc). The proposal is a single
run_jobs(jobs, max_concurrent), about 15 lines with a semaphore andgather. Each stage then only lists its jobs and says what one job does.Please comment on the five open questions at the end: resume granularity,
per_judgepools vs. a per-provider cap, progress output, failing fast when every job fails, and job-level retries.This is the roadmap's Shared engine phase. Code follows in separate PRs once the design is agreed, and this doc lands with the first of them.
🤖 Generated with Claude Code