feat: an event stream, so an agent reads one stream and not twenty answers

[stephenc]

Closes part of #10.

What it does

qex events --json

One JSON object on one line for each change of state, at the moment of the
change. An agent that drives twenty jobs reads one stream instead of asking
about each job in a loop.

{"event":"stream","time":...,"version":"0.8.0","pid":123,"coordinator_started_at":...,"first_seq":1,"last_seq":9}
{"event":"job","seq":10,"time":...,"id":"a1b2...","name":"build","state":"running","previous":"starting","change":"state","job":{ ...the whole record... }}
{"event":"gap","time":...,"missed":37,"next_seq":420,"reason":"..."}
{"event":"bye","time":...,"reason":"the coordinator stops, because ..."}

The socket protocol keeps its framing: Events is one more request name, and
the coordinator answers it with many Event responses on the same connection.
There is no second transport and no second format.

The decisions, and why

What counts as an event — a change of the record of a job, and nothing else.

• A change of state. This includes the FIRST state, so the admission of a job is
  the queued line with previous: null. A separate admitted event would say
  the same thing twice.
• A change of the reason that a queued job waits (change: "reason"). This one
  is not optional. The scheduler writes blocked_reason a moment AFTER the
  admission, so a reader that has the admission line alone sees
  blocked_reason: null and never learns that the job waits for memory, for a
  lock, or for a job that failed. This is the event that makes the stream
  usable.
• Three lines that are not job events, each answering a question the reader
  cannot answer itself: stream (which coordinator, which numbers), gap (you
  lost events, here is the count), bye (the coordinator stops, and why).
• A deleted record gives NO event. The reader deleted it; that is not news.

Catch-up: the default is catch-up, not live-only. Every event has a seq,
and --since N continues after it. --since start (the default) gives
everything the coordinator still holds, and --since now gives live only. A
stream that silently starts at "now" makes an agent that restarts lose the
results that arrived while it was away — exactly the class of fault this tool
exists to remove, so live-only is not the default.

The numbers belong to one coordinator. The stream line carries
coordinator_started_at, and a --since number that this coordinator never
issued gives a gap line that says the coordinator restarted. The stream never
pretends that a number from a dead coordinator was honoured.

A slow reader: drop, and COUNT the drop. The coordinator keeps the last 512
events in a ring behind the same lock as the jobs. It never waits for a reader
and its memory never grows for one. When the ring passes a reader, that reader
gets a gap line with missed. A silent gap is worse than a reported one: an
agent that loses the line failed and hears nothing waits for ever. A write to
a reader that has stopped reading gives up after 30 seconds and closes that one
connection; the reader then meets the end of the stream, and the coordinator
loses one thread and nothing else.

Capability negotiation. events is now in capabilities::ALL, and
qex events refuses a coordinator that does not report it: the message names
the coordinator, says what it cannot obey, and gives kill <pid>. The check
happens BEFORE the request, so the words come from qex and not from a parser.
This matters more here than anywhere else, because an empty stream and a stream
with no events look identical — a silent refusal makes the reader wait for ever.
A new function capabilities::check_command covers a command that needs a
request name, beside check, which covers a job that carries an option.

The coordinator still retires. An attached reader does NOT hold it open: a
stream that keeps a coordinator alive for ever is a leak. The coordinator stops
on the same rule as before, and it writes a bye line first, then waits up to
one second for its readers. A stream that ends with NO bye means that
something stopped the coordinator; qex events says so on stderr and exits 1. A
stream that dies in silence under its reader is a fault, because the reader
cannot tell it from a broken socket.

One writer for the stream. State::publish_changes compares the records
with the records it reported last, and every path that changes a job calls it —
the queue, the scheduler, qex cancel, qex kill, the supervisor, and the
refresh that reads the status files. Calling it twice for one change costs a
comparison and gives no repeated line, so a new code path cannot silently miss
the stream. The stream and qex list read one map, so they can never disagree.

While testing this, the shutdown of the coordinator was moved to delete its
socket file FIRST. The window in which a new command connects to a coordinator
that no longer accepts connections — and then reports "version unknown, pid 0" —
was already there; the wait for the readers would have widened it.

What I measured

• One event is one job record: about 1KB. The ring is 512 events, so the stream
  costs the coordinator about half a megabyte, whatever a reader does.
• Order, on a job that sleeps two seconds:
  queued -> starting -> running -> completed, one line each, flushed as it
  happens. A job of a few milliseconds gives queued -> starting -> completed:
  the stream reports what the coordinator holds, and it does not invent a state
  that the coordinator never saw.
• Retirement under a reader: with a three second idle limit, the reader received
  the bye line and exited 0. The coordinator did not stay.
• Gaps: with the ring reduced to three events, four jobs gave
  {"missed":6} for --since start and {"missed":5} for --since 1, and the
  stream continued after the gap.

Tests

New unit tests (src/events.rs, src/capabilities.rs, src/schema.rs,
src/proto.rs):

• each_event_fits_one_line_of_json — a reader splits on the newline.
• the_ring_holds_the_last_events_and_counts_the_others — a wrong count makes a
  gap line that lies.
• a_reader_receives_the_events_after_its_number_only
• the_numbers_never_repeat — a repeated number makes a restarted reader lose
  an event.
• each_change_gives_one_line_and_a_repeat_gives_none — several paths publish
  for one change; an agent must not act twice.
• each_cursor_form_survives_the_wire
• a_command_is_refused_by_a_coordinator_that_does_not_know_it and
  this_build_says_that_it_has_the_event_stream — the capability rule.
• the_event_schema_names_each_type_of_line

New end-to-end tests (tests/e2e.rs). Every reader in a test has a time limit,
so no test can hang:

• the_event_stream_reports_each_change_of_state_in_order — the reader starts
  BEFORE the job, and it gets queued, starting, running, completed in order,
  one JSON object per line, with increasing numbers and the whole record.
• a_reader_continues_from_the_number_that_it_read — no loss and no repetition.
• the_stream_counts_the_events_that_it_dropped — with a small ring, the reader
  gets a gap with missed > 0 and the stream continues.
• the_coordinator_retires_under_a_reader_and_says_goodbye — the reader does
  not hold the coordinator open, and it hears the stop.
• a_coordinator_that_has_no_event_stream_refuses_the_command — the test
  answers on the socket like an earlier version. qex events exits 1, names the
  remedy, and writes no stream. The test starts no qex coordinator, so it kills
  no process.

Results:

cargo fmt --all                                  clean
cargo clippy --all-targets -- -D warnings        clean
cargo test --bins                                181 passed, 0 failed
cargo test --test e2e -- --test-threads=2        76 passed, 0 failed (50s)

Also in this change

• docs/reference.md, docs/agents.md, docs/index.md, README.md and
  skills/qex/SKILL.md.
• qex help events, a new topic, and the command in the command list.
• qex schema event — the stream is an interface for a program, so it gets a
  schema like the other JSON that qex publishes.
• The version moves to 0.8.0. This repository generates no shell completions, so
  there was nothing to add there.

🤖 Generated with Claude Code

https://claude.ai/code/session_01KNvagiMEU3myn8EXGaGEM9

[stephenc]

Review findings addressed in 4956dc0.

1 — a reader that goes away on a quiet queue (thread and fd leak). stream() now tests the connection itself at each POLL tick: poll(POLLIN, 0), then a one-byte MSG_PEEK | MSG_DONTWAIT; a read of zero bytes, or POLLHUP/POLLERR/POLLNVAL, means the reader closed its end. The peek never removes a byte, so a later CLI that sends something still finds its bytes. A write cannot find this condition, because on a quiet queue there is nothing to write — that is the whole fault.

Measured with QEX_IDLE_EXIT_SECS=300 on an idle queue: baseline threads=3 fds=4; immediately after 15 bounded qex events --json --since now --timeout 1s readers, threads=4 fds=6 (the one that had just exited, inside the 250ms tick); two seconds later, threads=3 fds=4. wait_for_readers no longer burns its second, for the same reason.

New e2e test a_reader_that_goes_away_leaves_no_thread_behind counts /proc/<pid>/task and /proc/<pid>/fd after 8 bounded readers (#[cfg(target_os = "linux")]; macOS has no /proc). I checked that it pins the behaviour: with the new check disabled it FAILS, and it passes with it.

2 — --since N after a restart. I took the better fix; it was contained. Each stream now has a name: EventLog generates a stream_id (uuid v4) and the stream line carries it. Cursor::After became { seq, stream: Option<Uuid> }, and --since <stream_id>:<seq> sends both. When the name is given and differs, the coordinator emits the gap itself and continues from first_seq. stream_id rather than coordinator_started_at, because two coordinators can start in one second — a retiring coordinator and the one the next command starts, which is exactly the case in question — and a pid repeats.

A bare --since N still cannot be checked by the coordinator, so qex events now writes a warning naming the remedy, and docs/agents.md, skills/qex/SKILL.md, src/help.rs, docs/reference.md and src/schema.rs all say: keep two values, stream_id and seq, and give both. The "you then lose nothing" sentences are gone.

New e2e test a_number_from_a_coordinator_that_stopped_gives_a_gap: three jobs, the coordinator is killed, and --since <old_stream>:2 against the new one gives a gap with missed: null whose reason names the old stream, then continues from seq 1. a_reader_continues_from_the_number_that_it_read now uses the <stream_id>:<seq> form.

3 — the doc comment. info has its pgrep paragraph back; events keeps only its own.

4 — a state that the coordinator did not see. Said plainly in qex help events (a section of its own), in the state description of qex schema event, in the module comment, and in the three agent-facing pages: the stream reports what the coordinator OBSERVED, it reads each record twice a second, a shorter job goes starting → completed with no running line, and previous gives the true sequence. A line for a state qex did not see would be the fault, not its absence.

5 — missed: 0. missed is now Option<u64> and is null whenever the two streams have no common measure. The schema types it ["integer","null"] and says why; the text form writes "qex cannot count the events that you lost".

cargo fmt --all                             clean
cargo clippy --all-targets -- -D warnings   clean
cargo test --bins                           182 passed, 0 failed
cargo test --test e2e -- --test-threads=2   78 passed, 0 failed (57.4s)

Every reader in every test still has a time limit.