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

An event listener an Effection operation installs is state that operation
owns, but nothing about an event source enforces that. A handler attached to a
socket, a child process or a DOM target survived the operation that attached
it, and kept writing to state the abandoned owner left behind. Cleanup that
runs when the event fires is not cleanup: a cancelled wait is exactly the case
where the event never arrives.

`@effectionx/node` moves to 0.2.5, whose `once()` registers nothing until it is
interpreted and detaches in a `finally` — after delivery, a halt, or losing a
race. Every direct pin moves together; `@effectionx/process` stays at 0.8.1,
which leaves one audited transitive 0.2.4 lock entry that nothing here
resolves.

`local/require-scope-bound-event-registration` makes the policy a blocking
gate. It resolves sources, owners, `action` and `ensure` through their imports
rather than their spelling, and an owner is the nearest generator or `action()`
executor and nothing wider. Cleanup counts as established only when it is in
place before the thing it releases exists: a lexical `finally` entered before
the subscription, an `ensure()` that *completed* before it, or the cleanup an
`action()` returns. An `ensure()` yielded afterwards is rejected, because
entering it is itself a suspension — an owner halted there unwinds with no
cleanup registered at all, measured, with the listener still attached.

Every applicable registration is migrated onto one of those shapes. The three
owners that own a child process register their cleanup before spawning it, and
that cleanup keeps every handler attached until the child's `close` — the only
event that says the process and the pipes this run inherited are finished. An
assigned exit status is not that boundary, so none of them reads one: a
cancelled command still captures what its child wrote on the way out, and still
escalates SIGTERM to SIGKILL exactly as before.

Readiness races are interpreted inline, in the same synchronous run as the
`connect`, `listen` or `spawn` they watch, because a spawned race attaches its
arms a turn later and the source can settle in that turn. The foreground reaper
names its exit handler and releases it in the one funnel every settlement
passes through. `nodeResponseChannel` becomes an `Operation`, so both of its
handlers belong to the request task. The form responder keeps its request
observed until that request's own `close`, so the peer's hang-up reaches a
listener instead of the process. Web server teardown detaches every accepted
socket's close handler before it suspends, then ends request tasks before
destroying their connections. The workflow and CLI HTTP fixtures read request
bodies in tasks of the server's own scope.

Every distinct manual owner is cancelled while its listeners are live. Each
case measures the baseline the runtime was already holding, reads the vector
once with the owner running, halts it before the event that would have settled
it, reads the vector again before replaying anything, then replays and proves
no capture, callback or result moved. Where a source is a child process or
accepted socket an operation does not hand out, a package-private observation
seam carries it; no public package contract widened.

This is a lifecycle change. Event ordering, output aggregation, response bytes,
failure precedence, process signalling and every public result are unchanged.

Author: taras

.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",

deno.lock

@@ -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",

package.json

@@ -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/adapter-protocol.test.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.