fix(run-stats): count blocks server-side so blocks_processed is correct in every worker mode

[rhoadesScholar]

Third of three, replacing #70. This is the substantial one — it changes the wire protocol and an
internal Rust API — and it was the piece hardest to see under the other two in #70.

Targets v2.0 directly. It shares no files with #73, so the two can be reviewed in either order.
See the note at the bottom about running the reproduction, though.

Symptom

The Resource Utilization report showed blocks 0 per task for subprocess-mode runs — the default —
and empty per-worker counts. tests/test_run_stats.py had to pin itself to thread mode to assert
anything. This was the "known gap" section in docs/source/design/RUN_STATS.md.

import daisy

def process(block):
    pass                      # clean return = success

task = daisy.Task(
    task_id="stats_mwe",
    total_roi=daisy.Roi((0,), (60,)),      # 6 blocks of 10
    read_roi=daisy.Roi((0,), (10,)),
    write_roi=daisy.Roi((0,), (10,)),
    process_function=process, read_write_conflict=False, max_workers=2,
)
daisy.run_blockwise([task], multiprocessing=True, progress=True)

Both runs complete 6/6 blocks — the scheduler was always fine, only the accounting was wrong.

On v2.0:

  Per-task:
    task            blocks  max conc    mean ms ∠ slope         cpu busy      wall
    stats_mwe            0         2      0.43 ∠ -0.2131              2%     0.17s

On this branch:

  Per-task:
    task            blocks  max conc    mean ms ∠ slope         cpu busy      wall
    stats_mwe            6         2      0.46 ∠ -0.2110              2%     0.17s

Root cause

WorkerStats.blocks_processed was incremented only inside the in-process worker thread's block loop.
Subprocess shim workers and external cluster workers process blocks where the stats layer never looked.

Fix — count where every worker already reports

• New protocol message Register { task_id, worker_id }, sent by Client::connect right after the
  connection opens. Appended as the last enum variant, so the wire discriminants of the six existing
  messages are unchanged.
• The bookkeeper records each connection's registered identity; every valid block return
  (ReleaseBlock or BlockFailed) increments a RunTally counter per task and per worker.
• build_run_stats merges the per-worker counts into the exit-channel WorkerStats by
  (task_id, worker_id). Registered workers with no thread in the server process — fully external
  workers — get synthetic entries carrying their block counts.

Semantics

• Counts are now identical across thread, subprocess-shim and external workers (the shim child registers
  with the same worker id as its babysitter thread).
• Reclaimed attempts (block timeout, dead client) were never valid returns and are no longer counted
  — slightly more honest than the old thread-loop counter.
• Failed-but-returned blocks still count, matching the old accounting.
• Old client / new server: a client that never registers still yields correct per-task counts; only
  its per-worker attribution is absent.
• New client / old server: out of scope, since server and workers ship together. Flagging it
  explicitly rather than leaving it implicit: that assumption does not hold for every consumer — volara,
  for instance, spawns volara-cli workers that resolve daisy from their own environment, so a mixed
  install is possible there.
• Registration doubles as a liveness signal: registering clears any stale closed-connection marker
  left on an OS-recycled ephemeral port, and disconnecting drops the registration, so recycled ports
  cannot inherit stale identities or get their in-flight blocks falsely reclaimed.

Internal API change: Client::connect and _rs.SyncClient now take a worker_id. Both are
internal; daisy.Client reads it from DAISY_CONTEXT as before.

docs/source/design/RUN_STATS.md is updated — the "known gap" section is replaced by the Register
design, with daisy.profile_block kept as the future path for per-block CPU/RSS.

Tests

$ cargo test -p daisy-core
test result: ok. 41 passed
test result: ok. 5 passed      # integration_tcp

$ pytest tests/test_run_stats.py -q
5 passed

Rust: unit tests for RunTally counting and the build_run_stats merge (including synthetic
external-worker entries and the unregistered-client fallback), bookkeeper registration lifecycle and
recycled-port hygiene; integration_tcp.rs now asserts externally-connected registered workers appear in
run stats. Python: the previously thread-pinned tests now run the default subprocess mode and their
blocks_processed assertions pass; a new pinned test keeps thread-mode counting covered.

This also fixes tests/test_run_stats.py::test_slowing_workload_reports_positive_slope, which fails
reproducibly on v2.0 (3/3 runs on a tree that touches nothing in run_stats). I flagged it as
pre-existing on #73; it is fixed here.

One note on reproducing the MWE

The block function above deliberately does not reference the daisy module. If you write the more
natural block.status = daisy.BlockStatus.SUCCESS under an editable install, you will hit
TypeError: cannot pickle '_thread._local' object before reaching the stats — that is the separate bug
fixed in #73. Either apply #73 first, or keep the block function free of module globals as above.

[pattonw]

Thanks for the report. This was a problem and was due to the same values being counted in multiple places (blocks processed). There should really only be one source of ground truth so that this sort of thing doesn't happen. resource tracking was in a fairly proof-of-concept state anyway so I fleshed it out. I removed the duplicated block done tracking, and made the resource tracking optional and more generally applicable. It now fires in the Client acquire_block context manager meaning even custom methods that people call on e.g. slurm will be properly tracked if desired without any custom user code.
That fix is merged and now resource tracking is much more stable 🙏