feat(spa): pause and resume a download, in session

Stage 7a of ~/next/improve-downloads.md: pausing within a session, on the
targets that can actually do it. Resuming across a reload is 7b.

A paused transfer holds **nothing**. Its slot goes back to the node the moment
it stops and resuming rejoins the queue at the tail, because anything else lets
one member close a node by pausing four downloads and going to lunch. So the
lease is taken inside the run loop rather than before it, and pause is refused
outright for a transfer that could not ask for another one.

Resuming is exact rather than approximate: the pipeline stops between two
chunks and never inside one, so what is on disk is always a whole number of
chunks and `fromChunk` is a verified position. The failure mode being avoided
is a file that looks complete and is quietly corrupt.

The target has to survive it, so a pause no longer reaches the `abort()` that a
failure does -- that would delete Electron's `.part` or the file just created in
the granted folder, leaving nothing to continue. And the in-memory fallback
keeps its accumulated chunks rather than starting a second array.

The button is drawn only where the target says it can. A service-worker stream
says no, in its own code and for its own reasons: the browser is already writing
an HTTP response into its own download folder, not feeding it stalls that
download where we cannot see or resume it, and an idle worker is terminated
within seconds. Firefox and Safari therefore keep cancel and get no pause, which
is the decision recorded in §6.5.

Cancelling a paused transfer ends it. A paused run is parked on a promise;
without waking it the row said "cancelled" over work that had not stopped and a
target that was still open.

Six cases, each checked against the unfixed source. Hub suite 842 passed.

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

2 files changed, 228 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_downloads.py b/packages/meshbay-hub/tests/test_downloads.py
index dfb1433..41ae5d3 100644
--- a/packages/meshbay-hub/tests/test_downloads.py
+++ b/packages/meshbay-hub/tests/test_downloads.py
@@ -489,3 +489,71 @@ out.push(asked);
     assert batched == [False, True, True], (
         "the stuck one is still the only holder of the gesture, so the released "
         "openings must not try for a dialog of their own")
+
+
+def test_a_pause_falls_between_chunks_and_resumes_at_one(tmp_path):
+    """What makes resuming exact rather than approximate.
+
+    Everything written is a whole number of chunks, because the loop checks for
+    a pause between two of them and never inside one. So `fromChunk` is a
+    position, not an estimate, and a resumed download is never appended to at an
+    offset nobody verified — the failure mode being avoided is a file that looks
+    complete and is quietly corrupt.
+
+    The real `pipelinedDownload` is lifted out and run against stubs, on the
+    rule this repo follows for the video player: model the environment, never
+    the code under test.
+    """
+    src = (STATIC / "file-utils.js").read_text()
+    fn = src[src.index("async function pipelinedDownload"):]
+    fn = fn[:fn.index("\n}\n") + 2]
+
+    script = tmp_path / "pipeline.mjs"
+    script.write_text("""
+const CHUNK_SIZE = 8;
+const PIPELINE_WINDOW = 4;
+const written = [];
+// `ct` has to be truthy: chunk 0 with a falsy body is refused as undecryptable,
+// which is the guard working, not the harness.
+const _fetchChunkResilient = async (transport, fileId, i) =>
+  ({ ct: new Uint8Array([i & 0xff]), nonce: new Uint8Array(12) });
+const _writeOrStall = async (w, bytes, index) => { written.push(index); };
+globalThis.window = { MeshBayCrypto: {
+  // The plaintext carries its own index, so what lands where can be checked.
+  decryptChunkBin: async (k, id, index) => ({ byteLength: CHUNK_SIZE, index }),
+} };
+""" + fn + """
+const out = {};
+const signal = { aborted: false, paused: false };
+// Stop it part way, the way the store does.
+let seen = 0;
+const onChunk = () => { if (++seen === 3) signal.paused = true; };
+try {
+  await pipelinedDownload({}, 'k', 'file', 10, onChunk, {}, signal, '', 0);
+  out.threw = 'no';
+} catch (err) {
+  out.threw = err.name;
+}
+out.resumeFrom = signal.resumeFrom;
+out.writtenBeforePause = written.slice();
+
+// And again, from where it said.
+signal.paused = false;
+written.length = 0;
+await pipelinedDownload({}, 'k', 'file', 10, () => {}, {}, signal, '',
+                        out.resumeFrom);
+out.writtenAfterResume = written.slice();
+console.log(JSON.stringify(out));
+""")
+    proc = subprocess.run(["node", str(script)], capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    out = json.loads(proc.stdout)
+
+    assert out["threw"] == "PausedError", out
+    # Whole chunks only, in order, with nothing skipped.
+    assert out["writtenBeforePause"] == list(range(len(out["writtenBeforePause"])))
+    assert out["resumeFrom"] == len(out["writtenBeforePause"]), (
+        f"stopped after {len(out['writtenBeforePause'])} chunks but asked to "
+        f"resume at {out['resumeFrom']} — that gap is a hole in the file")
+    # The resumed run covers exactly the rest, and repeats nothing.
+    assert out["writtenAfterResume"] == list(range(out["resumeFrom"], 10)), out
diff --git a/packages/meshbay-hub/tests/test_transfers.py b/packages/meshbay-hub/tests/test_transfers.py
index c48e38c..035f569 100644
--- a/packages/meshbay-hub/tests/test_transfers.py
+++ b/packages/meshbay-hub/tests/test_transfers.py
@@ -553,3 +553,163 @@ def test_a_transport_is_not_closed_under_a_preparing_transfer(tmp_path):
     """, tmp_path)
     assert out[1] is False
     assert out[3] is True
+
+
+# ── Pause and resume ────────────────────────────────────────────────────────
+#
+# The rule the whole design turns on: **a paused transfer holds nothing.** Its
+# slot goes back to the node the moment it stops, and resuming rejoins the queue
+# at the tail. Anything else lets one member close a node by pausing four
+# downloads and going to lunch (§6.2 of ~/next/improve-downloads.md).
+
+
+def _pausable_run():
+    """A `run` that stops where it is told and reports where it resumed."""
+    return """
+const mkStore = () => {
+  const t = new TransferStore();
+  const leases = [];
+  const state = { starts: [], paused: null, aborted: false };
+  t.start({
+    kind: 'download', name: 'f', total: 1000,
+    prepare: async () => ({ name: 'f', pausable: true }),
+    makeLease: () => { const l = new L(); leases.push(l); return l; },
+    run: async ({ signal, from }) => {
+      state.starts.push(from);
+      state.running = true;
+      try {
+      // Runs until told to stop, one "chunk" at a time.
+      for (let i = from; i < 10; i++) {
+        await new Promise(r => setTimeout(r, 5));
+        if (signal.aborted) { const e = new Error('c'); e.name = 'AbortError'; throw e; }
+        if (signal.paused) {
+          signal.resumeFrom = i;
+          const e = new Error('p'); e.name = 'PausedError'; throw e;
+        }
+      }
+      } finally { state.running = false; }
+    },
+  });
+  return { t, leases, state };
+};
+"""
+
+
+def test_pausing_gives_the_slot_back(tmp_path):
+    """The node has to get it back at once, not when the person resumes: the
+    whole point of a queue is that a slot nobody is using is a slot somebody
+    else can have."""
+    out = _run(_lease_stub() + _pausable_run() + """
+      const { t, leases } = mkStore();
+      await new Promise(r => setTimeout(r, 5));
+      leases[0].grant();
+      await new Promise(r => setTimeout(r, 20));
+      say('before:' + t.list()[0].status);
+      t.pause(t.list()[0].id);
+      await new Promise(r => setTimeout(r, 30));
+      say('after:' + t.list()[0].status);
+      say('released:' + leases[0].released.join(','));
+      say('leases:' + leases.length);
+    """, tmp_path)
+    assert out[0] == "before:running"
+    assert out[1] == "after:paused"
+    assert out[2] == "released:paused", "a paused transfer kept its slot"
+    assert out[3] == "leases:1"
+
+
+def test_resuming_asks_for_a_new_slot_and_continues_where_it_stopped(tmp_path):
+    """Rejoining at the tail is the design, not an accident: a paused transfer
+    that could reclaim its old place would be a way to hold one."""
+    out = _run(_lease_stub() + _pausable_run() + """
+      const { t, leases, state } = mkStore();
+      await new Promise(r => setTimeout(r, 5));
+      leases[0].grant();
+      await new Promise(r => setTimeout(r, 20));
+      const id = t.list()[0].id;
+      t.pause(id);
+      await new Promise(r => setTimeout(r, 30));
+      t.resume(id);
+      await new Promise(r => setTimeout(r, 10));
+      say('queued:' + t.list()[0].status, 'leases:' + leases.length);
+      leases[1].grant();
+      await new Promise(r => setTimeout(r, 120));
+      say('end:' + t.list()[0].status);
+      say('starts:' + state.starts.join(','));
+    """, tmp_path)
+    assert out[0] == "queued:queued", "a resumed transfer skipped the queue"
+    assert out[1] == "leases:2", "resuming did not ask for a slot again"
+    assert out[2] == "end:done"
+    starts = out[3].split(":")[1].split(",")
+    assert starts[0] == "0" and int(starts[1]) > 0, (
+        f"resumed from {starts} — it started again from the beginning")
+
+
+def test_a_transfer_whose_target_cannot_pause_is_not_paused(tmp_path):
+    """A service-worker stream is a download the browser already owns: not
+    writing to it stalls it outside our control and an idle worker is killed
+    within seconds. A button that silently restarts from zero is worse than no
+    button, so `pause` refuses rather than pretending."""
+    out = _run(_lease_stub() + """
+      const t = new TransferStore();
+      const lease = new L();
+      t.start({ kind: 'download', name: 'f', total: 10, lease,
+                prepare: async () => ({ name: 'f' }),
+                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;
+      say('pausable:' + t.list()[0].pausable);
+      t.pause(id);
+      await new Promise(r => setTimeout(r, 20));
+      say('status:' + t.list()[0].status);
+      t.cancel(id);
+    """, tmp_path)
+    assert out == ["pausable:false", "status:running"]
+
+
+def test_cancelling_a_paused_transfer_actually_ends_it(tmp_path):
+    """A paused run is parked on a promise. Without waking it, cancel marks the
+    row and leaves the work parked for the life of the page, holding its target
+    open — a button that lies, in the same way the first test in this file
+    describes."""
+    out = _run(_lease_stub() + _pausable_run() + """
+      const { t, leases, state } = mkStore();
+      await new Promise(r => setTimeout(r, 5));
+      leases[0].grant();
+      await new Promise(r => setTimeout(r, 20));
+      const id = t.list()[0].id;
+      t.pause(id);
+      await new Promise(r => setTimeout(r, 30));
+      t.cancel(id);
+      // What matters is whether the store's own loop ends, not whether the row
+      // says so: the row is marked at once either way.
+      const settled = await Promise.race([
+        t._items[0].promise.then(() => 'settled', () => 'settled'),
+        new Promise(r => setTimeout(() => r('parked'), 60)),
+      ]);
+      say('status:' + t.list()[0].status);
+      say('loop:' + settled);
+      say('resumed:' + state.starts.length);
+    """, tmp_path)
+    assert out[0] == "status:cancelled"
+    assert out[1] == "loop:settled", (
+        "the run was still parked on the resume promise after a cancel — the "
+        "row said cancelled over work that had not stopped")
+    assert out[2] == "resumed:1", "cancelling started the work again"
+
+
+def test_a_paused_transfer_still_counts_as_live(tmp_path):
+    """It is not finished, and its transport must not be closed under it — the
+    person is coming back to it."""
+    out = _run(_lease_stub() + _pausable_run() + """
+      const { t, leases } = mkStore();
+      await new Promise(r => setTimeout(r, 5));
+      leases[0].grant();
+      await new Promise(r => setTimeout(r, 20));
+      t.pause(t.list()[0].id);
+      await new Promise(r => setTimeout(r, 30));
+      say('pending:' + t.pending);
+    """, tmp_path)
+    assert out == ["pending:1"]