From afb82815cc309829c484cd0c9c2b48ae7ebe7013 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:28:17 -0700 Subject: [PATCH 1/6] feat: workers/ job loop, with events and reporters (no callers yet) run_jobs skips jobs whose output is on disk, runs the rest with one concurrency cap per job kind (none for an unlimited kind), retries what a job says is retryable, and records failures as outcomes. Finished and skipped jobs hand on their follow-ups, each checked for done in turn. The loop emits job and stage events; code inside a job emits its own with report(). PlainLineReporter prints timestamped, flushed ASCII lines; EventsFileReporter writes events.jsonl. An import-linter contract keeps workers/ from importing the domain. Design: docs/design/job-loop.md and terminal-interface.md (#266, #271). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + pyproject.toml | 13 + src/vera_mh/workers/__init__.py | 7 + src/vera_mh/workers/events.py | 88 +++++++ src/vera_mh/workers/loop.py | 214 +++++++++++++++++ src/vera_mh/workers/reporters.py | 129 ++++++++++ tests/unit/workers/__init__.py | 0 tests/unit/workers/test_loop.py | 344 +++++++++++++++++++++++++++ tests/unit/workers/test_reporters.py | 121 ++++++++++ 9 files changed, 917 insertions(+) create mode 100644 src/vera_mh/workers/__init__.py create mode 100644 src/vera_mh/workers/events.py create mode 100644 src/vera_mh/workers/loop.py create mode 100644 src/vera_mh/workers/reporters.py create mode 100644 tests/unit/workers/__init__.py create mode 100644 tests/unit/workers/test_loop.py create mode 100644 tests/unit/workers/test_reporters.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f73905819..0243cbf9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - **Config files may leave out output and speed fields**—A `--config` file must still state every field that changes what a run produces (models, inputs or `target`, and for `generate` `turns`, `user_language`, `persona_speaks_first`, `sessions`, `max_total_words`; for `score` and `pool`, `personas` and `skip_risk_analysis`). Fields that only change where output goes or how fast a run goes (`output`, `max_concurrent`, `per_judge`) may now be left out and take the same defaults as the CLI flags, so adding such an option no longer breaks existing configs. `vera judge`'s config `output` now defaults beside the conversations, as `--output` does. Configs that state every field work unchanged. - **A `vera` command**—`uv sync` now installs a `vera` command, so `uv run vera …` (or `vera …` in an activated environment) works from anywhere in the repository; `uv run python vera.py …` still works. Generation now finds its shared prompt files relative to the repository instead of the working directory. The Docker image installs the command and runs it by default; its previous `CMD` used the system Python, which lacked the dependencies. - **User-side language**—`vera generate --user-language ` (config: `generation.user_language`) has the user-side LLM write in that language. Non-English adds one line to the persona prompt; English, the default, leaves the prompt unchanged. +- **Job loop (not used yet)**—`vera_mh.workers` adds the loop that generation and judging will move onto ([docs/design/job-loop.md](docs/design/job-loop.md)): `run_jobs` skips jobs whose output is on disk, runs the rest with one concurrency cap per job kind, retries, and records each job's outcome, and its follow-ups carry a conversation on to its next session and its evaluations. It reports what happens as events, as plain timestamped lines or `events.jsonl`. Nothing calls it yet, so runs are unchanged. ### Testing and CI diff --git a/pyproject.toml b/pyproject.toml index fd65b32a2..c8f9606af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -88,6 +88,19 @@ type = "forbidden" source_modules = ["vera_mh.score"] forbidden_modules = ["vera_mh.llm_clients"] +# docs/architecture.md, Layer model: the domain hands its jobs to the loop, +# never the reverse. +[[tool.importlinter.contracts]] +name = "workers does not import the domain" +type = "forbidden" +source_modules = ["vera_mh.workers"] +forbidden_modules = [ + "vera_mh.generate", + "vera_mh.judge", + "vera_mh.score", + "vera_mh.vera_cli", +] + [tool.pyright] typeCheckingMode = "basic" diff --git a/src/vera_mh/workers/__init__.py b/src/vera_mh/workers/__init__.py new file mode 100644 index 000000000..8a620ecf2 --- /dev/null +++ b/src/vera_mh/workers/__init__.py @@ -0,0 +1,7 @@ +"""The job loop: runs generation and judging jobs with a concurrency cap per kind. + +Import the submodule you need (`workers.loop`, `workers.events`, +`workers.reporters`). Nothing here imports a domain package (`generate`, +`judge`, `score`, `vera_cli`): the domain hands its jobs to the loop, never the +reverse. Design: docs/design/job-loop.md and docs/design/terminal-interface.md. +""" diff --git a/src/vera_mh/workers/events.py b/src/vera_mh/workers/events.py new file mode 100644 index 000000000..830285a25 --- /dev/null +++ b/src/vera_mh/workers/events.py @@ -0,0 +1,88 @@ +"""What happens during a run, as events a reporter shows or writes down. + +The loop emits the job events itself: started, finished, failed, retrying, +skipped, and the stage finished at the end. Code running inside a job emits +the rest through `report()` (a turn started, a judge question asked, a call's +token usage), without the loop or the job having to pass a reporter down. + +Every event's fields are JSON-safe, so the same event can be printed as a line +or written to `events.jsonl`. +""" + +from __future__ import annotations + +import contextlib +from contextvars import ContextVar +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Iterator, Protocol + +# Emitted by the loop. +JOB_STARTED = "job_started" +JOB_FINISHED = "job_finished" +JOB_FAILED = "job_failed" +JOB_RETRYING = "job_retrying" +JOB_SKIPPED = "job_skipped" +STAGE_FINISHED = "stage_finished" + +# Emitted from inside a job, with `report()`. +TURN_STARTED = "turn_started" # data: turn, speaker ("user" or "chatbot") +QUESTION_ASKED = "question_asked" # data: question_id +CALL_FINISHED = "call_finished" # data: model, input_tokens, output_tokens + + +@dataclass(frozen=True) +class Event: + """One thing that happened. `job` and `kind` are empty for a stage event.""" + + type: str + job: str | None = None + kind: str | None = None + data: dict[str, Any] = field(default_factory=dict) + time: datetime = field(default_factory=lambda: datetime.now().astimezone()) + + def to_dict(self) -> dict[str, Any]: + """The event as one flat JSON object, as written to `events.jsonl`.""" + return { + **self.data, + "time": self.time.isoformat(timespec="seconds"), + "type": self.type, + "kind": self.kind, + "job": self.job, + } + + +class Reporter(Protocol): + """Shows or records events: see `workers.reporters`.""" + + def emit(self, event: Event) -> None: ... + + +# The reporter and job of the job running in the current asyncio task. The loop +# sets it around each job's `run()`; each job runs in its own task, so +# concurrent jobs don't see each other's. +_running: ContextVar[tuple[Reporter, str, str] | None] = ContextVar( + "running_job", default=None +) + + +@contextlib.contextmanager +def running_job(reporter: Reporter, label: str, kind: str) -> Iterator[None]: + """While inside, `report()` emits events for this job to `reporter`.""" + token = _running.set((reporter, label, kind)) + try: + yield + finally: + _running.reset(token) + + +def report(event_type: str, **data: Any) -> None: + """Emit an event from inside a running job, such as `TURN_STARTED`. + + Outside a job (a unit test calling the code directly, say) it does nothing. + """ + running = _running.get() + if running is None: + return + reporter, label, kind = running + reporter.emit(Event(event_type, job=label, kind=kind, data=data)) diff --git a/src/vera_mh/workers/loop.py b/src/vera_mh/workers/loop.py new file mode 100644 index 000000000..aa342afd0 --- /dev/null +++ b/src/vera_mh/workers/loop.py @@ -0,0 +1,214 @@ +"""The job loop: all the concurrency code for generation and judging, in one place. + +A job is plain data (paths, model names, indices) plus four methods; see `Job`. +`run_jobs` skips every job whose output is already on disk, runs the rest with +one concurrency cap per job kind, retries what the job says is worth retrying, +and records each job's outcome. A job's failure is an outcome, never an +exception, so one bad conversation doesn't stop the others. + +Jobs that wait on another job are its follow-ups: the next session of a +conversation, or a finished conversation's evaluations. A job hands them on +when it finishes, and also when it is skipped as done, so resuming a run +reaches every missing job however far down the chain it is, and redoes nothing. + +Design: docs/design/job-loop.md. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import time +from collections import Counter +from dataclasses import dataclass +from typing import Awaitable, Iterable, Literal, Mapping, Protocol + +from vera_mh.workers.events import ( + JOB_FAILED, + JOB_FINISHED, + JOB_RETRYING, + JOB_SKIPPED, + JOB_STARTED, + STAGE_FINISHED, + Event, + Reporter, + running_job, +) +from vera_mh.workers.reporters import Reporters + + +class Job(Protocol): + """One unit of work: one session of a conversation, or one evaluation. + + Keep jobs plain data, with JSON-safe fields, so they could be sent to an + external queue one day (job-loop.md, decision 7). + """ + + kind: str # "session" or "evaluation": picks the concurrency cap + label: str # how the job is shown, such as "gpt-5.2 Brian" + + def done(self) -> bool: + """Is this job's output already on disk?""" + ... + + async def run(self) -> None: + """Do the work and write the output.""" + ... + + def follow_ups(self) -> list[Job]: + """The jobs waiting on this one, worked out from its fields and disk + alone, so a job skipped as done hands them on too.""" + ... + + def retryable(self, error: Exception) -> bool: + """Is running the job again worth it after this error?""" + ... + + +Status = Literal["finished", "failed", "skipped"] + + +@dataclass +class Outcome: + """What happened to one job.""" + + job: Job + status: Status + error: Exception | None = None + attempts: int = 0 + seconds: float = 0.0 + + @property + def finished(self) -> bool: + return self.status == "finished" + + @property + def failed(self) -> bool: + return self.status == "failed" + + @property + def skipped(self) -> bool: + return self.status == "skipped" + + +async def run_jobs( + jobs: Iterable[Job], + max_concurrent: Mapping[str, int | None], + attempts: int = 1, + *, + reporter: Reporter | None = None, + stage: str = "jobs", +) -> list[Outcome]: + """Run `jobs` and their follow-ups; return every job's outcome. + + `max_concurrent` caps how many jobs of each kind run at once; None or 0 + means no limit. Every job's kind must have an entry. `attempts` is how many + times a job may run, counting the first. `stage` names the run in the + final `STAGE_FINISHED` event. + """ + if attempts < 1: + raise ValueError(f"attempts must be at least 1, not {attempts}") + report = reporter or Reporters() + # No semaphore for an unlimited kind: one sized from `jobs` would throttle + # the follow-ups, which can outnumber them. + 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]: + try: + is_done = job.done() + except Exception as error: + outcome = Outcome(job, "failed", error=error) + else: + if is_done: + outcome = Outcome(job, "skipped") + else: + async with limits[job.kind]: + outcome = await _run_with_retries(job, attempts, report) + + follow_ups: list[Job] = [] + if not outcome.failed: + try: + follow_ups = job.follow_ups() + except Exception as error: + # The jobs waiting on this one can't be listed, so they would be + # missing without a trace: count the job as failed instead. + outcome = Outcome( + job, + "failed", + error=error, + attempts=outcome.attempts, + seconds=outcome.seconds, + ) + _emit_outcome(report, outcome) + return [ + outcome, + *await _gather_flat(run_one(follow_up) for follow_up in follow_ups), + ] + + outcomes = await _gather_flat(run_one(job) for job in jobs) + counts = Counter(outcome.status for outcome in outcomes) + report.emit( + Event( + STAGE_FINISHED, + data={ + "stage": stage, + "done": counts["finished"], + "failed": counts["failed"], + "skipped": counts["skipped"], + }, + ) + ) + return outcomes + + +async def _run_with_retries(job: Job, attempts: int, report: Reporter) -> Outcome: + report.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) + start = time.monotonic() + with running_job(report, job.label, job.kind): + for attempt in range(1, attempts + 1): + try: + await job.run() + except Exception as error: + if attempt == attempts or not job.retryable(error): + return Outcome( + job, + "failed", + error=error, + attempts=attempt, + seconds=_since(start), + ) + retry = {"attempt": attempt, "error": _describe(error)} + report.emit( + Event(JOB_RETRYING, job=job.label, kind=job.kind, data=retry) + ) + else: + return Outcome(job, "finished", attempts=attempt, seconds=_since(start)) + raise AssertionError("unreachable: run_jobs checks attempts >= 1") + + +def _emit_outcome(report: Reporter, outcome: Outcome) -> None: + job = outcome.job + if outcome.skipped: + report.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) + elif outcome.finished: + data = {"seconds": round(outcome.seconds, 3), "attempts": outcome.attempts} + report.emit(Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data)) + else: + data = {"error": _describe(outcome.error), "attempts": outcome.attempts} + report.emit(Event(JOB_FAILED, job=job.label, kind=job.kind, data=data)) + + +async def _gather_flat(runs: Iterable[Awaitable[list[Outcome]]]) -> list[Outcome]: + results = await asyncio.gather(*runs) + return [outcome for result in results for outcome in result] + + +def _since(start: float) -> float: + return time.monotonic() - start + + +def _describe(error: BaseException | None) -> str: + return f"{type(error).__name__}: {error}" if error else "" diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py new file mode 100644 index 000000000..3e55e8928 --- /dev/null +++ b/src/vera_mh/workers/reporters.py @@ -0,0 +1,129 @@ +"""Reporters: where events go. + +- `PlainLineReporter` prints one flushed, timestamped line per event, so output + piped to `tee` or a CI log is never silent while work is happening. ASCII only. +- `EventsFileReporter` appends each event to `events.jsonl` as one JSON object, + for scripts and notebooks. +- `Reporters` sends each event to several reporters. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import TextIO + +from vera_mh.workers.events import ( + JOB_FAILED, + JOB_FINISHED, + JOB_RETRYING, + JOB_SKIPPED, + JOB_STARTED, + QUESTION_ASKED, + STAGE_FINISHED, + TURN_STARTED, + Event, + Reporter, +) + + +class Reporters: + """Sends each event to every reporter given, in order.""" + + def __init__(self, *reporters: Reporter): + self.reporters = reporters + + def emit(self, event: Event) -> None: + for reporter in self.reporters: + reporter.emit(event) + + +class EventsFileReporter: + """Appends each event to a JSON Lines file, one object per line.""" + + def __init__(self, path: Path): + self.path = path + + def emit(self, event: Event) -> None: + # Opened per event: a run makes a few events per model call, so the cost + # is nothing, and a crash never leaves a buffered event unwritten. + with open(self.path, "a", encoding="utf-8") as f: + f.write(json.dumps(event.to_dict(), ensure_ascii=False, default=str)) + f.write("\n") + + +# The word each job event shows as, padded so the labels line up. +JOB_WORDS = { + JOB_STARTED: "start", + JOB_FINISHED: "done", + JOB_FAILED: "FAIL", + JOB_RETRYING: "retry", + JOB_SKIPPED: "skip", +} +# In-job events are shown only with `verbose`: a run has many of them. +IN_JOB_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"} + + +class PlainLineReporter: + """Prints one line per event, flushed at once: + + 19:44:06 session done gpt-5.2 Brian 2m41s + 19:45:10 session FAIL claude-opus-4-5 Abigail RateLimitError: ... + """ + + def __init__(self, stream: TextIO | None = None, verbose: bool = False): + self.stream = stream or sys.stdout + self.verbose = verbose + + def emit(self, event: Event) -> None: + line = self.format(event) + if line is not None: + print(line, file=self.stream, flush=True) + + def format(self, event: Event) -> str | None: + """The line for `event`, or None if it isn't shown.""" + clock = event.time.strftime("%H:%M:%S") + if event.type == STAGE_FINISHED: + counts = event.data + return ( + f"{clock} {counts.get('stage', '')} finished: " + f"{counts.get('done', 0)} done, {counts.get('failed', 0)} failed, " + f"{counts.get('skipped', 0)} skipped" + ) + if event.type in JOB_WORDS: + word = JOB_WORDS[event.type] + elif self.verbose and event.type in IN_JOB_WORDS: + word = IN_JOB_WORDS[event.type] + else: + return None + detail = _detail(event) + line = f"{clock} {event.kind or '':<11} {word:<6} {event.job or ''}" + return f"{line} {detail}" if detail else line + + +def _detail(event: Event) -> str: + data = event.data + if event.type == JOB_FINISHED: + return duration(data.get("seconds", 0)) + if event.type == JOB_FAILED: + return str(data.get("error", "")) + if event.type == JOB_RETRYING: + return f"attempt {data.get('attempt')} failed: {data.get('error', '')}" + if event.type == TURN_STARTED: + return f"turn {data.get('turn')} {data.get('speaker')} responding" + if event.type == QUESTION_ASKED: + return f"question {data.get('question_id')}" + return "" + + +def duration(seconds: float) -> str: + """`45s`, `2m41s`, `1h02m`.""" + seconds = int(round(seconds)) + if seconds < 60: + return f"{seconds}s" + minutes, seconds = divmod(seconds, 60) + if minutes < 60: + return f"{minutes}m{seconds:02d}s" + hours, minutes = divmod(minutes, 60) + return f"{hours}h{minutes:02d}m" diff --git a/tests/unit/workers/__init__.py b/tests/unit/workers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py new file mode 100644 index 000000000..2a12d22c7 --- /dev/null +++ b/tests/unit/workers/test_loop.py @@ -0,0 +1,344 @@ +"""The job loop: skipping, concurrency caps, retries, follow-ups, and events.""" + +from __future__ import annotations + +import asyncio +import json +from collections import Counter +from dataclasses import asdict, dataclass, field +from pathlib import Path + +import pytest + +from vera_mh.workers import events +from vera_mh.workers.events import Event +from vera_mh.workers.loop import Job, run_jobs + +UNLIMITED: dict[str, int | None] = {"session": None, "evaluation": None} + + +class Tracker: + """What the fake jobs did: the order things happened in, and how many jobs + of each kind ran at once.""" + + def __init__(self) -> None: + self.log: list[str] = [] + self.running: Counter[str] = Counter() + self.peak: Counter[str] = Counter() + + +@dataclass +class FakeJob: + label: str + tracker: Tracker + kind: str = "session" + is_done: bool = False + # run() raises this many times before it succeeds. + failures: int = 0 + error: type[Exception] = RuntimeError + retry_on: tuple[type[Exception], ...] = (RuntimeError,) + next_jobs: list[FakeJob] = field(default_factory=list) + # How long run() takes, so jobs overlap. + seconds: float = 0.0 + runs: int = 0 + + def done(self) -> bool: + return self.is_done + + async def run(self) -> None: + self.runs += 1 + self.tracker.log.append(f"start {self.label}") + self.tracker.running[self.kind] += 1 + self.tracker.peak[self.kind] = max( + self.tracker.peak[self.kind], self.tracker.running[self.kind] + ) + try: + await asyncio.sleep(self.seconds) + if self.runs <= self.failures: + raise self.error(f"{self.label} failed") + finally: + self.tracker.running[self.kind] -= 1 + self.tracker.log.append(f"end {self.label}") + + def follow_ups(self) -> list[Job]: + return list(self.next_jobs) + + def retryable(self, error: Exception) -> bool: + return isinstance(error, self.retry_on) + + +class ListReporter: + def __init__(self) -> None: + self.events: list[Event] = [] + + def emit(self, event: Event) -> None: + self.events.append(event) + + def types(self, label: str) -> list[str]: + return [event.type for event in self.events if event.job == label] + + +def statuses(outcomes) -> dict[str, str]: + return {outcome.job.label: outcome.status for outcome in outcomes} + + +async def test_a_done_job_is_skipped_and_not_run() -> None: + tracker = Tracker() + done = FakeJob("done", tracker, is_done=True) + todo = FakeJob("todo", tracker) + + outcomes = await run_jobs([done, todo], UNLIMITED) + + assert statuses(outcomes) == {"done": "skipped", "todo": "finished"} + assert done.runs == 0 + assert todo.runs == 1 + + +async def test_each_kind_has_its_own_cap() -> None: + tracker = Tracker() + sessions = [FakeJob(f"s{i}", tracker, seconds=0.02) for i in range(6)] + evaluations = [ + FakeJob(f"e{i}", tracker, kind="evaluation", seconds=0.02) for i in range(6) + ] + + outcomes = await run_jobs( + [*sessions, *evaluations], {"session": 2, "evaluation": 3} + ) + + assert all(outcome.finished for outcome in outcomes) + assert tracker.peak == Counter({"session": 2, "evaluation": 3}) + + +async def test_a_failure_is_an_outcome_not_an_exception() -> None: + tracker = Tracker() + bad = FakeJob("bad", tracker, failures=1, error=ValueError, retry_on=()) + good = FakeJob("good", tracker) + + outcomes = await run_jobs([bad, good], UNLIMITED) + + assert statuses(outcomes) == {"bad": "failed", "good": "finished"} + failed = next(outcome for outcome in outcomes if outcome.failed) + assert isinstance(failed.error, ValueError) + + +async def test_retries_stop_at_attempts() -> None: + tracker = Tracker() + job = FakeJob("flaky", tracker, failures=5) + + (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) + + assert outcome.failed + assert job.runs == 3 + assert outcome.attempts == 3 + + +async def test_a_retry_can_succeed() -> None: + tracker = Tracker() + job = FakeJob("flaky", tracker, failures=1) + + (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) + + assert outcome.finished + assert outcome.attempts == 2 + + +async def test_an_error_that_isnt_retryable_is_not_retried() -> None: + tracker = Tracker() + job = FakeJob("broken", tracker, failures=5, error=ValueError) + + (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) + + assert outcome.failed + assert job.runs == 1 + + +async def test_follow_ups_run_after_their_parent() -> None: + tracker = Tracker() + child = FakeJob("child", tracker, kind="evaluation") + parent = FakeJob("parent", tracker, next_jobs=[child]) + + outcomes = await run_jobs([parent], UNLIMITED) + + assert statuses(outcomes) == {"parent": "finished", "child": "finished"} + assert tracker.log == ["start parent", "end parent", "start child", "end child"] + + +async def test_follow_ups_dont_run_when_their_parent_fails() -> None: + tracker = Tracker() + child = FakeJob("child", tracker) + parent = FakeJob("parent", tracker, failures=1, next_jobs=[child]) + + outcomes = await run_jobs([parent], UNLIMITED) + + assert statuses(outcomes) == {"parent": "failed"} + assert child.runs == 0 + + +async def test_a_skipped_job_hands_on_its_follow_ups_each_checked_for_done() -> None: + """Resume: sessions 1 and 2 are on disk, session 3 isn't, and after the + last session come the evaluations. Only the missing work runs.""" + tracker = Tracker() + evaluation = FakeJob("evaluation", tracker, kind="evaluation") + session_3 = FakeJob("session 3", tracker, next_jobs=[evaluation]) + session_2 = FakeJob("session 2", tracker, is_done=True, next_jobs=[session_3]) + session_1 = FakeJob("session 1", tracker, is_done=True, next_jobs=[session_2]) + + outcomes = await run_jobs([session_1], UNLIMITED) + + assert statuses(outcomes) == { + "session 1": "skipped", + "session 2": "skipped", + "session 3": "finished", + "evaluation": "finished", + } + assert session_1.runs == session_2.runs == 0 + + +@pytest.mark.parametrize("unlimited", [None, 0]) +async def test_follow_ups_of_an_unlimited_kind_all_run_at_once(unlimited) -> None: + """One job hands on five follow-ups. With no cap they must all run at + once, even though the list `run_jobs` was given held a single job.""" + tracker = Tracker() + evaluations = [ + FakeJob(f"e{i}", tracker, kind="evaluation", seconds=0.05) for i in range(5) + ] + session = FakeJob("session", tracker, next_jobs=evaluations) + + await run_jobs([session], {"session": unlimited, "evaluation": unlimited}) + + assert tracker.peak["evaluation"] == 5 + + +@pytest.mark.parametrize("unlimited", [None, 0]) +async def test_none_and_zero_both_mean_no_limit(unlimited) -> None: + tracker = Tracker() + jobs = [FakeJob(f"s{i}", tracker, seconds=0.05) for i in range(4)] + + await run_jobs(jobs, {"session": unlimited}) + + assert tracker.peak["session"] == 4 + + +async def test_done_or_follow_ups_raising_counts_the_job_as_failed() -> None: + class BrokenDone(FakeJob): + def done(self) -> bool: + raise OSError("disk gone") + + class BrokenFollowUps(FakeJob): + def follow_ups(self) -> list[Job]: + raise OSError("disk gone") + + tracker = Tracker() + outcomes = await run_jobs( + [BrokenDone("done", tracker), BrokenFollowUps("follow-ups", tracker)], + UNLIMITED, + ) + + assert statuses(outcomes) == {"done": "failed", "follow-ups": "failed"} + + +async def test_attempts_must_be_at_least_one() -> None: + with pytest.raises(ValueError): + await run_jobs([], UNLIMITED, attempts=0) + + +# ------------------------------------------------------------------------ events + + +async def test_the_loop_reports_each_job_and_the_stage() -> None: + tracker = Tracker() + reporter = ListReporter() + jobs = [ + FakeJob("ok", tracker), + FakeJob("flaky", tracker, failures=1), + FakeJob("bad", tracker, failures=1, error=ValueError), + FakeJob("done", tracker, is_done=True), + ] + + await run_jobs(jobs, UNLIMITED, attempts=2, reporter=reporter, stage="generate") + + assert reporter.types("ok") == [events.JOB_STARTED, events.JOB_FINISHED] + assert reporter.types("flaky") == [ + events.JOB_STARTED, + events.JOB_RETRYING, + events.JOB_FINISHED, + ] + assert reporter.types("bad") == [events.JOB_STARTED, events.JOB_FAILED] + assert reporter.types("done") == [events.JOB_SKIPPED] + stage = reporter.events[-1] + assert stage.type == events.STAGE_FINISHED + assert stage.data == {"stage": "generate", "done": 2, "failed": 1, "skipped": 1} + + +async def test_a_job_reports_from_inside_its_run() -> None: + @dataclass + class Talker: + label: str + kind: str = "session" + + def done(self) -> bool: + return False + + async def run(self) -> None: + events.report(events.TURN_STARTED, turn=1, speaker="user") + + def follow_ups(self) -> list[Job]: + return [] + + def retryable(self, error: Exception) -> bool: + return False + + reporter = ListReporter() + + await run_jobs([Talker("a"), Talker("b")], UNLIMITED, reporter=reporter) + + turns = [event for event in reporter.events if event.type == events.TURN_STARTED] + assert {event.job for event in turns} == {"a", "b"} + assert all(event.kind == "session" for event in turns) + assert turns[0].data == {"turn": 1, "speaker": "user"} + + +def test_report_outside_a_job_does_nothing() -> None: + events.report(events.TURN_STARTED, turn=1, speaker="user") + + +# --------------------------------------------------------- jobs stay serializable + + +@dataclass +class SampleSession: + """What a real job looks like: JSON-safe fields, plus the four methods.""" + + label: str + persona: str + repeat: int + session: int + run_dir: str + kind: str = "session" + + def output(self) -> Path: + name = f"{self.persona}_run{self.repeat}_session{self.session}.txt" + return Path(self.run_dir) / name + + def done(self) -> bool: + return self.output().exists() + + async def run(self) -> None: + self.output().write_text("user: hi\n", encoding="utf-8") + + def follow_ups(self) -> list[Job]: + return [] + + def retryable(self, error: Exception) -> bool: + return False + + +async def test_a_job_survives_a_json_round_trip_and_runs(tmp_path: Path) -> None: + job = SampleSession("gpt-5.2 Brian", "Brian", 1, 2, str(tmp_path)) + + copy = SampleSession(**json.loads(json.dumps(asdict(job)))) + + assert copy == job + (outcome,) = await run_jobs([copy], UNLIMITED) + assert outcome.finished + assert job.done() diff --git a/tests/unit/workers/test_reporters.py b/tests/unit/workers/test_reporters.py new file mode 100644 index 000000000..6820c1174 --- /dev/null +++ b/tests/unit/workers/test_reporters.py @@ -0,0 +1,121 @@ +"""Reporters: the plain lines a person reads, and the events.jsonl scripts read.""" + +from __future__ import annotations + +import io +import json +from datetime import datetime, timezone +from pathlib import Path + +from vera_mh.workers import events +from vera_mh.workers.events import Event +from vera_mh.workers.reporters import ( + EventsFileReporter, + PlainLineReporter, + Reporters, + duration, +) + +AT = datetime(2026, 10, 30, 19, 44, 6, tzinfo=timezone.utc) + + +def job_event(event_type: str, **data) -> Event: + return Event(event_type, job="gpt-5.2 Brian", kind="session", data=data, time=AT) + + +def test_plain_lines_show_each_job_event() -> None: + reporter = PlainLineReporter(io.StringIO()) + + started = reporter.format(job_event(events.JOB_STARTED)) + finished = reporter.format(job_event(events.JOB_FINISHED, seconds=161)) + failed = reporter.format( + job_event(events.JOB_FAILED, error="RateLimitError: slow down") + ) + skipped = reporter.format(job_event(events.JOB_SKIPPED)) + + assert started == "19:44:06 session start gpt-5.2 Brian" + assert finished == "19:44:06 session done gpt-5.2 Brian 2m41s" + assert failed == ( + "19:44:06 session FAIL gpt-5.2 Brian RateLimitError: slow down" + ) + assert skipped == "19:44:06 session skip gpt-5.2 Brian" + + +def test_the_stage_line_counts_outcomes() -> None: + reporter = PlainLineReporter(io.StringIO()) + stage = Event( + events.STAGE_FINISHED, + data={"stage": "generate", "done": 196, "failed": 2, "skipped": 0}, + time=AT, + ) + + assert reporter.format(stage) == ( + "19:44:06 generate finished: 196 done, 2 failed, 0 skipped" + ) + + +def test_in_job_events_show_only_when_verbose() -> None: + turn = job_event(events.TURN_STARTED, turn=14, speaker="chatbot") + tokens = job_event(events.CALL_FINISHED, input_tokens=10, output_tokens=5) + + assert PlainLineReporter(io.StringIO()).format(turn) is None + verbose = PlainLineReporter(io.StringIO(), verbose=True) + assert verbose.format(turn) == ( + "19:44:06 session turn gpt-5.2 Brian turn 14 chatbot responding" + ) + # Token usage goes to events.jsonl, not the screen. + assert verbose.format(tokens) is None + + +def test_plain_lines_are_written_and_ascii() -> None: + stream = io.StringIO() + reporter = PlainLineReporter(stream) + + reporter.emit(job_event(events.JOB_STARTED)) + reporter.emit(job_event(events.TURN_STARTED, turn=1, speaker="user")) + reporter.emit(job_event(events.JOB_FINISHED, seconds=45)) + + lines = stream.getvalue().splitlines() + assert len(lines) == 2 + assert all(line.isascii() for line in lines) + + +def test_events_file_has_one_json_object_per_event(tmp_path: Path) -> None: + path = tmp_path / "events.jsonl" + reporter = EventsFileReporter(path) + + reporter.emit(job_event(events.JOB_STARTED)) + reporter.emit(job_event(events.CALL_FINISHED, input_tokens=10, output_tokens=5)) + + lines = path.read_text(encoding="utf-8").splitlines() + assert [json.loads(line) for line in lines] == [ + { + "time": "2026-10-30T19:44:06+00:00", + "type": "job_started", + "kind": "session", + "job": "gpt-5.2 Brian", + }, + { + "input_tokens": 10, + "output_tokens": 5, + "time": "2026-10-30T19:44:06+00:00", + "type": "call_finished", + "kind": "session", + "job": "gpt-5.2 Brian", + }, + ] + + +def test_reporters_send_each_event_to_all() -> None: + first, second = io.StringIO(), io.StringIO() + reporter = Reporters(PlainLineReporter(first), PlainLineReporter(second)) + + reporter.emit(job_event(events.JOB_STARTED)) + + assert first.getvalue() == second.getvalue() != "" + + +def test_durations() -> None: + assert duration(45) == "45s" + assert duration(161) == "2m41s" + assert duration(3720) == "1h02m" From a0f250f4fb5dc771e9992324652abf42ae44914f Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:46:55 -0700 Subject: [PATCH 2/6] refactor: simpler job loop One try around a job's checks instead of two; a retry loop with no unreachable line; Outcome compared by status; reporters index the fields our own events always carry; the running job as a named type. Co-Authored-By: Claude Opus 5.5 --- src/vera_mh/workers/events.py | 23 +++++++------- src/vera_mh/workers/loop.py | 51 ++++++++------------------------ src/vera_mh/workers/reporters.py | 15 +++++----- tests/unit/workers/test_loop.py | 12 ++++---- 4 files changed, 39 insertions(+), 62 deletions(-) diff --git a/src/vera_mh/workers/events.py b/src/vera_mh/workers/events.py index 830285a25..920e15d25 100644 --- a/src/vera_mh/workers/events.py +++ b/src/vera_mh/workers/events.py @@ -58,18 +58,23 @@ class Reporter(Protocol): def emit(self, event: Event) -> None: ... -# The reporter and job of the job running in the current asyncio task. The loop +@dataclass(frozen=True) +class _RunningJob: + reporter: Reporter + label: str + kind: str + + +# The job running in the current asyncio task, and where its events go. The loop # sets it around each job's `run()`; each job runs in its own task, so # concurrent jobs don't see each other's. -_running: ContextVar[tuple[Reporter, str, str] | None] = ContextVar( - "running_job", default=None -) +_running: ContextVar[_RunningJob | None] = ContextVar("running_job", default=None) @contextlib.contextmanager def running_job(reporter: Reporter, label: str, kind: str) -> Iterator[None]: """While inside, `report()` emits events for this job to `reporter`.""" - token = _running.set((reporter, label, kind)) + token = _running.set(_RunningJob(reporter, label, kind)) try: yield finally: @@ -81,8 +86,6 @@ def report(event_type: str, **data: Any) -> None: Outside a job (a unit test calling the code directly, say) it does nothing. """ - running = _running.get() - if running is None: - return - reporter, label, kind = running - reporter.emit(Event(event_type, job=label, kind=kind, data=data)) + job = _running.get() + if job is not None: + job.reporter.emit(Event(event_type, job=job.label, kind=job.kind, data=data)) diff --git a/src/vera_mh/workers/loop.py b/src/vera_mh/workers/loop.py index aa342afd0..3609cccb0 100644 --- a/src/vera_mh/workers/loop.py +++ b/src/vera_mh/workers/loop.py @@ -78,18 +78,6 @@ class Outcome: attempts: int = 0 seconds: float = 0.0 - @property - def finished(self) -> bool: - return self.status == "finished" - - @property - def failed(self) -> bool: - return self.status == "failed" - - @property - def skipped(self) -> bool: - return self.status == "skipped" - async def run_jobs( jobs: Iterable[Job], @@ -118,30 +106,17 @@ async def run_jobs( async def run_one(job: Job) -> list[Outcome]: try: - is_done = job.done() - except Exception as error: - outcome = Outcome(job, "failed", error=error) - else: - if is_done: + if job.done(): outcome = Outcome(job, "skipped") else: async with limits[job.kind]: outcome = await _run_with_retries(job, attempts, report) - - follow_ups: list[Job] = [] - if not outcome.failed: - try: - follow_ups = job.follow_ups() - except Exception as error: - # The jobs waiting on this one can't be listed, so they would be - # missing without a trace: count the job as failed instead. - outcome = Outcome( - job, - "failed", - error=error, - attempts=outcome.attempts, - seconds=outcome.seconds, - ) + follow_ups = job.follow_ups() if outcome.status != "failed" else [] + except Exception as error: + # `done()` or `follow_ups()` crashed, on a corrupt file say. Count the + # job as failed rather than stop the run, or lose the jobs waiting on it + # without a trace. + outcome, follow_ups = Outcome(job, "failed", error=error), [] _emit_outcome(report, outcome) return [ outcome, @@ -167,10 +142,12 @@ async def run_one(job: Job) -> list[Outcome]: async def _run_with_retries(job: Job, attempts: int, report: Reporter) -> Outcome: report.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) start = time.monotonic() + attempt = 1 with running_job(report, job.label, job.kind): - for attempt in range(1, attempts + 1): + while True: try: await job.run() + return Outcome(job, "finished", attempts=attempt, seconds=_since(start)) except Exception as error: if attempt == attempts or not job.retryable(error): return Outcome( @@ -184,16 +161,14 @@ async def _run_with_retries(job: Job, attempts: int, report: Reporter) -> Outcom report.emit( Event(JOB_RETRYING, job=job.label, kind=job.kind, data=retry) ) - else: - return Outcome(job, "finished", attempts=attempt, seconds=_since(start)) - raise AssertionError("unreachable: run_jobs checks attempts >= 1") + attempt += 1 def _emit_outcome(report: Reporter, outcome: Outcome) -> None: job = outcome.job - if outcome.skipped: + if outcome.status == "skipped": report.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) - elif outcome.finished: + elif outcome.status == "finished": data = {"seconds": round(outcome.seconds, 3), "attempts": outcome.attempts} report.emit(Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data)) else: diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py index 3e55e8928..09e9dad72 100644 --- a/src/vera_mh/workers/reporters.py +++ b/src/vera_mh/workers/reporters.py @@ -87,9 +87,8 @@ def format(self, event: Event) -> str | None: if event.type == STAGE_FINISHED: counts = event.data return ( - f"{clock} {counts.get('stage', '')} finished: " - f"{counts.get('done', 0)} done, {counts.get('failed', 0)} failed, " - f"{counts.get('skipped', 0)} skipped" + f"{clock} {counts['stage']} finished: {counts['done']} done, " + f"{counts['failed']} failed, {counts['skipped']} skipped" ) if event.type in JOB_WORDS: word = JOB_WORDS[event.type] @@ -105,15 +104,15 @@ def format(self, event: Event) -> str | None: def _detail(event: Event) -> str: data = event.data if event.type == JOB_FINISHED: - return duration(data.get("seconds", 0)) + return duration(data["seconds"]) if event.type == JOB_FAILED: - return str(data.get("error", "")) + return data["error"] if event.type == JOB_RETRYING: - return f"attempt {data.get('attempt')} failed: {data.get('error', '')}" + return f"attempt {data['attempt']} failed: {data['error']}" if event.type == TURN_STARTED: - return f"turn {data.get('turn')} {data.get('speaker')} responding" + return f"turn {data['turn']} {data['speaker']} responding" if event.type == QUESTION_ASKED: - return f"question {data.get('question_id')}" + return f"question {data['question_id']}" return "" diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py index 2a12d22c7..fd4cb09a2 100644 --- a/tests/unit/workers/test_loop.py +++ b/tests/unit/workers/test_loop.py @@ -105,7 +105,7 @@ async def test_each_kind_has_its_own_cap() -> None: [*sessions, *evaluations], {"session": 2, "evaluation": 3} ) - assert all(outcome.finished for outcome in outcomes) + assert all(outcome.status == "finished" for outcome in outcomes) assert tracker.peak == Counter({"session": 2, "evaluation": 3}) @@ -117,7 +117,7 @@ async def test_a_failure_is_an_outcome_not_an_exception() -> None: outcomes = await run_jobs([bad, good], UNLIMITED) assert statuses(outcomes) == {"bad": "failed", "good": "finished"} - failed = next(outcome for outcome in outcomes if outcome.failed) + failed = next(outcome for outcome in outcomes if outcome.status == "failed") assert isinstance(failed.error, ValueError) @@ -127,7 +127,7 @@ async def test_retries_stop_at_attempts() -> None: (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) - assert outcome.failed + assert outcome.status == "failed" assert job.runs == 3 assert outcome.attempts == 3 @@ -138,7 +138,7 @@ async def test_a_retry_can_succeed() -> None: (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) - assert outcome.finished + assert outcome.status == "finished" assert outcome.attempts == 2 @@ -148,7 +148,7 @@ async def test_an_error_that_isnt_retryable_is_not_retried() -> None: (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) - assert outcome.failed + assert outcome.status == "failed" assert job.runs == 1 @@ -340,5 +340,5 @@ async def test_a_job_survives_a_json_round_trip_and_runs(tmp_path: Path) -> None assert copy == job (outcome,) = await run_jobs([copy], UNLIMITED) - assert outcome.finished + assert outcome.status == "finished" assert job.done() From b3e042dda04a8b065a82238e0dbcc50564aca267 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 4 Nov 2026 11:27:00 -0800 Subject: [PATCH 3/6] refactor: job loop follows the approved design (JobLoop, no retries) - JobLoop: one queue and one cap per job kind; submit() never waits, run() starts queued jobs while their kind has room. Replaces run_jobs. - No job-level retries: LLM calls retry themselves (llm_clients/), and a failed job is recorded and rerun later with --into. attempt and retryable leave the Job protocol; JOB_RETRYING becomes CALL_RETRYING, emitted by clients from inside a job and always shown. - Each job gets exactly one outcome, even if follow_ups() raises after run() succeeded. - Event times come from workers/events.py's _now, allow-listed in test_run_root's CLOCK_ALLOWED (the CI failure). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- src/vera_mh/workers/events.py | 18 +- src/vera_mh/workers/loop.py | 252 ++++++++++++++------------- src/vera_mh/workers/reporters.py | 26 +-- tests/unit/utils/test_run_root.py | 1 + tests/unit/workers/test_loop.py | 91 +++------- tests/unit/workers/test_reporters.py | 14 ++ 7 files changed, 194 insertions(+), 210 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0243cbf9d..4a204b15f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,7 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - **Config files may leave out output and speed fields**—A `--config` file must still state every field that changes what a run produces (models, inputs or `target`, and for `generate` `turns`, `user_language`, `persona_speaks_first`, `sessions`, `max_total_words`; for `score` and `pool`, `personas` and `skip_risk_analysis`). Fields that only change where output goes or how fast a run goes (`output`, `max_concurrent`, `per_judge`) may now be left out and take the same defaults as the CLI flags, so adding such an option no longer breaks existing configs. `vera judge`'s config `output` now defaults beside the conversations, as `--output` does. Configs that state every field work unchanged. - **A `vera` command**—`uv sync` now installs a `vera` command, so `uv run vera …` (or `vera …` in an activated environment) works from anywhere in the repository; `uv run python vera.py …` still works. Generation now finds its shared prompt files relative to the repository instead of the working directory. The Docker image installs the command and runs it by default; its previous `CMD` used the system Python, which lacked the dependencies. - **User-side language**—`vera generate --user-language ` (config: `generation.user_language`) has the user-side LLM write in that language. Non-English adds one line to the persona prompt; English, the default, leaves the prompt unchanged. -- **Job loop (not used yet)**—`vera_mh.workers` adds the loop that generation and judging will move onto ([docs/design/job-loop.md](docs/design/job-loop.md)): `run_jobs` skips jobs whose output is on disk, runs the rest with one concurrency cap per job kind, retries, and records each job's outcome, and its follow-ups carry a conversation on to its next session and its evaluations. It reports what happens as events, as plain timestamped lines or `events.jsonl`. Nothing calls it yet, so runs are unchanged. +- **Job loop (not used yet)**—`vera_mh.workers` adds the loop that generation and judging will move onto ([docs/design/job-loop.md](docs/design/job-loop.md)): `JobLoop` keeps one queue and one concurrency cap per job kind, skips jobs whose output is on disk, and records each job's outcome; a finished job's follow-ups carry a conversation on to its next session and its evaluations. It never retries a job: LLM calls retry themselves, and a failed job is rerun later with `--into`. It reports what happens as events, as plain timestamped lines or `events.jsonl`. Nothing calls it yet, so runs are unchanged. ### Testing and CI diff --git a/src/vera_mh/workers/events.py b/src/vera_mh/workers/events.py index 920e15d25..b2c542a20 100644 --- a/src/vera_mh/workers/events.py +++ b/src/vera_mh/workers/events.py @@ -1,9 +1,10 @@ """What happens during a run, as events a reporter shows or writes down. -The loop emits the job events itself: started, finished, failed, retrying, -skipped, and the stage finished at the end. Code running inside a job emits -the rest through `report()` (a turn started, a judge question asked, a call's -token usage), without the loop or the job having to pass a reporter down. +The loop emits the job events itself: started, finished, failed, skipped, and +the stage finished at the end. Code running inside a job emits the rest through +`report()` (a turn started, a judge question asked, an LLM call retrying, a +call's token usage), without the loop or the job having to pass a reporter +down. Every event's fields are JSON-safe, so the same event can be printed as a line or written to `events.jsonl`. @@ -21,16 +22,21 @@ JOB_STARTED = "job_started" JOB_FINISHED = "job_finished" JOB_FAILED = "job_failed" -JOB_RETRYING = "job_retrying" JOB_SKIPPED = "job_skipped" STAGE_FINISHED = "stage_finished" # Emitted from inside a job, with `report()`. TURN_STARTED = "turn_started" # data: turn, speaker ("user" or "chatbot") QUESTION_ASKED = "question_asked" # data: question_id +# LLM calls retry themselves (llm_clients/); the loop never retries a job. +CALL_RETRYING = "call_retrying" # data: attempt, max_attempts, error CALL_FINISHED = "call_finished" # data: model, input_tokens, output_tokens +def _now() -> datetime: + return datetime.now().astimezone() + + @dataclass(frozen=True) class Event: """One thing that happened. `job` and `kind` are empty for a stage event.""" @@ -39,7 +45,7 @@ class Event: job: str | None = None kind: str | None = None data: dict[str, Any] = field(default_factory=dict) - time: datetime = field(default_factory=lambda: datetime.now().astimezone()) + time: datetime = field(default_factory=_now) def to_dict(self) -> dict[str, Any]: """The event as one flat JSON object, as written to `events.jsonl`.""" diff --git a/src/vera_mh/workers/loop.py b/src/vera_mh/workers/loop.py index 3609cccb0..e74acbf86 100644 --- a/src/vera_mh/workers/loop.py +++ b/src/vera_mh/workers/loop.py @@ -1,15 +1,17 @@ """The job loop: all the concurrency code for generation and judging, in one place. -A job is plain data (paths, model names, indices) plus four methods; see `Job`. -`run_jobs` skips every job whose output is already on disk, runs the rest with -one concurrency cap per job kind, retries what the job says is worth retrying, -and records each job's outcome. A job's failure is an outcome, never an -exception, so one bad conversation doesn't stop the others. - -Jobs that wait on another job are its follow-ups: the next session of a -conversation, or a finished conversation's evaluations. A job hands them on -when it finishes, and also when it is skipped as done, so resuming a run -reaches every missing job however far down the chain it is, and redoes nothing. +A job is plain data (paths, model names, indices) plus three methods; see `Job`. +Each job kind (`session`, `evaluation`) has its own queue and its own cap. +Everything that adds work goes through `JobLoop.submit`, which never waits: the +stage loading its first jobs, and a finished job handing on its follow-ups (the +next session of a conversation, or a finished conversation's evaluations). +`JobLoop.run` starts queued jobs while their kind has room, until nothing is +queued or running. + +A job whose output is already on disk is skipped, but still hands on its +follow-ups, so resuming a run reaches every missing job however far down the +chain it is, and redoes nothing. The loop never retries a job: LLM calls retry +themselves, inside it. A failed job is recorded and rerun later with `--into`. Design: docs/design/job-loop.md. """ @@ -17,16 +19,14 @@ from __future__ import annotations import asyncio -import contextlib import time -from collections import Counter +from collections import Counter, deque from dataclasses import dataclass -from typing import Awaitable, Iterable, Literal, Mapping, Protocol +from typing import Literal, Mapping, Protocol from vera_mh.workers.events import ( JOB_FAILED, JOB_FINISHED, - JOB_RETRYING, JOB_SKIPPED, JOB_STARTED, STAGE_FINISHED, @@ -44,7 +44,7 @@ class Job(Protocol): external queue one day (job-loop.md, decision 7). """ - kind: str # "session" or "evaluation": picks the concurrency cap + kind: str # "session" or "evaluation": picks the queue and the cap label: str # how the job is shown, such as "gpt-5.2 Brian" def done(self) -> bool: @@ -60,10 +60,6 @@ def follow_ups(self) -> list[Job]: alone, so a job skipped as done hands them on too.""" ... - def retryable(self, error: Exception) -> bool: - """Is running the job again worth it after this error?""" - ... - Status = Literal["finished", "failed", "skipped"] @@ -74,116 +70,122 @@ class Outcome: job: Job status: Status - error: Exception | None = None - attempts: int = 0 + error: BaseException | None = None seconds: float = 0.0 -async def run_jobs( - jobs: Iterable[Job], - max_concurrent: Mapping[str, int | None], - attempts: int = 1, - *, - reporter: Reporter | None = None, - stage: str = "jobs", -) -> list[Outcome]: - """Run `jobs` and their follow-ups; return every job's outcome. - - `max_concurrent` caps how many jobs of each kind run at once; None or 0 - means no limit. Every job's kind must have an entry. `attempts` is how many - times a job may run, counting the first. `stage` names the run in the - final `STAGE_FINISHED` event. +class JobLoop: + """Runs jobs with one queue and one concurrency cap per job kind. + + A class rather than a function because `submit` is called both by the stage + and by finished jobs handing on their follow-ups, and both need the same + queues. """ - if attempts < 1: - raise ValueError(f"attempts must be at least 1, not {attempts}") - report = reporter or Reporters() - # No semaphore for an unlimited kind: one sized from `jobs` would throttle - # the follow-ups, which can outnumber them. - 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]: + + def __init__( + self, + max_concurrent: Mapping[str, int | None], + *, + reporter: Reporter | None = None, + stage: str = "jobs", + ): + """`max_concurrent` caps how many jobs of each kind run at once; None + or 0 means no limit, and every job's kind must have an entry. `stage` + names the run in the final `STAGE_FINISHED` event.""" + self.caps = dict(max_concurrent) + self.queues: dict[str, deque[Job]] = {kind: deque() for kind in self.caps} + self.running: dict[asyncio.Task[None], tuple[Job, float]] = {} + self.outcomes: list[Outcome] = [] + self.report = reporter or Reporters() + self.stage = stage + + def submit(self, job: Job) -> None: + """Put a job on its kind's queue, or skip it if its output is on disk. + Never waits.""" try: - if job.done(): - outcome = Outcome(job, "skipped") - else: - async with limits[job.kind]: - outcome = await _run_with_retries(job, attempts, report) - follow_ups = job.follow_ups() if outcome.status != "failed" else [] + is_done = job.done() except Exception as error: - # `done()` or `follow_ups()` crashed, on a corrupt file say. Count the - # job as failed rather than stop the run, or lose the jobs waiting on it - # without a trace. - outcome, follow_ups = Outcome(job, "failed", error=error), [] - _emit_outcome(report, outcome) - return [ - outcome, - *await _gather_flat(run_one(follow_up) for follow_up in follow_ups), - ] - - outcomes = await _gather_flat(run_one(job) for job in jobs) - counts = Counter(outcome.status for outcome in outcomes) - report.emit( - Event( - STAGE_FINISHED, - data={ - "stage": stage, - "done": counts["finished"], - "failed": counts["failed"], - "skipped": counts["skipped"], - }, - ) - ) - return outcomes - - -async def _run_with_retries(job: Job, attempts: int, report: Reporter) -> Outcome: - report.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) - start = time.monotonic() - attempt = 1 - with running_job(report, job.label, job.kind): + # A corrupt file, say: fail this job rather than the whole run. + self._record(Outcome(job, "failed", error=error)) + return + if is_done: + self._hand_on(Outcome(job, "skipped")) + else: + self.queues[job.kind].append(job) + + async def run(self) -> list[Outcome]: + """Run until every queue is empty and nothing is running; return every + job's outcome, skipped ones included.""" while True: - try: - await job.run() - return Outcome(job, "finished", attempts=attempt, seconds=_since(start)) - except Exception as error: - if attempt == attempts or not job.retryable(error): - return Outcome( - job, - "failed", - error=error, - attempts=attempt, - seconds=_since(start), - ) - retry = {"attempt": attempt, "error": _describe(error)} - report.emit( - Event(JOB_RETRYING, job=job.label, kind=job.kind, data=retry) - ) - attempt += 1 - - -def _emit_outcome(report: Reporter, outcome: Outcome) -> None: - job = outcome.job - if outcome.status == "skipped": - report.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) - elif outcome.status == "finished": - data = {"seconds": round(outcome.seconds, 3), "attempts": outcome.attempts} - report.emit(Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data)) - else: - data = {"error": _describe(outcome.error), "attempts": outcome.attempts} - report.emit(Event(JOB_FAILED, job=job.label, kind=job.kind, data=data)) - - -async def _gather_flat(runs: Iterable[Awaitable[list[Outcome]]]) -> list[Outcome]: - results = await asyncio.gather(*runs) - return [outcome for result in results for outcome in result] - - -def _since(start: float) -> float: - return time.monotonic() - start - - -def _describe(error: BaseException | None) -> str: - return f"{type(error).__name__}: {error}" if error else "" + self._start_jobs_that_fit() + if not self.running: + break + finished, _ = await asyncio.wait( + self.running, return_when=asyncio.FIRST_COMPLETED + ) + for task in finished: + job, started = self.running.pop(task) + seconds = time.monotonic() - started + error = task.exception() + if error is None: + self._hand_on(Outcome(job, "finished", seconds=seconds)) + else: + self._record(Outcome(job, "failed", error=error, seconds=seconds)) + + counts = Counter(outcome.status for outcome in self.outcomes) + totals = { + "stage": self.stage, + "done": counts["finished"], + "failed": counts["failed"], + "skipped": counts["skipped"], + } + self.report.emit(Event(STAGE_FINISHED, data=totals)) + return self.outcomes + + 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.report.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) + task = asyncio.create_task(self._run_one(job)) + self.running[task] = (job, time.monotonic()) + + async def _run_one(self, job: Job) -> None: + # Each job runs in its own task, so `report()` inside it reaches the + # reporter with this job's name, and concurrent jobs don't mix. + with running_job(self.report, job.label, job.kind): + await job.run() + + 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 _hand_on(self, outcome: Outcome) -> None: + """Record a finished or skipped job, and submit its follow-ups.""" + try: + follow_ups = outcome.job.follow_ups() + except Exception as error: + # The jobs waiting on this one can't be listed, so they would go + # missing without a trace: count the job as failed instead. + outcome = Outcome( + outcome.job, "failed", error=error, seconds=outcome.seconds + ) + follow_ups = [] + self._record(outcome) + for follow_up in follow_ups: + self.submit(follow_up) + + def _record(self, outcome: Outcome) -> None: + self.outcomes.append(outcome) + job = outcome.job + if outcome.status == "skipped": + self.report.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) + elif outcome.status == "finished": + data = {"seconds": round(outcome.seconds, 3)} + self.report.emit( + Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data) + ) + else: + data = {"error": f"{type(outcome.error).__name__}: {outcome.error}"} + self.report.emit(Event(JOB_FAILED, job=job.label, kind=job.kind, data=data)) diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py index 09e9dad72..3fdee766e 100644 --- a/src/vera_mh/workers/reporters.py +++ b/src/vera_mh/workers/reporters.py @@ -15,9 +15,9 @@ from typing import TextIO from vera_mh.workers.events import ( + CALL_RETRYING, JOB_FAILED, JOB_FINISHED, - JOB_RETRYING, JOB_SKIPPED, JOB_STARTED, QUESTION_ASKED, @@ -53,16 +53,17 @@ def emit(self, event: Event) -> None: f.write("\n") -# The word each job event shows as, padded so the labels line up. -JOB_WORDS = { +# The word each event shows as, padded so the labels line up. A call retrying +# is always shown, since it's how a slow or failing provider shows up; turns +# and questions only with `verbose`, as a run has many of them. +WORDS = { JOB_STARTED: "start", JOB_FINISHED: "done", JOB_FAILED: "FAIL", - JOB_RETRYING: "retry", JOB_SKIPPED: "skip", + CALL_RETRYING: "retry", } -# In-job events are shown only with `verbose`: a run has many of them. -IN_JOB_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"} +VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"} class PlainLineReporter: @@ -90,10 +91,10 @@ def format(self, event: Event) -> str | None: f"{clock} {counts['stage']} finished: {counts['done']} done, " f"{counts['failed']} failed, {counts['skipped']} skipped" ) - if event.type in JOB_WORDS: - word = JOB_WORDS[event.type] - elif self.verbose and event.type in IN_JOB_WORDS: - word = IN_JOB_WORDS[event.type] + if event.type in WORDS: + word = WORDS[event.type] + elif self.verbose and event.type in VERBOSE_WORDS: + word = VERBOSE_WORDS[event.type] else: return None detail = _detail(event) @@ -107,8 +108,9 @@ def _detail(event: Event) -> str: return duration(data["seconds"]) if event.type == JOB_FAILED: return data["error"] - if event.type == JOB_RETRYING: - return f"attempt {data['attempt']} failed: {data['error']}" + if event.type == CALL_RETRYING: + attempt = f"{data['attempt']}/{data['max_attempts']}" + return f"call attempt {attempt} failed: {data['error']}" if event.type == TURN_STARTED: return f"turn {data['turn']} {data['speaker']} responding" if event.type == QUESTION_ASKED: diff --git a/tests/unit/utils/test_run_root.py b/tests/unit/utils/test_run_root.py index 8eb3b3a93..a4ceedbc5 100644 --- a/tests/unit/utils/test_run_root.py +++ b/tests/unit/utils/test_run_root.py @@ -47,6 +47,7 @@ ("llm_clients/llm_interface.py", "_set_response_metadata"): "response time", ("utils/naming.py", "add_timestamp_to_path"): "comparison chart file names", ("utils/run_root.py", "run_started_at"): "the run's one clock read", + ("workers/events.py", "_now"): "event time stamps", } diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py index fd4cb09a2..0f21627db 100644 --- a/tests/unit/workers/test_loop.py +++ b/tests/unit/workers/test_loop.py @@ -1,4 +1,4 @@ -"""The job loop: skipping, concurrency caps, retries, follow-ups, and events.""" +"""The job loop: skipping, concurrency caps, follow-ups, failures, and events.""" from __future__ import annotations @@ -12,11 +12,19 @@ from vera_mh.workers import events from vera_mh.workers.events import Event -from vera_mh.workers.loop import Job, run_jobs +from vera_mh.workers.loop import Job, JobLoop, Outcome UNLIMITED: dict[str, int | None] = {"session": None, "evaluation": None} +async def run_jobs(jobs, max_concurrent, **settings) -> list[Outcome]: + """Load `jobs` into a fresh loop, as a stage does, and run it.""" + loop = JobLoop(max_concurrent, **settings) + for job in jobs: + loop.submit(job) + return await loop.run() + + class Tracker: """What the fake jobs did: the order things happened in, and how many jobs of each kind ran at once.""" @@ -33,10 +41,8 @@ class FakeJob: tracker: Tracker kind: str = "session" is_done: bool = False - # run() raises this many times before it succeeds. - failures: int = 0 - error: type[Exception] = RuntimeError - retry_on: tuple[type[Exception], ...] = (RuntimeError,) + # run() raises this when set. + error: type[Exception] | None = None next_jobs: list[FakeJob] = field(default_factory=list) # How long run() takes, so jobs overlap. seconds: float = 0.0 @@ -54,7 +60,7 @@ async def run(self) -> None: ) try: await asyncio.sleep(self.seconds) - if self.runs <= self.failures: + if self.error is not None: raise self.error(f"{self.label} failed") finally: self.tracker.running[self.kind] -= 1 @@ -63,9 +69,6 @@ async def run(self) -> None: def follow_ups(self) -> list[Job]: return list(self.next_jobs) - def retryable(self, error: Exception) -> bool: - return isinstance(error, self.retry_on) - class ListReporter: def __init__(self) -> None: @@ -109,9 +112,11 @@ async def test_each_kind_has_its_own_cap() -> None: assert tracker.peak == Counter({"session": 2, "evaluation": 3}) -async def test_a_failure_is_an_outcome_not_an_exception() -> None: +async def test_a_failure_is_an_outcome_and_the_job_is_not_run_again() -> None: + """The loop never retries: LLM calls retry themselves, and a failed job is + rerun later with --into.""" tracker = Tracker() - bad = FakeJob("bad", tracker, failures=1, error=ValueError, retry_on=()) + bad = FakeJob("bad", tracker, error=ValueError) good = FakeJob("good", tracker) outcomes = await run_jobs([bad, good], UNLIMITED) @@ -119,37 +124,7 @@ async def test_a_failure_is_an_outcome_not_an_exception() -> None: assert statuses(outcomes) == {"bad": "failed", "good": "finished"} failed = next(outcome for outcome in outcomes if outcome.status == "failed") assert isinstance(failed.error, ValueError) - - -async def test_retries_stop_at_attempts() -> None: - tracker = Tracker() - job = FakeJob("flaky", tracker, failures=5) - - (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) - - assert outcome.status == "failed" - assert job.runs == 3 - assert outcome.attempts == 3 - - -async def test_a_retry_can_succeed() -> None: - tracker = Tracker() - job = FakeJob("flaky", tracker, failures=1) - - (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) - - assert outcome.status == "finished" - assert outcome.attempts == 2 - - -async def test_an_error_that_isnt_retryable_is_not_retried() -> None: - tracker = Tracker() - job = FakeJob("broken", tracker, failures=5, error=ValueError) - - (outcome,) = await run_jobs([job], UNLIMITED, attempts=3) - - assert outcome.status == "failed" - assert job.runs == 1 + assert bad.runs == 1 async def test_follow_ups_run_after_their_parent() -> None: @@ -166,7 +141,7 @@ async def test_follow_ups_run_after_their_parent() -> None: async def test_follow_ups_dont_run_when_their_parent_fails() -> None: tracker = Tracker() child = FakeJob("child", tracker) - parent = FakeJob("parent", tracker, failures=1, next_jobs=[child]) + parent = FakeJob("parent", tracker, error=RuntimeError, next_jobs=[child]) outcomes = await run_jobs([parent], UNLIMITED) @@ -197,7 +172,7 @@ async def test_a_skipped_job_hands_on_its_follow_ups_each_checked_for_done() -> @pytest.mark.parametrize("unlimited", [None, 0]) async def test_follow_ups_of_an_unlimited_kind_all_run_at_once(unlimited) -> None: """One job hands on five follow-ups. With no cap they must all run at - once, even though the list `run_jobs` was given held a single job.""" + once, even though the stage loaded a single job.""" tracker = Tracker() evaluations = [ FakeJob(f"e{i}", tracker, kind="evaluation", seconds=0.05) for i in range(5) @@ -235,11 +210,7 @@ def follow_ups(self) -> list[Job]: ) assert statuses(outcomes) == {"done": "failed", "follow-ups": "failed"} - - -async def test_attempts_must_be_at_least_one() -> None: - with pytest.raises(ValueError): - await run_jobs([], UNLIMITED, attempts=0) + assert len(outcomes) == 2 # one outcome per job, even when run() succeeded # ------------------------------------------------------------------------ events @@ -250,24 +221,18 @@ async def test_the_loop_reports_each_job_and_the_stage() -> None: reporter = ListReporter() jobs = [ FakeJob("ok", tracker), - FakeJob("flaky", tracker, failures=1), - FakeJob("bad", tracker, failures=1, error=ValueError), + FakeJob("bad", tracker, error=ValueError), FakeJob("done", tracker, is_done=True), ] - await run_jobs(jobs, UNLIMITED, attempts=2, reporter=reporter, stage="generate") + await run_jobs(jobs, UNLIMITED, reporter=reporter, stage="generate") assert reporter.types("ok") == [events.JOB_STARTED, events.JOB_FINISHED] - assert reporter.types("flaky") == [ - events.JOB_STARTED, - events.JOB_RETRYING, - events.JOB_FINISHED, - ] assert reporter.types("bad") == [events.JOB_STARTED, events.JOB_FAILED] assert reporter.types("done") == [events.JOB_SKIPPED] stage = reporter.events[-1] assert stage.type == events.STAGE_FINISHED - assert stage.data == {"stage": "generate", "done": 2, "failed": 1, "skipped": 1} + assert stage.data == {"stage": "generate", "done": 1, "failed": 1, "skipped": 1} async def test_a_job_reports_from_inside_its_run() -> None: @@ -285,9 +250,6 @@ async def run(self) -> None: def follow_ups(self) -> list[Job]: return [] - def retryable(self, error: Exception) -> bool: - return False - reporter = ListReporter() await run_jobs([Talker("a"), Talker("b")], UNLIMITED, reporter=reporter) @@ -307,7 +269,7 @@ def test_report_outside_a_job_does_nothing() -> None: @dataclass class SampleSession: - """What a real job looks like: JSON-safe fields, plus the four methods.""" + """What a real job looks like: JSON-safe fields, plus the three methods.""" label: str persona: str @@ -329,9 +291,6 @@ async def run(self) -> None: def follow_ups(self) -> list[Job]: return [] - def retryable(self, error: Exception) -> bool: - return False - async def test_a_job_survives_a_json_round_trip_and_runs(tmp_path: Path) -> None: job = SampleSession("gpt-5.2 Brian", "Brian", 1, 2, str(tmp_path)) diff --git a/tests/unit/workers/test_reporters.py b/tests/unit/workers/test_reporters.py index 6820c1174..d6217fdd5 100644 --- a/tests/unit/workers/test_reporters.py +++ b/tests/unit/workers/test_reporters.py @@ -67,6 +67,20 @@ def test_in_job_events_show_only_when_verbose() -> None: assert verbose.format(tokens) is None +def test_a_call_retrying_is_shown_even_when_not_verbose() -> None: + retry = job_event( + events.CALL_RETRYING, + attempt=2, + max_attempts=4, + error="RateLimitError: slow down", + ) + + assert PlainLineReporter(io.StringIO()).format(retry) == ( + "19:44:06 session retry gpt-5.2 Brian " + "call attempt 2/4 failed: RateLimitError: slow down" + ) + + def test_plain_lines_are_written_and_ascii() -> None: stream = io.StringIO() reporter = PlainLineReporter(stream) From 1810d0367544ea57f2e95903673694d8d58bc6c1 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:16:33 -0800 Subject: [PATCH 4/6] refactor: simpler job loop from review - One asyncio.Semaphore per capped kind instead of hand-written queues, _start_jobs_that_fit, and _has_room; run() waits for all tasks. - One try per job: any error, from done(), run() or follow_ups(), is that job's outcome. - JobLoop takes a list of reporters; the Reporters fan-out class goes. - timedelta formats durations; the duration() helper goes. - The in-job reporting context gets clearer names (reporting_for, _current_job) and a worked example. - __init__ documents its arguments. Co-Authored-By: Claude Opus 5.5 --- src/vera_mh/workers/events.py | 49 ++++++--- src/vera_mh/workers/loop.py | 147 ++++++++++++--------------- src/vera_mh/workers/reporters.py | 32 +----- tests/unit/workers/test_loop.py | 12 ++- tests/unit/workers/test_reporters.py | 19 +--- 5 files changed, 115 insertions(+), 144 deletions(-) diff --git a/src/vera_mh/workers/events.py b/src/vera_mh/workers/events.py index b2c542a20..b492d267a 100644 --- a/src/vera_mh/workers/events.py +++ b/src/vera_mh/workers/events.py @@ -16,7 +16,7 @@ from contextvars import ContextVar from dataclasses import dataclass, field from datetime import datetime -from typing import Any, Iterator, Protocol +from typing import Any, Callable, Iterator, Protocol # Emitted by the loop. JOB_STARTED = "job_started" @@ -64,34 +64,55 @@ class Reporter(Protocol): def emit(self, event: Event) -> None: ... +# Reporting from inside a job +# +# Code deep inside a job, such as an LLM client retrying a call, has no handle +# on the loop, and passing one down through the simulator, the judge, and every +# client would touch all of them. Instead the loop records which job the +# current asyncio task is running, and `report()` reads it back: +# +# # in the loop, around each job (each job runs in its own task) +# with reporting_for(loop.emit, "gpt-5.2 Brian", "session"): +# await job.run() +# +# # anywhere inside that job, however deep +# report(CALL_RETRYING, attempt=2, max_attempts=4, error="RateLimitError") +# +# # every reporter then receives +# Event(type="call_retrying", job="gpt-5.2 Brian", kind="session", +# data={"attempt": 2, "max_attempts": 4, "error": "RateLimitError"}) +# +# A `ContextVar` is a variable whose value is per asyncio task, so ten jobs +# running at once each see their own job, never another's. + + @dataclass(frozen=True) -class _RunningJob: - reporter: Reporter +class _CurrentJob: + emit: Callable[[Event], None] label: str kind: str -# The job running in the current asyncio task, and where its events go. The loop -# sets it around each job's `run()`; each job runs in its own task, so -# concurrent jobs don't see each other's. -_running: ContextVar[_RunningJob | None] = ContextVar("running_job", default=None) +_current_job: ContextVar[_CurrentJob | None] = ContextVar("current_job", default=None) @contextlib.contextmanager -def running_job(reporter: Reporter, label: str, kind: str) -> Iterator[None]: - """While inside, `report()` emits events for this job to `reporter`.""" - token = _running.set(_RunningJob(reporter, label, kind)) +def reporting_for( + emit: Callable[[Event], None], label: str, kind: str +) -> Iterator[None]: + """Inside this block, `report()` sends events to `emit`, as this job's.""" + token = _current_job.set(_CurrentJob(emit, label, kind)) try: yield finally: - _running.reset(token) + _current_job.reset(token) def report(event_type: str, **data: Any) -> None: - """Emit an event from inside a running job, such as `TURN_STARTED`. + """Emit an event from inside a running job, such as `CALL_RETRYING`. Outside a job (a unit test calling the code directly, say) it does nothing. """ - job = _running.get() + job = _current_job.get() if job is not None: - job.reporter.emit(Event(event_type, job=job.label, kind=job.kind, data=data)) + job.emit(Event(event_type, job=job.label, kind=job.kind, data=data)) diff --git a/src/vera_mh/workers/loop.py b/src/vera_mh/workers/loop.py index e74acbf86..3c0d2e004 100644 --- a/src/vera_mh/workers/loop.py +++ b/src/vera_mh/workers/loop.py @@ -1,12 +1,10 @@ """The job loop: all the concurrency code for generation and judging, in one place. A job is plain data (paths, model names, indices) plus three methods; see `Job`. -Each job kind (`session`, `evaluation`) has its own queue and its own cap. -Everything that adds work goes through `JobLoop.submit`, which never waits: the -stage loading its first jobs, and a finished job handing on its follow-ups (the -next session of a conversation, or a finished conversation's evaluations). -`JobLoop.run` starts queued jobs while their kind has room, until nothing is -queued or running. +A stage creates a `JobLoop`, hands it its first jobs with `submit`, and awaits +`run`. Each job kind (`session`, `evaluation`) has its own concurrency cap. A +finished job's follow-ups (the next session of a conversation, or a finished +conversation's evaluations) are submitted the same way, as soon as it finishes. A job whose output is already on disk is skipped, but still hands on its follow-ups, so resuming a run reaches every missing job however far down the @@ -19,10 +17,11 @@ from __future__ import annotations import asyncio +import contextlib import time -from collections import Counter, deque +from collections import Counter from dataclasses import dataclass -from typing import Literal, Mapping, Protocol +from typing import Literal, Mapping, Protocol, Sequence from vera_mh.workers.events import ( JOB_FAILED, @@ -32,9 +31,8 @@ STAGE_FINISHED, Event, Reporter, - running_job, + reporting_for, ) -from vera_mh.workers.reporters import Reporters class Job(Protocol): @@ -44,7 +42,7 @@ class Job(Protocol): external queue one day (job-loop.md, decision 7). """ - kind: str # "session" or "evaluation": picks the queue and the cap + kind: str # "session" or "evaluation": picks the concurrency cap label: str # how the job is shown, such as "gpt-5.2 Brian" def done(self) -> bool: @@ -75,62 +73,54 @@ class Outcome: class JobLoop: - """Runs jobs with one queue and one concurrency cap per job kind. + """Runs jobs with one concurrency cap per job kind. A class rather than a function because `submit` is called both by the stage and by finished jobs handing on their follow-ups, and both need the same - queues. + caps and the same set of running jobs. """ def __init__( self, max_concurrent: Mapping[str, int | None], *, - reporter: Reporter | None = None, + reporters: Sequence[Reporter] = (), stage: str = "jobs", ): - """`max_concurrent` caps how many jobs of each kind run at once; None - or 0 means no limit, and every job's kind must have an entry. `stage` - names the run in the final `STAGE_FINISHED` event.""" - self.caps = dict(max_concurrent) - self.queues: dict[str, deque[Job]] = {kind: deque() for kind in self.caps} - self.running: dict[asyncio.Task[None], tuple[Job, float]] = {} - self.outcomes: list[Outcome] = [] - self.report = reporter or Reporters() + """ + Args: + max_concurrent: How many jobs of each kind may run at once, such as + `{"session": 10, "evaluation": 20}`. None or 0 means no limit. + Every kind a job can have needs an entry. + reporters: Where events go, such as a `PlainLineReporter` for the + terminal and an `EventsFileReporter` for `events.jsonl`. Every + event goes to each of them. + stage: The name the final `STAGE_FINISHED` event gives this run, + such as "generate". + """ + # A semaphore per capped kind: a job waits for a free slot of its kind. + self.limits = { + kind: asyncio.Semaphore(cap) if cap else contextlib.nullcontext() + for kind, cap in max_concurrent.items() + } + self.reporters = reporters self.stage = stage + self.tasks: set[asyncio.Task[None]] = set() + self.outcomes: list[Outcome] = [] def submit(self, job: Job) -> None: - """Put a job on its kind's queue, or skip it if its output is on disk. - Never waits.""" - try: - is_done = job.done() - except Exception as error: - # A corrupt file, say: fail this job rather than the whole run. - self._record(Outcome(job, "failed", error=error)) - return - if is_done: - self._hand_on(Outcome(job, "skipped")) - else: - self.queues[job.kind].append(job) + """Start a job in the background; it waits for a free slot of its kind. + Never waits itself. Call it from inside a running event loop.""" + 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; return every + """Wait until every job has finished, follow-ups included; return every job's outcome, skipped ones included.""" - while True: - self._start_jobs_that_fit() - if not self.running: - break - finished, _ = await asyncio.wait( - self.running, return_when=asyncio.FIRST_COMPLETED - ) - for task in finished: - job, started = self.running.pop(task) - seconds = time.monotonic() - started - error = task.exception() - if error is None: - self._hand_on(Outcome(job, "finished", seconds=seconds)) - else: - self._record(Outcome(job, "failed", error=error, seconds=seconds)) + while self.tasks: + # Waits for all the jobs started so far. Their follow-ups add new + # tasks while it waits, so go round until no task is left. + finished, _ = await asyncio.wait(self.tasks) + self.tasks -= finished counts = Counter(outcome.status for outcome in self.outcomes) totals = { @@ -139,40 +129,33 @@ async def run(self) -> list[Outcome]: "failed": counts["failed"], "skipped": counts["skipped"], } - self.report.emit(Event(STAGE_FINISHED, data=totals)) + self.emit(Event(STAGE_FINISHED, data=totals)) return self.outcomes - 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.report.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) - task = asyncio.create_task(self._run_one(job)) - self.running[task] = (job, time.monotonic()) + def emit(self, event: Event) -> None: + for reporter in self.reporters: + reporter.emit(event) async def _run_one(self, job: Job) -> None: - # Each job runs in its own task, so `report()` inside it reaches the - # reporter with this job's name, and concurrent jobs don't mix. - with running_job(self.report, job.label, job.kind): - await job.run() - - 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 _hand_on(self, outcome: Outcome) -> None: - """Record a finished or skipped job, and submit its follow-ups.""" + # Any error, from `done()`, `run()` or `follow_ups()`, becomes this + # job's outcome: one job failing never stops the others. + seconds = 0.0 try: - follow_ups = outcome.job.follow_ups() + if job.done(): + status: Status = "skipped" + else: + async with self.limits[job.kind]: + self.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) + start = time.monotonic() + with reporting_for(self.emit, job.label, job.kind): + await job.run() + seconds = time.monotonic() - start + status = "finished" + follow_ups = job.follow_ups() except Exception as error: - # The jobs waiting on this one can't be listed, so they would go - # missing without a trace: count the job as failed instead. - outcome = Outcome( - outcome.job, "failed", error=error, seconds=outcome.seconds - ) - follow_ups = [] - self._record(outcome) + self._record(Outcome(job, "failed", error=error, seconds=seconds)) + return + self._record(Outcome(job, status, seconds=seconds)) for follow_up in follow_ups: self.submit(follow_up) @@ -180,12 +163,10 @@ def _record(self, outcome: Outcome) -> None: self.outcomes.append(outcome) job = outcome.job if outcome.status == "skipped": - self.report.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) + self.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) elif outcome.status == "finished": data = {"seconds": round(outcome.seconds, 3)} - self.report.emit( - Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data) - ) + self.emit(Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data)) else: data = {"error": f"{type(outcome.error).__name__}: {outcome.error}"} - self.report.emit(Event(JOB_FAILED, job=job.label, kind=job.kind, data=data)) + self.emit(Event(JOB_FAILED, job=job.label, kind=job.kind, data=data)) diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py index 3fdee766e..4cfe046dd 100644 --- a/src/vera_mh/workers/reporters.py +++ b/src/vera_mh/workers/reporters.py @@ -4,13 +4,15 @@ piped to `tee` or a CI log is never silent while work is happening. ASCII only. - `EventsFileReporter` appends each event to `events.jsonl` as one JSON object, for scripts and notebooks. -- `Reporters` sends each event to several reporters. + +A `JobLoop` takes a list of them and sends every event to each. """ from __future__ import annotations import json import sys +from datetime import timedelta from pathlib import Path from typing import TextIO @@ -24,21 +26,9 @@ STAGE_FINISHED, TURN_STARTED, Event, - Reporter, ) -class Reporters: - """Sends each event to every reporter given, in order.""" - - def __init__(self, *reporters: Reporter): - self.reporters = reporters - - def emit(self, event: Event) -> None: - for reporter in self.reporters: - reporter.emit(event) - - class EventsFileReporter: """Appends each event to a JSON Lines file, one object per line.""" @@ -69,7 +59,7 @@ def emit(self, event: Event) -> None: class PlainLineReporter: """Prints one line per event, flushed at once: - 19:44:06 session done gpt-5.2 Brian 2m41s + 19:44:06 session done gpt-5.2 Brian 0:02:41 19:45:10 session FAIL claude-opus-4-5 Abigail RateLimitError: ... """ @@ -105,7 +95,7 @@ def format(self, event: Event) -> str | None: def _detail(event: Event) -> str: data = event.data if event.type == JOB_FINISHED: - return duration(data["seconds"]) + return str(timedelta(seconds=round(data["seconds"]))) if event.type == JOB_FAILED: return data["error"] if event.type == CALL_RETRYING: @@ -116,15 +106,3 @@ def _detail(event: Event) -> str: if event.type == QUESTION_ASKED: return f"question {data['question_id']}" return "" - - -def duration(seconds: float) -> str: - """`45s`, `2m41s`, `1h02m`.""" - seconds = int(round(seconds)) - if seconds < 60: - return f"{seconds}s" - minutes, seconds = divmod(seconds, 60) - if minutes < 60: - return f"{minutes}m{seconds:02d}s" - hours, minutes = divmod(minutes, 60) - return f"{hours}h{minutes:02d}m" diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py index 0f21627db..3f5cd4549 100644 --- a/tests/unit/workers/test_loop.py +++ b/tests/unit/workers/test_loop.py @@ -225,7 +225,7 @@ async def test_the_loop_reports_each_job_and_the_stage() -> None: FakeJob("done", tracker, is_done=True), ] - await run_jobs(jobs, UNLIMITED, reporter=reporter, stage="generate") + await run_jobs(jobs, UNLIMITED, reporters=[reporter], stage="generate") assert reporter.types("ok") == [events.JOB_STARTED, events.JOB_FINISHED] assert reporter.types("bad") == [events.JOB_STARTED, events.JOB_FAILED] @@ -252,7 +252,7 @@ def follow_ups(self) -> list[Job]: reporter = ListReporter() - await run_jobs([Talker("a"), Talker("b")], UNLIMITED, reporter=reporter) + await run_jobs([Talker("a"), Talker("b")], UNLIMITED, reporters=[reporter]) turns = [event for event in reporter.events if event.type == events.TURN_STARTED] assert {event.job for event in turns} == {"a", "b"} @@ -260,6 +260,14 @@ def follow_ups(self) -> list[Job]: assert turns[0].data == {"turn": 1, "speaker": "user"} +async def test_every_event_goes_to_each_reporter() -> None: + first, second = ListReporter(), ListReporter() + + await run_jobs([FakeJob("ok", Tracker())], UNLIMITED, reporters=[first, second]) + + assert first.events == second.events != [] + + def test_report_outside_a_job_does_nothing() -> None: events.report(events.TURN_STARTED, turn=1, speaker="user") diff --git a/tests/unit/workers/test_reporters.py b/tests/unit/workers/test_reporters.py index d6217fdd5..9fab4c038 100644 --- a/tests/unit/workers/test_reporters.py +++ b/tests/unit/workers/test_reporters.py @@ -12,8 +12,6 @@ from vera_mh.workers.reporters import ( EventsFileReporter, PlainLineReporter, - Reporters, - duration, ) AT = datetime(2026, 10, 30, 19, 44, 6, tzinfo=timezone.utc) @@ -34,7 +32,7 @@ def test_plain_lines_show_each_job_event() -> None: skipped = reporter.format(job_event(events.JOB_SKIPPED)) assert started == "19:44:06 session start gpt-5.2 Brian" - assert finished == "19:44:06 session done gpt-5.2 Brian 2m41s" + assert finished == "19:44:06 session done gpt-5.2 Brian 0:02:41" assert failed == ( "19:44:06 session FAIL gpt-5.2 Brian RateLimitError: slow down" ) @@ -118,18 +116,3 @@ def test_events_file_has_one_json_object_per_event(tmp_path: Path) -> None: "job": "gpt-5.2 Brian", }, ] - - -def test_reporters_send_each_event_to_all() -> None: - first, second = io.StringIO(), io.StringIO() - reporter = Reporters(PlainLineReporter(first), PlainLineReporter(second)) - - reporter.emit(job_event(events.JOB_STARTED)) - - assert first.getvalue() == second.getvalue() != "" - - -def test_durations() -> None: - assert duration(45) == "45s" - assert duration(161) == "2m41s" - assert duration(3720) == "1h02m" From 6056041d47aa2880da270972fa79c63cf17319e9 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:16:31 -0800 Subject: [PATCH 5/6] refactor: job types in workers/job.py, and review names - The Job protocol, Status, and Outcome move to workers/job.py: they describe jobs, not the loop. Status sits at the top, with what each value means. - Outcome.seconds is now elapsed (and the event's "seconds" field too). - The words each event shows as on the terminal move to events.py, next to the event names. - Comments on what the semaphores do inside `async with`, how reporting_for's yield works, why _now is a named function, and why PlainLineReporter looks sys.stdout up late. Co-Authored-By: Claude Opus 5.5 --- src/vera_mh/workers/__init__.py | 2 +- src/vera_mh/workers/events.py | 25 +++++++++++- src/vera_mh/workers/job.py | 48 +++++++++++++++++++++++ src/vera_mh/workers/loop.py | 58 ++++++---------------------- src/vera_mh/workers/reporters.py | 21 +++------- tests/unit/workers/test_loop.py | 3 +- tests/unit/workers/test_reporters.py | 4 +- 7 files changed, 94 insertions(+), 67 deletions(-) create mode 100644 src/vera_mh/workers/job.py diff --git a/src/vera_mh/workers/__init__.py b/src/vera_mh/workers/__init__.py index 8a620ecf2..103a8546e 100644 --- a/src/vera_mh/workers/__init__.py +++ b/src/vera_mh/workers/__init__.py @@ -1,6 +1,6 @@ """The job loop: runs generation and judging jobs with a concurrency cap per kind. -Import the submodule you need (`workers.loop`, `workers.events`, +Import the submodule you need (`workers.job`, `workers.loop`, `workers.events`, `workers.reporters`). Nothing here imports a domain package (`generate`, `judge`, `score`, `vera_cli`): the domain hands its jobs to the loop, never the reverse. Design: docs/design/job-loop.md and docs/design/terminal-interface.md. diff --git a/src/vera_mh/workers/events.py b/src/vera_mh/workers/events.py index b492d267a..d0506b2a3 100644 --- a/src/vera_mh/workers/events.py +++ b/src/vera_mh/workers/events.py @@ -32,8 +32,23 @@ CALL_RETRYING = "call_retrying" # data: attempt, max_attempts, error CALL_FINISHED = "call_finished" # data: model, input_tokens, output_tokens +# The word each event shows as on the terminal (`PlainLineReporter`). A call +# retrying is always shown, since it's how a slow or failing provider shows up; +# turns and questions only with `verbose`, as a run has many of them. Token +# usage only goes to `events.jsonl`. +WORDS = { + JOB_STARTED: "start", + JOB_FINISHED: "done", + JOB_FAILED: "FAIL", + JOB_SKIPPED: "skip", + CALL_RETRYING: "retry", +} +VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"} + def _now() -> datetime: + # A named function, not a lambda, so test_run_root's clock rule can allow + # this one `datetime.now()` call by name. return datetime.now().astimezone() @@ -86,6 +101,8 @@ def emit(self, event: Event) -> None: ... # running at once each see their own job, never another's. +# Many jobs run at once, but each in its own asyncio task, and a `ContextVar` +# holds one value per task: inside any one task there is exactly one current job. @dataclass(frozen=True) class _CurrentJob: emit: Callable[[Event], None] @@ -100,7 +117,13 @@ class _CurrentJob: def reporting_for( emit: Callable[[Event], None], label: str, kind: str ) -> Iterator[None]: - """Inside this block, `report()` sends events to `emit`, as this job's.""" + """Inside this block, `report()` sends events to `emit`, as this job's. + + As with any `@contextmanager`, `with reporting_for(...):` runs the code + before `yield` when the block starts (set the current job), then the + block's body at `yield`, then the `finally` when the block ends, even if it + raised (put back whatever was current before). + """ token = _current_job.set(_CurrentJob(emit, label, kind)) try: yield diff --git a/src/vera_mh/workers/job.py b/src/vera_mh/workers/job.py new file mode 100644 index 000000000..8e65a7643 --- /dev/null +++ b/src/vera_mh/workers/job.py @@ -0,0 +1,48 @@ +"""What a job is, and what can happen to one. + +`JobLoop` (workers/loop.py) runs jobs; generation and judging define them: one +session of a conversation, or one evaluation. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal, Protocol + +# What happened to a job: it ran and wrote its output, it raised, or its output +# was already on disk. +Status = Literal["finished", "failed", "skipped"] + + +class Job(Protocol): + """One unit of work: one session of a conversation, or one evaluation. + + Keep jobs plain data, with JSON-safe fields, so they could be sent to an + external queue one day (job-loop.md, decision 7). + """ + + kind: str # "session" or "evaluation": picks the concurrency cap + label: str # how the job is shown, such as "gpt-5.2 Brian" + + def done(self) -> bool: + """Is this job's output already on disk?""" + ... + + async def run(self) -> None: + """Do the work and write the output.""" + ... + + def follow_ups(self) -> list[Job]: + """The jobs waiting on this one, worked out from its fields and disk + alone, so a job skipped as done hands them on too.""" + ... + + +@dataclass +class Outcome: + """What happened to one job.""" + + job: Job + status: Status + error: BaseException | None = None + elapsed: float = 0.0 # seconds the job ran; 0 if it was skipped diff --git a/src/vera_mh/workers/loop.py b/src/vera_mh/workers/loop.py index 3c0d2e004..0af1cc439 100644 --- a/src/vera_mh/workers/loop.py +++ b/src/vera_mh/workers/loop.py @@ -1,6 +1,7 @@ """The job loop: all the concurrency code for generation and judging, in one place. -A job is plain data (paths, model names, indices) plus three methods; see `Job`. +A job is plain data (paths, model names, indices) plus three methods; see +`workers/job.py`. A stage creates a `JobLoop`, hands it its first jobs with `submit`, and awaits `run`. Each job kind (`session`, `evaluation`) has its own concurrency cap. A finished job's follow-ups (the next session of a conversation, or a finished @@ -20,8 +21,7 @@ import contextlib import time from collections import Counter -from dataclasses import dataclass -from typing import Literal, Mapping, Protocol, Sequence +from typing import Mapping, Sequence from vera_mh.workers.events import ( JOB_FAILED, @@ -33,43 +33,7 @@ Reporter, reporting_for, ) - - -class Job(Protocol): - """One unit of work: one session of a conversation, or one evaluation. - - Keep jobs plain data, with JSON-safe fields, so they could be sent to an - external queue one day (job-loop.md, decision 7). - """ - - kind: str # "session" or "evaluation": picks the concurrency cap - label: str # how the job is shown, such as "gpt-5.2 Brian" - - def done(self) -> bool: - """Is this job's output already on disk?""" - ... - - async def run(self) -> None: - """Do the work and write the output.""" - ... - - def follow_ups(self) -> list[Job]: - """The jobs waiting on this one, worked out from its fields and disk - alone, so a job skipped as done hands them on too.""" - ... - - -Status = Literal["finished", "failed", "skipped"] - - -@dataclass -class Outcome: - """What happened to one job.""" - - job: Job - status: Status - error: BaseException | None = None - seconds: float = 0.0 +from vera_mh.workers.job import Job, Outcome, Status class JobLoop: @@ -98,7 +62,9 @@ def __init__( stage: The name the final `STAGE_FINISHED` event gives this run, such as "generate". """ - # A semaphore per capped kind: a job waits for a free slot of its kind. + # Each job runs inside `async with self.limits[job.kind]:`. A semaphore + # lets in up to `cap` jobs of its kind at once and makes the rest wait + # their turn; `nullcontext()` lets everything in, for a kind with no cap. self.limits = { kind: asyncio.Semaphore(cap) if cap else contextlib.nullcontext() for kind, cap in max_concurrent.items() @@ -139,7 +105,7 @@ def emit(self, event: Event) -> None: async def _run_one(self, job: Job) -> None: # Any error, from `done()`, `run()` or `follow_ups()`, becomes this # job's outcome: one job failing never stops the others. - seconds = 0.0 + elapsed = 0.0 try: if job.done(): status: Status = "skipped" @@ -149,13 +115,13 @@ async def _run_one(self, job: Job) -> None: start = time.monotonic() with reporting_for(self.emit, job.label, job.kind): await job.run() - seconds = time.monotonic() - start + elapsed = time.monotonic() - start status = "finished" follow_ups = job.follow_ups() except Exception as error: - self._record(Outcome(job, "failed", error=error, seconds=seconds)) + self._record(Outcome(job, "failed", error=error, elapsed=elapsed)) return - self._record(Outcome(job, status, seconds=seconds)) + self._record(Outcome(job, status, elapsed=elapsed)) for follow_up in follow_ups: self.submit(follow_up) @@ -165,7 +131,7 @@ def _record(self, outcome: Outcome) -> None: if outcome.status == "skipped": self.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) elif outcome.status == "finished": - data = {"seconds": round(outcome.seconds, 3)} + data = {"elapsed": round(outcome.elapsed, 3)} self.emit(Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data)) else: data = {"error": f"{type(outcome.error).__name__}: {outcome.error}"} diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py index 4cfe046dd..1a6f967ea 100644 --- a/src/vera_mh/workers/reporters.py +++ b/src/vera_mh/workers/reporters.py @@ -20,11 +20,11 @@ CALL_RETRYING, JOB_FAILED, JOB_FINISHED, - JOB_SKIPPED, - JOB_STARTED, QUESTION_ASKED, STAGE_FINISHED, TURN_STARTED, + VERBOSE_WORDS, + WORDS, Event, ) @@ -43,19 +43,6 @@ def emit(self, event: Event) -> None: f.write("\n") -# The word each event shows as, padded so the labels line up. A call retrying -# is always shown, since it's how a slow or failing provider shows up; turns -# and questions only with `verbose`, as a run has many of them. -WORDS = { - JOB_STARTED: "start", - JOB_FINISHED: "done", - JOB_FAILED: "FAIL", - JOB_SKIPPED: "skip", - CALL_RETRYING: "retry", -} -VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"} - - class PlainLineReporter: """Prints one line per event, flushed at once: @@ -64,6 +51,8 @@ class PlainLineReporter: """ def __init__(self, stream: TextIO | None = None, verbose: bool = False): + # Looked up when the reporter is made, not when Python reads this file, + # so a test (or anything else) that swaps sys.stdout still gets the lines. self.stream = stream or sys.stdout self.verbose = verbose @@ -95,7 +84,7 @@ def format(self, event: Event) -> str | None: def _detail(event: Event) -> str: data = event.data if event.type == JOB_FINISHED: - return str(timedelta(seconds=round(data["seconds"]))) + return str(timedelta(seconds=round(data["elapsed"]))) if event.type == JOB_FAILED: return data["error"] if event.type == CALL_RETRYING: diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py index 3f5cd4549..1a9a4b868 100644 --- a/tests/unit/workers/test_loop.py +++ b/tests/unit/workers/test_loop.py @@ -12,7 +12,8 @@ from vera_mh.workers import events from vera_mh.workers.events import Event -from vera_mh.workers.loop import Job, JobLoop, Outcome +from vera_mh.workers.job import Job, Outcome +from vera_mh.workers.loop import JobLoop UNLIMITED: dict[str, int | None] = {"session": None, "evaluation": None} diff --git a/tests/unit/workers/test_reporters.py b/tests/unit/workers/test_reporters.py index 9fab4c038..c2db8a486 100644 --- a/tests/unit/workers/test_reporters.py +++ b/tests/unit/workers/test_reporters.py @@ -25,7 +25,7 @@ def test_plain_lines_show_each_job_event() -> None: reporter = PlainLineReporter(io.StringIO()) started = reporter.format(job_event(events.JOB_STARTED)) - finished = reporter.format(job_event(events.JOB_FINISHED, seconds=161)) + finished = reporter.format(job_event(events.JOB_FINISHED, elapsed=161)) failed = reporter.format( job_event(events.JOB_FAILED, error="RateLimitError: slow down") ) @@ -85,7 +85,7 @@ def test_plain_lines_are_written_and_ascii() -> None: reporter.emit(job_event(events.JOB_STARTED)) reporter.emit(job_event(events.TURN_STARTED, turn=1, speaker="user")) - reporter.emit(job_event(events.JOB_FINISHED, seconds=45)) + reporter.emit(job_event(events.JOB_FINISHED, elapsed=45)) lines = stream.getvalue().splitlines() assert len(lines) == 2 From 6d830dd895a25f8e0b0358e3d790c058d0228557 Mon Sep 17 00:00:00 2001 From: Luca Belli <129434630+sator-labs@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:59:19 -0800 Subject: [PATCH 6/6] refactor: EventType enum, and a plain set_current_job - Event types are an EventType StrEnum instead of loose string constants; each member is still its plain string in events.jsonl. - The reporting_for context manager (token, yield, reset) becomes set_current_job(): each job already runs in its own asyncio task, and a ContextVar set inside a task stays in that task and ends with it, so there is nothing to undo. - Clearer docs on what a stage event is and on the ContextVar. Co-Authored-By: Claude Opus 5.5 --- src/vera_mh/workers/events.py | 108 ++++++++++++++------------- src/vera_mh/workers/loop.py | 33 ++++---- src/vera_mh/workers/reporters.py | 24 ++---- tests/unit/workers/test_loop.py | 16 ++-- tests/unit/workers/test_reporters.py | 31 ++++---- 5 files changed, 101 insertions(+), 111 deletions(-) diff --git a/src/vera_mh/workers/events.py b/src/vera_mh/workers/events.py index d0506b2a3..a3012e492 100644 --- a/src/vera_mh/workers/events.py +++ b/src/vera_mh/workers/events.py @@ -12,38 +12,45 @@ from __future__ import annotations -import contextlib from contextvars import ContextVar from dataclasses import dataclass, field from datetime import datetime -from typing import Any, Callable, Iterator, Protocol - -# Emitted by the loop. -JOB_STARTED = "job_started" -JOB_FINISHED = "job_finished" -JOB_FAILED = "job_failed" -JOB_SKIPPED = "job_skipped" -STAGE_FINISHED = "stage_finished" - -# Emitted from inside a job, with `report()`. -TURN_STARTED = "turn_started" # data: turn, speaker ("user" or "chatbot") -QUESTION_ASKED = "question_asked" # data: question_id -# LLM calls retry themselves (llm_clients/); the loop never retries a job. -CALL_RETRYING = "call_retrying" # data: attempt, max_attempts, error -CALL_FINISHED = "call_finished" # data: model, input_tokens, output_tokens +from enum import StrEnum +from typing import Any, Callable, Protocol + + +class EventType(StrEnum): + """Every kind of event. A `StrEnum`, so each one is also its plain string + value ("job_started"), which is what `events.jsonl` records.""" + + # Emitted by the loop, about one job. + JOB_STARTED = "job_started" + JOB_FINISHED = "job_finished" # data: elapsed + JOB_FAILED = "job_failed" # data: error + JOB_SKIPPED = "job_skipped" + # Emitted by the loop once, when `JobLoop.run()` ends: about the whole run. + STAGE_FINISHED = "stage_finished" # data: stage, done, failed, skipped + + # Emitted from inside a job, with `report()`. + TURN_STARTED = "turn_started" # data: turn, speaker ("user" or "chatbot") + QUESTION_ASKED = "question_asked" # data: question_id + # LLM calls retry themselves (llm_clients/); the loop never retries a job. + CALL_RETRYING = "call_retrying" # data: attempt, max_attempts, error + CALL_FINISHED = "call_finished" # data: model, input_tokens, output_tokens + # The word each event shows as on the terminal (`PlainLineReporter`). A call # retrying is always shown, since it's how a slow or failing provider shows up; # turns and questions only with `verbose`, as a run has many of them. Token # usage only goes to `events.jsonl`. WORDS = { - JOB_STARTED: "start", - JOB_FINISHED: "done", - JOB_FAILED: "FAIL", - JOB_SKIPPED: "skip", - CALL_RETRYING: "retry", + EventType.JOB_STARTED: "start", + EventType.JOB_FINISHED: "done", + EventType.JOB_FAILED: "FAIL", + EventType.JOB_SKIPPED: "skip", + EventType.CALL_RETRYING: "retry", } -VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"} +VERBOSE_WORDS = {EventType.TURN_STARTED: "turn", EventType.QUESTION_ASKED: "ask"} def _now() -> datetime: @@ -54,9 +61,14 @@ def _now() -> datetime: @dataclass(frozen=True) class Event: - """One thing that happened. `job` and `kind` are empty for a stage event.""" + """One thing that happened. - type: str + `job` and `kind` name the job it's about. `STAGE_FINISHED` is the one event + that isn't about a job but about the whole run (its counts are in `data`), + so it has neither. + """ + + type: EventType job: str | None = None kind: str | None = None data: dict[str, Any] = field(default_factory=dict) @@ -67,7 +79,7 @@ def to_dict(self) -> dict[str, Any]: return { **self.data, "time": self.time.isoformat(timespec="seconds"), - "type": self.type, + "type": self.type.value, "kind": self.kind, "job": self.job, } @@ -83,26 +95,27 @@ def emit(self, event: Event) -> None: ... # # Code deep inside a job, such as an LLM client retrying a call, has no handle # on the loop, and passing one down through the simulator, the judge, and every -# client would touch all of them. Instead the loop records which job the -# current asyncio task is running, and `report()` reads it back: +# client would touch all of them. So the loop writes down, at the start of each +# job, which job is running and where its events go, and `report()` reads that +# back: # -# # in the loop, around each job (each job runs in its own task) -# with reporting_for(loop.emit, "gpt-5.2 Brian", "session"): -# await job.run() +# # in the loop, at the start of each job +# set_current_job(loop.emit, "gpt-5.2 Brian", "session") +# await job.run() # # # anywhere inside that job, however deep -# report(CALL_RETRYING, attempt=2, max_attempts=4, error="RateLimitError") +# report(EventType.CALL_RETRYING, attempt=2, max_attempts=4, error="...") # # # every reporter then receives # Event(type="call_retrying", job="gpt-5.2 Brian", kind="session", -# data={"attempt": 2, "max_attempts": 4, "error": "RateLimitError"}) +# data={"attempt": 2, "max_attempts": 4, "error": "..."}) # -# A `ContextVar` is a variable whose value is per asyncio task, so ten jobs -# running at once each see their own job, never another's. +# Where it's written down is a `ContextVar`: a variable with a separate value in +# each asyncio task. The loop runs every job in its own task, so ten jobs +# running at once each set and read their own value, never another job's, and a +# job's value disappears with its task when the job ends. -# Many jobs run at once, but each in its own asyncio task, and a `ContextVar` -# holds one value per task: inside any one task there is exactly one current job. @dataclass(frozen=True) class _CurrentJob: emit: Callable[[Event], None] @@ -110,28 +123,17 @@ class _CurrentJob: kind: str +# The running job's `_CurrentJob`, one per asyncio task; None outside a job. _current_job: ContextVar[_CurrentJob | None] = ContextVar("current_job", default=None) -@contextlib.contextmanager -def reporting_for( - emit: Callable[[Event], None], label: str, kind: str -) -> Iterator[None]: - """Inside this block, `report()` sends events to `emit`, as this job's. - - As with any `@contextmanager`, `with reporting_for(...):` runs the code - before `yield` when the block starts (set the current job), then the - block's body at `yield`, then the `finally` when the block ends, even if it - raised (put back whatever was current before). - """ - token = _current_job.set(_CurrentJob(emit, label, kind)) - try: - yield - finally: - _current_job.reset(token) +def set_current_job(emit: Callable[[Event], None], label: str, kind: str) -> None: + """From now on in this asyncio task, `report()` sends events to `emit` as + this job's. The loop calls it at the start of each job's task.""" + _current_job.set(_CurrentJob(emit, label, kind)) -def report(event_type: str, **data: Any) -> None: +def report(event_type: EventType, **data: Any) -> None: """Emit an event from inside a running job, such as `CALL_RETRYING`. Outside a job (a unit test calling the code directly, say) it does nothing. diff --git a/src/vera_mh/workers/loop.py b/src/vera_mh/workers/loop.py index 0af1cc439..2b8b24f5d 100644 --- a/src/vera_mh/workers/loop.py +++ b/src/vera_mh/workers/loop.py @@ -23,16 +23,7 @@ from collections import Counter from typing import Mapping, Sequence -from vera_mh.workers.events import ( - JOB_FAILED, - JOB_FINISHED, - JOB_SKIPPED, - JOB_STARTED, - STAGE_FINISHED, - Event, - Reporter, - reporting_for, -) +from vera_mh.workers.events import Event, EventType, Reporter, set_current_job from vera_mh.workers.job import Job, Outcome, Status @@ -95,7 +86,7 @@ async def run(self) -> list[Outcome]: "failed": counts["failed"], "skipped": counts["skipped"], } - self.emit(Event(STAGE_FINISHED, data=totals)) + self.emit(Event(EventType.STAGE_FINISHED, data=totals)) return self.outcomes def emit(self, event: Event) -> None: @@ -111,10 +102,14 @@ async def _run_one(self, job: Job) -> None: status: Status = "skipped" else: async with self.limits[job.kind]: - self.emit(Event(JOB_STARTED, job=job.label, kind=job.kind)) + self.emit( + Event(EventType.JOB_STARTED, job=job.label, kind=job.kind) + ) start = time.monotonic() - with reporting_for(self.emit, job.label, job.kind): - await job.run() + # This coroutine is the job's own task (see `submit`), so + # this applies to this job only, and ends with it. + set_current_job(self.emit, job.label, job.kind) + await job.run() elapsed = time.monotonic() - start status = "finished" follow_ups = job.follow_ups() @@ -129,10 +124,14 @@ def _record(self, outcome: Outcome) -> None: self.outcomes.append(outcome) job = outcome.job if outcome.status == "skipped": - self.emit(Event(JOB_SKIPPED, job=job.label, kind=job.kind)) + self.emit(Event(EventType.JOB_SKIPPED, job=job.label, kind=job.kind)) elif outcome.status == "finished": data = {"elapsed": round(outcome.elapsed, 3)} - self.emit(Event(JOB_FINISHED, job=job.label, kind=job.kind, data=data)) + self.emit( + Event(EventType.JOB_FINISHED, job=job.label, kind=job.kind, data=data) + ) else: data = {"error": f"{type(outcome.error).__name__}: {outcome.error}"} - self.emit(Event(JOB_FAILED, job=job.label, kind=job.kind, data=data)) + self.emit( + Event(EventType.JOB_FAILED, job=job.label, kind=job.kind, data=data) + ) diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py index 1a6f967ea..2de0cd898 100644 --- a/src/vera_mh/workers/reporters.py +++ b/src/vera_mh/workers/reporters.py @@ -16,17 +16,7 @@ from pathlib import Path from typing import TextIO -from vera_mh.workers.events import ( - CALL_RETRYING, - JOB_FAILED, - JOB_FINISHED, - QUESTION_ASKED, - STAGE_FINISHED, - TURN_STARTED, - VERBOSE_WORDS, - WORDS, - Event, -) +from vera_mh.workers.events import VERBOSE_WORDS, WORDS, Event, EventType class EventsFileReporter: @@ -64,7 +54,7 @@ def emit(self, event: Event) -> None: def format(self, event: Event) -> str | None: """The line for `event`, or None if it isn't shown.""" clock = event.time.strftime("%H:%M:%S") - if event.type == STAGE_FINISHED: + if event.type == EventType.STAGE_FINISHED: counts = event.data return ( f"{clock} {counts['stage']} finished: {counts['done']} done, " @@ -83,15 +73,15 @@ def format(self, event: Event) -> str | None: def _detail(event: Event) -> str: data = event.data - if event.type == JOB_FINISHED: + if event.type == EventType.JOB_FINISHED: return str(timedelta(seconds=round(data["elapsed"]))) - if event.type == JOB_FAILED: + if event.type == EventType.JOB_FAILED: return data["error"] - if event.type == CALL_RETRYING: + if event.type == EventType.CALL_RETRYING: attempt = f"{data['attempt']}/{data['max_attempts']}" return f"call attempt {attempt} failed: {data['error']}" - if event.type == TURN_STARTED: + if event.type == EventType.TURN_STARTED: return f"turn {data['turn']} {data['speaker']} responding" - if event.type == QUESTION_ASKED: + if event.type == EventType.QUESTION_ASKED: return f"question {data['question_id']}" return "" diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py index 1a9a4b868..c273a23a9 100644 --- a/tests/unit/workers/test_loop.py +++ b/tests/unit/workers/test_loop.py @@ -11,7 +11,7 @@ import pytest from vera_mh.workers import events -from vera_mh.workers.events import Event +from vera_mh.workers.events import Event, EventType from vera_mh.workers.job import Job, Outcome from vera_mh.workers.loop import JobLoop @@ -228,11 +228,11 @@ async def test_the_loop_reports_each_job_and_the_stage() -> None: await run_jobs(jobs, UNLIMITED, reporters=[reporter], stage="generate") - assert reporter.types("ok") == [events.JOB_STARTED, events.JOB_FINISHED] - assert reporter.types("bad") == [events.JOB_STARTED, events.JOB_FAILED] - assert reporter.types("done") == [events.JOB_SKIPPED] + assert reporter.types("ok") == [EventType.JOB_STARTED, EventType.JOB_FINISHED] + assert reporter.types("bad") == [EventType.JOB_STARTED, EventType.JOB_FAILED] + assert reporter.types("done") == [EventType.JOB_SKIPPED] stage = reporter.events[-1] - assert stage.type == events.STAGE_FINISHED + assert stage.type == EventType.STAGE_FINISHED assert stage.data == {"stage": "generate", "done": 1, "failed": 1, "skipped": 1} @@ -246,7 +246,7 @@ def done(self) -> bool: return False async def run(self) -> None: - events.report(events.TURN_STARTED, turn=1, speaker="user") + events.report(EventType.TURN_STARTED, turn=1, speaker="user") def follow_ups(self) -> list[Job]: return [] @@ -255,7 +255,7 @@ def follow_ups(self) -> list[Job]: await run_jobs([Talker("a"), Talker("b")], UNLIMITED, reporters=[reporter]) - turns = [event for event in reporter.events if event.type == events.TURN_STARTED] + turns = [event for event in reporter.events if event.type == EventType.TURN_STARTED] assert {event.job for event in turns} == {"a", "b"} assert all(event.kind == "session" for event in turns) assert turns[0].data == {"turn": 1, "speaker": "user"} @@ -270,7 +270,7 @@ async def test_every_event_goes_to_each_reporter() -> None: def test_report_outside_a_job_does_nothing() -> None: - events.report(events.TURN_STARTED, turn=1, speaker="user") + events.report(EventType.TURN_STARTED, turn=1, speaker="user") # --------------------------------------------------------- jobs stay serializable diff --git a/tests/unit/workers/test_reporters.py b/tests/unit/workers/test_reporters.py index c2db8a486..7da0f7118 100644 --- a/tests/unit/workers/test_reporters.py +++ b/tests/unit/workers/test_reporters.py @@ -7,8 +7,7 @@ from datetime import datetime, timezone from pathlib import Path -from vera_mh.workers import events -from vera_mh.workers.events import Event +from vera_mh.workers.events import Event, EventType from vera_mh.workers.reporters import ( EventsFileReporter, PlainLineReporter, @@ -17,19 +16,19 @@ AT = datetime(2026, 10, 30, 19, 44, 6, tzinfo=timezone.utc) -def job_event(event_type: str, **data) -> Event: +def job_event(event_type: EventType, **data) -> Event: return Event(event_type, job="gpt-5.2 Brian", kind="session", data=data, time=AT) def test_plain_lines_show_each_job_event() -> None: reporter = PlainLineReporter(io.StringIO()) - started = reporter.format(job_event(events.JOB_STARTED)) - finished = reporter.format(job_event(events.JOB_FINISHED, elapsed=161)) + started = reporter.format(job_event(EventType.JOB_STARTED)) + finished = reporter.format(job_event(EventType.JOB_FINISHED, elapsed=161)) failed = reporter.format( - job_event(events.JOB_FAILED, error="RateLimitError: slow down") + job_event(EventType.JOB_FAILED, error="RateLimitError: slow down") ) - skipped = reporter.format(job_event(events.JOB_SKIPPED)) + skipped = reporter.format(job_event(EventType.JOB_SKIPPED)) assert started == "19:44:06 session start gpt-5.2 Brian" assert finished == "19:44:06 session done gpt-5.2 Brian 0:02:41" @@ -42,7 +41,7 @@ def test_plain_lines_show_each_job_event() -> None: def test_the_stage_line_counts_outcomes() -> None: reporter = PlainLineReporter(io.StringIO()) stage = Event( - events.STAGE_FINISHED, + EventType.STAGE_FINISHED, data={"stage": "generate", "done": 196, "failed": 2, "skipped": 0}, time=AT, ) @@ -53,8 +52,8 @@ def test_the_stage_line_counts_outcomes() -> None: def test_in_job_events_show_only_when_verbose() -> None: - turn = job_event(events.TURN_STARTED, turn=14, speaker="chatbot") - tokens = job_event(events.CALL_FINISHED, input_tokens=10, output_tokens=5) + turn = job_event(EventType.TURN_STARTED, turn=14, speaker="chatbot") + tokens = job_event(EventType.CALL_FINISHED, input_tokens=10, output_tokens=5) assert PlainLineReporter(io.StringIO()).format(turn) is None verbose = PlainLineReporter(io.StringIO(), verbose=True) @@ -67,7 +66,7 @@ def test_in_job_events_show_only_when_verbose() -> None: def test_a_call_retrying_is_shown_even_when_not_verbose() -> None: retry = job_event( - events.CALL_RETRYING, + EventType.CALL_RETRYING, attempt=2, max_attempts=4, error="RateLimitError: slow down", @@ -83,9 +82,9 @@ def test_plain_lines_are_written_and_ascii() -> None: stream = io.StringIO() reporter = PlainLineReporter(stream) - reporter.emit(job_event(events.JOB_STARTED)) - reporter.emit(job_event(events.TURN_STARTED, turn=1, speaker="user")) - reporter.emit(job_event(events.JOB_FINISHED, elapsed=45)) + reporter.emit(job_event(EventType.JOB_STARTED)) + reporter.emit(job_event(EventType.TURN_STARTED, turn=1, speaker="user")) + reporter.emit(job_event(EventType.JOB_FINISHED, elapsed=45)) lines = stream.getvalue().splitlines() assert len(lines) == 2 @@ -96,8 +95,8 @@ def test_events_file_has_one_json_object_per_event(tmp_path: Path) -> None: path = tmp_path / "events.jsonl" reporter = EventsFileReporter(path) - reporter.emit(job_event(events.JOB_STARTED)) - reporter.emit(job_event(events.CALL_FINISHED, input_tokens=10, output_tokens=5)) + reporter.emit(job_event(EventType.JOB_STARTED)) + reporter.emit(job_event(EventType.CALL_FINISHED, input_tokens=10, output_tokens=5)) lines = path.read_text(encoding="utf-8").splitlines() assert [json.loads(line) for line in lines] == [