feat(direct-control): caller-owned shared session seam + graceful FIN close + timeout stats

packages/civ7-direct-control/src/index.ts

@@ -382,6 +382,7 @@ export type {
   Civ7DirectControlEndpoint,
   Civ7DirectControlHealth,
   Civ7DirectControlOptions,
+  Civ7DirectControlSessionStats,
   Civ7TunerState,
   Civ7TunerStateRole,
   Civ7TunerStateSelection,

packages/civ7-direct-control/src/session/session.ts

@@ -17,10 +17,14 @@ import type {
   Civ7CommandResult,
   Civ7DirectControlEndpoint,
   Civ7DirectControlOptions,
+  Civ7DirectControlSessionStats,
   Civ7TunerState,
   Civ7TunerStateSelection,
 } from "./types.js";
 
+/** How long a graceful FIN gets before `close()` falls back to `destroy()`. */
+const GRACEFUL_CLOSE_TIMEOUT_MS = 1_000;
+
 type PendingCiv7TunerRequest = {
   resolve: (frame: Civ7TunerFrame) => void;
   reject: (err: Error) => void;
@@ -36,6 +40,7 @@ export class Civ7DirectControlSession {
   private endpointValue: Civ7DirectControlEndpoint | undefined;
   private buffer = Buffer.alloc(0);
   private readonly pending = new Map<number, PendingCiv7TunerRequest>();
+  private consecutiveResponseTimeouts = 0;
 
   constructor(options: Civ7DirectControlOptions = {}) {
     this.config = resolveCiv7DirectControlConfig(options);
@@ -45,6 +50,16 @@ export class Civ7DirectControlSession {
     return this.endpointValue;
   }
 
+  /**
+   * Health counters observed on this socket — the one vantage point that
+   * sees ALL traffic on a shared session (every consumer's requests). A
+   * sustained run of response-timeouts is the wedged/busy-tuner signature
+   * the studio's backoff gate keys on.
+   */
+  get stats(): Civ7DirectControlSessionStats {
+    return { consecutiveResponseTimeouts: this.consecutiveResponseTimeouts };
+  }
+
   async connect(): Promise<Civ7DirectControlEndpoint> {
     if (this.socket && !this.socket.destroyed && this.endpointValue) {
       return this.endpointValue;
@@ -84,13 +99,30 @@ export class Civ7DirectControlSession {
     );
   }
 
+  /**
+   * Graceful close: FIN first (`socket.end()`), so the game can release its
+   * descriptor cleanly — abrupt `destroy()` teardown is the suspected driver
+   * of the game-side fd leak that wedges the tuner after long sessions. The
+   * destroy fallback only fires if the peer never completes the handshake.
+   */
   async close(): Promise<void> {
     const socket = this.socket;
     this.socket = undefined;
     this.endpointValue = undefined;
     this.buffer = Buffer.alloc(0);
     this.rejectPending(new Civ7DirectControlError("socket-closed", "Civ7 tuner socket closed"));
-    if (socket && !socket.destroyed) socket.destroy();
+    if (!socket || socket.destroyed) return;
+    await new Promise<void>((resolve) => {
+      const timer = setTimeout(() => socket.destroy(), GRACEFUL_CLOSE_TIMEOUT_MS);
+      socket.once("close", () => {
+        clearTimeout(timer);
+        resolve();
+      });
+      // Post-end errors (peer reset during the handshake) must not surface as
+      // unhandled; "close" always follows.
+      socket.on("error", () => {});
+      socket.end();
+    });
   }
 
   async queryStates(options: { timeoutMs?: number } = {}): Promise<ReadonlyArray<Civ7TunerState>> {
@@ -132,6 +164,7 @@ export class Civ7DirectControlSession {
     const response = new Promise<Civ7TunerFrame>((resolve, reject) => {
       const timer = setTimeout(() => {
         this.pending.delete(listenerId);
+        this.consecutiveResponseTimeouts += 1;
         reject(
           new Civ7DirectControlError(
             "response-timeout",
@@ -155,6 +188,7 @@ export class Civ7DirectControlSession {
       if (!pending) continue;
       clearTimeout(pending.timer);
       this.pending.delete(parsed.frame.listenerId);
+      this.consecutiveResponseTimeouts = 0;
       pending.resolve(parsed.frame);
     }
   }
@@ -180,6 +214,11 @@ export async function withCiv7DirectControlSession<T>(
   options: Civ7DirectControlOptions,
   run: (session: Civ7DirectControlSession) => Promise<T>,
 ): Promise<T> {
+  // Caller-owned shared session: reuse, never close — the owner (e.g. the
+  // studio daemon's Effect-scoped service) manages acquisition/release.
+  if (options.session) {
+    return await run(options.session);
+  }
   const session = new Civ7DirectControlSession(options);
   try {
     return await run(session);

packages/civ7-direct-control/src/session/types.ts

@@ -1,4 +1,5 @@
 import type { Civ7DirectControlError } from "../direct-control-error.js";
+import type { Civ7DirectControlSession } from "./session.js";
 
 export type Civ7TunerState = Readonly<{
   id: string;
@@ -26,6 +27,20 @@ export type Civ7DirectControlOptions = Readonly<{
   port?: number;
   timeoutMs?: number;
   env?: NodeJS.ProcessEnv;
+  /**
+   * Caller-owned shared session. When present, procedures reuse it (the
+   * connection is multiplexed by listenerId, so concurrent calls are fine)
+   * and NEVER close it — lifecycle belongs to the owner (e.g. the studio
+   * daemon's Effect-scoped `Civ7TunerSession`). When absent, behavior is
+   * unchanged: a fresh session per call, closed in `finally`.
+   */
+  session?: Civ7DirectControlSession;
+}>;
+
+/** Read-only health counters observed on a session's socket (all traffic). */
+export type Civ7DirectControlSessionStats = Readonly<{
+  /** Response-timeouts since the last successfully resolved frame. */
+  consecutiveResponseTimeouts: number;
 }>;
 
 export type Civ7CommandResult = Readonly<{

packages/civ7-direct-control/test/shared-session.test.ts

@@ -0,0 +1,200 @@
+import { createServer, type Socket } from "node:net";
+import { afterEach, describe, expect, test } from "vitest";
+
+import { Civ7DirectControlSession, executeCiv7Command } from "../src/index";
+
+// Pins for the shared-session seam (mapgen-studio-tuner-session workstream):
+// a caller-owned session is reused across procedure calls and never closed by
+// the callee; `close()` is a graceful FIN (the abrupt-destroy teardown is the
+// suspected driver of the game-side fd leak); `stats` tracks consecutive
+// response-timeouts and resets on success.
+
+type SharedSessionServer = Readonly<{
+  port: number;
+  connections: () => number;
+  finReceived: () => boolean;
+  close: () => Promise<void>;
+}>;
+
+const openServers: Array<() => Promise<void>> = [];
+
+afterEach(async () => {
+  await Promise.all(openServers.splice(0).map((close) => close()));
+});
+
+describe("caller-owned shared tuner session", () => {
+  test("procedures reuse the injected session over one connection and do not close it", async () => {
+    const server = await startSharedSessionServer();
+    const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port });
+    try {
+      const first = await executeCiv7Command({
+        port: server.port,
+        session,
+        command: "first",
+        timeoutMs: 1_000,
+      });
+      const second = await executeCiv7Command({
+        port: server.port,
+        session,
+        command: "second",
+        timeoutMs: 1_000,
+      });
+
+      expect(first.output).toEqual(["null"]);
+      expect(second.output).toEqual(["null"]);
+      expect(server.connections()).toBe(1);
+      // The wrapper must NOT have closed the caller-owned session.
+      expect(session.endpoint).toEqual({ host: "127.0.0.1", port: server.port });
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("without an injected session, each call opens and closes its own connection", async () => {
+    const server = await startSharedSessionServer();
+    await executeCiv7Command({
+      host: "127.0.0.1",
+      port: server.port,
+      command: "first",
+      timeoutMs: 1_000,
+    });
+    await executeCiv7Command({
+      host: "127.0.0.1",
+      port: server.port,
+      command: "second",
+      timeoutMs: 1_000,
+    });
+    expect(server.connections()).toBe(2);
+  });
+
+  test("close() delivers a FIN handshake, not an abrupt teardown", async () => {
+    const server = await startSharedSessionServer();
+    const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port });
+    await session.queryStates({ timeoutMs: 1_000 });
+
+    await session.close();
+    // Give the server's event loop a beat to observe the FIN.
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(server.finReceived()).toBe(true);
+  });
+
+  test("stats: consecutive response-timeouts accumulate and reset on success", async () => {
+    const server = await startSharedSessionServer({ silentCommands: ["slow"] });
+    const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port });
+    try {
+      expect(session.stats.consecutiveResponseTimeouts).toBe(0);
+
+      await expect(
+        executeCiv7Command({ port: server.port, session, command: "slow", timeoutMs: 50 }),
+      ).rejects.toMatchObject({ code: "response-timeout" });
+      await expect(
+        executeCiv7Command({ port: server.port, session, command: "slow", timeoutMs: 50 }),
+      ).rejects.toMatchObject({ code: "response-timeout" });
+
+      // Each failed call performs LSQ (succeeds → reset) then CMD (times out),
+      // so the counter reflects the trailing run of timeouts on this socket.
+      expect(session.stats.consecutiveResponseTimeouts).toBe(1);
+
+      const ok = await executeCiv7Command({
+        port: server.port,
+        session,
+        command: "fast",
+        timeoutMs: 1_000,
+      });
+      expect(ok.output).toEqual(["null"]);
+      expect(session.stats.consecutiveResponseTimeouts).toBe(0);
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("stats: an unresponsive tuner accumulates across consecutive requests", async () => {
+    const server = await startSharedSessionServer({ silent: true });
+    const session = new Civ7DirectControlSession({ host: "127.0.0.1", port: server.port });
+    try {
+      await expect(session.queryStates({ timeoutMs: 50 })).rejects.toMatchObject({
+        code: "response-timeout",
+      });
+      await expect(session.queryStates({ timeoutMs: 50 })).rejects.toMatchObject({
+        code: "response-timeout",
+      });
+      await expect(session.queryStates({ timeoutMs: 50 })).rejects.toMatchObject({
+        code: "response-timeout",
+      });
+      expect(session.stats.consecutiveResponseTimeouts).toBe(3);
+    } finally {
+      await session.close();
+    }
+  });
+});
+
+async function startSharedSessionServer(
+  options: Readonly<{ silent?: boolean; silentCommands?: readonly string[] }> = {},
+): Promise<SharedSessionServer> {
+  let connections = 0;
+  let finReceived = false;
+  const sockets = new Set<Socket>();
+
+  const server = createServer((socket) => {
+    connections += 1;
+    sockets.add(socket);
+    socket.on("close", () => sockets.delete(socket));
+    socket.on("end", () => {
+      finReceived = true;
+    });
+    socket.on("error", () => {});
+    let buffer = Buffer.alloc(0);
+    socket.on("data", (chunk) => {
+      buffer = Buffer.concat([buffer, chunk]);
+      for (;;) {
+        if (buffer.length < 8) return;
+        const messageLength = buffer.readUInt32LE(0);
+        const bytesRead = 8 + messageLength;
+        if (buffer.length < bytesRead) return;
+        const listenerId = buffer.readUInt32LE(4);
+        const message = buffer.subarray(8, bytesRead).toString("utf8").replace(/\0$/, "");
+        buffer = buffer.subarray(bytesRead);
+
+        if (options.silent) continue;
+        const isSilent = options.silentCommands?.some((command) =>
+          message.includes(`:${command}`),
+        );
+        if (isSilent) continue;
+        if (message === "LSQ:") {
+          socket.write(encodeResponse(listenerId, ["65535", "App UI", "1", "Tuner"]));
+        } else {
+          socket.write(encodeResponse(listenerId, ["null"]));
+        }
+      }
+    });
+  });
+
+  await new Promise<void>((resolve, reject) => {
+    server.listen(0, "127.0.0.1", () => resolve());
+    server.on("error", reject);
+  });
+
+  const close = () =>
+    new Promise<void>((resolve) => {
+      for (const socket of sockets) socket.destroy();
+      server.close(() => resolve());
+    });
+  openServers.push(close);
+
+  return {
+    port: (server.address() as { port: number }).port,
+    connections: () => connections,
+    finReceived: () => finReceived,
+    close,
+  };
+}
+
+function encodeResponse(listenerId: number, parts: readonly string[]): Buffer {
+  const message = Buffer.from(`${parts.join("\0")}\0`, "utf8");
+  const frame = Buffer.alloc(8 + message.length);
+  frame.writeUInt32LE(message.length, 0);
+  frame.writeUInt32LE(listenerId, 4);
+  message.copy(frame, 8);
+  return frame;
+}