From 9257198eecd68893d14c67669d8b1f9d3e76aedb Mon Sep 17 00:00:00 2001
From: "Timothy J. Aveni" <tja@berkeley.edu>
Date: Sun, 20 Sep 2026 19:58:05 -0700
Subject: [PATCH] Add cooperative cancellation for asynchronous package
 invocations

---
 portable-service.mjs | 38 +++++++++++++++++++++++++++++---------
 1 file changed, 29 insertions(+), 9 deletions(-)

diff --git a/portable-service.mjs b/portable-service.mjs
index 402a5e1..812437e 100644
--- a/portable-service.mjs
+++ b/portable-service.mjs
@@ -20,14 +20,21 @@ export function servePortableRegistry(
     code: typeof error?.code === "string" ? error.code.slice(0, 1024) : "PACKAGE_ERROR",
     message: String(error?.message ?? error).slice(0, 4096),
   });
-  const cancel = (id, reason) => {
-    active.delete(id);
+  const rejectPorts = (id, reason) => {
     for (const [key, call] of pending)
       if (call.invocation === id) {
         pending.delete(key);
         call.reject(reason);
       }
   };
+  const cancel = (id, reason) => {
+    const token = active.get(id);
+    if (token) {
+      token.cancelled = true;
+      token.controller.abort(reason);
+    }
+    rejectPorts(id, reason);
+  };
   const fatal = (error) => {
     if (stopped) return;
     stopped = true;
@@ -45,13 +52,17 @@ export function servePortableRegistry(
       return;
     }
     if (message.kind === "cancel") {
+      const token = active.get(message.invocation);
+      if (token && token.frame !== message.frame) throw Error("INVALID_CANCELLATION_FRAME");
       cancel(message.invocation, Error("INVOCATION_CANCELLED"));
+      // A completed result may already be in flight. Acknowledge without resurrecting its context.
+      if (!token) send({ abi, kind: "cancelled", invocation: message.invocation, frame: message.frame });
       return;
     }
     if (message.kind === "port-result") {
       const key = JSON.stringify([message.invocation, message.sequence]);
       const call = pending.get(key);
-      if (!call && !active.has(message.invocation)) return; // A canceled call may already have a reply in flight.
+      if (!call && (!active.has(message.invocation) || active.get(message.invocation).cancelled)) return;
       if (!call || message.frame !== call.frame) throw Error("UNEXPECTED_PORT_RESULT");
       pending.delete(key);
       if (message.status === "returned") call.resolve(message.output);
@@ -75,11 +86,12 @@ export function servePortableRegistry(
     )
       throw Error("INVALID_INVOCATION");
     if (active.has(message.invocation) || active.size >= maxInvocations) throw Error("INVOCATION_LIMIT_OR_DUPLICATE");
-    const token = {};
+    const token = { frame: message.frame, controller: new AbortController(), cancelled: false };
     active.set(message.invocation, token);
     let sequence = 0;
     const channel = (request) => {
-      if (active.get(message.invocation) !== token) return Promise.reject(Error("EXPIRED_INVOCATION"));
+      if (active.get(message.invocation) !== token || token.cancelled)
+        return Promise.reject(Error("EXPIRED_INVOCATION"));
       if (++sequence > maxPortCalls || pending.size >= maxPortCalls) return Promise.reject(Error("PORT_CALL_LIMIT"));
       const key = JSON.stringify([message.invocation, sequence]);
       return new Promise((resolve, reject) => {
@@ -105,21 +117,29 @@ export function servePortableRegistry(
       .invoke(
         { exportId: message.exportId, workspace: message.workspace, receiver: message.receiver, input: message.input },
         channel,
+        Object.freeze({ signal: token.controller.signal, effect: message.effect }),
       )
       .then(
         (output) => {
-          if (active.get(message.invocation) === token)
+          if (active.get(message.invocation) === token && !token.cancelled)
             send({ abi, invocation: message.invocation, status: "returned", output });
         },
         (error) => {
-          if (active.get(message.invocation) === token)
+          if (active.get(message.invocation) === token && !token.cancelled)
             send({ abi, invocation: message.invocation, status: "failed", error: errorBody(error) });
         },
       )
       .catch(fatal)
       .finally(() => {
-        if (active.get(message.invocation) === token) cancel(message.invocation, Error("INVOCATION_ENDED"));
-      });
+        if (active.get(message.invocation) === token) {
+          active.delete(message.invocation);
+          token.controller.abort(Error("INVOCATION_ENDED"));
+          rejectPorts(message.invocation, Error("INVOCATION_ENDED"));
+          if (token.cancelled && !stopped)
+            send({ abi, kind: "cancelled", invocation: message.invocation, frame: token.frame });
+        }
+      })
+      .catch(fatal);
   };
   process.stdin.on("data", (bytes) => {
     if (stopped) return;