feat: workers/ job loop, with events and reporters (no callers yet) by luca-belli · Pull Request #308 · SpringCare/VERA-MH

luca-belli · 2026-10-31T00:41:03Z

PR 3 of the "Steps 4 and 5, PR by PR" table (#307). It implements the loop exactly as approved in the job-loop design (#266) and the event layer from the terminal-interface design (#271). Nothing calls it yet, so runs are unchanged.

What's in it

workers/loop.py, the JobLoop class: one concurrency cap per job kind (session, evaluation), each an asyncio.Semaphore.
- submit(job) never waits: it starts the job as a task, and the task waits for a free slot of its kind, in order.
- A job whose output is on disk is skipped, but still hands on its follow-ups. A finished job hands on its follow-ups too, such as the next session or a conversation's evaluations.
- run() waits until every task has finished, follow-ups included.
- A kind with cap None or 0 has no limit. Follow-ups take a slot when they start, so they're never throttled by how few jobs the stage loaded.
- No job-level retries (decision 2). LLM calls retry themselves in llm_clients/. A failed job is recorded and rerun later with --into.
- One try per job: any error, from done(), run() or follow_ups(), is that job's single outcome, and never stops the other jobs.
- JobLoop(..., reporters=[...]) sends every event to each reporter.
The Job protocol: kind, label, done(), run(), follow_ups(). Jobs are plain JSON-safe data (decision 7).
workers/events.py:
- the Event type (JSON-safe) and the Reporter protocol;
- report(...) for events from inside a job: turn started, question asked, call finished with token usage, and call_retrying, which the clients will emit when an LLM call retries.
workers/reporters.py:
- the plain-line reporter: timestamped, flushed, ASCII only, with durations as 0:02:41 (timedelta). A call retrying is always shown; turns and questions only with verbose.
- the events.jsonl writer.
pyproject.toml: an import-linter contract that workers/ imports no domain package.
test_run_root.py: event times are allow-listed in CLOCK_ALLOWED. The earlier CI failure was this rule.
Tests: tests/unit/workers/ covers the design's testing list (skip-on-done, cap per kind, failures recorded and never resubmitted, follow-up ordering and resume chains, unlimited kinds, a JSON round trip) and the reporter formats.

Decisions made here, for review

In-job events use a context variable, set around each job's task. events.py has a worked example. Threading a reporter through ConversationSimulator and LLMJudge would be the alternative, and jobs can't hold one since they must stay plain data.
The loop emits the final stage event, named by JobLoop(..., stage=...).
Plain lines show the job kind (session/evaluation), since the loop doesn't know stage names; jobs put model and persona in label.

Checks

ruff and pyright clean; import-linter: 3 contracts kept
full-run check on the committed slice: matches
New tests written but not run locally (CI runs them)

🤖 Generated with Claude Code

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 <noreply@anthropic.com>

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 <noreply@anthropic.com>

- 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 <noreply@anthropic.com>

luca-belli · 2026-11-02T19:30:38Z

+        ...
+
+
+Status = Literal["finished", "failed", "skipped"]


should this be on top?

Superseded: that was the retryable method on the old Job protocol. Since the rewrite (b3e042d), the loop has no retries at all (#266 decision 2), so it's gone.

is it? I am taling about the Status var

also isn't status a property of the jjob? not the loop

i guess this file should maybe be called types or something?

or the job related ones should so somewhere else

Got it, sorry: you meant Status itself. In 6056041 it's at the top of its own module, with a comment on what each value means.

Agreed. It describes what happened to a job, so it now lives with Outcome and the Job protocol, not with the loop (6056041).

Close to that: they're now workers/job.py, "what a job is, and what can happen to one", and loop.py holds only JobLoop (6056041).

Done, same commit: Job, Status, and Outcome are in workers/job.py.

luca-belli · 2026-11-02T19:31:32Z

+    seconds: float = 0.0
+
+
+async def run_jobs(


should this be in a class? maybe not

It is now: JobLoop (b3e042d), as the approved #266 design says. A class because submit is called both by the stage and by finished jobs handing on their follow-ups, and both need the same caps and running set.

luca-belli · 2026-11-02T19:32:38Z

+        for kind, cap in max_concurrent.items()
+    }
+
+    async def run_one(job: Job) -> list[Outcome]:


do we need to define a function insidea another one? it's not really readable

Agreed, and it's gone: no function is defined inside another anymore. Each step is its own method on JobLoop (submit, run, _run_one, _record).

luca-belli · 2026-11-02T19:33:45Z

+            else:
+                async with limits[job.kind]:
+                    outcome = await _run_with_retries(job, attempts, report)
+            follow_ups = job.follow_ups() if outcome.status != "failed" else []


should the follow up job be added to the queue, instead of being processed later?

Yes, that's how it works now: a finished job's follow-ups go back through submit() right away, which never waits. Since 1810d03 they simply start as tasks that wait for a free slot of their kind.

luca-belli · 2026-11-04T22:58:17Z

+# 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)


i don't understand what this is supposed to do

Fair, it was opaque. In 1810d03 it has clearer names (_current_job, reporting_for) and a comment block with a worked example. In short: code deep inside a job (say an LLM client retrying a call) needs to report an event as that job's, but has no handle on the loop. Rather than pass one down through the simulator, the judge, and every client, the loop records which job the current asyncio task is running, and report() reads it back. A ContextVar holds one value per asyncio task, so ten jobs running at once each see their own.

luca-belli · 2026-11-04T23:08:12Z

+        with running_job(self.report, job.label, job.kind):
+            await job.run()
+
+    def _has_room(self, kind: str) -> bool:


do we need to re-define this? isn't there something that already exists?

Yes, asyncio.Semaphore already does it. Switched in 1810d03: one semaphore per capped kind, and the hand-written queues, _start_jobs_that_fit, and _has_room are gone. The tasks waiting on a kind's semaphore are that kind's queue, served in order. The #266 sketch is updated to match (728edfb). The loop is 172 lines with docs.

luca-belli · 2026-11-04T23:08:52Z

+)
+
+
+class Reporters:


what's the advantage of having a separate report class?

None worth a class. Removed in 1810d03: JobLoop takes a list of reporters (reporters=[PlainLineReporter(), EventsFileReporter(path)]) and sends every event to each.

luca-belli · 2026-11-04T23:09:31Z

+VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"}
+
+
+class PlainLineReporter:


is this logging?

Not logging. It's the terminal channel for a person watching, one of #271's three channels (terminal, log files, events.jsonl). The log files keep everything; this prints one short flushed line per job event. It could sit on Python's logging (a StreamHandler flushes per record too), but a print(..., flush=True) to a chosen stream is simpler and easy to test. Happy to switch if you'd rather --quiet/--verbose map to logging levels.

luca-belli · 2026-11-04T23:09:59Z

+    return ""
+
+
+def duration(seconds: float) -> str:


can't we use an existing function?

Yes: datetime.timedelta. Switched in 1810d03, so a duration shows as 0:02:41, the same format as #271's elapsed time. The duration() helper and its test are gone.

luca-belli · 2026-11-04T23:11:01Z

+    return await loop.run()
+
+
+class Tracker:


why are we defining new tings in the test? can't we use what we already have?

There's nothing to reuse yet: no real job type exists until PR 4 and PR 5 add Evaluation and Session. tests/mocks/ has LLM mocks, not jobs, and the loop's tests shouldn't need LLMs. FakeJob is the smallest thing that implements the protocol and records what happened (Tracker: order and peak concurrency per kind). Once real jobs exist, their own tests will run them through the loop with the mock LLMs.

- 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 <noreply@anthropic.com>

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

luca-belli · 2026-11-05T21:37:49Z

+CALL_FINISHED = "call_finished"  # data: model, input_tokens, output_tokens
+
+
+def _now() -> datetime:


what's the advantage of this?

None, besides the test. It's the one place the code reads the clock for event times, and test_run_root's clock rule allows datetime.now() calls by the name of their enclosing function. A lambda has no name the rule can list, so it's a tiny named function. Said so in a comment in 6056041.

luca-belli · 2026-11-05T21:41:04Z

+
+
+@dataclass(frozen=True)
+class _CurrentJob:


is. it possible that there are more than current jobs, if we do them async?

Yes, many jobs run at once, but each in its own asyncio task, and a ContextVar holds one value per task. So across the run there are many current jobs, one per task, and inside any one task there's exactly one. When a job calls report(), it reads its own task's value and never sees another job's. Comment added above _CurrentJob in 6056041.

should that be clarified better?

Yes. In 6d830dd the comment block explains it in order: why (code deep inside a job can't reach the loop), what happens (the loop writes down which job is running; report() reads it back), an example, and only then what a ContextVar is (a variable with a separate value in each asyncio task).

luca-belli · 2026-11-05T21:42:00Z

+    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))


i don't understand how the empty yield. works

It's how @contextmanager works: the generator is split at yield. with reporting_for(...): runs the code before yield when the block starts (set the current job), then runs the body of the with block at the yield, then runs the finally when the block ends, even if it raised (put back the previous value). Nothing is yielded because the block needs no value; the yield only marks where the body goes. The docstring says this now (6056041).

so the whole purpuose is to emit the token?

No: the token was only the receipt for putting the old value back when the block ended. Your question made me check whether that's needed at all, and it isn't. Each job already runs in its own asyncio task, and a value set inside a task stays in that task and disappears when the task ends. So in 6d830dd the context manager, the token, the yield, and the reset are gone: the loop calls set_current_job(...) at the start of each job's task, and that's all.

luca-belli · 2026-11-05T21:44:34Z

+        ...
+
+
+Status = Literal["finished", "failed", "skipped"]


is it? I am taling about the Status var

luca-belli · 2026-11-05T21:45:34Z

+        ...
+
+
+Status = Literal["finished", "failed", "skipped"]


also isn't status a property of the jjob? not the loop

luca-belli · 2026-11-05T21:48:31Z

+        """
+        # 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()


do we expect those to be run within a context? is that what it si happenign?

Yes. Each job runs inside async with self.limits[job.kind]: (in _run_one). A semaphore used that way lets in up to cap jobs of its kind at once and makes the rest wait their turn. For a kind with no cap, nullcontext() is a context manager that lets everything in, so _run_one needn't special-case it. Comment added in 6056041.

luca-belli · 2026-11-05T21:49:54Z

+            kind: asyncio.Semaphore(cap) if cap else contextlib.nullcontext()
+            for kind, cap in max_concurrent.items()
+        }
+        self.reporters = reporters


why are the reporter per loop? shouldn't be per job?

Per loop on purpose. A reporter is a destination for the whole run: the terminal, or events.jsonl. Every job's events go to the same places, and each event already says which job it's from (job and kind). Per-job reporters would mean every job holding the terminal and the file, and jobs must stay plain data (decision 7). A job still reports its own events through report().

luca-belli · 2026-11-05T21:57:15Z

+
+- `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,


does it mean tht we have logging for free now?

if we listen to those events

Partly, yes. events.jsonl already is a structured log of every event. A third reporter that forwards each event to Python's logging would be about five lines, and every job event would then land in the run's log too. What we don't get for free is the detail the current per-conversation logs have (prompts, full responses), since those aren't events.

Right, that's the idea: anything that wants them implements emit(event) and is passed in reporters=[...]. The terminal and events.jsonl are two such listeners; a logging one or a live display (#271's step 4) are more. Nothing in the loop changes to add one.

luca-belli · 2026-11-05T22:01:44Z

+    JOB_SKIPPED: "skip",
+    CALL_RETRYING: "retry",
+}
+VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"}


those should be defined where the strings are defined, not here

Agreed, moved: the words each event shows as on the terminal are now in events.py, right under the event names (6056041).

luca-belli · 2026-11-05T22:02:13Z

+    """
+
+    def __init__(self, stream: TextIO | None = None, verbose: bool = False):
+        self.stream = stream or sys.stdout


why not having directly stdout in the init?

Because a default argument is evaluated once, when Python first reads the file. stream: TextIO = sys.stdout would capture whatever sys.stdout was at import, so anything that later swaps it (pytest's output capture, a redirect) wouldn't get the lines. stream or sys.stdout looks it up when the reporter is made. Comment added in 6056041.

- 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 <noreply@anthropic.com>

luca-belli · 2026-11-05T23:45:42Z

+from typing import Any, Callable, Iterator, Protocol
+
+# Emitted by the loop.
+JOB_STARTED = "job_started"


should those be literals? Enums? States?

An enum. In 6d830dd they're an EventType StrEnum: one type that groups them and that pyright checks (report() and Event take an EventType, so a typo is an error, not a silent new event). Each member is also its plain string ("job_started"), so events.jsonl is unchanged. A Literal would type-check too, but code would still pass bare strings around; the enum gives them names.

luca-belli · 2026-11-05T23:46:00Z

+
+@dataclass(frozen=True)
+class Event:
+    """One thing that happened. `job` and `kind` are empty for a stage event."""


what's a stage event?

The one event that isn't about a job: STAGE_FINISHED, emitted once when JobLoop.run() ends, with the run's counts (done, failed, skipped). That's why it has no job or kind. The Event docstring says so now (6d830dd), and EventType groups it separately: "emitted by the loop once, about the whole run".

luca-belli · 2026-11-05T23:46:50Z

+
+
+@dataclass(frozen=True)
+class _CurrentJob:


should that be clarified better?

luca-belli · 2026-11-05T23:46:59Z

+    kind: str
+
+
+_current_job: ContextVar[_CurrentJob | None] = ContextVar("current_job", default=None)


what does it even mean?

That line declares the variable that holds the running job: _current_job is a ContextVar whose value is that job's name, kind, and where its events go, or None outside a job. Because it's a ContextVar, every asyncio task has its own value, so concurrent jobs don't overwrite each other. It now has a one-line comment saying exactly that (6d830dd).

luca-belli · 2026-11-05T23:47:34Z

+    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))


so the whole purpuose is to emit the token?

- 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 <noreply@anthropic.com>

luca-belli and others added 3 commits October 30, 2026 17:28

refactor: simpler job loop

a0f250f

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 <noreply@anthropic.com>

luca-belli commented Nov 4, 2026

View reviewed changes

luca-belli added a commit that referenced this pull request Nov 4, 2026

docs: job-loop sketch uses a semaphore per kind, as built in #308

728edfb

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

luca-belli commented Nov 5, 2026

View reviewed changes

		VERBOSE_WORDS = {TURN_STARTED: "turn", QUESTION_ASKED: "ask"}


		class PlainLineReporter:

		CALL_FINISHED = "call_finished" # data: model, input_tokens, output_tokens


		def _now() -> datetime:

		kind: str


		_current_job: ContextVar[_CurrentJob \| None] = ContextVar("current_job", default=None)

Conversation

luca-belli commented Oct 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What's in it

Decisions made here, for review

Checks

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

luca-belli commented Oct 31, 2026 •

edited

Loading