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.
11 kB · 283 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284import { createHash } from "node:crypto";import { existsSync, mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs";import { tmpdir } from "node:os";import { join, resolve } from "node:path";import { lockSync } from "proper-lockfile";import { afterEach, describe, expect, it, vi } from "vitest";import { acquireSessionLease, canonicalSessionPath, getWindowsProcessStartId, isRenameTargetContention, SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV, SessionAlreadyActiveError,} from "../src/core/session-lease.js";
const tempDirs: string[] = [];
afterEach(() => { for (const directory of tempDirs.splice(0)) { rmSync(directory, { recursive: true, force: true }); }});
function createTempDir(): string { const directory = mkdtempSync(join(tmpdir(), "prime-session-lease-test-")); tempDirs.push(directory); return directory;}
function enabledEnvironment(owner: string): NodeJS.ProcessEnv { return { [SESSION_LEASES_ENABLED_ENV]: "1", [SESSION_LEASE_OWNER_ID_ENV]: owner, };}
describe("session leases", () => { it("reads an invariant process start identity on Windows", () => { const calls: Array<{ command: string; args: string[] }> = []; const processStartId = getWindowsProcessStartId(42, (command, args) => { calls.push({ command, args }); return "638880485801234567\r\n"; });
expect(processStartId).toBe("win:638880485801234567"); expect(calls).toEqual([ { command: "powershell.exe", args: [ "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", "([System.Diagnostics.Process]::GetProcessById(42)).StartTime.ToUniversalTime().Ticks", ], }, ]); });
it("rejects invalid Windows process start identities", () => { let queryCount = 0; const query = () => { queryCount++; return "not-a-start-time"; };
expect(getWindowsProcessStartId(42, query)).toBeUndefined(); expect(getWindowsProcessStartId(0, query)).toBeUndefined(); expect(queryCount).toBe(1); });
it("rejects a second live owner with a typed active-session error", () => { const agentDir = createTempDir(); const sessionPath = join(agentDir, "session.jsonl"); const first = acquireSessionLease(sessionPath, agentDir, enabledEnvironment("resident-a"));
expect(() => acquireSessionLease(sessionPath, agentDir, enabledEnvironment("owned-b"))).toThrow( SessionAlreadyActiveError, ); try { acquireSessionLease(sessionPath, agentDir, enabledEnvironment("owned-b")); } catch (error) { expect(error).toMatchObject({ code: "session_already_active", activeSessionId: "resident-a", sessionPath: canonicalSessionPath(sessionPath), }); }
first?.release(); const second = acquireSessionLease(sessionPath, agentDir, enabledEnvironment("owned-b")); expect(second?.sessionPath).toBe(canonicalSessionPath(sessionPath)); second?.release(); });
it("reclaims a lease whose owner process is gone", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(resolve(agentDir, "stale.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); mkdirSync(lockDirectory, { recursive: true }); writeFileSync( join(lockDirectory, "owner.json"), JSON.stringify({ version: 1, token: "stale", pid: 2_147_483_647, activeSessionId: "dead-owner", sessionPath, createdAt: new Date(0).toISOString(), }), );
const lease = acquireSessionLease(sessionPath, agentDir, enabledEnvironment("replacement")); expect(lease?.sessionPath).toBe(sessionPath); lease?.release(); });
it("never reclaims a lease whose owner file cannot be read", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(resolve(agentDir, "unreadable.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); // owner.json as a directory: every read fails with a non-ENOENT error, the // same shape as a transient EPERM/EBUSY on Windows. That may be a LIVE // lease, so acquisition must fail instead of destroying it. mkdirSync(join(lockDirectory, "owner.json"), { recursive: true });
expect(() => acquireSessionLease(sessionPath, agentDir, enabledEnvironment("intruder"))).toThrow( "Could not acquire session lease", ); expect(existsSync(join(lockDirectory, "owner.json"))).toBe(true); });
it("reports guard contention as a coordination failure", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(join(agentDir, "session.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const leaseRoot = join(agentDir, "session-leases"); const lockDirectory = join(leaseRoot, `${key}.lock`); mkdirSync(leaseRoot, { recursive: true }); const release = lockSync(lockDirectory, { realpath: false, lockfilePath: `${lockDirectory}.guard`, stale: 5000, }); // Keep the owner fresh while exercising the bounded synchronous retry count. const wait = vi.spyOn(Atomics, "wait").mockReturnValue("timed-out"); try { let thrown: unknown; try { acquireSessionLease(sessionPath, agentDir, enabledEnvironment("resident-a")); } catch (error) { thrown = error; } expect(thrown).toBeInstanceOf(Error); expect(thrown).not.toBeInstanceOf(SessionAlreadyActiveError); expect((thrown as Error).message).toContain("Could not coordinate session lease"); } finally { wait.mockRestore(); release(); } });
it("treats symlink aliases as the same persisted session", () => { const agentDir = createTempDir(); const sessionPath = join(agentDir, "session.jsonl"); const aliasPath = join(agentDir, "session-alias.jsonl"); writeFileSync(sessionPath, ""); symlinkSync(sessionPath, aliasPath); const first = acquireSessionLease(sessionPath, agentDir, enabledEnvironment("resident-a"));
expect(() => acquireSessionLease(aliasPath, agentDir, enabledEnvironment("owned-b"))).toThrow( SessionAlreadyActiveError, ); first?.release(); });
it("reclaims a lease after its pid has been reused", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(resolve(agentDir, "reused-pid.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); mkdirSync(lockDirectory, { recursive: true }); writeFileSync( join(lockDirectory, "owner.json"), JSON.stringify({ version: 1, token: "stale", pid: process.pid, processStartId: "different-process", activeSessionId: "old-owner", sessionPath, createdAt: new Date(0).toISOString(), }), );
const lease = acquireSessionLease(sessionPath, agentDir, enabledEnvironment("replacement")); expect(lease?.sessionPath).toBe(sessionPath); lease?.release(); });
it("fails closed on corrupt owner.json instead of reclaiming the lease", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(resolve(agentDir, "corrupt.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); mkdirSync(lockDirectory, { recursive: true }); writeFileSync(join(lockDirectory, "owner.json"), "this is not json"); expect(() => acquireSessionLease(sessionPath, agentDir, enabledEnvironment("replacement"))).toThrow("Corrupt"); });
it("fails closed on owner.json with missing required fields", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(resolve(agentDir, "partial.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); mkdirSync(lockDirectory, { recursive: true }); writeFileSync(join(lockDirectory, "owner.json"), JSON.stringify({ version: 1, token: "orphan" })); expect(() => acquireSessionLease(sessionPath, agentDir, enabledEnvironment("replacement"))).toThrow("Corrupt"); });
it("reclaims a lease when owner.json is absent", () => { const agentDir = createTempDir(); const sessionPath = canonicalSessionPath(resolve(agentDir, "absent-lock.jsonl")); const key = createHash("sha256").update(sessionPath).digest("hex"); const lockDirectory = join(agentDir, "session-leases", `${key}.lock`); mkdirSync(lockDirectory, { recursive: true }); const lease = acquireSessionLease(sessionPath, agentDir, enabledEnvironment("replacement")); expect(lease?.sessionPath).toBe(sessionPath); lease?.release(); });
it("isRenameTargetContention returns true for EEXIST and ENOTEMPTY", () => { expect(isRenameTargetContention("/tmp", "EEXIST")).toBe(true); expect(isRenameTargetContention("/tmp", "ENOTEMPTY")).toBe(true); });
it("isRenameTargetContention returns false for EPERM on a nonexistent target", () => { const dir = createTempDir(); const missing = join(dir, "nonexistent.lock"); // Target does not exist, so EPERM is a real permission error. // No existing target and platform does not matter for that case. expect(isRenameTargetContention(missing, "EPERM")).toBe(false); });
it("isRenameTargetContention returns true for EPERM on an existing target", () => { const dir = createTempDir(); const target = join(dir, "existing.lock"); mkdirSync(target, { recursive: true }); // Target exists, so EPERM from renameSync means contention on Windows. expect(isRenameTargetContention(target, "EPERM", "win32")).toBe(true); expect(isRenameTargetContention(target, "EPERM", "darwin")).toBe(false); expect(isRenameTargetContention(target, "EPERM", "linux")).toBe(false); });
it("isRenameTargetContention returns true for EACCES on an existing target", () => { const dir = createTempDir(); const target = join(dir, "existing-eacces.lock"); mkdirSync(target, { recursive: true }); expect(isRenameTargetContention(target, "EACCES", "win32")).toBe(true); expect(isRenameTargetContention(target, "EACCES", "darwin")).toBe(false); expect(isRenameTargetContention(target, "EACCES", "linux")).toBe(false); });
it("isRenameTargetContention returns false for EACCES on a nonexistent target", () => { const dir = createTempDir(); const missing = join(dir, "missing-eacces.lock"); expect(isRenameTargetContention(missing, "EACCES")).toBe(false); });
it("isRenameTargetContention returns false for unrelated error codes", () => { expect(isRenameTargetContention("/tmp", "EIO")).toBe(false); expect(isRenameTargetContention("/tmp", "EBUSY")).toBe(false); expect(isRenameTargetContention("/tmp", undefined)).toBe(false); });
it("is inert for direct SDK runtimes unless worker isolation enables it", () => { const agentDir = createTempDir(); expect(acquireSessionLease(join(agentDir, "session.jsonl"), agentDir, {})).toBeUndefined(); });});