diff --git a/CHANGELOG.md b/CHANGELOG.md index f7390581..4a204b15 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)): `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/pyproject.toml b/pyproject.toml index fd65b32a..c8f9606a 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 00000000..103a8546 --- /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.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 new file mode 100644 index 00000000..a3012e49 --- /dev/null +++ b/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)) diff --git a/src/vera_mh/workers/job.py b/src/vera_mh/workers/job.py new file mode 100644 index 00000000..8e65a764 --- /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 new file mode 100644 index 00000000..2b8b24f5 --- /dev/null +++ b/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) + ) diff --git a/src/vera_mh/workers/reporters.py b/src/vera_mh/workers/reporters.py new file mode 100644 index 00000000..2de0cd89 --- /dev/null +++ b/src/vera_mh/workers/reporters.py @@ -0,0 +1,87 @@ +"""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. + +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 + +from vera_mh.workers.events import VERBOSE_WORDS, WORDS, Event, EventType + + +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") + + +class PlainLineReporter: + """Prints one line per event, flushed at once: + + 19:44:06 session done gpt-5.2 Brian 0:02:41 + 19:45:10 session FAIL claude-opus-4-5 Abigail RateLimitError: ... + """ + + 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 + + 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 == EventType.STAGE_FINISHED: + counts = event.data + return ( + f"{clock} {counts['stage']} finished: {counts['done']} done, " + f"{counts['failed']} failed, {counts['skipped']} skipped" + ) + 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) + 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 == EventType.JOB_FINISHED: + return str(timedelta(seconds=round(data["elapsed"]))) + if event.type == EventType.JOB_FAILED: + return data["error"] + if event.type == EventType.CALL_RETRYING: + attempt = f"{data['attempt']}/{data['max_attempts']}" + return f"call attempt {attempt} failed: {data['error']}" + if event.type == EventType.TURN_STARTED: + return f"turn {data['turn']} {data['speaker']} responding" + if event.type == EventType.QUESTION_ASKED: + return f"question {data['question_id']}" + return "" diff --git a/tests/unit/utils/test_run_root.py b/tests/unit/utils/test_run_root.py index 8eb3b3a9..a4ceedbc 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/__init__.py b/tests/unit/workers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/workers/test_loop.py b/tests/unit/workers/test_loop.py new file mode 100644 index 00000000..c273a23a --- /dev/null +++ b/tests/unit/workers/test_loop.py @@ -0,0 +1,312 @@ +"""The job loop: skipping, concurrency caps, follow-ups, failures, 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, EventType +from vera_mh.workers.job import Job, Outcome +from vera_mh.workers.loop import JobLoop + +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.""" + + 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 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 + 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.error is not None: + 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) + + +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.status == "finished" for outcome in outcomes) + assert tracker.peak == Counter({"session": 2, "evaluation": 3}) + + +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, error=ValueError) + 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.status == "failed") + assert isinstance(failed.error, ValueError) + assert bad.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, error=RuntimeError, 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 stage loaded 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"} + assert len(outcomes) == 2 # one outcome per job, even when run() succeeded + + +# ------------------------------------------------------------------------ events + + +async def test_the_loop_reports_each_job_and_the_stage() -> None: + tracker = Tracker() + reporter = ListReporter() + jobs = [ + FakeJob("ok", tracker), + FakeJob("bad", tracker, error=ValueError), + FakeJob("done", tracker, is_done=True), + ] + + await run_jobs(jobs, UNLIMITED, reporters=[reporter], stage="generate") + + 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 == EventType.STAGE_FINISHED + assert stage.data == {"stage": "generate", "done": 1, "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(EventType.TURN_STARTED, turn=1, speaker="user") + + def follow_ups(self) -> list[Job]: + return [] + + reporter = ListReporter() + + await run_jobs([Talker("a"), Talker("b")], UNLIMITED, reporters=[reporter]) + + 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"} + + +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(EventType.TURN_STARTED, turn=1, speaker="user") + + +# --------------------------------------------------------- jobs stay serializable + + +@dataclass +class SampleSession: + """What a real job looks like: JSON-safe fields, plus the three 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 [] + + +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.status == "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 00000000..7da0f711 --- /dev/null +++ b/tests/unit/workers/test_reporters.py @@ -0,0 +1,117 @@ +"""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.events import Event, EventType +from vera_mh.workers.reporters import ( + EventsFileReporter, + PlainLineReporter, +) + +AT = datetime(2026, 10, 30, 19, 44, 6, tzinfo=timezone.utc) + + +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(EventType.JOB_STARTED)) + finished = reporter.format(job_event(EventType.JOB_FINISHED, elapsed=161)) + failed = reporter.format( + job_event(EventType.JOB_FAILED, error="RateLimitError: slow down") + ) + 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" + 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( + EventType.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(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) + 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_a_call_retrying_is_shown_even_when_not_verbose() -> None: + retry = job_event( + EventType.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) + + 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 + 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(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] == [ + { + "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", + }, + ]