fix(server): serialize teardown against input; escalate only a delivered SIGTERM

CodeRabbit review on PR #157, server-side findings.

Serialize input with teardown (#10). `closeSession` left `closed === false` with
the live adapter installed while awaiting `adapter.close()` and `kill()`, so a
concurrent `send.message` could pass `ensureLiveForInput` and prompt an agent
mid-reap — the turn lost or errored. Teardown is now published as a per-session
promise: concurrent closes collapse onto one, and `ensureLiveForInput` awaits it
before reopening.

Drop queued input on close (#7). An adapter's `close()` cancels its turn and
reports `idle`, which is exactly the signal `flushNext()` waits for — so a queued
mid-turn message was prompted into the process about to be reaped. Closing now
clears the queue first, matching `cancel`'s "stop means stop" (SPEC-35).

Escalate only a delivered SIGTERM (#6). `ChildProcess.kill()` returns false when
the signal was not delivered (the child is already gone); scheduling the SIGKILL
anyway could fire it at a pid the OS had since recycled, hitting an unrelated
process. The test double now reports delivery like Node does, and covers the
false path.

Correct the close() contract in subprocess-adapter's doc (#8): it said "must
never throw", contradicting `AgentAdapter.close`, which states the opposite now
that the manager owns bounding and swallowing.

Cover the reaper's untested paths (#9): the default sweep interval and its 30s
floor (which stops a short window busy-looping), and the overlapping-sweep guard.

Both races are red-then-green: each new test reproduces the interleaving before
the fix. 1304 server tests pass.

Author: leduckhc

server/src/adapters/child_transport.test.ts

@@ -33,9 +33,14 @@ function fakeSpawn(): {
     child.stdout = new EventEmitter();
     child.stderr = new EventEmitter();
     child.signals = [] as string[];
+    // Mirrors Node: true when the signal was delivered, false when it was not
+    // (already-exited child). `killDelivers` lets a test take the false path.
+    child.killDelivers = true;
     child.kill = (signal?: string) => {
-      child.killed = true;
       child.signals.push(signal ?? "SIGTERM");
+      if (!child.killDelivers) return false;
+      child.killed = true;
+      return true;
     };
     children.push(child);
     return child as unknown as ChildProcess;
@@ -345,3 +350,20 @@ test("pid is undefined when the spawn faults (no child pid)", () => {
   const t = spawnLineProcess({ command: "x", cwd: "/tmp", label: "t", spawn });
   assert.equal(t.pid, undefined);
 });
+
+/**
+ * `kill()` returning false means the signal was not delivered — the child is
+ * already gone. Escalating anyway would fire a SIGKILL at a pid the OS may have
+ * recycled, hitting an unrelated process.
+ */
+test("dispose does not escalate when the SIGTERM was not delivered", async () => {
+  const { spawn, children } = fakeSpawn();
+  const t = spawnLineProcess({ command: "x", cwd: "/tmp", label: "t", spawn, killGraceMs: 5 });
+  children[0]!.killDelivers = false;
+
+  t.dispose();
+  assert.deepEqual(children[0]!.signals, ["SIGTERM"], "the attempt is still made");
+
+  await new Promise((r) => setTimeout(r, 25));
+  assert.deepEqual(children[0]!.signals, ["SIGTERM"], "but no SIGKILL follows an undelivered signal");
+});

server/src/adapters/child_transport.ts

@@ -287,11 +287,16 @@ export function spawnLineProcess(opts: SpawnLineOptions): ChildLineTransport {
       // Already reaped by the OS — nothing to signal, and SIGKILLing a pid that
       // has been recycled would hit an unrelated process.
       if (settled) return;
+      // `kill()` returns false when the signal was NOT delivered (the child is
+      // already gone). Escalating on that would schedule a SIGKILL against a pid
+      // the OS may have recycled by then — a signal to an unrelated process.
+      let termed = false;
       try {
-        child.kill("SIGTERM");
+        termed = child.kill("SIGTERM");
       } catch {
         /* ignore */
       }
+      if (!termed) return;
       // Escalate once. An agent that ignores SIGTERM (or is wedged mid-turn)
       // would otherwise stay resident indefinitely, holding its whole RSS.
       if (killTimer) return;

server/src/adapters/subprocess-adapter.ts

@@ -56,7 +56,9 @@ export abstract class SubprocessAdapter extends EventEmitter implements AgentAda
    * Graceful agent-side session release before the reap. Defaults to a no-op so
    * a transport without the primitive degrades to "just kill it"; subclasses
    * with a real close (ACP `session/close`, codex `thread/unsubscribe`)
-   * override. Must never throw — see {@link AgentAdapter.close}.
+   * override. An override implements the plain request only: it MAY reject and
+   * MAY hang, because `SessionManager.closeSession` bounds it and reaps
+   * regardless — see {@link AgentAdapter.close}.
    */
   async close(): Promise<void> {}
   abstract kill(signal?: NodeJS.Signals): Promise<void>;

server/src/idle_reaper.test.ts

@@ -173,3 +173,56 @@ test("resolveIdleCloseMs reads minutes, defaults, and honours 0", () => {
   assert.equal(resolveIdleCloseMs({ MAKIT_IDLE_CLOSE_MIN: "soon" } as NodeJS.ProcessEnv), DEFAULT_IDLE_CLOSE_MS);
   assert.equal(resolveIdleCloseMs({ MAKIT_IDLE_CLOSE_MIN: "-3" } as NodeJS.ProcessEnv), DEFAULT_IDLE_CLOSE_MS);
 });
+
+/**
+ * The interval derivation and its floor were untested (review). The floor is the
+ * load-bearing half: a short window must not turn the sweep into a busy loop.
+ */
+test("the default sweep interval is a quarter of the window, floored at 30s", () => {
+  const armedFor = (idleCloseMs: number) => {
+    let armed: number | undefined;
+    new IdleReaper({
+      sessions: () => [],
+      close: async () => {},
+      idleCloseMs,
+      setTimer: (_fn, ms) => {
+        armed = ms;
+        return "h";
+      },
+    }).start();
+    return armed;
+  };
+  assert.equal(armedFor(60 * 60_000), 15 * 60_000, "an hour window sweeps every 15min");
+  assert.equal(armedFor(60_000), 30_000, "a 1min window is floored, not swept every 15s");
+});
+
+/**
+ * A sweep awaits an agent per close, so a slow one can still be running when the
+ * next tick fires. Without the re-entrancy guard the same session closes twice.
+ */
+test("a sweep already in flight suppresses the next one", async () => {
+  let releaseFirst = () => {};
+  const gate = new Promise<void>((r) => {
+    releaseFirst = r;
+  });
+  const s = session({ lastActivityAt: 0 });
+  const closed: string[] = [];
+  const r = new IdleReaper({
+    sessions: () => [s],
+    close: async (id) => {
+      closed.push(id);
+      await gate;
+      (s as { closed: boolean }).closed = true;
+    },
+    idleCloseMs: 60_000,
+    now: () => 10 * 60_000,
+  });
+
+  const first = r.sweep();
+  await new Promise((res) => setTimeout(res, 5));
+  assert.deepEqual(await r.sweep(), [], "the overlapping sweep is a no-op");
+
+  releaseFirst();
+  assert.deepEqual(await first, [s.id]);
+  assert.deepEqual(closed, [s.id], "closed exactly once");
+});

server/src/manager.test.ts

@@ -2510,3 +2510,99 @@ test("closeSession reaps even when the agent's close() never settles", async ()
     store.close();
   }
 });
+
+/**
+ * Teardown must be serialized against input (review: "Serialize input with
+ * session teardown"). While `closeSession` awaits `adapter.close()` and `kill()`,
+ * the session still reads `closed === false` with the live adapter installed, so
+ * a racing `send.message` would pass `ensureLiveForInput` and prompt an agent
+ * that is being reaped — the turn is lost or errors.
+ */
+test("a message racing a close waits for teardown, then reopens the session", async () => {
+  const store = new SqliteEventStore();
+  const cwd = mkdtempSync(join(tmpdir(), "makit-close-race-"));
+  try {
+    const order: string[] = [];
+    let releaseClose = () => {};
+    const gate = new Promise<void>((r) => {
+      releaseClose = r;
+    });
+    const mgr = new SessionManager({
+      projects: [cwd],
+      store,
+      adapterFactory: () => {
+        const a: any = stubAdapter([]);
+        a.close = async () => {
+          order.push("close:start");
+          await gate;
+          order.push("close:end");
+        };
+        return a;
+      },
+    });
+    const projectId = mgr.listProjects()[0].id;
+    const s = await mgr.spawnPiSession(projectId, "racer", "pi");
+
+    const closing = mgr.closeSession(s.id); // not awaited: teardown in flight
+    await new Promise((r) => setTimeout(r, 10));
+    assert.deepEqual(order, ["close:start"], "teardown is mid-flight");
+
+    const input = (async () => {
+      await mgr.ensureLiveForInput(s.id);
+      order.push("input:live");
+    })();
+
+    releaseClose();
+    await Promise.all([closing, input]);
+
+    assert.deepEqual(
+      order,
+      ["close:start", "close:end", "input:live"],
+      "input must not be serviced until teardown finished",
+    );
+    assert.equal(mgr.getSession(s.id)!.closed, false, "the racing message reopened it");
+  } finally {
+    rmSync(cwd, { recursive: true, force: true });
+    store.close();
+  }
+});
+
+/**
+ * Closing drops queued mid-turn messages, mirroring `cancel`'s "stop means stop"
+ * (SPEC-35). Without this, the `idle` an adapter's `close()` emits triggers
+ * `flushNext()` and a queued message is sent into the agent being reaped.
+ */
+test("closeSession drops queued mid-turn messages instead of flushing them into a dying agent", async () => {
+  const store = new SqliteEventStore();
+  const cwd = mkdtempSync(join(tmpdir(), "makit-close-queue-"));
+  try {
+    const sends: string[] = [];
+    const mgr = new SessionManager({
+      projects: [cwd],
+      store,
+      adapterFactory: () => {
+        const a: any = stubAdapter([]);
+        a.send = async (input: { text: string }) => {
+          sends.push(input.text);
+        };
+        a.steer = async () => false; // unsteerable → mid-turn input queues
+        a.close = async () => a.emit("status", "idle"); // as a real cancel does
+        return a;
+      },
+    });
+    const projectId = mgr.listProjects()[0].id;
+    const s = await mgr.spawnPiSession(projectId, "queued", "pi");
+    await s.sendUserMessage("first");
+    s.adapter.emit("status", "running");
+    await s.sendUserMessage("queued-while-busy");
+    assert.equal(s.queuedMessages.length, 1, "precondition: one message is queued");
+
+    await mgr.closeSession(s.id);
+
+    assert.equal(s.queuedMessages.length, 0, "the queue is dropped, not flushed");
+    assert.deepEqual(sends, ["first"], "no queued message reached the closing agent");
+  } finally {
+    rmSync(cwd, { recursive: true, force: true });
+    store.close();
+  }
+});

server/src/manager.ts

@@ -184,6 +184,12 @@ export class SessionManager extends EventEmitter {
   private readonly sessions = new Map<string, Session>();
   /** pi session uuid → live makit session id, so re-attach reuses the process. */
   private readonly attachedByPi = new Map<string, string>();
+  /**
+   * In-flight closes, keyed by session id. Teardown awaits the agent, during
+   * which the session still reads `closed === false` with its live adapter
+   * installed — so input paths MUST await this rather than racing it.
+   */
+  private readonly closeInFlight = new Map<string, Promise<void>>();
   /** In-flight attaches, so concurrent attach calls collapse onto one process. */
   private readonly attachInFlight = new Map<string, Promise<Session>>();
   /** In-flight draft promotions, keyed by session id, so concurrent first
@@ -1204,6 +1210,27 @@ export class SessionManager extends EventEmitter {
     const session = this.sessions.get(id);
     if (!session) throw new Error(`no such session: ${id}`);
     if (session.closed) return;
+    // Collapse concurrent closes (a user tap racing an idle sweep) onto one
+    // teardown, and publish it so input paths can wait instead of prompting an
+    // agent that is being reaped.
+    const already = this.closeInFlight.get(id);
+    if (already) return already;
+    const run = this.runClose(id, session);
+    this.closeInFlight.set(id, run);
+    try {
+      await run;
+    } finally {
+      this.closeInFlight.delete(id);
+    }
+  }
+
+  /** The teardown itself; {@link closeSession} owns de-duplication. */
+  private async runClose(id: string, session: Session): Promise<void> {
+    // Stop means stop (SPEC-35), as with `cancel`: drop pending mid-turn input
+    // before the agent is asked to close. Otherwise the `idle` an adapter emits
+    // while cancelling its turn triggers `flushNext()` and a queued message is
+    // prompted into the process we are about to reap.
+    session.clearQueue();
     // Best-effort, in order: closing is also the recovery path in the
     // removeWorktree loop (called after git already deleted the tree), so
     // neither a rejecting close() nor a rejecting kill() may abort the close —
@@ -1263,6 +1290,11 @@ export class SessionManager extends EventEmitter {
    * sessions for exactly that reason.
    */
   async ensureLiveForInput(sessionId: string): Promise<void> {
+    // A close in flight still reads `closed === false` with the live adapter
+    // installed, so acting now would prompt an agent mid-reap. Let teardown
+    // finish, then bring the session back deliberately.
+    const closing = this.closeInFlight.get(sessionId);
+    if (closing) await closing.catch(() => {});
     const session = this.sessions.get(sessionId);
     if (session?.closed) {
       await this.reopenSession(sessionId);