…th the writer thread

test_cleanup_all_drains_queue failed ~20% of the time, in isolation
and in suites. The race:

  T0 save() puts item in queue
  T1 writer queue.get() pulls item (already off the queue)
  T2 cleanup_all() drains queue — finds it empty
  T3 cleanup_all() rmtree(snapshot_dir) + mkdir(snapshot_dir)
  T4 writer mkdir(req_dir) + temp write + rename(final)

Result: ``req-X/N.safetensors`` survives the supposed cleanup, the
caller (scheduler / shutdown) believes the snapshot dir is empty, and
the leftover file is the disk shadow of a request that should have
been forgotten.

Add ``_writer_busy`` (threading.Lock). The writer holds it for the
entire duration of each ``_process_write_item`` call. ``cleanup_all()``
drains the queue first (no new items can start) and then acquires
``_writer_busy``, so any item the writer had already pulled finishes
before rmtree runs.

Same class of race exists for ``cleanup_request()`` — writer pulls a
request's item before cleanup_request sets the cancelled-counter, then
mkdir's the req_dir back into existence after the rmtree, and the
late-rename rescue at the tail of ``_process_write_item`` catches it
in most timings but a small window remains. Apply the same
``_writer_busy`` barrier (bounded, see below) so the two cleanup paths
are symmetric.

Per-path timeouts so cleanup_request can yield faster than
cleanup_all:

- ``_CLEANUP_ALL_TIMEOUT_S = 5.0`` — cleanup_all runs at reset /
  startup where blocking longer is tolerable in exchange for a
  stronger orphan-avoidance guarantee.
- ``_CLEANUP_REQUEST_TIMEOUT_S = 2.0`` — cleanup_request is called
  from the scheduler's abort hot path (~3 sites) where bounding
  latency matters more than chasing one in-flight write.

The bounded acquires have a logged-warning fallback. Without that
bound a slow disk inside ``_write_safetensors_no_mx`` could deadlock
the scheduler's hot path — cleanup_all() is called from request abort
/ reset (scheduler.py:5400 / 5596 / 5737), not just shutdown. The
worst-case fallback behaviour is an orphaned file under the recreated
snapshot dir until next startup's constructor cleanup, which is the
exact same state the pre-fix code produced on every cleanup_all.

Counter pop on cleanup_request now only runs when the
``_writer_busy`` acquire succeeded. On the timeout fallback the
writer is still mid-item and may not yet have consulted
``_is_cancelled``; popping the counter there would defeat the late-
rename rescue that the docstring advertises as the timeout-
fallback safety net. The previous code popped unconditionally.

Counter bump itself is now additive (``get + count``) and skipped
entirely when ``count == 0``. Two distinct bugs that closes:

- Skip-on-zero: cleanup_request("X") for an rid with NO pending
  items previously wrote ``cancelled[X] = 0`` then popped on the
  acquired path. On the timeout fallback the pop never ran and the
  ``X: 0`` entry lingered for the process lifetime — every later
  ``save()`` under that rid (or any reuse of the string) was
  silently discarded by the writer's ``_is_cancelled`` gates,
  which check key membership not value > 0. Counter must only
  exist when there is at least one in-flight item to drain it.
  Regression: test_cleanup_request_no_pending_does_not_pin_counter
  _on_timeout (new).
- Additive: a re-entrant cleanup_request("X") for an rid that
  already has an in-flight cancellation must NOT overwrite the
  previous count. The writer's ``cleared_by_cleanup`` branch +
  ``_writer_busy`` lock together close the file-write race today
  (so no orphan file slips out), but the per-item dec_cancelled
  bookkeeping has to balance — overwriting drops remaining decs on
  the floor and on the next ``save()`` under the same rid the
  writer would see a non-zero counter from the earlier batch and
  silently discard the new item.

shutdown() now accepts ``cleanup=True`` so callers that want both
operations can express the cleanup-before-shutdown ordering in one
call instead of sequencing them manually. The cleanup_all() warning
at the top of the function catches misordered callers that still
call cleanup_all() AFTER shutdown — the writer no longer reacquires
``_writer_busy`` past the sentinel, so a post-shutdown cleanup
degrades to an in-memory-only clear.

Tests:
- ``test_cleanup_all_drains_queue`` — no longer sleeps; relies on the
  lock to guarantee ordering, runs deterministically.
- ``test_cleanup_all_blocks_until_writer_finishes_pinned_item`` —
  monkey-patches ``_write_safetensors_no_mx`` to block on an Event,
  pins the writer mid-item, asserts cleanup_all does NOT return until
  the writer releases. Without the lock this test fails
  deterministically rather than the original ~20% flake rate.
- ``test_cleanup_request_blocks_until_writer_finishes_pinned_item`` —
  symmetric for cleanup_request.
- ``test_cleanup_request_keeps_counter_on_timeout`` — regression for
  the bug where the timeout-fallback pop dropped the late-rename
  rescue's safety net.
- ``test_cleanup_request_timeout_drains_counter_on_writer_early_return``
  — the writer's cleared_by_cleanup early-return must still
  dec_cancelled or the counter pins the rid forever.
- ``test_cleanup_request_no_pending_does_not_pin_counter_on_timeout``
  (new) — regression for the skip-on-zero bug above. Pins the writer
  with an unrelated save, calls cleanup_request("never-saved-rid")
  past the 0.1s timeout, asserts the rid does NOT appear in
  ``_cancelled_requests`` AND that a subsequent save under the same
  rid is not silently discarded.
- ``test_shutdown_cleanup_true_runs_cleanup_before_setting_flag`` —
  pins the cleanup-before-shutdown ordering of the new
  ``shutdown(cleanup=True)`` path.

24/24 boundary-store tests pass.