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.
74 kB · 1863 lines
TypeScript
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864import { type ChildProcess, spawn } from "node:child_process";import { createHash, randomUUID } from "node:crypto";import { chmodSync, existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync,} from "node:fs";import { createServer } from "node:net";import { tmpdir } from "node:os";import { dirname, join, resolve } from "node:path";import { afterEach, describe, expect, it } from "vitest";import { ENV_AGENT_DIR, getCronJobsPath } from "../src/config.js";import { AgentCronJobStore } from "../src/core/cron-jobs.js";import { readActiveOrphanProcesses } from "../src/core/orphan-process-journal.js";import { acquireSessionLease, getPsProcessStartId, SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV,} from "../src/core/session-lease.js";import { readSessionInfo, SessionManager } from "../src/core/session-manager.js";import { DaemonAgentConnection } from "../src/modes/agent-connection/daemon-agent-connection.js";import { DaemonClient, getDaemonSocketCloseReason } from "../src/modes/daemon/daemon-client.js";import type { SessionSummary } from "../src/modes/daemon/daemon-session-list.js";import { type DaemonWorkerDescriptor, type DaemonWorkerFrameHeader, isDaemonWorkerFrameHeader,} from "../src/modes/daemon/daemon-worker-protocol.js";import { encodePrivateFrame, PrivateFrameDecoder } from "../src/modes/session-worker/private-framing.js";import { fixtureEnv } from "./utilities.js";
const cliPath = resolve(__dirname, "../src/cli.ts");const tsxPath = resolve(__dirname, "../../../node_modules/tsx/dist/cli.mjs");const blockingProcessPath = resolve(__dirname, "fixtures/blocking-process.mjs");const tempDirs: string[] = [];const children = new Set<ChildProcess>();const workerPids = new Set<number>();const daemonSockets = new Set<string>();const childDiagnostics = new WeakMap<ChildProcess, { stdout: string; stderr: string }>();const PROCESS_STRESS_WORKERS = Number.parseInt(process.env.PRIME_AGENT_STRESS_WORKERS ?? "10", 10);
afterEach(async () => { for (const socketPath of daemonSockets) { const client = new DaemonClient(socketPath); try { await client.connect(250); await client.request({ type: "shutdown" }, 2000); } catch { // Already gone. } finally { client.close(); } } daemonSockets.clear(); for (const child of children) { if (child.exitCode === null && child.signalCode === null) { child.kill("SIGTERM"); } } // Await owned exits before rmSync: a dying worker's log writer otherwise races it into ENOTEMPTY. await Promise.all([...children].map((child) => waitForExit(child).catch(() => undefined))); children.clear(); for (const pid of workerPids) { try { process.kill(pid, "SIGCONT"); } catch { // Already gone. } try { process.kill(-pid, "SIGTERM"); } catch { try { process.kill(pid, "SIGTERM"); } catch { // Already gone. } } } await Promise.all([...workerPids].map((pid) => waitForProcessGone(pid).catch(() => undefined))); workerPids.clear(); for (const directory of tempDirs.splice(0)) { rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); }});
function tempDir(): string { const directory = mkdtempSync(join(tmpdir(), "prime-daemon-supervisor-test-")); tempDirs.push(directory); return directory;}
function spawnSupervisor( agentDir: string, socketPath: string, cwd: string, extraArgs: readonly string[] = [], extraEnv: NodeJS.ProcessEnv = {},): ChildProcess { daemonSockets.add(socketPath); const child = spawn( process.execPath, [tsxPath, cliPath, "--mode", "daemon", "--daemon-socket", socketPath, "--offline", ...extraArgs], { cwd, env: { ...process.env, ...fixtureEnv(), ...extraEnv, [ENV_AGENT_DIR]: agentDir, PI_OFFLINE: "1", TSX_TSCONFIG_PATH: resolve(__dirname, "../../../tsconfig.json"), }, stdio: ["ignore", "pipe", "pipe"], }, ); children.add(child); const diagnostics = { stdout: "", stderr: "" }; childDiagnostics.set(child, diagnostics); child.stdout?.on("data", (chunk: Buffer) => { diagnostics.stdout += chunk.toString("utf8"); }); child.stderr?.on("data", (chunk: Buffer) => { diagnostics.stderr += chunk.toString("utf8"); }); return child;}
function readDaemonLogs(agentDir: string): string { const logsDir = join(agentDir, "logs"); try { return readdirSync(logsDir) .map((name) => `${name}:\n${readFileSync(join(logsDir, name), "utf8")}`) .join("\n"); } catch { return "no daemon logs"; }}
function readWorkerDescriptor(agentDir: string): DaemonWorkerDescriptor { const workersRoot = join(agentDir, "daemon-workers"); for (const directory of readdirSync(workersRoot)) { const descriptorDirectory = join(workersRoot, directory); for (const name of readdirSync(descriptorDirectory)) { if (name.endsWith(".json")) { return JSON.parse(readFileSync(join(descriptorDirectory, name), "utf8")) as DaemonWorkerDescriptor; } } } throw new Error("Worker descriptor was not persisted");}
function countWorkerDescriptors(agentDir: string): number { const workersRoot = join(agentDir, "daemon-workers"); try { return readdirSync(workersRoot).reduce((count, directory) => { return count + readdirSync(join(workersRoot, directory)).filter((name) => name.endsWith(".json")).length; }, 0); } catch { return 0; }}
async function waitForWorkerStopTombstone(agentDir: string): Promise<DaemonWorkerDescriptor> { const deadline = Date.now() + 5000; while (Date.now() < deadline) { try { const descriptor = readWorkerDescriptor(agentDir); if (descriptor.stopRequestedAt) { return descriptor; } } catch { // The descriptor directory may still be appearing. } await new Promise((resolveDelay) => setTimeout(resolveDelay, 10)); } throw new Error("Worker stop tombstone was not persisted");}
function readSupervisorConfig(agentDir: string): { defaultSessionConfig?: { sessionDir?: string; noTools?: boolean } } { const workersRoot = join(agentDir, "daemon-workers"); for (const directory of readdirSync(workersRoot)) { const path = join(workersRoot, directory, "supervisor-config"); try { return JSON.parse(readFileSync(path, "utf8")) as { defaultSessionConfig?: { sessionDir?: string; noTools?: boolean }; }; } catch { // Continue looking for the descriptor directory for this daemon socket. } } throw new Error("Supervisor config was not persisted");}
async function connectEventually(socketPath: string, child?: ChildProcess): Promise<DaemonClient> { const deadline = Date.now() + 15_000; let lastError: unknown; while (Date.now() < deadline) { if (child && (child.exitCode !== null || child.signalCode !== null)) { const diagnostics = childDiagnostics.get(child); throw new Error( `Supervisor exited before becoming ready (code ${child.exitCode}, signal ${child.signalCode})\n` + `stdout:\n${diagnostics?.stdout ?? ""}\nstderr:\n${diagnostics?.stderr ?? ""}`, ); } const client = new DaemonClient(socketPath); try { await client.connect(250); await client.waitForHello(1000); return client; } catch (error) { lastError = error; client.close(); await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } } throw new Error(`Timed out waiting for supervisor: ${String(lastError)}`);}
async function waitForSocketGone(socketPath: string): Promise<void> { const deadline = Date.now() + 10_000; while (Date.now() < deadline) { const client = new DaemonClient(socketPath); try { await client.connect(100); } catch { client.close(); return; } client.close(); await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } throw new Error("Daemon supervisor socket remained available after shutdown");}
function requireSummary(responseData: unknown): SessionSummary { if (!responseData || typeof responseData !== "object") { throw new Error("Missing daemon session summary"); } return responseData as SessionSummary;}
function requireSessionList(responseData: unknown): SessionSummary[] { if (!responseData || typeof responseData !== "object" || !("sessions" in responseData)) { throw new Error("Missing daemon session list"); } const sessions = (responseData as { sessions: unknown }).sessions; if (!Array.isArray(sessions)) { throw new Error("Invalid daemon session list"); } return sessions as SessionSummary[];}
async function waitForExit(child: ChildProcess): Promise<void> { if (child.exitCode !== null || child.signalCode !== null) { return; } await new Promise<void>((resolveExit, reject) => { const timeout = setTimeout(() => reject(new Error("Timed out waiting for process exit")), 10_000); child.once("exit", () => { clearTimeout(timeout); resolveExit(); }); });}
async function waitForProcessGone(pid: number): Promise<void> { const deadline = Date.now() + 10_000; while (Date.now() < deadline) { try { process.kill(pid, 0); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") { return; } } await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } throw new Error(`Worker ${pid} remained alive after daemon shutdown`);}
async function waitForCondition(predicate: () => boolean, failureMessage: string, timeoutMs = 10_000): Promise<void> { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { if (predicate()) { return; } await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } throw new Error(failureMessage);}
function quoteShellArgument(argument: string): string { return `'${argument.replaceAll("'", `'"'"'`)}'`;}
async function startBlockingBash(client: DaemonClient, activeSessionId: string, readyPath: string): Promise<void> { const command = [process.execPath, blockingProcessPath, readyPath].map(quoteShellArgument).join(" "); const response = await client.request({ type: "execute_bash", activeSessionId, command }); if (!response.success) { throw new Error(response.error); } await waitForCondition(() => existsSync(readyPath), `Blocking bash process did not become ready: ${readyPath}`);}
describe("daemon supervisor resident workers", () => { it("accepts the canonical socket path when launched with duplicate slashes", async () => { if (process.platform === "win32") return; const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const socketPath = join(root, "daemon.sock"); mkdirSync(projectDir, { recursive: true });
const supervisor = spawnSupervisor(agentDir, `${root}//daemon.sock`, projectDir); const client = await connectEventually(socketPath, supervisor); const response = await client.request({ type: "list" });
expect(response.success).toBe(true); expect(requireSessionList(response.success ? response.data : undefined)).toHaveLength(0); await client.request({ type: "shutdown", force: true }); client.close(); await waitForSocketGone(socketPath); }, 60_000);
it("creates top-level sessions at depth zero when the supervisor inherits a child depth", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-depth-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir, [], { RLM_DEPTH: "1" }); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) throw new Error(created.error); const summary = requireSummary(created.data); if (!summary.workerPid || !summary.sessionFile) throw new Error("Worker did not expose its session identity"); workerPids.add(summary.workerPid);
expect(summary).toMatchObject({ runtimeKind: "top-level", rlmDepth: 0 }); expect(await readSessionInfo(summary.sessionFile)).toMatchObject({ rlmDepth: 0, parentSessionPath: undefined });
await client.request({ type: "shutdown" }); client.close(); await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); await waitForSocketGone(socketPath); }, 60_000);
it("restarts an adopted pre-roster worker from the current binary", async () => { const directory = tempDir(); const agentDir = join(directory, "agent"); const projectDir = join(directory, "project"); const sessionDir = join(agentDir, "sessions"); mkdirSync(projectDir, { recursive: true }); const manager = SessionManager.create(projectDir, sessionDir); manager.appendMessage({ role: "user", content: "pre-roster fixture", timestamp: 1 }); manager.flushNow(); const sessionPath = manager.getSessionFile(); const sessionId = manager.getSessionId(); if (!sessionPath) throw new Error("Fixture session did not persist");
// A long-lived stand-in process plays the pre-roster worker's pid. const legacyProcess = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000);"], { stdio: "ignore" }); children.add(legacyProcess); if (!legacyProcess.pid) throw new Error("Missing legacy process pid");
// A fake worker socket that authenticates without advertising the roster capability. const workerSocketPath = join(directory, "legacy-worker.sock"); const fakeWorker = createServer((socket) => { const decoder = new PrivateFrameDecoder(isDaemonWorkerFrameHeader); socket.write( encodePrivateFrame<DaemonWorkerFrameHeader>( { kind: "outbound", outboundType: "daemon_hello" }, Buffer.from(`${JSON.stringify({ type: "daemon_hello" })}\n`), ), ); socket.on("data", (chunk: Buffer) => { for (const frame of decoder.push(chunk)) { if (frame.header.kind !== "command") continue; const command = JSON.parse(frame.payload.toString("utf8")) as { id: string; type: string }; const data = command.type === "list" ? { sessions: [ { id: "legacy-root-active", activeSessionId: "legacy-root-active", sessionId, sessionFile: sessionPath, lifecycle: "live", activity: "idle", isSessionActive: false, cwd: projectDir, isStreaming: false, isCompacting: false, attachedClients: 0, messageCount: 1, sessionActions: { queuedCount: 0, steering: [], followUps: [] }, }, ], } : {}; socket.write( encodePrivateFrame<DaemonWorkerFrameHeader>( { kind: "outbound", outboundType: "response", requestId: frame.header.requestId }, Buffer.from( `${JSON.stringify({ id: command.id, type: "response", command: command.type, success: true, data })}\n`, ), ), ); } }); }); const socketPath = join(directory, "daemon.sock"); await new Promise<void>((resolveListen) => fakeWorker.listen(workerSocketPath, resolveListen)); const descriptorDir = join( agentDir, "daemon-workers", createHash("sha256").update(socketPath).digest("hex").slice(0, 12), ); mkdirSync(descriptorDir, { recursive: true }); const now = new Date().toISOString(); writeFileSync( join(descriptorDir, "legacy-worker.json"), `${JSON.stringify({ version: 2, workerId: "legacy-worker", pid: legacyProcess.pid, socketPath: workerSocketPath, recoveryJournalPath: join(descriptorDir, "legacy-worker.recovery.jsonl"), supervisorSocketPath: socketPath, authenticationToken: "legacy-token", rootActiveSessionId: "legacy-root-active", rootSessionId: sessionId, sessionFile: sessionPath, sessionDir, createdAt: now, updatedAt: now, lifecycle: "ready", createCommand: { type: "create", sessionPath }, consecutiveFailures: 0, })}\n`, );
// Once the supervisor kills the old pid, its socket goes quiet exactly like a dead worker's. legacyProcess.once("exit", () => { fakeWorker.close(); rmSync(workerSocketPath, { force: true }); }); const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); let restarted: SessionSummary | undefined; const deadline = Date.now() + 30_000; while (Date.now() < deadline) { const listed = await client.request({ type: "list" }); restarted = requireSessionList(listed.success ? listed.data : undefined).find( (candidate) => candidate.sessionId === sessionId, ); if (restarted?.workerState === "ready" && restarted.workerPid !== undefined) break; restarted = undefined; await new Promise((resolveDelay) => setTimeout(resolveDelay, 100)); } if (!restarted?.workerPid) { throw new Error(`Pre-roster worker was not restarted:\n${readDaemonLogs(agentDir)}`); } workerPids.add(restarted.workerPid); // A fresh current-binary worker owns the reloaded idle session; the fake pre-roster pid is not adopted. expect(restarted.workerPid).not.toBe(legacyProcess.pid); expect(restarted.isSessionActive).toBe(false); // The seeded user message plus the harness digest injected on resume. expect(restarted.messageCount).toBe(2); await waitForProcessGone(legacyProcess.pid); fakeWorker.close(); client.close(); }, 90_000);
it("lists, creates, and attaches passive children through their owning worker", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-passive-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true });
const parentManager = SessionManager.create(projectDir, sessionDir); parentManager.appendMessage({ role: "user", content: "parent fixture", timestamp: 1 }); parentManager.flushNow(); const parentSessionFile = parentManager.getSessionFile(); const parentArtifactDir = parentManager.getSessionArtifactDir(); if (!parentSessionFile || !parentArtifactDir) throw new Error("Missing parent fixture paths");
const makeChild = (childId: string, sessionName: string, timestamp: number) => { const childSessionDir = join(parentArtifactDir, childId); const manager = SessionManager.create(projectDir, childSessionDir); manager.newSession({ parentSession: parentSessionFile }); manager.appendSessionInfo(sessionName); manager.appendMessage({ role: "user", content: `completed ${childId} fixture`, timestamp }); manager.flushNow(); const sessionFile = manager.getSessionFile(); if (!sessionFile) throw new Error("Missing child fixture path"); return { childId, sessionName, childSessionDir, manager, sessionFile }; }; const child = makeChild("passive-child", "passive-child-worker", 2); const createChild = makeChild("passive-create-child", "passive-create-worker", 3); writeFileSync( join(parentArtifactDir, "rlm-subagents.jsonl"), `${[child, createChild] .map((fixture) => JSON.stringify({ type: "rlm_subagent", childId: fixture.childId, sessionName: fixture.sessionName, sessionDir: fixture.childSessionDir, sessionFile: fixture.sessionFile, parentSessionId: parentManager.getSessionId(), parentSessionFile, rlmDepth: 1, rlmMaxDepth: 4, rlmParentNodeId: fixture.childId, status: "completed", createdAt: 1, updatedAt: "2026-01-01T00:00:00.000Z", }), ) .join("\n")}`, );
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", sessionPath: parentSessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); expect(created.success).toBe(true); const parentSummary = requireSummary(created.success ? created.data : undefined); if (!parentSummary.workerPid) throw new Error("Parent worker did not expose its pid"); workerPids.add(parentSummary.workerPid);
const beforeAttach = await client.request({ type: "list" }); expect(beforeAttach.success).toBe(true); const passiveSummary = requireSessionList(beforeAttach.success ? beforeAttach.data : undefined).find( (summary) => summary.sessionFile === child.sessionFile, ); expect(passiveSummary).toMatchObject({ sessionId: child.manager.getSessionId(), sessionName: "passive-child-worker", runtimeKind: "subagent", rlmChildId: child.childId, workerPid: parentSummary.workerPid, }); expect(passiveSummary?.activeSessionId).toBeUndefined();
const createdChild = await client.request({ type: "create", sessionPath: createChild.sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!createdChild.success) throw new Error(createdChild.error); expect(requireSummary(createdChild.data)).toMatchObject({ workerPid: parentSummary.workerPid, rlmChildId: createChild.childId, }); expect(countWorkerDescriptors(agentDir)).toBe(1);
const attached = await client.request({ type: "attach", activeSessionId: child.manager.getSessionId(), capabilities: ["attach_snapshot", "event_sequence", "slim_attach"], }); if (!attached.success) throw new Error(attached.error);
const afterAttach = await client.request({ type: "list" }); const hydratedSummary = requireSessionList(afterAttach.success ? afterAttach.data : undefined).find( (summary) => summary.sessionFile === child.sessionFile, ); expect(hydratedSummary).toMatchObject({ activeSessionId: expect.any(String), workerPid: parentSummary.workerPid, rlmChildId: child.childId, }); expect(countWorkerDescriptors(agentDir)).toBe(1);
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); }, 60_000);
it("keeps client-owned workers hidden and removes them without archiving", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-owned-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "owned worker fixture", timestamp: 1 }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); }
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const launchEnvSentinel = `owned-env-${randomUUID()}`; const created = await client.request({ type: "create", sessionPath: sessionFile, lifecycle: "client_owned", launchEnv: { PRIME_AGENT_OWNED_TEST: launchEnvSentinel }, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); expect(created.success).toBe(true); const summary = requireSummary(created.success ? created.data : undefined); if (!summary.workerPid || !summary.activeSessionId) { throw new Error("Client-owned worker did not expose its process identity"); } workerPids.add(summary.workerPid);
const publicList = await client.request({ type: "list" }); expect(publicList.success).toBe(true); expect(requireSessionList(publicList.success ? publicList.data : undefined)).toEqual([]); const internalList = await client.request({ type: "list", includeClientOwned: true }); expect(internalList.success).toBe(true); expect(requireSessionList(internalList.success ? internalList.data : undefined)).toHaveLength(1); const otherClient = await connectEventually(socketPath); const deniedList = await otherClient.request({ type: "list", includeClientOwned: true }); expect(deniedList).toMatchObject({ success: true, data: { sessions: [], busyClientOwnedSessionCount: 0 }, }); const privateSelector = summary.sessionId.slice(-8); const deniedAttach = await otherClient.request({ type: "attach", activeSessionId: privateSelector }); expect(deniedAttach).toMatchObject({ success: false, error: `Unknown active session: ${privateSelector}`, }); const deniedCron = await otherClient.request({ type: "cron_add", activeSessionId: summary.activeSessionId, schedule: "every 1h", prompt: "unauthorized", }); expect(deniedCron).toMatchObject({ success: false, error: `Unknown active session: ${summary.activeSessionId}`, }); otherClient.close(); const connection = await DaemonAgentConnection.attach(client, summary.activeSessionId, { ownedSession: true, supportsExtensionUi: false, }); await expect(connection.listHeartbeats()).resolves.toEqual([]); await expect(connection.listCronJobs()).resolves.toEqual([]); const cronResponse = await client.request({ type: "cron_add", activeSessionId: summary.activeSessionId, schedule: "every 1h", prompt: "check status", }); if (!cronResponse.success || !cronResponse.data || typeof cronResponse.data !== "object") { throw new Error(cronResponse.success ? "Cron response was missing its job" : cronResponse.error); } const cronJob = (cronResponse.data as { job: { id: string } }).job; await expect(connection.listCronJobs()).resolves.toEqual([expect.objectContaining({ id: cronJob.id })]); await expect(connection.cancelCronJob(cronJob.id)).resolves.toMatchObject({ id: cronJob.id });
const descriptor = readWorkerDescriptor(agentDir); expect(descriptor.ownerClientId).toEqual(expect.any(String)); expect(JSON.stringify(descriptor)).not.toContain(launchEnvSentinel); expect(descriptor.createCommand).not.toHaveProperty("launchEnv"); expect(descriptor.createCommand).not.toHaveProperty("lifecycle");
await connection.dispose(); await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); await waitForCondition( () => countWorkerDescriptors(agentDir) === 0, "Client-owned worker descriptor was not removed", ); expect((await readSessionInfo(sessionFile))?.state?.status).not.toBe("archived");
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); }, 60_000);
it("releases an adopted client-owned worker when disposal races supervisor replacement", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const socketPath = join(tmpdir(), `prime-supervisor-owned-adopt-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", lifecycle: "client_owned", noSession: true, launchEnv: { TSX_TSCONFIG_PATH: resolve(__dirname, "../../../tsconfig.json") }, config: { cwd: projectDir, agentDir, noTools: true, noExtensions: true }, }); if (!created.success) { throw new Error(created.error); } const summary = requireSummary(created.data); if (!summary.workerPid || !summary.activeSessionId) { throw new Error("Client-owned worker did not expose its process identity"); } workerPids.add(summary.workerPid); const connection = await DaemonAgentConnection.attach(client, summary.activeSessionId, { closeClientOnDispose: true, ownedSession: true, recoverDaemon: async () => {}, supportsExtensionUi: false, });
supervisor.kill("SIGTERM"); await waitForExit(supervisor); children.delete(supervisor); await connection.dispose();
await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); await waitForCondition( () => countWorkerDescriptors(agentDir) === 0, "Adopted client-owned worker descriptor was not removed", ); const replacementClient = await connectEventually(socketPath); await replacementClient.request({ type: "shutdown" }); replacementClient.close(); await waitForSocketGone(socketPath); }, 60_000);
it("delivers agent-origin messages between root siblings on separate workers", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join( tmpdir(), `prime-supervisor-root-message-${process.pid}-${randomUUID().slice(0, 8)}.sock`, ); mkdirSync(projectDir, { recursive: true });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const createRoot = async (name: string) => { const response = await client.request({ type: "create", name, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); expect(response.success).toBe(true); return requireSummary(response.success ? response.data : undefined); }; const source = await createRoot("source-root"); const target = await createRoot("target-root"); expect(source.workerPid).not.toBe(target.workerPid); await startBlockingBash(client, target.activeSessionId ?? target.id, join(root, "target-root-blocker.ready"));
const response = await client.request({ type: "send_message", fromActiveSessionId: source.activeSessionId ?? source.id, targetActiveSessionId: target.activeSessionId ?? target.id, message: "hello sibling root", agentOrigin: true, }); expect(response.success, JSON.stringify(response)).toBe(true); expect(response).toMatchObject({ success: true, data: { source: "agent_message", target: { activeSessionId: target.activeSessionId ?? target.id }, message: "hello sibling root", deliveryStatus: "queued", }, }); const shutdown = await client.request({ type: "shutdown" }, 10_000); expect(shutdown.success).toBe(true); client.close(); await waitForSocketGone(socketPath); }, 30_000);
it("cancels an archived session heartbeat without spawning a worker", { tags: ["process-stress"] }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join( tmpdir(), `prime-supervisor-archived-cron-${process.pid}-${randomUUID().slice(0, 8)}.sock`, ); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "do not revive me", timestamp: 1 }); sessionManager.appendSessionState({ status: "active" }); sessionManager.appendSessionState({ status: "archived" }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); } const cronStore = new AgentCronJobStore(getCronJobsPath(agentDir)); const heartbeat = cronStore.createHeartbeat({ activeSessionId: "old-active-session", sessionId: sessionManager.getSessionId(), sessionFile, cwd: projectDir, scheduleText: "every 10s", prompt: "continue old work", now: new Date(Date.now() - 20_000), });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const migratedStore = AgentCronJobStore.forSessionArtifacts(); migratedStore.registerSessionArtifact(sessionManager.getSessionId(), sessionManager.getSessionArtifactDir()!); await waitForCondition( () => migratedStore.list().find((job) => job.id === heartbeat.id)?.status === "cancelled", "Archived heartbeat was not cancelled", );
expect(countWorkerDescriptors(agentDir)).toBe(0); const listed = await client.request({ type: "list" }); expect(listed.success).toBe(true); expect( requireSessionList(listed.success ? listed.data : undefined).filter( (session) => session.activeSessionId || session.workerPid, ), ).toEqual([]); expect((await readSessionInfo(sessionFile))?.state).toEqual({ status: "archived" });
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); });
it("cancels an orphan heartbeat instead of recreating a descriptorless active session", { tags: ["process-stress"], }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-orphan-cron-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "old scheduled work", timestamp: 1 }); sessionManager.appendSessionState({ status: "active" }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); } const cronStore = new AgentCronJobStore(getCronJobsPath(agentDir)); const heartbeat = cronStore.createHeartbeat({ activeSessionId: "deleted-worker", sessionId: sessionManager.getSessionId(), sessionFile, cwd: projectDir, scheduleText: "every 10s", prompt: "continue old work", now: new Date(Date.now() - 20_000), });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const migratedStore = AgentCronJobStore.forSessionArtifacts(); migratedStore.registerSessionArtifact(sessionManager.getSessionId(), sessionManager.getSessionArtifactDir()!); await waitForCondition( () => migratedStore.list().find((job) => job.id === heartbeat.id)?.status === "cancelled", "Orphan heartbeat was not cancelled", );
expect(countWorkerDescriptors(agentDir)).toBe(0); const listed = await client.request({ type: "list" }); expect(listed.success).toBe(true); expect( requireSessionList(listed.success ? listed.data : undefined).filter( (session) => session.activeSessionId || session.workerPid, ), ).toEqual([]);
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); });
it("restarts an empty supervisor without requiring a resident worker", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(root, "custom-sessions"); const socketPath = join(tmpdir(), `prime-supervisor-restart-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir, ["--session-dir", sessionDir, "--no-tools"]); const client = await connectEventually(socketPath, supervisor); const restarted = await client.request({ type: "restart" }); expect(restarted.success).toBe(true); client.close(); await waitForExit(supervisor); children.delete(supervisor);
const replacementClient = await connectEventually(socketPath); const listed = await replacementClient.request({ type: "list" }); expect(listed.success).toBe(true); const persistedConfig = readSupervisorConfig(agentDir); expect(persistedConfig).toMatchObject({ defaultSessionConfig: { sessionDir } }); expect(persistedConfig.defaultSessionConfig).not.toHaveProperty("noTools"); await replacementClient.request({ type: "shutdown" }); replacementClient.close(); await waitForSocketGone(socketPath); });
it("survives a worker process spawn error", { tags: ["process-stress"] }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const missingCwd = join(root, "missing-project"); const socketPath = join(tmpdir(), `prime-supervisor-spawn-error-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const failed = await client.request({ type: "create", config: { cwd: missingCwd, agentDir, noTools: true, noExtensions: true }, });
expect(failed).toMatchObject({ success: false }); expect(supervisor.exitCode).toBeNull(); const listed = await client.request({ type: "list" }); expect(listed.success).toBe(true); expect(requireSessionList(listed.success ? listed.data : undefined)).toEqual([]);
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); });
it("archives resident roots and cancels their heartbeats on explicit daemon shutdown", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-shutdown-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "stop with daemon", timestamp: 1 }); sessionManager.appendSessionState({ status: "active" }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); }
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", sessionPath: sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) { throw new Error(created.error); } const summary = requireSummary(created.data); if (!summary.workerPid) { throw new Error("Resident worker did not expose its pid"); } workerPids.add(summary.workerPid); const heartbeatResponse = await client.request({ type: "heartbeat_set", activeSessionId: summary.activeSessionId ?? summary.id, schedule: "every 1h", prompt: "continue old work", }); if (!heartbeatResponse.success || !heartbeatResponse.data || typeof heartbeatResponse.data !== "object") { throw new Error(heartbeatResponse.success ? "Heartbeat response was missing data" : heartbeatResponse.error); } const heartbeat = (heartbeatResponse.data as { heartbeat: { id: string } }).heartbeat; const cronStore = AgentCronJobStore.forSessionArtifacts(); cronStore.registerSessionArtifact(summary.sessionId, sessionManager.getSessionArtifactDir()!); const observer = await connectEventually(socketPath, supervisor); const observerClosed = new Promise<Error>((resolveClose) => observer.onClose(resolveClose));
expect((await client.request({ type: "shutdown" })).success).toBe(true); client.close(); expect(getDaemonSocketCloseReason(await observerClosed)).toBe("shutdown"); observer.close(); await waitForSocketGone(socketPath); await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); expect(countWorkerDescriptors(agentDir)).toBe(0); expect((await readSessionInfo(sessionFile))?.state).toEqual({ status: "archived" }); expect(cronStore.list().find((job) => job.id === heartbeat.id)).toMatchObject({ status: "cancelled" });
const replacement = spawnSupervisor(agentDir, socketPath, projectDir); const replacementClient = await connectEventually(socketPath, replacement); const listed = await replacementClient.request({ type: "list" }); expect(listed.success).toBe(true); expect( requireSessionList(listed.success ? listed.data : undefined).filter( (session) => session.activeSessionId || session.workerPid, ), ).toEqual([]); await replacementClient.request({ type: "shutdown" }); replacementClient.close(); await waitForSocketGone(socketPath); }, 30_000);
it("finalizes a timed-out worker stop by force-stopping the process and removing its registration", { tags: ["process-stress"], timeout: 45_000, }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join( tmpdir(), `prime-supervisor-stop-finalize-${process.pid}-${randomUUID().slice(0, 8)}.sock`, ); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "finalize me", timestamp: 1 }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); }
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", sessionPath: sessionFile, lifecycle: "client_owned", config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) { throw new Error(created.error); } const summary = requireSummary(created.data); if (!summary.workerPid) { throw new Error("Resident worker did not expose its pid"); } workerPids.add(summary.workerPid); const activeSessionId = summary.activeSessionId ?? summary.id;
// A suspended worker cannot exit within the stop deadline, so the stop // times out and used to leave a tombstoned registration behind forever. process.kill(summary.workerPid, "SIGSTOP"); const stopResult = await client.request({ type: "complete_owned_session", activeSessionId }, 30_000); expect(stopResult).toMatchObject({ success: false, error: expect.stringContaining("did not stop"), }); const tombstone = readWorkerDescriptor(agentDir); expect(tombstone.stopRequestedAt).toEqual(expect.any(String));
// The supervisor finishes the interrupted stop on its own: it escalates // to SIGKILL, waits for the process to die, and removes the registration. await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); await waitForCondition( () => countWorkerDescriptors(agentDir) === 0, "Timed-out worker stop was not finalized", 20_000, );
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); });
it("resumes a saved session immediately after a worker stop fails and the process dies", { tags: ["process-stress"], timeout: 45_000, }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-resume-heal-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "resume me", timestamp: 1 }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); }
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", sessionPath: sessionFile, lifecycle: "client_owned", config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) { throw new Error(created.error); } const summary = requireSummary(created.data); if (!summary.workerPid) { throw new Error("Resident worker did not expose its pid"); } workerPids.add(summary.workerPid); const activeSessionId = summary.activeSessionId ?? summary.id;
// A suspended worker forces the stop past its deadline, leaving a // tombstoned registration for a process that dies moments later. process.kill(summary.workerPid, "SIGSTOP"); const stopResult = await client.request({ type: "complete_owned_session", activeSessionId }, 30_000); expect(stopResult).toMatchObject({ success: false, error: expect.stringContaining("did not stop"), }); process.kill(summary.workerPid, "SIGKILL"); await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid);
// Resuming the saved transcript must not be blocked by the stale // registration. Whichever cleanup wins the race — the background stop // finalizer or the resume-time reclaim (each covered deterministically // by unit tests) — the user-visible guarantee is the same: the resume // below must succeed with a fresh worker. const resumed = await client.request({ type: "create", sessionPath: sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); expect(resumed.success).toBe(true); const resumedSummary = requireSummary(resumed.success ? resumed.data : undefined); expect(resumedSummary.sessionId).toBe(summary.sessionId); expect(resumedSummary.workerPid).not.toBe(summary.workerPid); expect(resumedSummary.workerState).toBe("ready"); if (resumedSummary.workerPid) { workerPids.add(resumedSummary.workerPid); }
const attached = await client.request({ type: "attach", activeSessionId: resumedSummary.activeSessionId ?? resumedSummary.id, }); expect(attached.success).toBe(true);
await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); if (resumedSummary.workerPid) { await waitForProcessGone(resumedSummary.workerPid); workerPids.delete(resumedSummary.workerPid); } });
it("does not resurrect an intentionally stopped root when the supervisor dies during kill", { tags: ["process-stress"], timeout: 30_000, }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-stop-race-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "stop me", timestamp: 1 }); sessionManager.appendSessionState({ status: "active" }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); }
const firstSupervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, firstSupervisor); const firstSupervisorPid = client.hello?.supervisorPid; if (!firstSupervisorPid) { throw new Error("Daemon hello did not expose its supervisor pid"); } const created = await client.request({ type: "create", sessionPath: sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) { throw new Error(created.error); } const summary = requireSummary(created.data); if (!summary.workerPid) { throw new Error("Resident worker did not expose its pid"); } workerPids.add(summary.workerPid); const activeSessionId = summary.activeSessionId ?? summary.id; const heartbeatResponse = await client.request({ type: "heartbeat_set", activeSessionId, schedule: "every 1h", prompt: "continue old work", }); if (!heartbeatResponse.success || !heartbeatResponse.data || typeof heartbeatResponse.data !== "object") { throw new Error(heartbeatResponse.success ? "Heartbeat response was missing data" : heartbeatResponse.error); } const heartbeat = (heartbeatResponse.data as { heartbeat: { id: string } }).heartbeat; const cronStore = AgentCronJobStore.forSessionArtifacts(); cronStore.registerSessionArtifact(summary.sessionId, sessionManager.getSessionArtifactDir()!); process.kill(summary.workerPid, "SIGSTOP"); const killResult = client.request({ type: "kill", activeSessionId }).catch((error: unknown) => error); const tombstone = await waitForWorkerStopTombstone(agentDir); expect(tombstone.stopRequestedAt).toEqual(expect.any(String)); expect(tombstone.archiveOnStop).toBe(true);
process.kill(firstSupervisorPid, "SIGKILL"); await waitForExit(firstSupervisor); children.delete(firstSupervisor); client.close(); await expect(killResult).resolves.toBeInstanceOf(Error);
const replacementSupervisor = spawnSupervisor(agentDir, socketPath, projectDir); const replacementClient = await connectEventually(socketPath, replacementSupervisor); const listed = await replacementClient.request({ type: "list" }); expect(listed.success).toBe(true); const sessions = requireSessionList(listed.success ? listed.data : undefined); expect(sessions.filter((session) => session.activeSessionId || session.workerPid)).toEqual([]); await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); await waitForCondition( () => countWorkerDescriptors(agentDir) === 0, "Intentional worker stop descriptor was not removed", ); expect(countWorkerDescriptors(agentDir)).toBe(0); expect((await readSessionInfo(sessionFile))?.state).toEqual({ status: "archived" }); expect(cronStore.list().find((job) => job.id === heartbeat.id)).toMatchObject({ status: "cancelled" });
await replacementClient.request({ type: "shutdown" }); replacementClient.close(); await waitForSocketGone(socketPath); });
it("hosts and adopts isolated worker processes", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-smoke-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionFiles = Array.from({ length: 2 }, (_, index) => { const manager = SessionManager.create(projectDir, sessionDir); manager.appendMessage({ role: "user", content: `smoke root ${index}`, timestamp: index + 1 }); const sessionFile = manager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); } return sessionFile; });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await Promise.all( sessionFiles.map((sessionPath) => client.request({ type: "create", sessionPath, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }), ), ); const summaries = created.map((response) => { if (!response.success) { throw new Error(response.error); } return requireSummary(response.data); }); const pids = summaries.map((summary) => summary.workerPid); expect(new Set(pids).size).toBe(2); expect(pids).not.toContain(supervisor.pid); for (const pid of pids) { if (!pid) { throw new Error("Resident root did not expose a worker pid"); } workerPids.add(pid); }
const listed = await client.request({ type: "list" }); expect(listed.success).toBe(true); expect(requireSessionList(listed.success ? listed.data : undefined)).toHaveLength(2); supervisor.kill("SIGTERM"); await waitForExit(supervisor); children.delete(supervisor); client.close();
const replacementClient = await connectEventually(socketPath); const adopted = await replacementClient.request({ type: "list" }); expect(adopted.success).toBe(true); expect( new Set(requireSessionList(adopted.success ? adopted.data : undefined).map((summary) => summary.workerPid)), ).toEqual(new Set(pids)); await replacementClient.request({ type: "shutdown" }); replacementClient.close(); await waitForSocketGone(socketPath); await Promise.all( pids.map(async (pid) => { if (pid) { await waitForProcessGone(pid); workerPids.delete(pid); } }), ); }, 60_000);
it("hosts resident roots in isolated worker processes without a session cap", { tags: ["process-stress"], timeout: 180_000, }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-many-roots-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionFiles = Array.from({ length: PROCESS_STRESS_WORKERS }, (_, index) => { const manager = SessionManager.create(projectDir, sessionDir); manager.appendMessage({ role: "user", content: `root ${index}`, timestamp: index + 1 }); const sessionFile = manager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); } return sessionFile; });
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const externalLease = acquireSessionLease(sessionFiles[0], agentDir, { [SESSION_LEASES_ENABLED_ENV]: "1", [SESSION_LEASE_OWNER_ID_ENV]: "external-owner", }); const conflict = await client.request({ type: "create", sessionPath: sessionFiles[0], config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); expect(conflict).toMatchObject({ success: false, errorInfo: { code: "session_already_active", activeSessionId: "external-owner" }, }); const emptyAfterConflict = await client.request({ type: "list" }); expect(requireSessionList(emptyAfterConflict.success ? emptyAfterConflict.data : undefined)).toHaveLength(0); externalLease?.release(); const created = await Promise.all( sessionFiles.map((sessionPath) => client.request({ type: "create", sessionPath, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }), ), ); const summaries = created.map((response) => { if (!response.success) { throw new Error(response.error); } return requireSummary(response.data); }); const pids = summaries.map((summary) => summary.workerPid); expect(new Set(pids).size).toBe(PROCESS_STRESS_WORKERS); expect(pids).not.toContain(supervisor.pid); for (const pid of pids) { if (!pid) { throw new Error("Resident root did not expose a worker pid"); } workerPids.add(pid); } const firstActiveSessionId = summaries[0]!.activeSessionId ?? summaries[0]!.id; const addedCron = await client.request({ type: "cron_add", activeSessionId: firstActiveSessionId, schedule: "every 1h", prompt: "check status", }); expect(addedCron.success).toBe(true); const cronJob = (addedCron.success ? addedCron.data : undefined) as { job?: { id?: string } } | undefined; if (!cronJob?.job?.id) { throw new Error("Supervisor did not persist the cron job"); } const listedCron = await client.request({ type: "cron_list", activeSessionId: firstActiveSessionId }); expect(listedCron).toMatchObject({ success: true, data: { jobs: [{ id: cronJob.job.id }] } }); const cancelledCron = await client.request({ type: "cron_cancel", jobId: cronJob.job.id }); expect(cancelledCron.success).toBe(true); const activeSessionIds = summaries.map((summary) => summary.activeSessionId ?? summary.id); await Promise.all( activeSessionIds.map((activeSessionId, index) => startBlockingBash(client, activeSessionId, join(root, `stress-blocker-${index}.ready`)), ), ); const heartbeats = await Promise.all( activeSessionIds.map((activeSessionId, index) => client.request({ type: "heartbeat_set", activeSessionId, schedule: "every 10s", prompt: `heartbeat ${index}`, }), ), ); expect(heartbeats.every((response) => response.success)).toBe(true); await waitForCondition( () => { const stores = summaries.map((summary, index) => { const store = AgentCronJobStore.forSessionArtifacts(); store.registerSessionArtifact( summary.sessionId, join(dirname(dirname(sessionFiles[index]!)), "session-artifacts", summary.sessionId), ); return store; }); return stores.every((store) => store.list().some((job) => job.lastSkippedAt !== undefined)); }, "Session workers did not advance their heartbeats independently", 15_000, );
const listed = await client.request({ type: "list" }); expect(listed.success).toBe(true); expect(requireSessionList(listed.success ? listed.data : undefined)).toHaveLength(PROCESS_STRESS_WORKERS); supervisor.kill("SIGTERM"); await waitForExit(supervisor); children.delete(supervisor); client.close(); const replacementClient = await connectEventually(socketPath); const adopted = await replacementClient.request({ type: "list" }); expect(adopted.success).toBe(true); expect( new Set(requireSessionList(adopted.success ? adopted.data : undefined).map((summary) => summary.workerPid)), ).toEqual(new Set(pids)); await replacementClient.request({ type: "shutdown" }); replacementClient.close(); await waitForSocketGone(socketPath); await Promise.all( pids.map(async (pid) => { if (pid) { await waitForProcessGone(pid); workerPids.delete(pid); } }), ); });
it("isolates a root, streams a chunked snapshot, and adopts the same worker after restart", async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-supervisor-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); const largePrompt = `large:${"x".repeat(600 * 1024)}`; sessionManager.appendMessage({ role: "user", content: largePrompt, timestamp: 1 }); sessionManager.appendMessage({ role: "assistant", content: [{ type: "text", text: "done" }], api: "openai-responses", provider: "faux", model: "faux", usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason: "stop", timestamp: 2, }); const sessionFile = sessionManager.getSessionFile(); if (!sessionFile) { throw new Error("Fixture session did not persist"); }
const firstSupervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, firstSupervisor); const created = await client.request({ type: "create", sessionPath: sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true, }, }); if (!created.success) { const diagnostics = childDiagnostics.get(firstSupervisor); throw new Error( `${created.error}\nsupervisor stdout:\n${diagnostics?.stdout ?? ""}\nsupervisor stderr:\n${diagnostics?.stderr ?? ""}\n${readDaemonLogs(agentDir)}`, ); } expect(created.success).toBe(true); const createdSummary = requireSummary(created.data); expect(createdSummary.workerState).toBe("ready"); expect(createdSummary.workerPid).not.toBe(firstSupervisor.pid); if (!createdSummary.workerPid) { throw new Error("Resident worker did not expose its pid"); } workerPids.add(createdSummary.workerPid);
const connection = await DaemonAgentConnection.attach( client, createdSummary.activeSessionId ?? createdSummary.id, { recoverDaemon: async () => {} }, ); const connectionEvents: string[] = []; const replacementMessageCounts: number[] = []; connection.subscribe((event) => { connectionEvents.push(event.type === "connection_status" ? `${event.type}:${event.status}` : event.type); if (event.type === "session_replaced") { // Count conversation messages only; every session carries a harness digest. replacementMessageCounts.push( event.messages.filter( (message) => !( message.role === "custom" && (message as { customType?: string }).customType === "harness_digest" ), ).length, ); } }); const snapshot = await connection.getInitialSnapshot(); const snapshotConversation = snapshot.messages.filter( (message) => !(message.role === "custom" && (message as { customType?: string }).customType === "harness_digest"), ); expect(snapshotConversation).toHaveLength(2); expect(snapshotConversation[0]).toMatchObject({ role: "user", content: largePrompt });
const activeSessionId = createdSummary.activeSessionId ?? createdSummary.id; const createdNew = await client.request({ type: "new_session", activeSessionId }); expect(createdNew.success).toBe(true); const emptyReplacementDeadline = Date.now() + 5000; while (!replacementMessageCounts.includes(0) && Date.now() < emptyReplacementDeadline) { await new Promise((resolveDelay) => setTimeout(resolveDelay, 10)); } expect(replacementMessageCounts).toContain(0); const switchedBack = await client.request({ type: "switch_session", activeSessionId, sessionPath: sessionFile }); expect(switchedBack.success).toBe(true); const restoredReplacementDeadline = Date.now() + 5000; while (replacementMessageCounts.at(-1) !== 2 && Date.now() < restoredReplacementDeadline) { await new Promise((resolveDelay) => setTimeout(resolveDelay, 10)); } expect(replacementMessageCounts.at(-1)).toBe(2);
firstSupervisor.kill("SIGTERM"); await waitForExit(firstSupervisor); children.delete(firstSupervisor);
const reconnectDeadline = Date.now() + 15_000; while (!connectionEvents.includes("connection_status:connected") && Date.now() < reconnectDeadline) { await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } expect(connectionEvents).toContain("connection_status:reconnecting"); // The direct worker link held through the supervisor swap, so no resync is warranted. expect(connectionEvents).not.toContain("session_resynced"); expect(connectionEvents).toContain("connection_status:connected"); expect(connectionEvents).not.toContain("closed"); await expect(connection.getState()).resolves.toMatchObject({ activeSessionId: createdSummary.activeSessionId, sessionId: createdSummary.sessionId, }); const listed = await client.request({ type: "list", all: true, sessionDir }); expect(listed.success).toBe(true); if (!listed.success) { throw new Error(listed.error); } const adopted = requireSessionList(listed.data).find( (summary) => (summary.activeSessionId ?? summary.id) === (createdSummary.activeSessionId ?? createdSummary.id), ); expect(adopted).toMatchObject({ workerState: "ready", workerPid: createdSummary.workerPid, });
const descriptor = readWorkerDescriptor(agentDir); await startBlockingBash(client, activeSessionId, join(root, "orphan-blocker.ready")); if (!descriptor.orphanProcessJournalPath) { throw new Error("Resident worker did not persist its orphan-process journal path"); } let orphanPids: number[] = []; const orphanDeadline = Date.now() + 5000; while (orphanPids.length === 0 && Date.now() < orphanDeadline) { orphanPids = readActiveOrphanProcesses(descriptor.orphanProcessJournalPath, descriptor.pid).map( (orphan) => orphan.pid, ); if (orphanPids.length === 0) { await new Promise((resolveDelay) => setTimeout(resolveDelay, 10)); } } expect(orphanPids.length).toBeGreaterThan(0); for (const pid of orphanPids) { workerPids.add(pid); } process.kill(createdSummary.workerPid, "SIGKILL"); await waitForProcessGone(createdSummary.workerPid); workerPids.delete(createdSummary.workerPid); await Promise.all(orphanPids.map((pid) => waitForProcessGone(pid))); for (const pid of orphanPids) { workerPids.delete(pid); }
let failed: SessionSummary | undefined; const recoveryDeadline = Date.now() + 20_000; while (Date.now() < recoveryDeadline) { const response = await client.request({ type: "list" }); if (response.success) { failed = requireSessionList(response.data).find( (summary) => (summary.activeSessionId ?? summary.id) === (createdSummary.activeSessionId ?? createdSummary.id), ); if (failed?.workerState === "failed") break; } await new Promise((resolveDelay) => setTimeout(resolveDelay, 50)); } expect(failed).toMatchObject({ workerState: "failed", activeSessionId: createdSummary.activeSessionId }); await connection.dispose();
const reopened = await client.request({ type: "create", sessionPath: sessionFile, continueRecent: false, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, launchEnv: { PRIME_AGENT_TEST_FRESH_CONTEXT: "1" }, }); if (!reopened.success) throw new Error(reopened.error); const recovered = requireSummary(reopened.data); if (!recovered.workerPid) throw new Error("Recovered worker did not expose its pid"); workerPids.add(recovered.workerPid); const recoveredConnection = await DaemonAgentConnection.attach( client, recovered.activeSessionId ?? recovered.id, { recoverDaemon: async () => {} }, ); const recoveredSnapshot = await recoveredConnection.getInitialSnapshot(); expect(recoveredSnapshot.messages).toEqual( expect.arrayContaining([ expect.objectContaining({ role: "user" }), expect.objectContaining({ role: "assistant" }), ]), ); await expect(recoveredConnection.getState()).resolves.toMatchObject({ sessionId: createdSummary.sessionId });
await recoveredConnection.dispose(); await client.request({ type: "shutdown" }); client.close(); await waitForSocketGone(socketPath); await waitForProcessGone(recovered.workerPid); workerPids.delete(recovered.workerPid); });
it("runs a session-artifact cron job while the supervisor is being replaced", { tags: ["process-stress"], timeout: 30_000, }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const sessionDir = join(agentDir, "sessions"); const socketPath = join(tmpdir(), `prime-worker-cron-${process.pid}-${randomUUID().slice(0, 8)}.sock`); mkdirSync(projectDir, { recursive: true }); const sessionManager = SessionManager.create(projectDir, sessionDir); sessionManager.appendMessage({ role: "user", content: "scheduled work", timestamp: 1 }); sessionManager.appendSessionState({ status: "active" }); const sessionFile = sessionManager.getSessionFile(); const artifactDir = sessionManager.getSessionArtifactDir(); if (!sessionFile || !artifactDir) { throw new Error("Fixture session did not persist"); }
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); const client = await connectEventually(socketPath, supervisor); const created = await client.request({ type: "create", sessionPath: sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) { throw new Error(created.error); } const summary = requireSummary(created.data); if (!summary.workerPid) { throw new Error("Resident worker did not expose its pid"); } workerPids.add(summary.workerPid); await startBlockingBash(client, summary.activeSessionId ?? summary.id, join(root, "heartbeat-blocker.ready")); const scheduled = await client.request({ type: "heartbeat_set", activeSessionId: summary.activeSessionId ?? summary.id, schedule: "every 10s", prompt: "continue without the supervisor", }); if (!scheduled.success || !scheduled.data || typeof scheduled.data !== "object") { throw new Error(scheduled.success ? "Heartbeat response was missing its job" : scheduled.error); } const job = (scheduled.data as { heartbeat: { id: string } }).heartbeat;
client.close(); supervisor.kill("SIGKILL"); await waitForExit(supervisor); children.delete(supervisor);
const store = AgentCronJobStore.forSessionArtifacts(); store.registerSessionArtifact(sessionManager.getSessionId(), artifactDir); await waitForCondition( () => store.list().find((candidate) => candidate.id === job.id)?.lastSkippedAt !== undefined, "Resident worker did not advance its heartbeat without the supervisor", 15_000, ); expect(store.list().find((candidate) => candidate.id === job.id)).toBeDefined();
const replacement = await connectEventually(socketPath); await replacement.request({ type: "shutdown" }); replacement.close(); await waitForSocketGone(socketPath); await waitForProcessGone(summary.workerPid); workerPids.delete(summary.workerPid); });
it("exits an orphaned session worker when no replacement supervisor can come up", { timeout: 120_000 }, async () => { const root = tempDir(); const agentDir = join(root, "agent"); const projectDir = join(root, "project"); const registryDir = join(root, "supervisor-registry"); // The socket path must stay short: macOS rejects listen() with EINVAL // past the sun_path limit, and the per-test temp dir nests too deep. const socketDir = mkdtempSync(join(tmpdir(), "prime-orphan-gc-")); tempDirs.push(socketDir); const sessionDir = join(agentDir, "sessions"); const socketPath = join(socketDir, "supervisor.sock"); mkdirSync(projectDir, { recursive: true });
const manager = SessionManager.create(projectDir, sessionDir); manager.appendMessage({ role: "user", content: "orphan gc fixture", timestamp: 1 }); manager.flushNow(); const sessionFile = manager.getSessionFile(); if (!sessionFile) throw new Error("Missing fixture session path");
const supervisor = spawnSupervisor(agentDir, socketPath, projectDir, [], { PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_REGISTRY_DIR: registryDir, PRIME_AGENT_INTERNAL_WORKER_SUPERVISOR_LOST_EXIT_MS: "3000", }); // Cold tsx startup can exceed the shared helper's 15s readiness deadline. let client: DaemonClient | undefined; const readyDeadline = Date.now() + 60_000; while (client === undefined && Date.now() < readyDeadline) { if (supervisor.exitCode !== null || supervisor.signalCode !== null) { const diagnostics = childDiagnostics.get(supervisor); throw new Error( `Supervisor exited before becoming ready\nstdout:\n${diagnostics?.stdout ?? ""}\nstderr:\n${diagnostics?.stderr ?? ""}`, ); } const candidate = new DaemonClient(socketPath); try { await candidate.connect(250); await candidate.waitForHello(1000); client = candidate; } catch { candidate.close(); await new Promise((resolveDelay) => setTimeout(resolveDelay, 250)); } } if (!client) throw new Error("Supervisor did not become ready in time"); const created = await client.request({ type: "create", sessionPath: sessionFile, config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, }); if (!created.success) throw new Error(created.error); const summary = requireSummary(created.data); if (!summary.workerPid) throw new Error("Session worker did not spawn"); workerPids.add(summary.workerPid);
// SIGKILL the registry-recorded supervisor pid: the spawned child can be // a tsx wrapper, and killing the wrapper alone leaves the supervisor // listening on its socket, so the worker would never orphan. const ownerDirectories = readdirSync(registryDir).filter((name) => name.endsWith(".owner")); expect(ownerDirectories).toHaveLength(1); const owner = JSON.parse(readFileSync(join(registryDir, ownerDirectories[0], "owner.json"), "utf8")) as { pid?: number; }; if (!owner.pid) throw new Error("Supervisor owner record is missing its pid"); const supervisorPid = owner.pid; const workerPid = summary.workerPid; client.close();
// A read-only socket directory makes every replacement-launch attempt // fail (lock/bind EACCES): the real-world orphan precondition where a // worker cannot bring its supervisor back. chmodSync(socketDir, 0o555); try { process.kill(supervisorPid, "SIGKILL"); await waitForCondition( () => { try { process.kill(workerPid, 0); return false; } catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; } }, "Orphaned session worker did not exit after the supervisor-lost window", 60_000, ); } finally { chmodSync(socketDir, 0o755); } workerPids.delete(workerPid); });});
describe("issue #879 stable daemon process identity across timezone changes", () => { it("pins the portable process query to UTC across caller timezone changes", () => { const calls: Array<{ command: string; args: string[]; env: NodeJS.ProcessEnv | undefined }> = []; const query = (command: string, args: string[], options?: { env?: NodeJS.ProcessEnv }) => { calls.push({ command, args, env: options?.env }); return options?.env?.TZ === "UTC" ? "Sat Aug 29 20:55:18 2026\n" : "Sat Aug 29 16:55:18 2026\n"; }; const originalTimezone = process.env.TZ; let before: string | undefined; let after: string | undefined; try { process.env.TZ = "America/Los_Angeles"; before = getPsProcessStartId(42, query); process.env.TZ = "America/New_York"; after = getPsProcessStartId(42, query); } finally { if (originalTimezone === undefined) delete process.env.TZ; else process.env.TZ = originalTimezone; }
expect(before).toBe("ps:Sat Aug 29 20:55:18 2026"); expect(after).toBe(before); expect(calls).toHaveLength(2); for (const call of calls) { expect(call).toMatchObject({ command: "ps", args: ["-p", "42", "-o", "lstart="], env: { LC_ALL: "C", LC_TIME: "C", LANG: "C", TZ: "UTC" }, }); } });});