🔒 Require scope-bound event listeners in Effection code (#748)

.oxlintrc.json

@@ -19,7 +19,8 @@
     "local/no-sync-filesystem": "error",
     "local/no-yield-in-finally": "error",
     "local/prefer-effection-operation": "error",
-    "local/prefer-effection-result": "error"
+    "local/prefer-effection-result": "error",
+    "local/require-scope-bound-event-registration": "error"
   },
 
   "overrides": [

AGENTS.md

@@ -447,6 +447,28 @@ the corpus or enabling in-shard concurrency is not an answer to a miss.
     structural values for composition data and a contextual Api for operations.
     Security enforcement, durable identity, and reconciliation never trust
     replaceable context state.
+16. An event listener an Effection operation installs has that operation's
+    lifetime. Wait for one event with `once()` from `@effectionx/node/events`,
+    never `emitter.once()` or `addEventListener(..., { once: true })`: cleanup
+    that waits for the event is no cleanup for a wait that is cancelled. A
+    longer subscription binds a stable handler and removes that same handler,
+    from that same receiver and event, in the owner's own teardown — `.off()`
+    for Node, `.removeEventListener()` with the matching capture mode for the
+    DOM. Removal is synchronous, so it belongs in a `finally` around the
+    subscription, an `ensure()` that **completed before** the subscription, or
+    the cleanup an `action()` returns; where teardown must wait on the event
+    itself, keep the handler through the wait and remove it in a synchronous
+    `finally` inside that same `ensure()`. `yield* ensure(...)` is itself a
+    suspension: an owner halted while it registers unwinds with no cleanup on
+    it at all, so an `ensure()` yielded *after* the subscription has not
+    established anything — nor may a native resource be created before the
+    cleanup that releases it, and only the resource's own closing event proves
+    it is finished, never an assigned exit status. The listener ordering is
+    enforced by the `local/require-scope-bound-event-registration` Oxlint rule
+    (`scripts/oxlint-rules/`), which does not autofix: which owner, which
+    handler and which order are lifecycle decisions. The resource half — a
+    child spawned before the cleanup that reaps it — is not something that rule
+    can see, and is held by each owner's focused lifecycle regression instead.
 
 ## Writing Guide
 

architecture.md

@@ -2422,6 +2422,16 @@ hidden inside library objects that accumulate. One exception: metadata an
 author declares at module evaluation, about a value the author owns, may live
 on that value.
 
+A callback registered with something outside the process — a listener on a
+socket, a child process, a stream or a DOM target — is state of exactly this
+kind, and the source it is attached to knows nothing about the operation that
+attached it. The event arriving is not teardown: a cancelled wait is precisely
+the case where it never arrives, and a losing race arm is one that will never
+be told. So the owner detaches its handlers, on completion, failure, halt and
+race loss alike, before it is considered closed, and an event delivered after
+that reaches nothing and changes nothing. `local/require-scope-bound-event-registration`
+holds source to it.
+
 A Repository selection is composition data and is therefore replaceable: a
 document may bind one, render one, hand one to a child, and construct one that
 looks exactly like it. Nothing a repository provider does is authorized by the

deno.json

@@ -33,7 +33,7 @@
     "@effectionx/fetch": "npm:@effectionx/fetch@0.2.1",
     "@effectionx/fs": "npm:@effectionx/fs@0.3.0",
     "@effectionx/middleware": "npm:@effectionx/middleware@0.1.1",
-    "@effectionx/node": "npm:@effectionx/node@0.2.4",
+    "@effectionx/node": "npm:@effectionx/node@0.2.5",
     "@effectionx/process": "npm:@effectionx/process@0.8.1",
     "@effectionx/scope-eval": "npm:@effectionx/scope-eval@0.1.3",
     "@effectionx/stream-helpers": "npm:@effectionx/stream-helpers@0.8.3",

package.json

@@ -30,7 +30,7 @@
     "@effectionx/fetch": "0.2.1",
     "@effectionx/fs": "0.3.0",
     "@effectionx/middleware": "0.1.1",
-    "@effectionx/node": "0.2.4",
+    "@effectionx/node": "0.2.5",
     "@effectionx/process": "0.8.1",
     "@effectionx/scope-eval": "0.1.3",
     "@effectionx/stream-helpers": "0.8.3",

packages/acp/tests/adapter-protocol.test.ts

@@ -18,9 +18,18 @@
  */
 import { describe, it } from "@executablemd/test-support/bdd";
 import { expect } from "@executablemd/test-support/expect";
-import { all, ensure, resource, spawn as effectionSpawn, until, withResolvers } from "effection";
+import {
+  all,
+  ensure,
+  resource,
+  scoped,
+  spawn as effectionSpawn,
+  until,
+  withResolvers,
+} from "effection";
 import type { Operation } from "effection";
 import { rm } from "@effectionx/fs";
+import { Buffer } from "node:buffer";
 import { spawn } from "node:child_process";
 import type { ChildProcess } from "node:child_process";
 import { mkdtemp } from "node:fs/promises";
@@ -36,6 +45,8 @@ const FAKE_CLAUDE = join(FIXTURES, "fake-claude-cli.cjs");
 
 /** One ACP conversation with a spawned adapter. */
 interface Adapter {
+  /** The adapter process, so a case can say what the resource still observes. */
+  readonly child: ChildProcess;
   request(method: string, params: unknown): Operation<Record<string, unknown>>;
   /**
    * Every `session/update` notification the adapter sent, in arrival order.
@@ -75,7 +86,7 @@ function useAdapter(provider: string, environment: Record<string, string>): Oper
     let next = 1;
     let buffer = "";
 
-    child.stdout?.on("data", (chunk: Buffer) => {
+    const onStdout = (chunk: Buffer): void => {
       buffer += chunk.toString("utf8");
       let index = buffer.indexOf("\n");
       while (index >= 0) {
@@ -110,9 +121,20 @@ function useAdapter(provider: string, environment: Record<string, string>): Oper
           child.stdin?.write(`${JSON.stringify({ jsonrpc: "2.0", id, result: {} })}\n`);
         }
       }
+    };
+
+    // Detached before the kill above, because destructors unwind in reverse:
+    // the adapter stops being read before the process it is reading is ended.
+    // Established before the subscription, because `yield* ensure(...)` is
+    // itself a suspension an owner can be halted at.
+    yield* ensure(() => {
+      child.stdout?.off("data", onStdout);
     });
 
+    child.stdout?.on("data", onStdout);
+
     yield* provide({
+      child,
       updates,
       *request(method: string, params: unknown): Operation<Record<string, unknown>> {
         const id = next++;
@@ -330,4 +352,33 @@ describe("Tier EA — the embedded adapters' prompt-response metadata", () => {
     expect(metaOf(first, "codex")).toEqual({ turnId: `turn:${sessionId}:1` });
     expect(metaOf(second, "codex")).toEqual({ turnId: `turn:${sessionId}:2` });
   });
+  /**
+   * The adapter resource reads one child's stdout for as long as it holds it.
+   * The count is read after the resource has been torn down and before the
+   * event is replayed, because a handler removed by its own event would leave
+   * the same count behind as one the resource released.
+   */
+  it("releases the adapter's output handler with the resource", function* () {
+    let child: ChildProcess | undefined;
+    let live = 0;
+    let before = 0;
+
+    yield* scoped(function* () {
+      const adapter = yield* useAdapter("claude", {});
+      child = adapter.child;
+      before = 0;
+      live = adapter.child.stdout?.listenerCount("data") ?? 0;
+    });
+
+    if (!child) {
+      throw new Error("the adapter never started");
+    }
+
+    expect(live).toBeGreaterThanOrEqual(before + 1);
+    expect(child.stdout?.listenerCount("data") ?? 0).toBe(live - 1);
+
+    child.stdout?.emit("data", Buffer.from("after the adapter was torn down"));
+
+    expect(child.stdout?.listenerCount("data") ?? 0).toBe(live - 1);
+  });
 });

packages/acp/tests/fixtures/claude-native-launch-proof.ts

@@ -354,30 +354,52 @@ function runChild(
       options.live.delete(running.pid);
     });
 
-    child = spawnChild(command, args, { cwd: options.cwd, stdio: ["pipe", "pipe", "pipe"] });
-    if (child.pid) {
-      options.live.add(child.pid);
+    const started = spawnChild(command, args, {
+      cwd: options.cwd,
+      stdio: ["pipe", "pipe", "pipe"],
+    });
+    child = started;
+    if (started.pid) {
+      options.live.add(started.pid);
     }
     let stdout = "";
     let stderr = "";
-    child.stdout?.on("data", (chunk: Buffer) => {
+
+    const onStdout = (chunk: Buffer): void => {
       stdout += chunk.toString();
-    });
-    child.stderr?.on("data", (chunk: Buffer) => {
+    };
+    const onStderr = (chunk: Buffer): void => {
       stderr += chunk.toString();
-    });
-    child.once("error", (error: Error) => failed.reject(error));
-    child.once("close", (code: number | null, signal: string | null) => {
-      if (child?.pid) {
-        options.live.delete(child.pid);
+    };
+    const onError = (error: Error): void => failed.reject(error);
+    const onClose = (code: number | null, signal: string | null): void => {
+      if (started.pid) {
+        options.live.delete(started.pid);
       }
       settled.resolve({ code, signal, stdout, stderr });
+    };
+
+    // Registered after the cleanup above and so torn down before it: the child
+    // stops being observed before it is signalled, and a race this arm loses
+    // leaves nothing attached to a process somebody else is still reading.
+    // Established before the subscriptions, because `yield* ensure(...)` is
+    // itself a suspension an owner can be halted at.
+    yield* ensure(() => {
+      started.stdout?.off("data", onStdout);
+      started.stderr?.off("data", onStderr);
+      started.off("error", onError);
+      started.off("close", onClose);
     });
 
+    started.stdout?.on("data", onStdout);
+    started.stderr?.on("data", onStderr);
+    started.on("error", onError);
+    started.on("close", onClose);
+
     if (options.input !== undefined) {
-      child.stdin?.write(options.input);
+      started.stdin?.write(options.input);
     }
-    child.stdin?.end();
+    started.stdin?.end();
 
     return yield* race([settled.operation, failed.operation]);
   })();
@@ -502,16 +524,17 @@ function ptyRun<T>(
       options.live.delete(running.pid);
     });
 
-    child = spawnChild("/usr/bin/script", ["-q", "/dev/null", command, ...args], {
+    const started = spawnChild("/usr/bin/script", ["-q", "/dev/null", command, ...args], {
       cwd: options.cwd,
       env: options.env,
       stdio: ["pipe", "pipe", "pipe"],
     });
-    if (child.pid) {
-      options.live.add(child.pid);
+    child = started;
+    if (started.pid) {
+      options.live.add(started.pid);
     }
 
-    const react = (chunk: Buffer) => {
+    const react = (chunk: Buffer): void => {
       text += chunk.toString();
       const waiter = pending;
       if (waiter && waiter.predicate(text.slice(consumed))) {
@@ -520,17 +543,31 @@ function ptyRun<T>(
         waiter.resolve("");
       }
     };
-    child.stdout?.on("data", react);
-    child.stderr?.on("data", react);
-    child.once("error", (error: Error) => failed.reject(error));
-    child.once("close", (status: number | null) => {
+    const onError = (error: Error): void => failed.reject(error);
+    const onClose = (status: number | null): void => {
       code = status ?? -1;
-      if (child?.pid) {
-        options.live.delete(child.pid);
+      if (started.pid) {
+        options.live.delete(started.pid);
       }
       settled.resolve();
+    };
+
+    // Registered after the interrupt cleanup above and so torn down before it:
+    // the terminal stops being read before the process holding it is signalled.
+    // Established before the subscriptions, because `yield* ensure(...)` is
+    // itself a suspension an owner can be halted at.
+    yield* ensure(() => {
+      started.stdout?.off("data", react);
+      started.stderr?.off("data", react);
+      started.off("error", onError);
+      started.off("close", onClose);
     });
 
+    started.stdout?.on("data", react);
+    started.stderr?.on("data", react);
+    started.on("error", onError);
+    started.on("close", onClose);
+
     const write = (bytes: string) => {
       // Everything already on screen belongs to the surface being answered, so
       // the next wait reads only what this write provoked.