feat: workers/ job loop, with events and reporters (no callers yet)

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 <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)): `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
 

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"
 

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.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.
+"""

src/vera_mh/workers/events.py

@@ -0,0 +1,143 @@
+"""What happens during a run, as events a reporter shows or writes 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`.
+"""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from datetime import datetime
+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 = {
+    EventType.JOB_STARTED: "start",
+    EventType.JOB_FINISHED: "done",
+    EventType.JOB_FAILED: "FAIL",
+    EventType.JOB_SKIPPED: "skip",
+    EventType.CALL_RETRYING: "retry",
+}
+VERBOSE_WORDS = {EventType.TURN_STARTED: "turn", EventType.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()
+
+
+@dataclass(frozen=True)
+class Event:
+    """One thing that happened.
+
+    `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)
+    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`."""
+        return {
+            **self.data,
+            "time": self.time.isoformat(timespec="seconds"),
+            "type": self.type.value,
+            "kind": self.kind,
+            "job": self.job,
+        }
+
+
+class Reporter(Protocol):
+    """Shows or records events: see `workers.reporters`."""
+
+    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. 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, 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(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": "..."})
+#
+# 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.
+
+
+@dataclass(frozen=True)
+class _CurrentJob:
+    emit: Callable[[Event], None]
+    label: str
+    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)
+
+
+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: 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.
+    """
+    job = _current_job.get()
+    if job is not None:
+        job.emit(Event(event_type, job=job.label, kind=job.kind, data=data))

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

src/vera_mh/workers/loop.py

@@ -0,0 +1,137 @@
+"""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
+`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
+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
+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.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import time
+from collections import Counter
+from typing import Mapping, Sequence
+
+from vera_mh.workers.events import Event, EventType, Reporter, set_current_job
+from vera_mh.workers.job import Job, Outcome, Status
+
+
+class JobLoop:
+    """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
+    caps and the same set of running jobs.
+    """
+
+    def __init__(
+        self,
+        max_concurrent: Mapping[str, int | None],
+        *,
+        reporters: Sequence[Reporter] = (),
+        stage: str = "jobs",
+    ):
+        """
+        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".
+        """
+        # 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()
+        }
+        self.reporters = reporters
+        self.stage = stage
+        self.tasks: set[asyncio.Task[None]] = set()
+        self.outcomes: list[Outcome] = []
+
+    def submit(self, job: Job) -> None:
+        """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]:
+        """Wait until every job has finished, follow-ups included; return every
+        job's outcome, skipped ones included."""
+        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 = {
+            "stage": self.stage,
+            "done": counts["finished"],
+            "failed": counts["failed"],
+            "skipped": counts["skipped"],
+        }
+        self.emit(Event(EventType.STAGE_FINISHED, data=totals))
+        return self.outcomes
+
+    def emit(self, event: Event) -> None:
+        for reporter in self.reporters:
+            reporter.emit(event)
+
+    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.
+        elapsed = 0.0
+        try:
+            if job.done():
+                status: Status = "skipped"
+            else:
+                async with self.limits[job.kind]:
+                    self.emit(
+                        Event(EventType.JOB_STARTED, job=job.label, kind=job.kind)
+                    )
+                    start = time.monotonic()
+                    # 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()
+        except Exception as error:
+            self._record(Outcome(job, "failed", error=error, elapsed=elapsed))
+            return
+        self._record(Outcome(job, status, elapsed=elapsed))
+        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.emit(Event(EventType.JOB_SKIPPED, job=job.label, kind=job.kind))
+        elif outcome.status == "finished":
+            data = {"elapsed": round(outcome.elapsed, 3)}
+            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(EventType.JOB_FAILED, job=job.label, kind=job.kind, data=data)
+            )