pgw#1002/#1003/#1004/#1005: a publish failure costs the UPLOAD, not the CAST (th#1653 client half)

[PaulFidika]

The client half of the th#1653 upload audit. Four defects on one exit path
which, together, turned a recoverable transient at the end of a 2h16m fp8 cast
into total loss — and which hid each other, so fixing any one alone would have
measured as no improvement.

pgw#1002 (P0) — honor the hub's own classification

HubPublishError carries the hub's th#1301 retryable bit and _map_exception
had no branch for the type at all, so a failure the hub explicitly asked us
to retry reported JOB_STATUS_FATAL. Only RETRYABLE is requeued, so the
orchestrator's five-attempt budget was never spent on the final artifact of a
two-hour job — while intermediate checkpoints, which raise
ArtifactTransferError with an honest flag, were requeued. That asymmetry
is what named the bug.

retryable=True → RETRYABLE; False (a repudiation) and None ("the hub named
nothing") stay FATAL. Provenance decides the class, never our reading of the
message (th#1259); the hub's code leads the detail so refusals group by a
stable token.

And the abort. DELETE /publishes/:id is not a tidy-up — hub-side it runs
cleanupCASPublishV2Staging and deletes every staged chunk. The handler sent it
on any exception, so a blip on chunk 590 of 590 threw away 37 GB to keep a
session row from looking in-flight. It now fires only for a refusal the hub
itself classified terminal. The default is KEEP.

pgw#1003 (P0) — the artifact survives

• IncrementalSafetensorsWriter.close() closed a handle: no fsync, no atomic
  rename, on the path that produces multi-GB cast outputs. Now temp → fsync →
  os.replace → fsync(dir), matching the download side since gw#408. An
  incomplete tensor set, or a body that raised, is discarded rather than left
  under the real name for the publisher to hash and ship.
• New convert/publish_journal.py. publish_id existed only as a local
  variable, so no successor could name the session whose staged bytes it should
  resume. The journal records it — plus the declared object set, the declared
  paths, and an opaque producer state — beside the produced bytes, before the
  first PUT, and clears it only on promotion. publish_v2 re-adopts a
  journalled session instead of declaring a fresh one; a session the hub no
  longer accepts costs one round trip and falls through to a fresh declare.
• clone.py closes the loop: its finally rmtree'd the produced tree on every
  exit path ("only the downloaded source is resumable", stated outright in the
  code). It now retains a workdir that still holds a resumable session — a
  machine-checkable condition, not a mode, bounded by the scratch sweeper that
  already exists — and a tree a predecessor finished casting and declaring
  is re-published rather than re-cast. Identity is checked twice: the declared
  path set on disk, then publish_v2 re-hashing every file.

Resume-exploitation design, and the hub-lane dependency

Scope is stated rather than implied: this is a same-pod durability record.
The disk dies with the pod. It covers the case that motivated the issue — a
blip, a hub restart, a completion timeout, an orchestrator requeue onto the same
worker — and deliberately not cross-pod resume, which needs the id to travel
through the hub (th#1654's own open question).

It does not block on th#1654. The client re-plans the adopted session
through POST .../publishes/:id/grants and uploads only what the need set still
names; th#1654 is what makes that set come back smaller (staged-but-unpromoted
objects are invisible to the planner until it lands). Before then the client
costs one extra round trip; the day it lands, a blip costs a chunk. Not aborting
the session on a transport failure is the precondition that makes th#1654 worth
anything at all — staging that is deleted on the way out can never be resumed.

th#1655 does not apply to this lane. th#1655 says so itself: "the repo-CAS
v2 path is NOT affected by this one — it has /grants." The CAS path's presign
re-mint is that existing route, so pgw#1004 §C is implemented against it with no
cross-lane gate.

pgw#1004 (P1) — the chunk data plane grows up

• _put_one had five attempts and no time.sleep anywhere: four threads
  hammering a store that had just answered 429. Now classify-before-charging (a
  terminal status raises on the spot and spends nothing) plus the repo's one
  decorrelated-jitter helper, promoted to _upload_transport.backoff_sleep_s
  rather than reimplemented a fourth time.
• UploadGrant carries expires_at — on the wire since th#1303, dropped on the
  floor. Expired (or inside the margin) re-plans on its own budget instead of
  consuming a re-upload pass, and a 403 on an already-expired grant is finally
  distinguishable from a substituted claim.
• Every completed object feeds upload:bytes + note_progress(), in the data
  plane itself rather than in a callback a caller can forget.
  convert/publish.py emits per-leg convert_publish events where it
  previously emitted none.
• Concurrency 4 → 8 under a 16-PUT ceiling, anchored to the two measured in-repo
  reference points (the hub's 16-window ranged reader; the SDK path's 10
  workers). The slot is taken before the span is read, so it bounds buffer
  residency rather than socket count.
• _concurrent_upload.py's claim to own a parallel_map_uploads that has never
  existed is deleted rather than reasserted.

pgw#1005 (P2) — verified digests and real tests

• upload_file_with_grant returned the caller's unverified claim and forwarded
  it into /complete. It now proves the digest from the bytes, refuses an
  expired scoped credential (expires_at: parsed, never read), and stops
  multiplying four outer attempts by botocore's ten. The docstring states what
  the check is and is not — this path is SDK multipart, where R2 does not
  enforce x-amz-checksum-sha256 on UploadPart, which is exactly why the CAS
  path keeps per-chunk presigns and the platform's strongest integrity property.
  That property is untouched.
• _upload_transport gets its first functional coverage: re-open-per-attempt at
  the part's true offset, fresh pool per retry + discard_connections, every
  classifier branch, the 2xx-without-ETag refusal.
• The four adversarial injectors tests/convert/fake_hub.py has shipped since
  the day it was written, which no test ever switched on, are wired to the
  v2 chunk surface and driven: mid-PUT reset, 5xx-then-success, expired presign,
  terminal repudiation.
• One row at the real 64 MiB chunk size — every existing upload test
  monkeypatches it to 4096.

Tests

New: tests/test_chunk_upload_robustness_pgw1004.py (11),
tests/test_upload_transport_pgw1005.py (30),
tests/test_grant_upload_digest_pgw1005.py (7),
tests/convert/test_publish_survival_pgw1003.py (12),
tests/convert/test_publish_status_mapping_pgw1002.py (8).

All integration-grade: real sockets, real files, real threads, the real
publish_v2 and the real _map_exception. No wall-clock assertions —
backoff is proven by injecting the sleep and reading what the loop asked for.

One existing row flips with the abort rule:
test_corrupt_local_bytes_are_refused_BEFORE_the_transfer asserted
hub.aborts and now asserts the opposite, with the reason inline.

Gates: mypy clean (236 files), ruff clean, lint_http_timeouts,
lint_unreached_surface, lint_config_reads all green.

No version bump — rides the next train.