Teach CLI and local server shared connections

apps/cli/package.json

@@ -18,6 +18,7 @@
     "build:publish": "bun run src/build.ts publish",
     "release:publish:dry-run": "bun run src/release.ts --dry-run",
     "release:publish": "bun run src/release.ts",
+    "test": "bun --bun vitest run",
     "typecheck": "tsgo --noEmit",
     "typecheck:slow": "tsc --noEmit"
   },
@@ -27,12 +28,15 @@
     "@executor-js/integrations-registry": "workspace:*",
     "@executor-js/local": "workspace:*",
     "@executor-js/runtime-quickjs": "workspace:*",
+    "@executor-js/sdk": "workspace:*",
     "@jitl/quickjs-wasmfile-release-sync": "catalog:",
     "effect": "catalog:",
     "quickjs-emscripten": "catalog:"
   },
   "devDependencies": {
+    "@effect/vitest": "catalog:",
     "bun-types": "catalog:",
-    "typescript": "catalog:"
+    "typescript": "catalog:",
+    "vitest": "catalog:"
   }
 }

apps/cli/src/local-server-manifest.test.ts

@@ -0,0 +1,100 @@
+import { afterEach, describe, expect, it } from "@effect/vitest";
+import { BunServices } from "@effect/platform-bun";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { Path } from "effect";
+import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+
+import { normalizeExecutorServerConnection } from "@executor-js/sdk/shared";
+import {
+  acquireLocalServerStartLock,
+  readLocalServerManifest,
+  releaseLocalServerStartLock,
+  removeLocalServerManifestIfOwnedBy,
+  resolveExecutorDataDir,
+  writeLocalServerManifest,
+} from "./local-server-manifest";
+
+const previousDataDir = process.env.EXECUTOR_DATA_DIR;
+
+afterEach(() => {
+  if (previousDataDir === undefined) {
+    delete process.env.EXECUTOR_DATA_DIR;
+  } else {
+    process.env.EXECUTOR_DATA_DIR = previousDataDir;
+  }
+});
+
+describe("local server manifest", () => {
+  it.effect("round-trips the active local server owner", () =>
+    Effect.gen(function* () {
+      const dataDir = mkdtempSync(join(tmpdir(), "executor-local-server-"));
+      process.env.EXECUTOR_DATA_DIR = dataDir;
+
+      try {
+        const manifest = {
+          version: 1 as const,
+          kind: "cli-daemon" as const,
+          pid: process.pid,
+          startedAt: "2026-05-28T00:00:00.000Z",
+          dataDir,
+          scopeDir: dataDir,
+          connection: normalizeExecutorServerConnection({
+            origin: "http://localhost:4788",
+          }),
+          owner: {
+            client: "cli" as const,
+            version: "1.2.3",
+            executablePath: "/usr/local/bin/executor",
+          },
+        };
+
+        yield* writeLocalServerManifest(manifest);
+        expect((yield* readLocalServerManifest())?.connection.origin).toBe("http://localhost:4788");
+
+        yield* removeLocalServerManifestIfOwnedBy({ pid: process.pid + 1 });
+        expect(yield* readLocalServerManifest()).not.toBeNull();
+
+        yield* removeLocalServerManifestIfOwnedBy({ pid: process.pid });
+        expect(yield* readLocalServerManifest()).toBeNull();
+      } finally {
+        rmSync(dataDir, { recursive: true, force: true });
+      }
+    }).pipe(Effect.provide(BunServices.layer)),
+  );
+
+  it.effect("serializes local startup with a stale-aware lock", () =>
+    Effect.gen(function* () {
+      const dataDir = mkdtempSync(join(tmpdir(), "executor-local-server-lock-"));
+      process.env.EXECUTOR_DATA_DIR = dataDir;
+
+      try {
+        const first = yield* acquireLocalServerStartLock();
+        const second = yield* Effect.exit(acquireLocalServerStartLock());
+
+        expect(Exit.isFailure(second)).toBe(true);
+        yield* releaseLocalServerStartLock(first);
+        const third = yield* acquireLocalServerStartLock();
+        yield* releaseLocalServerStartLock(third);
+      } finally {
+        rmSync(dataDir, { recursive: true, force: true });
+      }
+    }).pipe(Effect.provide(BunServices.layer)),
+  );
+
+  it.effect("resolves the data dir from EXECUTOR_DATA_DIR", () =>
+    Effect.gen(function* () {
+      const dataDir = mkdtempSync(join(tmpdir(), "executor-local-server-dir-"));
+      process.env.EXECUTOR_DATA_DIR = dataDir;
+
+      try {
+        const path = yield* Path.Path;
+        expect(resolveExecutorDataDir(path)).toBe(dataDir);
+      } finally {
+        rmSync(dataDir, { recursive: true, force: true });
+      }
+    }).pipe(Effect.provide(BunServices.layer)),
+  );
+});

apps/cli/src/local-server-manifest.ts

@@ -0,0 +1,130 @@
+import { homedir } from "node:os";
+import { resolve } from "node:path";
+import { FileSystem, Option, Path, Schema } from "effect";
+import type { PlatformError } from "effect/PlatformError";
+import * as Effect from "effect/Effect";
+
+import {
+  parseExecutorLocalServerManifest,
+  serializeExecutorLocalServerManifest,
+  type ExecutorLocalServerManifest,
+} from "@executor-js/sdk/shared";
+import { isPidAlive } from "./daemon-state";
+
+export interface LocalServerStartLock {
+  readonly path: string;
+}
+
+export const resolveExecutorDataDir = (path: Path.Path): string =>
+  resolve(process.env.EXECUTOR_DATA_DIR ?? path.join(homedir(), ".executor"));
+
+const serverControlDir = (path: Path.Path): string =>
+  path.join(resolveExecutorDataDir(path), "server-control");
+
+const localServerManifestPath = (path: Path.Path): string =>
+  path.join(serverControlDir(path), "server.json");
+
+const localServerStartLockPath = (path: Path.Path): string =>
+  path.join(serverControlDir(path), "startup.lock");
+
+export const readLocalServerManifest = (): Effect.Effect<
+  ExecutorLocalServerManifest | null,
+  never,
+  FileSystem.FileSystem | Path.Path
+> =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const raw = yield* fs
+      .readFileString(localServerManifestPath(path))
+      .pipe(Effect.catchCause(() => Effect.succeed(null)));
+    if (raw === null) return null;
+    return parseExecutorLocalServerManifest(raw);
+  });
+
+export const writeLocalServerManifest = (
+  manifest: ExecutorLocalServerManifest,
+): Effect.Effect<void, PlatformError, FileSystem.FileSystem | Path.Path> =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    yield* fs.makeDirectory(serverControlDir(path), { recursive: true });
+    yield* fs.writeFileString(
+      localServerManifestPath(path),
+      serializeExecutorLocalServerManifest(manifest),
+    );
+  });
+
+export const removeLocalServerManifestIfOwnedBy = (input: {
+  readonly pid: number;
+}): Effect.Effect<void, PlatformError, FileSystem.FileSystem | Path.Path> =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const manifestPath = localServerManifestPath(path);
+    const raw = yield* fs
+      .readFileString(manifestPath)
+      .pipe(Effect.catchCause(() => Effect.succeed(null)));
+    if (raw === null) return;
+    const manifest = parseExecutorLocalServerManifest(raw);
+    if (manifest?.pid !== input.pid) return;
+    yield* fs.remove(manifestPath, { force: true });
+  });
+
+const StartupLockPayload = Schema.Struct({
+  pid: Schema.Number,
+});
+
+const decodeStartupLockPayload = Schema.decodeUnknownOption(
+  Schema.fromJsonString(StartupLockPayload),
+);
+
+const parseLockPid = (raw: string): number | null => {
+  const decoded = decodeStartupLockPayload(raw);
+  return Option.isSome(decoded) ? decoded.value.pid : null;
+};
+
+export const acquireLocalServerStartLock = (): Effect.Effect<
+  LocalServerStartLock,
+  Error,
+  FileSystem.FileSystem | Path.Path
+> =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    yield* fs.makeDirectory(serverControlDir(path), { recursive: true });
+
+    const lockPath = localServerStartLockPath(path);
+    const lockPayload = `${JSON.stringify({ pid: process.pid, startedAt: new Date().toISOString() }, null, 2)}\n`;
+
+    const tryAcquire = () =>
+      fs.writeFileString(lockPath, lockPayload, { flag: "wx" }).pipe(
+        Effect.as(true),
+        Effect.catchCause(() => Effect.succeed(false)),
+      );
+
+    if (yield* tryAcquire()) return { path: lockPath };
+
+    const existingRaw = yield* fs
+      .readFileString(lockPath)
+      .pipe(Effect.catchCause(() => Effect.succeed(null)));
+    if (existingRaw !== null) {
+      const existingPid = parseLockPid(existingRaw);
+      if (existingPid !== null && !isPidAlive(existingPid)) {
+        yield* fs.remove(lockPath, { force: true });
+        if (yield* tryAcquire()) return { path: lockPath };
+      }
+    }
+
+    return yield* Effect.fail(
+      new Error("Another local Executor server startup is already in progress."),
+    );
+  });
+
+export const releaseLocalServerStartLock = (
+  lock: LocalServerStartLock,
+): Effect.Effect<void, PlatformError, FileSystem.FileSystem | Path.Path> =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    yield* fs.remove(lock.path, { force: true });
+  });