Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
9.5 kB · 332 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333import { chmodSync, existsSync, lstatSync, mkdirSync, unlinkSync } from "node:fs";import { createConnection } from "node:net";import { tmpdir } from "node:os";import { dirname, join } from "node:path";import lockfile from "proper-lockfile";
export { normalizeSocketPath } from "../../utils/daemon-socket-path.js";
const DAEMON_SOCKET_MODE = 0o600;const DAEMON_SOCKET_DIR_MODE = 0o700;const DAEMON_SOCKET_RELEASE_GRACE_MS = 1000;const DAEMON_SOCKET_RELEASE_POLL_MS = 25;const DAEMON_SOCKET_LOCK_STALE_MS = 5000;const DAEMON_SOCKET_LOCK_UPDATE_MS = 1000;
type DaemonSocketCompromiseListener = (error: Error) => void;
export class DaemonSocketPathLease { private released = false; private compromisedError?: Error; private readonly compromiseListeners = new Set<DaemonSocketCompromiseListener>();
constructor( readonly socketPath: string, private readonly releaseLock: () => Promise<void>, ) {}
get compromise(): Error | undefined { return this.compromisedError; }
onCompromised(listener: DaemonSocketCompromiseListener): () => void { if (this.compromisedError) { this.notifyCompromiseListener(listener, this.compromisedError); return () => {}; } this.compromiseListeners.add(listener); return () => this.compromiseListeners.delete(listener); }
recordCompromise(error: Error): void { if (this.compromisedError) return; this.compromisedError = error; const listeners = [...this.compromiseListeners]; this.compromiseListeners.clear(); for (const listener of listeners) this.notifyCompromiseListener(listener, error); }
async release(): Promise<void> { if (this.released) return; this.released = true; this.compromiseListeners.clear(); await this.releaseLock(); }
private notifyCompromiseListener(listener: DaemonSocketCompromiseListener, error: Error): void { try { listener(error); } catch { // A lease callback must not rethrow from proper-lockfile's refresh callback. } }}
export interface DaemonSocketIdentity { dev: number; ino: number;}
export function defaultDaemonSocketPath(): string { if (process.platform === "win32") { return "\\\\.\\pipe\\prime-agent-daemon"; } return join(defaultDaemonSocketDir(), "daemon.sock");}
export async function acquireDaemonSocketPathLease(socketPath: string): Promise<DaemonSocketPathLease | undefined> { ensureDefaultDaemonSocketDir(socketPath); if (process.platform === "win32") { return undefined; } let lease: DaemonSocketPathLease | undefined; let pendingCompromise: Error | undefined; const releaseLock = await lockfile.lock(socketPath, { realpath: false, stale: DAEMON_SOCKET_LOCK_STALE_MS, update: DAEMON_SOCKET_LOCK_UPDATE_MS, onCompromised: (error) => { if (lease) lease.recordCompromise(error); else pendingCompromise = error; }, retries: { retries: 600, factor: 1, minTimeout: DAEMON_SOCKET_RELEASE_POLL_MS, maxTimeout: DAEMON_SOCKET_RELEASE_POLL_MS, }, }); lease = new DaemonSocketPathLease(socketPath, releaseLock); if (pendingCompromise) lease.recordCompromise(pendingCompromise); return lease;}
export async function prepareDaemonSocketPath(socketPath: string, lease?: DaemonSocketPathLease): Promise<void> { ensureDefaultDaemonSocketDir(socketPath);
if (process.platform === "win32") { return; } if (lease) { assertSocketLease(socketPath, lease); assertSocketLeaseHeld(socketPath, lease); await prepareUnixDaemonSocketPath(socketPath, lease); return; } if (!existsSync(socketPath)) { return; } if (await canConnectToUnixSocket(socketPath)) { throw new Error(`Daemon socket already in use: ${socketPath}`); } const ownedLease = await acquireDaemonSocketPathLease(socketPath); try { await prepareUnixDaemonSocketPath(socketPath, ownedLease); } finally { await ownedLease?.release(); }}
async function prepareUnixDaemonSocketPath(socketPath: string, lease?: DaemonSocketPathLease): Promise<void> { if (!existsSync(socketPath)) { return; }
let stat: ReturnType<typeof lstatSync>; try { stat = lstatSync(socketPath); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { return; } throw error; } if (!stat.isSocket()) { throw new Error(`Daemon socket path exists and is not a socket: ${socketPath}`); }
const staleIdentity: DaemonSocketIdentity = { dev: stat.dev, ino: stat.ino }; if (await canConnectToUnixSocket(socketPath)) { throw new Error(`Daemon socket already in use: ${socketPath}`); } const deadline = Date.now() + DAEMON_SOCKET_RELEASE_GRACE_MS; while (Date.now() < deadline) { await delay(DAEMON_SOCKET_RELEASE_POLL_MS); if (!existsSync(socketPath)) { return; } let currentIdentity: DaemonSocketIdentity | undefined; try { currentIdentity = getDaemonSocketIdentity(socketPath); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") { return; } throw error; } if (!currentIdentity || currentIdentity.dev !== staleIdentity.dev || currentIdentity.ino !== staleIdentity.ino) { throw new Error(`Daemon socket changed ownership while waiting for cleanup: ${socketPath}`); } if (await canConnectToUnixSocket(socketPath)) { throw new Error(`Daemon socket already in use: ${socketPath}`); } }
if (lease) assertSocketLeaseHeld(socketPath, lease); unlinkSync(socketPath);}
export function restrictDaemonSocketPath(socketPath: string): void { if (process.platform === "win32") { return; } chmodSync(socketPath, DAEMON_SOCKET_MODE);}
export function getDaemonSocketIdentity(socketPath: string): DaemonSocketIdentity | undefined { if (process.platform === "win32") { return undefined; } const stat = lstatSync(socketPath); return { dev: stat.dev, ino: stat.ino };}
export function cleanupDaemonSocketPath( socketPath: string, expectedIdentity?: DaemonSocketIdentity, lease?: DaemonSocketPathLease,): void { if (process.platform === "win32") { return; } if (lease) { assertSocketLease(socketPath, lease); if (lease.compromise) { return; } try { cleanupUnixDaemonSocketPath(socketPath, expectedIdentity); } catch { // Best effort cleanup; shutdown should not be blocked by socket unlink failures. } return; } let releaseLock: (() => void) | undefined; let lockCompromised = false; try { releaseLock = lockfile.lockSync(socketPath, { realpath: false, stale: DAEMON_SOCKET_LOCK_STALE_MS, update: DAEMON_SOCKET_LOCK_UPDATE_MS, onCompromised: () => { lockCompromised = true; }, retries: 0, }); } catch { return; } if (lockCompromised) { try { releaseLock(); } catch { // Best effort release. } return; } try { cleanupUnixDaemonSocketPath(socketPath, expectedIdentity); } catch { // Best effort cleanup; shutdown should not be blocked by socket unlink failures. } finally { try { releaseLock(); } catch { // Best effort cleanup; a failed release is recoverable as a stale lock. } }}
function cleanupUnixDaemonSocketPath(socketPath: string, expectedIdentity?: DaemonSocketIdentity): void { if (!existsSync(socketPath)) { return; } if (expectedIdentity) { const currentIdentity = getDaemonSocketIdentity(socketPath); if ( !currentIdentity || currentIdentity.dev !== expectedIdentity.dev || currentIdentity.ino !== expectedIdentity.ino ) { return; } } unlinkSync(socketPath);}
function assertSocketLease(socketPath: string, lease: DaemonSocketPathLease): void { if (lease.socketPath !== socketPath) { throw new Error(`Daemon socket lease does not match ${socketPath}`); }}
function assertSocketLeaseHeld(socketPath: string, lease: DaemonSocketPathLease): void { if (lease.compromise) { throw new Error(`Daemon socket lease for ${socketPath} was compromised: ${lease.compromise.message}`); }}
export function defaultDaemonSocketDir(): string { const suffix = typeof process.getuid === "function" ? String(process.getuid()) : "user"; return join(tmpdir(), `prime-agent-${suffix}`);}
function ensureDefaultDaemonSocketDir(socketPath: string): void { if (process.platform === "win32" || dirname(socketPath) !== defaultDaemonSocketDir()) { return; }
if (!existsSync(defaultDaemonSocketDir())) { mkdirSync(defaultDaemonSocketDir(), { recursive: true, mode: DAEMON_SOCKET_DIR_MODE }); }
const stat = lstatSync(defaultDaemonSocketDir()); if (!stat.isDirectory()) { throw new Error(`Daemon socket directory exists and is not a directory: ${defaultDaemonSocketDir()}`); }
if (typeof process.getuid === "function" && stat.uid !== process.getuid()) { throw new Error(`Daemon socket directory is not owned by the current user: ${defaultDaemonSocketDir()}`); }
chmodSync(defaultDaemonSocketDir(), DAEMON_SOCKET_DIR_MODE);}
function canConnectToUnixSocket(socketPath: string): Promise<boolean> { return new Promise((resolveConnect) => { const socket = createConnection(socketPath); let settled = false; let timeoutId: ReturnType<typeof setTimeout> | undefined;
const finish = (canConnect: boolean) => { if (settled) { return; } settled = true; if (timeoutId) { clearTimeout(timeoutId); } socket.removeAllListeners(); socket.destroy(); resolveConnect(canConnect); };
timeoutId = setTimeout(() => finish(false), 250); socket.once("connect", () => finish(true)); socket.once("error", () => finish(false)); });}
function delay(ms: number): Promise<void> { return new Promise((resolve) => setTimeout(resolve, ms));}