feat: resume an interrupted upload, and pause one

Stage 8 of ~/next/improve-downloads.md, second half, plus the gap it exposed in
stage 7.

**Asking where to resume.** The node identifies an upload by (member,
directory, filename), so a client resuming one has to name the file — and
`transfer_open`, the obvious place to ask, travels in clear. Naming it there
would undo exactly what sealing this path bought in MNP 2.0: before it, the same
file was ciphertext leaving a node and plaintext arriving at one. So the
question is asked inside the seal that already exists, as an ordinary
`file_upload` with no bytes and `chunk_index: -1`. The node writes nothing,
creates no state, reserves no name, and answers with `resume_from` in the sealed
ack. A node that predates it refuses the index, which the client reads as "start
from the beginning" — the behaviour it had anyway — and the wait is bounded so
one that answers neither does not strand an upload.

The probe is answered after every check the write path makes, so it cannot ask
questions about a directory the caller may not write to, and it answers only
about the member who asks: otherwise one member could measure another's
progress on a file they never sent, and worse, resume it.

**Pausing an upload.** Reported: no pause button on an upload, even in the
desktop app. Stage 7 built pause around the download path — a target declares
whether it can be stopped — and an upload has no local target to ask. It was
also refused by design, since a transfer handed a lease it cannot re-create must
not be offered a button that would drop its slot for good. Uploads now ask for
their slot rather than being handed one, and say they are pausable outright: a
File is seekable and the node keeps the position. Resuming re-probes rather than
trusting the client's own memory, so it works across a reconnect too.

**And the slot they hold.** `_do_file_upload` never called `slots.touch(tr)`.
Chunks are not gated by the lease, so the file arrived — but the node reclaimed
a grant nobody appeared to be using after thirty seconds, twice, then abandoned
it, and the widget follows the lease. Measured from the journal: a 3.5 GB upload
read "waiting, 0 ahead" for a minute and a half while it was transferring. The
download twin of this was fixed on 2026-09-08; the same omission was still here,
invisible until uploads took a real lease.

`test_the_upload_itself_is_sealed` now checks every message `uploadFile` sends
rather than the first. Adding the probe put a second one in front of the one it
was written for, and it would have kept passing while guarding nothing.

Node suite 1202 passed, hub suite 850 passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

1 files changed, 129 insertions, 0 deletions

diff --git a/packages/meshbay-node/tests/test_partial_uploads.py b/packages/meshbay-node/tests/test_partial_uploads.py
index ea5637c..f5b6602 100644
--- a/packages/meshbay-node/tests/test_partial_uploads.py
+++ b/packages/meshbay-node/tests/test_partial_uploads.py
@@ -28,6 +28,10 @@ from meshbay_node.indexer.group_index import GroupIndex
 from meshbay_node.roots import Root, RootSet
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
+from meshbay_common.protocol import (
+    UPLOAD_PROBE_INDEX, file_upload_ack_payload,
+)
+
 from conftest import one_root, sealed_upload
 
 from meshbay_node.uploads import (
@@ -358,3 +362,128 @@ def test_an_upload_in_flight_is_known_to_the_reaper(tmp_path):
     assert len(live) == 1
     assert next(iter(live)).name == "film.mkv.part"
     assert next(iter(live)).exists()
+
+
+# ── asking where to resume ──────────────────────────────────────────────────
+
+
+def _acks(session, ctx):
+    return [file_upload_ack_payload(ctx["gek"], GROUP, m)
+            for m in session.sent if m.get("type") == "file_upload_ack"]
+
+
+def _probe(session, filename: str) -> dict:
+    """The question, asked exactly as the client asks it: an ordinary sealed
+    upload chunk with no bytes and the probe index."""
+    return sealed_upload(session, filename=filename, data=b"",
+                         chunk_index=UPLOAD_PROBE_INDEX, total_chunks=1)
+
+
+def test_a_probe_for_an_unknown_file_says_start_at_the_beginning(tmp_path):
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(_probe(peer, "film.mkv"))
+    assert _errors(peer) == []
+    assert _acks(peer, ctx)[0]["resume_from"] == 0
+
+
+def test_a_probe_reports_what_the_node_already_holds(tmp_path):
+    """The point of the whole stage: the client learns it has 2 chunks there and
+    sends the third, instead of sending a film again."""
+    ctx = _group_ctx(tmp_path)
+    first = _peer(ctx)
+    for i in range(2):
+        first._do_file_upload(sealed_upload(first, filename="film.mkv",
+                                            data=b"xxxx", chunk_index=i,
+                                            total_chunks=5))
+    assert _errors(first) == []
+
+    reconnected = _peer(ctx)
+    reconnected._do_file_upload(_probe(reconnected, "film.mkv"))
+    ack = _acks(reconnected, ctx)[0]
+    assert ack["resume_from"] == 2
+    assert ack["stored_as"] == "film.mkv"
+
+
+def test_a_probe_writes_nothing_and_reserves_nothing(tmp_path):
+    """It has to be free of consequence: a client that asks and goes away must
+    leave no file, no state and no name taken."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(_probe(peer, "film.mkv"))
+    root = ctx["roots"].roots[0]
+    assert list(root.path.iterdir()) == []
+    assert len(ctx.get("partial_uploads") or []) == 0
+    # And it promises no destination it has not taken.
+    assert _acks(peer, ctx)[0]["stored_as"] == ""
+
+
+def test_a_probe_answers_only_about_the_member_who_asks(tmp_path):
+    """Same keying as the upload itself. Otherwise one member could measure
+    another's progress on a file they never sent — and worse, resume it."""
+    ctx = _group_ctx(tmp_path)
+    alice = _peer(ctx, "alice")
+    alice._do_file_upload(sealed_upload(alice, filename="film.mkv",
+                                        data=b"xxxx", chunk_index=0,
+                                        total_chunks=5))
+    bob = _peer(ctx, "bob")
+    bob._do_file_upload(_probe(bob, "film.mkv"))
+    assert _acks(bob, ctx)[0]["resume_from"] == 0
+
+
+def test_an_ordinary_ack_carries_no_resume_field(tmp_path):
+    """So a client can tell a probe's answer from a chunk's without looking at
+    the index it echoed."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="a.bin", data=b"x",
+                                       chunk_index=0, total_chunks=2))
+    assert "resume_from" not in _acks(peer, ctx)[0]
+
+
+def test_a_probe_is_refused_where_an_upload_would_be(tmp_path):
+    """Every check the write path makes has already run when the probe is
+    answered, so it cannot be used to ask questions about somewhere the caller
+    may not write."""
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    peer._do_file_upload(sealed_upload(peer, filename="../escape",
+                                       data=b"", chunk_index=UPLOAD_PROBE_INDEX,
+                                       total_chunks=1))
+    assert [m.get("code") for m in _errors(peer)] == ["invalid_filename"]
+    assert _acks(peer, ctx) == []
+
+
+# ── the slot an upload holds ────────────────────────────────────────────────
+
+def test_an_upload_chunk_says_its_slot_is_in_use(tmp_path):
+    """A grant nobody takes up is reclaimed after thirty seconds and abandoned
+    on the third miss. Uploads are not gated by the lease, so the file arrived
+    anyway — but the widget follows the lease, and a 3.5 GB upload therefore
+    read "waiting, 0 ahead" for a minute and a half while it was transferring,
+    with three reclaims logged against it.
+
+    The download twin of this was fixed a day earlier; the same omission was
+    still here, invisible until uploads took a real lease.
+    """
+    from meshbay_node.transfers import TransferSlots, UPLOAD
+
+    ctx = _group_ctx(tmp_path)
+    peer = _peer(ctx)
+    slots = TransferSlots()
+    peer._ctx = dict(ctx)
+    peer._ctx["_transfer_slots"] = slots
+    peer._registry_key = "session-1"
+    lease, err = slots.open(tr="up-1", kind=UPLOAD, session_key="session-1",
+                            user_id="user-1", group_id=GROUP, bytes=10, chunks=2)
+    assert not err and lease.state == "granted"
+    assert lease.used is False
+
+    msg = sealed_upload(peer, filename="film.mkv", data=b"xxxx",
+                        chunk_index=0, total_chunks=2)
+    msg["tr"] = "up-1"
+    peer._do_file_upload(msg)
+
+    assert _errors(peer) == []
+    assert slots.leases["up-1"].used is True, (
+        "the node still believes nobody took this slot up, and will reclaim it")