refactor(infra): share jsonl socket requester

src/infra/exec-approvals.ts

@@ -1,9 +1,9 @@
 import crypto from "node:crypto";
 import fs from "node:fs";
-import net from "node:net";
 import os from "node:os";
 import path from "node:path";
 import { DEFAULT_AGENT_ID } from "../routing/session-key.js";
+import { requestJsonlSocket } from "./jsonl-socket.js";
 export * from "./exec-approvals-analysis.js";
 export * from "./exec-approvals-allowlist.js";
 
@@ -500,56 +500,23 @@ export async function requestExecApprovalViaSocket(params: {
     return null;
   }
   const timeoutMs = params.timeoutMs ?? 15_000;
-  return await new Promise((resolve) => {
+  const payload = JSON.stringify({
-    const client = new net.Socket();
+    type: "request",
-    let settled = false;
+    token,
-    let buffer = "";
+    id: crypto.randomUUID(),
-    const finish = (value: ExecApprovalDecision | null) => {
+    request,
-      if (settled) {
+  });
-        return;
-      }
-      settled = true;
-      try {
-        client.destroy();
-      } catch {
-        // ignore
-      }
-      resolve(value);
-    };
 
-    const timer = setTimeout(() => finish(null), timeoutMs);
+  return await requestJsonlSocket({
-    const payload = JSON.stringify({
+    socketPath,
-      type: "request",
+    payload,
-      token,
+    timeoutMs,
-      id: crypto.randomUUID(),
+    accept: (value) => {
-      request,
+      const msg = value as { type?: string; decision?: ExecApprovalDecision };
-    });
+      if (msg?.type === "decision" && msg.decision) {
-
+        return msg.decision;
-    client.on("error", () => finish(null));
-    client.connect(socketPath, () => {
-      client.write(`${payload}\n`);
-    });
-    client.on("data", (data) => {
-      buffer += data.toString("utf8");
-      let idx = buffer.indexOf("\n");
-      while (idx !== -1) {
-        const line = buffer.slice(0, idx).trim();
-        buffer = buffer.slice(idx + 1);
-        idx = buffer.indexOf("\n");
-        if (!line) {
-          continue;
-        }
-        try {
-          const msg = JSON.parse(line) as { type?: string; decision?: ExecApprovalDecision };
-          if (msg?.type === "decision" && msg.decision) {
-            clearTimeout(timer);
-            finish(msg.decision);
-            return;
-          }
-        } catch {
-          // ignore
-        }
       }
-    });
+      return undefined;
+    },
   });
 }

src/infra/exec-host.ts

@@ -1,5 +1,5 @@
 import crypto from "node:crypto";
-import net from "node:net";
+import { requestJsonlSocket } from "./jsonl-socket.js";
 
 export type ExecHostRequest = {
   command: string[];
@@ -43,79 +43,38 @@ export async function requestExecHostViaSocket(params: {
     return null;
   }
   const timeoutMs = params.timeoutMs ?? 20_000;
-  return await new Promise((resolve) => {
+  const requestJson = JSON.stringify(request);
-    const client = new net.Socket();
+  const nonce = crypto.randomBytes(16).toString("hex");
-    let settled = false;
+  const ts = Date.now();
-    let buffer = "";
+  const hmac = crypto
-    const finish = (value: ExecHostResponse | null) => {
+    .createHmac("sha256", token)
-      if (settled) {
+    .update(`${nonce}:${ts}:${requestJson}`)
-        return;
+    .digest("hex");
-      }
+  const payload = JSON.stringify({
-      settled = true;
+    type: "exec",
-      try {
+    id: crypto.randomUUID(),
-        client.destroy();
+    nonce,
-      } catch {
+    ts,
-        // ignore
+    hmac,
-      }
+    requestJson,
-      resolve(value);
+  });
-    };
 
-    const requestJson = JSON.stringify(request);
+  return await requestJsonlSocket({
-    const nonce = crypto.randomBytes(16).toString("hex");
+    socketPath,
-    const ts = Date.now();
+    payload,
-    const hmac = crypto
+    timeoutMs,
-      .createHmac("sha256", token)
+    accept: (value) => {
-      .update(`${nonce}:${ts}:${requestJson}`)
+      const msg = value as { type?: string; ok?: boolean; payload?: unknown; error?: unknown };
-      .digest("hex");
+      if (msg?.type !== "exec-res") {
-    const payload = JSON.stringify({
+        return undefined;
-      type: "exec",
-      id: crypto.randomUUID(),
-      nonce,
-      ts,
-      hmac,
-      requestJson,
-    });
-
-    const timer = setTimeout(() => finish(null), timeoutMs);
-
-    client.on("error", () => finish(null));
-    client.connect(socketPath, () => {
-      client.write(`${payload}\n`);
-    });
-    client.on("data", (data) => {
-      buffer += data.toString("utf8");
-      let idx = buffer.indexOf("\n");
-      while (idx !== -1) {
-        const line = buffer.slice(0, idx).trim();
-        buffer = buffer.slice(idx + 1);
-        idx = buffer.indexOf("\n");
-        if (!line) {
-          continue;
-        }
-        try {
-          const msg = JSON.parse(line) as {
-            type?: string;
-            ok?: boolean;
-            payload?: unknown;
-            error?: unknown;
-          };
-          if (msg?.type === "exec-res") {
-            clearTimeout(timer);
-            if (msg.ok === true && msg.payload) {
-              finish({ ok: true, payload: msg.payload as ExecHostRunResult });
-              return;
-            }
-            if (msg.ok === false && msg.error) {
-              finish({ ok: false, error: msg.error as ExecHostError });
-              return;
-            }
-            finish(null);
-            return;
-          }
-        } catch {
-          // ignore
-        }
       }
-    });
+      if (msg.ok === true && msg.payload) {
+        return { ok: true, payload: msg.payload as ExecHostRunResult };
+      }
+      if (msg.ok === false && msg.error) {
+        return { ok: false, error: msg.error as ExecHostError };
+      }
+      return null;
+    },
   });
 }

src/infra/jsonl-socket.ts

@@ -0,0 +1,59 @@
+import net from "node:net";
+
+export async function requestJsonlSocket<T>(params: {
+  socketPath: string;
+  payload: string;
+  timeoutMs: number;
+  accept: (msg: unknown) => T | null | undefined;
+}): Promise<T | null> {
+  const { socketPath, payload, timeoutMs, accept } = params;
+  return await new Promise((resolve) => {
+    const client = new net.Socket();
+    let settled = false;
+    let buffer = "";
+
+    const finish = (value: T | null) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      try {
+        client.destroy();
+      } catch {
+        // ignore
+      }
+      resolve(value);
+    };
+
+    const timer = setTimeout(() => finish(null), timeoutMs);
+
+    client.on("error", () => finish(null));
+    client.connect(socketPath, () => {
+      client.write(`${payload}\n`);
+    });
+    client.on("data", (data) => {
+      buffer += data.toString("utf8");
+      let idx = buffer.indexOf("\n");
+      while (idx !== -1) {
+        const line = buffer.slice(0, idx).trim();
+        buffer = buffer.slice(idx + 1);
+        idx = buffer.indexOf("\n");
+        if (!line) {
+          continue;
+        }
+        try {
+          const msg = JSON.parse(line) as unknown;
+          const result = accept(msg);
+          if (result === undefined) {
+            continue;
+          }
+          clearTimeout(timer);
+          finish(result);
+          return;
+        } catch {
+          // ignore
+        }
+      }
+    });
+  });
+}