fix: put presigned transfer work inside one owned abort scope

TR-RR-05. The shared abortable-operation primitive now distinguishes VALUE,
REJECTED and TERMINAL, so a collaborator's own rejection is no longer forged
into a cancellation and race() always names the same first owner terminal()
reports. The caller signal and scheduler are captured once, so replacing a
method after construction cannot change how an in-flight operation is bounded.
A scheduler that cannot install the deadline is itself terminal: previously it
released the caller listener and left no owner, which made every later abort
invisible. Late values are compensated exactly once.

Both presigned subsystems, which each carried their own copy of these
mechanics, are now projections of that primitive — giving it real production
importers rather than a shared helper nobody used.

TR-RR-01. close() on an active download aborts the scope instead of only
dropping listeners, so a fetch or read already in flight actually stops. The
consumer's stream signal joins the operation's ownership before any I/O begins,
so an already-aborted consumer no longer causes one network request first.

TR-RR-02. The upload abort scope is created before the digest, and the digest
races the caller and the deadline like every other step. A non-settling hash can
no longer hold put() open, and the vault claim and the network call happen only
after the owner is re-checked.

TR-RR-03. A capability registration is a versioned exact union validated at
registration time: the protocol version, HTTPS only, exact own-data fields, a
2xx expected status, and no ambient credential or cookie header — a presigned
URL carries its own authorization, and a session header alongside it would send
the user's credentials to that origin.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/unit/abortable-operation.test.ts

@@ -81,13 +81,75 @@ describe("shared abortable operation mechanics", () => {
     expect(operation.terminal()).toBe("CLOSED");
   });
 
-  it("observes a late rejection instead of leaking it", async () => {
+  /**
+   * TR-RR-05. A collaborator's own rejection is evidence about the work.
+   * Reporting it as `TERMINAL/CLOSED` erased the reason the operation failed
+   * and made `race()` disagree with `terminal()`, which still said no owner.
+   */
+  it("keeps a rejection distinct from a terminal owner", async () => {
     const operation = createAbortableOperation();
-    const rejected = Promise.reject(new Error("late"));
-    await expect(operation.race(rejected)).resolves.toEqual({
+    const reason = new Error("upstream failed");
+    await expect(
+      operation.race(Promise.reject(reason)),
+    ).resolves.toEqual({ kind: "REJECTED", reason });
+    expect(operation.terminal()).toBeNull();
+  });
+
+  it("agrees with terminal() on the first owner", async () => {
+    const caller = new AbortController();
+    const operation = createAbortableOperation({ signal: caller.signal });
+    const pending = operation.race(new Promise<never>(() => {}));
+    caller.abort();
+    // A later close cannot overwrite the first owner in either place.
+    operation.close();
+    await expect(pending).resolves.toEqual({
       kind: "TERMINAL",
-      terminal: "CLOSED",
+      terminal: "CALLER_ABORT",
     });
+    expect(operation.terminal()).toBe("CALLER_ABORT");
+  });
+
+  it("compensates a late value exactly once instead of admitting it", async () => {
+    const caller = new AbortController();
+    const operation = createAbortableOperation({ signal: caller.signal });
+    const compensated: string[] = [];
+    let release: ((value: string) => void) | undefined;
+    const pending = operation.race(
+      new Promise<string>((resolve) => {
+        release = resolve;
+      }),
+      (value) => compensated.push(value),
+    );
+    caller.abort();
+    expect(await pending).toEqual({
+      kind: "TERMINAL",
+      terminal: "CALLER_ABORT",
+    });
+    release?.("late-value");
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(compensated).toEqual(["late-value"]);
+  });
+
+  it("ignores a scheduler method replaced after construction", async () => {
+    const scheduler = {
+      setTimer: (callback: () => void, delayMs: number) =>
+        setTimeout(callback, delayMs),
+      clearTimer: (handle: unknown) => {
+        clearTimeout(handle as ReturnType<typeof setTimeout>);
+      },
+    };
+    const operation = createAbortableOperation({
+      timeoutMs: 5,
+      setTimer: scheduler.setTimer,
+      clearTimer: scheduler.clearTimer,
+    });
+    scheduler.setTimer = () => {
+      throw new TypeError("replaced after construction");
+    };
+    await expect(
+      operation.race(new Promise<never>(() => {})),
+    ).resolves.toEqual({ kind: "TERMINAL", terminal: "DEADLINE" });
   });
 
   it("close is idempotent and releases listeners and timers exactly once", () => {
@@ -109,7 +171,13 @@ describe("shared abortable operation mechanics", () => {
     expect(clearTimer).toHaveBeenCalledTimes(1);
   });
 
-  it("survives a throwing scheduler without leaking a listener", () => {
+  /**
+   * TR-RR-05. A scheduler that cannot install the deadline leaves the operation
+   * unbounded. Removing the caller listener and leaving no terminal owner made
+   * every later abort invisible, so installation failure is itself terminal and
+   * closes atomically with the resources already created.
+   */
+  it("closes atomically when the scheduler cannot install the deadline", async () => {
     const caller = new AbortController();
     const remove = vi.spyOn(caller.signal, "removeEventListener");
     const operation = createAbortableOperation({
@@ -121,8 +189,13 @@ describe("shared abortable operation mechanics", () => {
       clearTimer: () => {},
     });
 
-    expect(operation.terminal()).toBeNull();
+    expect(operation.terminal()).toBe("CLOSED");
     expect(remove).toHaveBeenCalledTimes(1);
+    // The operation is bounded, so a caller abort afterwards cannot be lost.
+    caller.abort();
+    await expect(
+      operation.race(new Promise<never>(() => {})),
+    ).resolves.toEqual({ kind: "TERMINAL", terminal: "CLOSED" });
   });
 
   it("compensates a late native handle without changing the outcome", async () => {

tests/unit/presigned-transfer.test.ts

@@ -801,7 +801,13 @@ describe("presigned transfer", () => {
     void payload;
   });
 
-  it("does not leak an abort listener when the scheduler throws", async () => {
+  /**
+   * TR-RR-05. A scheduler that cannot install the deadline leaves the operation
+   * unbounded. Releasing the caller listener and continuing anyway meant a
+   * later abort was invisible, so an install failure is itself terminal: the
+   * request fails closed with a typed Result and its resources are released.
+   */
+  it("fails closed when the scheduler cannot install the deadline", async () => {
     const bytes = new Uint8Array([1, 2, 3]);
     const payload = downloadCapabilityPayload(bytes);
     const caller = new AbortController();
@@ -827,7 +833,7 @@ describe("presigned transfer", () => {
         resourceId: "resource-1",
         signal: caller.signal,
       }),
-    ).resolves.toMatchObject({ ok: true });
+    ).resolves.toMatchObject({ ok: false });
     expect(remove).toHaveBeenCalled();
   });
 
@@ -838,6 +844,32 @@ describe("presigned transfer", () => {
     { label: "maxBytes below byteLength", patch: { maxBytes: 0 } },
     { label: "malformed digest", patch: { expectedSha256: "not-a-digest" } },
     { label: "non-positive expiry", patch: { expiresAtEpochMs: 0 } },
+    // TR-RR-03. The registration is a versioned exact union: plaintext, an
+    // ambient credential header, an unknown protocol version and any extra
+    // own field are all refused at the issuer seam.
+    { label: "unknown protocol version", patch: { protocol: "PRESIGNED_TRANSFER_V0" } },
+    { label: "missing protocol version", patch: { protocol: undefined } },
+    {
+      label: "plaintext target",
+      patch: {
+        href: `http://objects.example${DOWNLOAD_PATH}`,
+        origin: "http://objects.example",
+      },
+    },
+    {
+      label: "ambient credential header",
+      patch: {
+        requestHeaders: [{ name: "authorization", value: "Bearer leak" }],
+      },
+    },
+    {
+      label: "cookie response header",
+      patch: {
+        requiredResponseHeaders: [{ name: "set-cookie", value: "a=b" }],
+      },
+    },
+    { label: "status outside 2xx", patch: { expectedStatus: 302 } },
+    { label: "extra own field", patch: { injected: true } },
   ])(
     "rejects a malformed registration at the vault issuer seam ($label)",
     ({ patch }) => {
@@ -848,6 +880,7 @@ describe("presigned transfer", () => {
         maxActiveCapabilities: 4,
       });
       const base = {
+        protocol: PRESIGNED_TRANSFER_PROTOCOL,
         capabilityReceipt: "capability-direct-1",
         method: "GET" as const,
         binding: { kind: "DOWNLOAD" as const, resourceId: "resource-1" },