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

4 files changed, 141 insertions, 12 deletions

diff --git a/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs b/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs
index 0b77e42..a6008c2 100644
--- a/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs
+++ b/packages/meshbay-hub/tests/harness/upload_seal_probe.mjs
@@ -72,13 +72,34 @@ tp._nodeVersion = input.node_version;
 
 const frames = [];
 let uploadId = null;
+let answered = 0;
 tp._send = (msg) => {
   frames.push(toHex(msgpack_encode(msg)));
   if (msg.upload_id) uploadId = msg.upload_id;
-  if (input.mode !== 'receive') return;
+  if (input.mode !== 'receive') {
+    // Nothing answers in this mode -- except the probe, which the client waits
+    // five seconds for. A node that predates it refuses the index, and that
+    // refusal is a plain error rather than a sealed ack, so the harness can
+    // produce it honestly. It is also the degradation path worth exercising.
+    if (msg.chunk_index === -1) {
+      // With `probe_ack`, answer it the way a node holding part of this file
+      // does; without, the way one that predates the probe does.
+      const reply = input.probe_ack
+        ? Object.assign(msgpack_decode(hex(input.probe_ack)),
+                        { upload_id: uploadId })
+        : { type: 'error', upload_id: uploadId,
+            code: 'bad_chunk_index', detail: 'Unexpected chunk index' };
+      setImmediate(() => tp._dispatch(reply));
+    }
+    return;
+  }
   // Answer as the node did, on the next turn of the loop so the send path
   // finishes first — which is also how a real ack arrives.
-  const ack = msgpack_decode(hex(input.acks[msg.chunk_index]));
+  //
+  // By position, not by `chunk_index`: the node answers every frame including
+  // the probe, whose index is -1, and the two lists are built from the same
+  // sequence of frames.
+  const ack = msgpack_decode(hex(input.acks[answered++]));
   ack.upload_id = uploadId;
   // Through the real `_dispatch`, so the routing under test — matching an
   // ack to its uploader by `upload_id` — is the shipped one.
diff --git a/packages/meshbay-hub/tests/test_transfers.py b/packages/meshbay-hub/tests/test_transfers.py
index 035f569..f27d4fd 100644
--- a/packages/meshbay-hub/tests/test_transfers.py
+++ b/packages/meshbay-hub/tests/test_transfers.py
@@ -713,3 +713,60 @@ def test_a_paused_transfer_still_counts_as_live(tmp_path):
       say('pending:' + t.pending);
     """, tmp_path)
     assert out == ["pending:1"]
+
+
+def test_an_upload_can_be_paused_without_a_prepare_step(tmp_path):
+    """A download learns whether it can pause from its target, because only the
+    target knows. An upload has no target to ask: a `File` is seekable and the
+    node keeps the position, so it says so outright.
+
+    This was missed when pause shipped — the button appeared on downloads and
+    nowhere else, including in the desktop app where everything else works.
+    """
+    out = _run(_lease_stub() + """
+      const t = new TransferStore();
+      const leases = [];
+      t.start({
+        kind: 'upload', name: 'f', total: 100, pausable: true,
+        makeLease: () => { const l = new L(); leases.push(l); return l; },
+        run: async ({ signal, from }) => {
+          for (let i = from || 0; i < 10; i++) {
+            await new Promise(r => setTimeout(r, 5));
+            if (signal.paused) {
+              signal.resumeFrom = i;
+              const e = new Error('p'); e.name = 'PausedError'; throw e;
+            }
+          }
+        },
+      });
+      await new Promise(r => setTimeout(r, 5));
+      leases[0].grant();
+      await new Promise(r => setTimeout(r, 20));
+      say('pausable:' + t.list()[0].pausable);
+      t.pause(t.list()[0].id);
+      await new Promise(r => setTimeout(r, 30));
+      say('status:' + t.list()[0].status);
+      say('released:' + leases[0].released.join(','));
+    """, tmp_path)
+    assert out == ["pausable:true", "status:paused", "released:paused"]
+
+
+def test_an_upload_handed_a_lease_it_cannot_recreate_is_not_offered_pause(tmp_path):
+    """Pausing gives the slot back. A transfer that cannot ask for another one
+    would pause once and wait for ever, so the button is refused instead."""
+    out = _run(_lease_stub() + """
+      const t = new TransferStore();
+      const lease = new L();
+      t.start({ kind: 'upload', name: 'f', total: 100, pausable: true, lease,
+                run: async ({ signal }) => {
+                  while (!signal.aborted) await new Promise(r => setTimeout(r, 5));
+                } });
+      lease.grant();
+      await new Promise(r => setTimeout(r, 20));
+      const id = t.list()[0].id;
+      t.pause(id);
+      await new Promise(r => setTimeout(r, 20));
+      say('status:' + t.list()[0].status);
+      t.cancel(id);
+    """, tmp_path)
+    assert out == ["status:running"]
diff --git a/packages/meshbay-hub/tests/test_transport_contracts.py b/packages/meshbay-hub/tests/test_transport_contracts.py
index 879062b..fe550f9 100644
--- a/packages/meshbay-hub/tests/test_transport_contracts.py
+++ b/packages/meshbay-hub/tests/test_transport_contracts.py
@@ -349,16 +349,29 @@ def test_the_upload_itself_is_sealed(transport):
         "the upload must be sealed under the group key")
     assert "openGroup(" in body and "'file_upload_ack'" in body, (
         "the ack carries the stored name and must be opened, not read")
-    # The message the node actually receives: everything between `this._send({`
-    # and its close. Read on its own, because the same field names appear a few
-    # lines above inside `msgpack_encode({...})`, which is the sealed half.
-    sent = body[body.index("this._send({"):]
-    sent = sent[:sent.index("});")]
-    assert "filename" not in sent, "the filename is on the message in clear"
-    assert "data" not in sent, "the bytes are on the message in clear"
-    assert "dir" not in sent and "root" not in sent, (
-        "the destination is on the message in clear")
-    assert "...sealed," in sent, "the message must carry the sealed pair"
+    # The messages the node actually receives: everything between each
+    # `this._send({` and its close. Read on their own, because the same field
+    # names appear a few lines above inside `msgpack_encode({...})`, which is
+    # the sealed half.
+    #
+    # Every one of them, not the first: `uploadFile` sends a probe chunk before
+    # the file ("where am I?", UPLOAD_PROBE_INDEX) and it names the file too, so
+    # a check that stopped at the first message would have moved off the one it
+    # was written for the day the second appeared.
+    sends = []
+    rest = body
+    while "this._send({" in rest:
+        rest = rest[rest.index("this._send({"):]
+        sends.append(rest[:rest.index("});")])
+        rest = rest[len("this._send({"):]
+    assert len(sends) >= 2, "the probe and the chunks are both sent from here"
+    for sent in sends:
+        assert "filename" not in sent, "the filename is on the message in clear"
+        assert "data" not in sent, "the bytes are on the message in clear"
+        assert "dir" not in sent and "root" not in sent, (
+            "the destination is on the message in clear")
+        assert "...sealed," in sent or "...probeSealed," in sent, (
+            "the message must carry the sealed pair")
     assert "supportsSealedUpload" in body, (
         "an older node must be refused before a chunk is sent, not after")
 
diff --git a/packages/meshbay-hub/tests/test_upload_seal_client.py b/packages/meshbay-hub/tests/test_upload_seal_client.py
index d6f9156..2e4bfb5 100644
--- a/packages/meshbay-hub/tests/test_upload_seal_client.py
+++ b/packages/meshbay-hub/tests/test_upload_seal_client.py
@@ -167,3 +167,41 @@ def test_the_client_refuses_an_older_node_before_sending_a_chunk(_gek):
     assert result["state"] == "rejected"
     assert "older MeshBay" in result["message"]
     assert result["frames"] == [], "a chunk was sent to a node that cannot open it"
+
+
+def test_an_interrupted_upload_resumes_where_the_node_stopped(tmp_path, _gek):
+    """
+    The browser asks, the node answers, and the second attempt sends only what
+    is missing.
+
+    Both halves are the shipped ones: the frames come from the real
+    `uploadFile`, the answer comes from the real node handler. What is asserted
+    is the thing that used to be impossible — an upload interrupted at chunk two
+    of five that sends three chunks instead of five.
+    """
+    body = bytes(range(256)) * ((CHUNK * 5) // 256 + 1)
+    body = body[:CHUNK * 5]
+    first = _run_probe(_probe_input(_gek, "send",
+                                    file={"name": "film.mkv", "data": body.hex()}))
+    frames = [msgpack.unpackb(bytes.fromhex(f), raw=False)
+              for f in first["frames"]]
+    assert [f["chunk_index"] for f in frames] == [-1, 0, 1, 2, 3, 4]
+
+    # The link drops after two chunks.
+    session = _node_session(tmp_path, _gek)
+    for frame in frames[1:3]:
+        session._do_file_upload(frame)
+    assert not [m for m in session.sent if m.get("type") == "error"]
+
+    # It comes back and asks.
+    session.sent.clear()
+    session._do_file_upload(frames[0])
+    probe_ack = msgpack.packb(session.sent[-1], use_bin_type=True).hex()
+
+    second = _run_probe(_probe_input(
+        _gek, "send", file={"name": "film.mkv", "data": body.hex()},
+        probe_ack=probe_ack))
+    resumed = [msgpack.unpackb(bytes.fromhex(f), raw=False)["chunk_index"]
+               for f in second["frames"]]
+    assert resumed == [-1, 2, 3, 4], (
+        f"sent {resumed} — the answer to the probe was not used")