mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-08 19:19:42 +00:00
fix(cursor): close ACP list-models race and false exit 143 window (#1518)
* fix(cursor): close ACP list-models race and false exit 143 window Register the agent-acp-active guard before spawn, hold it until stdio close (not bare exit), record the ACP child PID, and align lock/cache home with resolveHapiHomeDir so runner and session children agree. Richer exit attribution distinguishes live-PID transport disruption from confirmed child death. Fixes residual #1472 after #835. Co-authored-by: Cursor <cursoragent@cursor.com> * test(cursor): isolate ACP guard teardown from ~/.hapi Reset afterEach under the temp HAPI_HOME only, and restore the isolated home before teardown in the unset-HAPI_HOME case, so tests cannot wipe a live agent-acp-active lock. Use distinct child PIDs in registration tests. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cursor): publish ACP lock pid before count Fail-closed reservation order: write pids/<hostPid> before count so concurrent reconcile cannot treat a mid-register lock as stale and clear it for list-models. Keep a short mtime grace only when pids/ is missing (mkdir gap). Regression covers mid-publish readers. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cursor): keep empty pids/ ACP reservation fail-closed Between mkdir(pids) and the host pid writeFile, reconcile could see liveCount=0 and clear the lock. Keep that window when count is still absent and the lock is fresh; re-scan for pids published mid-reconcile. Regression hooks the mkdir/write gap. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cursor): keep ACP lock across last-unregister publish race Write a short-lived registering marker before pids/count so empty pids with leftover count cannot erase a concurrent mid-addLockPid reservation. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(cursor): use per-pid ACP registering markers Crash/reboot must not pin list-models forever on a bare registering file; prune dead owners and only keep live registrar PIDs. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -2,7 +2,17 @@ import { afterEach, describe, expect, test, vi } from 'vitest';
|
||||
|
||||
const guard = vi.hoisted(() => ({
|
||||
register: vi.fn(),
|
||||
unregister: vi.fn()
|
||||
unregister: vi.fn(),
|
||||
recordChildPid: vi.fn(),
|
||||
getLockDir: vi.fn(() => '/tmp/test-hapi/locks/agent-acp-active'),
|
||||
isActive: vi.fn(() => true),
|
||||
describeState: vi.fn((childPid?: number | null) => ({
|
||||
lockDir: '/tmp/test-hapi/locks/agent-acp-active',
|
||||
inProcessCount: 1,
|
||||
childPid: childPid ?? null,
|
||||
childAlive: false,
|
||||
guardActive: true
|
||||
}))
|
||||
}));
|
||||
|
||||
const spawnState = vi.hoisted(() => ({
|
||||
@@ -12,12 +22,21 @@ const spawnState = vi.hoisted(() => ({
|
||||
stdinEnd: vi.fn(),
|
||||
stdinWrite: vi.fn<(chunk: string) => boolean>(() => true),
|
||||
kill: vi.fn(),
|
||||
exitCode: null as number | null
|
||||
exitCode: null as number | null,
|
||||
pid: 424242 as number | undefined,
|
||||
spawnCallOrder: [] as string[]
|
||||
}));
|
||||
|
||||
vi.mock('./agentCliGuard', () => ({
|
||||
registerActiveAcpTransport: guard.register,
|
||||
unregisterActiveAcpTransport: guard.unregister
|
||||
registerActiveAcpTransport: (...args: unknown[]) => {
|
||||
spawnState.spawnCallOrder.push('register');
|
||||
return guard.register(...args);
|
||||
},
|
||||
unregisterActiveAcpTransport: guard.unregister,
|
||||
recordActiveAcpChildPid: guard.recordChildPid,
|
||||
getAgentAcpLockDir: guard.getLockDir,
|
||||
isAgentAcpTransportActive: guard.isActive,
|
||||
describeAgentAcpGuardState: guard.describeState
|
||||
}));
|
||||
|
||||
vi.mock('@/utils/process', () => ({
|
||||
@@ -26,11 +45,15 @@ vi.mock('@/utils/process', () => ({
|
||||
|
||||
vi.mock('node:child_process', () => ({
|
||||
spawn: vi.fn(() => {
|
||||
spawnState.spawnCallOrder.push('spawn');
|
||||
spawnState.exitHandlers = [];
|
||||
spawnState.closeHandlers = [];
|
||||
spawnState.stdoutDataHandlers = [];
|
||||
const handlers = new Map<string, Array<(...args: unknown[]) => void>>();
|
||||
const proc = {
|
||||
get pid() {
|
||||
return spawnState.pid;
|
||||
},
|
||||
get exitCode() {
|
||||
return spawnState.exitCode;
|
||||
},
|
||||
@@ -81,12 +104,25 @@ describe('AcpStdioTransport agent CLI guard', () => {
|
||||
afterEach(() => {
|
||||
guard.register.mockClear();
|
||||
guard.unregister.mockClear();
|
||||
guard.recordChildPid.mockClear();
|
||||
guard.getLockDir.mockClear();
|
||||
guard.isActive.mockClear();
|
||||
guard.describeState.mockClear();
|
||||
guard.describeState.mockImplementation((childPid?: number | null) => ({
|
||||
lockDir: '/tmp/test-hapi/locks/agent-acp-active',
|
||||
inProcessCount: 1,
|
||||
childPid: childPid ?? null,
|
||||
childAlive: false,
|
||||
guardActive: true
|
||||
}));
|
||||
spawnState.stdinWrite.mockReset();
|
||||
spawnState.stdinWrite.mockReturnValue(true);
|
||||
spawnState.stdinEnd.mockClear();
|
||||
spawnState.kill.mockClear();
|
||||
vi.mocked(killProcessByChildProcess).mockClear();
|
||||
spawnState.exitCode = null;
|
||||
spawnState.pid = 424242;
|
||||
spawnState.spawnCallOrder = [];
|
||||
spawnState.exitHandlers = [];
|
||||
spawnState.closeHandlers = [];
|
||||
spawnState.stdoutDataHandlers = [];
|
||||
@@ -95,16 +131,45 @@ describe('AcpStdioTransport agent CLI guard', () => {
|
||||
test('registers cross-process guard only for Cursor agent command', async () => {
|
||||
const transport = new AcpStdioTransport({ command: 'agent', args: ['acp'] });
|
||||
expect(guard.register).toHaveBeenCalledTimes(1);
|
||||
expect(guard.recordChildPid).toHaveBeenCalledWith(424242);
|
||||
await transport.close();
|
||||
expect(guard.unregister).toHaveBeenCalledTimes(1);
|
||||
expect(guard.unregister).toHaveBeenCalledWith({ childPid: 424242 });
|
||||
});
|
||||
|
||||
test('registers the ACP guard before spawn so list-models cannot race the new child', () => {
|
||||
spawnState.spawnCallOrder = [];
|
||||
new AcpStdioTransport({ command: 'agent', args: ['acp'] });
|
||||
expect(spawnState.spawnCallOrder.indexOf('register')).toBeGreaterThanOrEqual(0);
|
||||
expect(spawnState.spawnCallOrder.indexOf('spawn')).toBeGreaterThan(
|
||||
spawnState.spawnCallOrder.indexOf('register')
|
||||
);
|
||||
});
|
||||
|
||||
test('keeps the ACP guard held across exit until close drains stdio', () => {
|
||||
new AcpStdioTransport({ command: 'agent', args: ['acp'] });
|
||||
guard.unregister.mockClear();
|
||||
|
||||
for (const handler of spawnState.exitHandlers) {
|
||||
handler(143, null);
|
||||
}
|
||||
expect(guard.unregister).not.toHaveBeenCalled();
|
||||
|
||||
for (const handler of spawnState.closeHandlers) {
|
||||
handler(143, null);
|
||||
}
|
||||
expect(guard.unregister).toHaveBeenCalledTimes(1);
|
||||
expect(guard.unregister).toHaveBeenCalledWith({ childPid: 424242 });
|
||||
});
|
||||
|
||||
test('does not register guard for non-agent ACP backends', () => {
|
||||
for (const command of ['gemini', 'opencode', 'kimi']) {
|
||||
guard.register.mockClear();
|
||||
guard.unregister.mockClear();
|
||||
guard.recordChildPid.mockClear();
|
||||
new AcpStdioTransport({ command });
|
||||
expect(guard.register).not.toHaveBeenCalled();
|
||||
expect(guard.recordChildPid).not.toHaveBeenCalled();
|
||||
expect(guard.unregister).not.toHaveBeenCalled();
|
||||
}
|
||||
});
|
||||
@@ -258,7 +323,7 @@ describe('AcpStdioTransport closed stdin writes', () => {
|
||||
}
|
||||
|
||||
await expect(transport.sendRequest('session/load')).rejects.toThrow(
|
||||
/ACP process exited \(code=1, signal=null\)\. stderr: Cannot use this model: grok-4\.5\[fast=true\]/
|
||||
/ACP process exited \(code=1, signal=null(?:, childPid=\d+, lock=[^)]+)?\)\. stderr: Cannot use this model: grok-4\.5\[fast=true\]/
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -2,7 +2,13 @@ import { spawn, type ChildProcessWithoutNullStreams, type SpawnOptions } from 'n
|
||||
import { logger } from '@/ui/logger';
|
||||
import { killProcessByChildProcess } from '@/utils/process';
|
||||
import { GEMINI_MODEL_PRESETS } from '@hapi/protocol';
|
||||
import { registerActiveAcpTransport, unregisterActiveAcpTransport } from './agentCliGuard';
|
||||
import {
|
||||
describeAgentAcpGuardState,
|
||||
getAgentAcpLockDir,
|
||||
recordActiveAcpChildPid,
|
||||
registerActiveAcpTransport,
|
||||
unregisterActiveAcpTransport
|
||||
} from './agentCliGuard';
|
||||
import { matchesAcpHttp2Cancel, matchesAcpRetryBackoff } from './acpStderrErrors';
|
||||
|
||||
interface JsonRpcRequest {
|
||||
@@ -73,6 +79,8 @@ export class AcpStdioTransport {
|
||||
/** True after process 'exit'; blocks new writes until 'close' drains stderr. */
|
||||
private exited = false;
|
||||
private exitError: Error | null = null;
|
||||
/** ACP child PID when known (for lock attribution / exit logs). */
|
||||
private childPid: number | null = null;
|
||||
|
||||
/** Rolling join window for stderr before close-time classification. */
|
||||
private static readonly RECENT_STDERR_WINDOW = 8_000;
|
||||
@@ -85,6 +93,12 @@ export class AcpStdioTransport {
|
||||
env?: Record<string, string>;
|
||||
}) {
|
||||
this.shouldGuardAgentCli = options.command === 'agent';
|
||||
// Register before spawn so runner/list-models cannot observe an unlocked
|
||||
// window between process creation and lock write (#1472).
|
||||
if (this.shouldGuardAgentCli) {
|
||||
registerActiveAcpTransport();
|
||||
}
|
||||
|
||||
this.process = spawn(
|
||||
options.command,
|
||||
options.args ?? [],
|
||||
@@ -92,7 +106,12 @@ export class AcpStdioTransport {
|
||||
) as ChildProcessWithoutNullStreams;
|
||||
|
||||
if (this.shouldGuardAgentCli) {
|
||||
registerActiveAcpTransport();
|
||||
const childPid = typeof this.process.pid === 'number' ? this.process.pid : null;
|
||||
this.childPid = childPid;
|
||||
if (childPid !== null) {
|
||||
recordActiveAcpChildPid(childPid);
|
||||
}
|
||||
logger.debug('[ACP] agent CLI guard armed', describeAgentAcpGuardState(childPid));
|
||||
}
|
||||
|
||||
this.process.stdout.setEncoding('utf8');
|
||||
@@ -128,12 +147,24 @@ export class AcpStdioTransport {
|
||||
|
||||
// Block new stdin writes as soon as the process exits, but defer markClosed
|
||||
// until 'close' so final stderr chunks can still enrich the failure.
|
||||
// Do NOT release the agent CLI guard here — exit→close is exactly when
|
||||
// list-models can race another `agent` and SIGTERM remaining ACP children.
|
||||
this.process.on('exit', (code, signal) => {
|
||||
this.releaseAgentCliGuard();
|
||||
this.exited = true;
|
||||
this.exitError = new Error(
|
||||
`ACP process exited (code=${code ?? 'null'}, signal=${signal ?? 'null'})`
|
||||
);
|
||||
const attribution = this.formatExitAttribution(code, signal);
|
||||
const guardState = describeAgentAcpGuardState(this.childPid);
|
||||
logger.debug(`[ACP] process exit ${attribution}`, guardState);
|
||||
if (guardState.childAlive === true) {
|
||||
// Node reported exit, but the recorded ACP PID is still alive —
|
||||
// likely a Cursor-internal worker/stdio quirk. Do not claim a
|
||||
// definitive process death in the error string operators grep.
|
||||
this.exitError = new Error(
|
||||
`ACP transport reported exit (${attribution}) but OS PID ${this.childPid} is still alive ` +
|
||||
`(lock=${getAgentAcpLockDir()}); treating as transport disruption, not confirmed child death`
|
||||
);
|
||||
} else {
|
||||
this.exitError = new Error(`ACP process exited (${attribution})`);
|
||||
}
|
||||
});
|
||||
|
||||
// Use 'close' (not only 'exit') so final stderr chunks are drained before we
|
||||
@@ -141,12 +172,17 @@ export class AcpStdioTransport {
|
||||
this.process.on('close', (code, signal) => {
|
||||
this.releaseAgentCliGuard();
|
||||
this.flushStderrParseBuffer();
|
||||
const attribution = this.formatExitAttribution(code, signal);
|
||||
const guardState = describeAgentAcpGuardState(this.childPid);
|
||||
const stderr = this.stderrForCloseError();
|
||||
let message = `ACP process exited (code=${code ?? 'null'}, signal=${signal ?? 'null'})`;
|
||||
let message = guardState.childAlive === true
|
||||
? `ACP transport closed (${attribution}) but OS PID ${this.childPid} is still alive ` +
|
||||
`(lock=${getAgentAcpLockDir()})`
|
||||
: `ACP process exited (${attribution})`;
|
||||
if (stderr) {
|
||||
message = `${message}. stderr: ${stderr}`;
|
||||
}
|
||||
logger.debug(message);
|
||||
logger.debug(message, guardState);
|
||||
const error = new Error(message);
|
||||
if (stderr) {
|
||||
(error as Error & { stderr?: string }).stderr = stderr;
|
||||
@@ -249,12 +285,23 @@ export class AcpStdioTransport {
|
||||
this.markClosed(new Error('ACP transport closed'));
|
||||
}
|
||||
|
||||
private formatExitAttribution(code: number | null, signal: NodeJS.Signals | null): string {
|
||||
const base = `code=${code ?? 'null'}, signal=${signal ?? 'null'}`;
|
||||
if (!this.shouldGuardAgentCli) {
|
||||
return base;
|
||||
}
|
||||
const child = this.childPid ?? this.process.pid ?? 'unknown';
|
||||
return `${base}, childPid=${child}, lock=${getAgentAcpLockDir()}`;
|
||||
}
|
||||
|
||||
private releaseAgentCliGuard(): void {
|
||||
if (!this.shouldGuardAgentCli || this.guardReleased) {
|
||||
return;
|
||||
}
|
||||
this.guardReleased = true;
|
||||
unregisterActiveAcpTransport();
|
||||
unregisterActiveAcpTransport(
|
||||
this.childPid !== null ? { childPid: this.childPid } : undefined
|
||||
);
|
||||
}
|
||||
|
||||
private handleStdout(chunk: string): void {
|
||||
|
||||
@@ -1,10 +1,15 @@
|
||||
import { existsSync, mkdirSync, writeFileSync } from 'node:fs';
|
||||
import { existsSync, mkdirSync, readFileSync, readdirSync, utimesSync, writeFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { homedir, tmpdir } from 'node:os';
|
||||
import { afterEach, describe, expect, test } from 'vitest';
|
||||
import {
|
||||
_resetAgentCliGuardForTests,
|
||||
_setActiveAcpTransportCountForTests,
|
||||
_setAddLockPidHookForTests,
|
||||
_setRegisterPublishHookForTests,
|
||||
getAgentAcpLockDir,
|
||||
isAgentAcpTransportActive,
|
||||
recordActiveAcpChildPid,
|
||||
registerActiveAcpTransport,
|
||||
unregisterActiveAcpTransport
|
||||
} from './agentCliGuard';
|
||||
@@ -34,6 +39,9 @@ describe('agentCliGuard', () => {
|
||||
const previousHome = process.env.HAPI_HOME;
|
||||
|
||||
afterEach(() => {
|
||||
// Always tear down under the isolated test home — never while HAPI_HOME
|
||||
// is unset (that would resolve ~/.hapi and could wipe a live ACP guard).
|
||||
process.env.HAPI_HOME = testHome;
|
||||
_resetAgentCliGuardForTests();
|
||||
if (previousHome === undefined) {
|
||||
delete process.env.HAPI_HOME;
|
||||
@@ -98,11 +106,25 @@ describe('agentCliGuard', () => {
|
||||
const dir = lockDir();
|
||||
mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(join(dir, 'count'), '1', 'utf8');
|
||||
// Age the lock past the pre-spawn grace so missing pids is truly stale.
|
||||
const aged = Date.now() - 60_000;
|
||||
utimesSync(dir, aged / 1000, aged / 1000);
|
||||
|
||||
expect(isAgentAcpTransportActive()).toBe(false);
|
||||
expect(existsSync(dir)).toBe(false);
|
||||
});
|
||||
|
||||
test('keeps a fresh count-without-pids lock fail-closed during pre-spawn grace', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
const dir = lockDir();
|
||||
mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(join(dir, 'count'), '1', 'utf8');
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(dir)).toBe(true);
|
||||
});
|
||||
|
||||
test('clears refcount lock when all pid entries are stale', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
writeTestAcpLock({ count: 2, pids: [99999998, 99999999] });
|
||||
@@ -118,4 +140,160 @@ describe('agentCliGuard', () => {
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(lockDir())).toBe(true);
|
||||
});
|
||||
|
||||
test('records the ACP child PID when provided, not only the HAPI host PID', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
// Distinct from process.pid so host + child markers are both asserted.
|
||||
const childPid = process.pid + 1_000_000;
|
||||
registerActiveAcpTransport({ childPid });
|
||||
|
||||
const dir = lockDir();
|
||||
expect(existsSync(join(dir, 'pids', String(process.pid)))).toBe(true);
|
||||
expect(existsSync(join(dir, 'pids', String(childPid)))).toBe(true);
|
||||
expect(readFileSync(join(dir, 'child-pid'), 'utf8').trim()).toBe(String(childPid));
|
||||
|
||||
unregisterActiveAcpTransport({ childPid });
|
||||
expect(existsSync(dir)).toBe(false);
|
||||
});
|
||||
|
||||
test('recordActiveAcpChildPid upgrades a pre-spawn reservation to the real child PID', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
registerActiveAcpTransport();
|
||||
const childPid = process.pid + 1_000_001;
|
||||
recordActiveAcpChildPid(childPid);
|
||||
|
||||
const dir = lockDir();
|
||||
expect(existsSync(join(dir, 'pids', String(process.pid)))).toBe(true);
|
||||
expect(existsSync(join(dir, 'pids', String(childPid)))).toBe(true);
|
||||
expect(readFileSync(join(dir, 'child-pid'), 'utf8').trim()).toBe(String(childPid));
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
|
||||
unregisterActiveAcpTransport({ childPid });
|
||||
expect(isAgentAcpTransportActive()).toBe(false);
|
||||
});
|
||||
|
||||
test('uses ~/.hapi lock home when HAPI_HOME is unset (not /tmp/hapi)', () => {
|
||||
delete process.env.HAPI_HOME;
|
||||
try {
|
||||
const expected = join(homedir(), '.hapi', 'locks', 'agent-acp-active');
|
||||
expect(getAgentAcpLockDir()).toBe(expected);
|
||||
expect(getAgentAcpLockDir()).not.toContain(join(tmpdir(), 'hapi'));
|
||||
} finally {
|
||||
// Restore isolated home before afterEach reset (belt + suspenders).
|
||||
process.env.HAPI_HOME = testHome;
|
||||
}
|
||||
});
|
||||
|
||||
test('publishes host pid marker before count so mid-register readers stay active', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
const steps: string[] = [];
|
||||
_setRegisterPublishHookForTests((step) => {
|
||||
steps.push(step);
|
||||
if (step === 'after-host-pid') {
|
||||
// Cross-process reader: no in-process reservation yet for them.
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
expect(existsSync(join(lockDir(), 'pids', String(process.pid)))).toBe(true);
|
||||
expect(existsSync(join(lockDir(), 'count'))).toBe(false);
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(lockDir())).toBe(true);
|
||||
}
|
||||
if (step === 'after-mkdir') {
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
// Grace keeps the mkdir-only reservation fail-closed.
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(lockDir())).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
registerActiveAcpTransport();
|
||||
expect(steps).toEqual(['after-mkdir', 'after-host-pid', 'after-count']);
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
_setRegisterPublishHookForTests(null);
|
||||
unregisterActiveAcpTransport();
|
||||
});
|
||||
|
||||
test('host-pid-without-count reservation is not cleared as stale by reconcile', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
const dir = lockDir();
|
||||
mkdirSync(join(dir, 'pids'), { recursive: true });
|
||||
writeFileSync(join(dir, 'pids', String(process.pid)), String(process.pid), 'utf8');
|
||||
// No count file — the old race window after count-before-pids, inverted.
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(dir)).toBe(true);
|
||||
expect(existsSync(join(dir, 'pids', String(process.pid)))).toBe(true);
|
||||
});
|
||||
|
||||
test('empty pids/ mid-addLockPid stays active for concurrent readers', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
let sawEmptyPids = false;
|
||||
_setAddLockPidHookForTests((phase) => {
|
||||
if (phase !== 'after-pids-mkdir') {
|
||||
return;
|
||||
}
|
||||
sawEmptyPids = true;
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
const dir = lockDir();
|
||||
expect(existsSync(join(dir, 'pids'))).toBe(true);
|
||||
expect(existsSync(join(dir, 'count'))).toBe(false);
|
||||
expect(readdirSync(join(dir, 'pids'))).toEqual([]);
|
||||
expect(existsSync(join(dir, 'registering', String(process.pid)))).toBe(true);
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(dir)).toBe(true);
|
||||
});
|
||||
|
||||
registerActiveAcpTransport();
|
||||
expect(sawEmptyPids).toBe(true);
|
||||
_setAddLockPidHookForTests(null);
|
||||
unregisterActiveAcpTransport();
|
||||
expect(isAgentAcpTransportActive()).toBe(false);
|
||||
});
|
||||
|
||||
test('last unregister does not erase concurrent mid-addLockPid registration', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
registerActiveAcpTransport();
|
||||
expect(readFileSync(join(lockDir(), 'count'), 'utf8')).toBe('1');
|
||||
|
||||
let sawRace = false;
|
||||
_setAddLockPidHookForTests((phase) => {
|
||||
if (phase !== 'after-pids-mkdir') {
|
||||
return;
|
||||
}
|
||||
sawRace = true;
|
||||
// Prior transport's last unregister while the new registrar has
|
||||
// empty-or-about-to-rewrite pids/ and a live `registering/<pid>`.
|
||||
// Force last-unregister semantics (in-process count → 0).
|
||||
_setActiveAcpTransportCountForTests(1);
|
||||
unregisterActiveAcpTransport();
|
||||
expect(existsSync(join(lockDir(), 'registering', String(process.pid)))).toBe(true);
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
expect(existsSync(lockDir())).toBe(true);
|
||||
});
|
||||
|
||||
registerActiveAcpTransport();
|
||||
expect(sawRace).toBe(true);
|
||||
_setAddLockPidHookForTests(null);
|
||||
_setActiveAcpTransportCountForTests(1);
|
||||
expect(existsSync(join(lockDir(), 'pids', String(process.pid)))).toBe(true);
|
||||
expect(existsSync(join(lockDir(), 'registering', String(process.pid)))).toBe(false);
|
||||
expect(isAgentAcpTransportActive()).toBe(true);
|
||||
unregisterActiveAcpTransport();
|
||||
expect(isAgentAcpTransportActive()).toBe(false);
|
||||
});
|
||||
|
||||
test('prunes crash-stale registering/<deadPid> so list-models is not pinned', () => {
|
||||
process.env.HAPI_HOME = testHome;
|
||||
const dir = lockDir();
|
||||
mkdirSync(join(dir, 'registering'), { recursive: true });
|
||||
mkdirSync(join(dir, 'pids'), { recursive: true });
|
||||
writeFileSync(join(dir, 'count'), '1', 'utf8');
|
||||
// Unlikely-to-be-alive PID — marker left by SIGKILL mid-publish.
|
||||
writeFileSync(join(dir, 'registering', '999999'), '1', 'utf8');
|
||||
_setActiveAcpTransportCountForTests(0);
|
||||
|
||||
expect(isAgentAcpTransportActive()).toBe(false);
|
||||
expect(existsSync(dir)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -4,10 +4,11 @@ import {
|
||||
readdirSync,
|
||||
readFileSync,
|
||||
rmSync,
|
||||
statSync,
|
||||
writeFileSync
|
||||
} from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { resolveHapiHomeDir } from '@/configuration';
|
||||
|
||||
/**
|
||||
* Cursor's `agent` CLI appears to allow only one active process at a time.
|
||||
@@ -16,18 +17,115 @@ import { tmpdir } from 'node:os';
|
||||
*
|
||||
* In-process ref counting covers RPC handlers in the same process; a HAPI_HOME
|
||||
* lock directory covers runner vs session child processes.
|
||||
*
|
||||
* Prefer recording the ACP child PID (not only the HAPI host PID) so stale
|
||||
* cleanup and logs attribute the real `agent` process. Register the lock
|
||||
* before spawn, and keep it held until stdio `close` — releasing on bare
|
||||
* `exit` opens a window where list-models can start another `agent`.
|
||||
*
|
||||
* Filesystem publish order is fail-closed: host PID marker under `pids/` is
|
||||
* written before `count`, so concurrent reconcile never sees a lock with no
|
||||
* pids and clears it mid-reservation. Per-host `registering/<pid>` markers
|
||||
* cover the mkdir→pid gap even when a prior transport left a positive
|
||||
* `count` (last-unregister vs concurrent register); dead-owner markers are
|
||||
* pruned so a crash cannot pin list-models forever. Mtime grace is a
|
||||
* backstop for the tiny window before that marker lands.
|
||||
*/
|
||||
let activeAcpTransportCount = 0;
|
||||
|
||||
/** @internal Test hook fired between register publish steps. */
|
||||
let registerPublishHook: ((step: 'after-mkdir' | 'after-host-pid' | 'after-count') => void) | null = null;
|
||||
|
||||
/** @internal Test hook inside addLockPid (mkdir vs write gap). */
|
||||
let addLockPidHook: ((phase: 'after-pids-mkdir' | 'after-pid-write') => void) | null = null;
|
||||
|
||||
/** Fail-closed window while mkdir → first pid file is in flight. */
|
||||
const PRESPAWN_RESERVATION_GRACE_MS = 5_000;
|
||||
|
||||
const REGISTERING_MARKER = 'registering';
|
||||
|
||||
export type AgentAcpGuardPidOptions = {
|
||||
/** Spawned `agent` child PID when known. */
|
||||
childPid?: number;
|
||||
};
|
||||
|
||||
function normalizePid(pid: number | undefined): number | null {
|
||||
if (pid === undefined || !Number.isInteger(pid) || pid <= 0) {
|
||||
return null;
|
||||
}
|
||||
return pid;
|
||||
}
|
||||
|
||||
export function getAgentAcpLockDir(): string {
|
||||
return join(resolveHapiHomeDir(), 'locks', 'agent-acp-active');
|
||||
}
|
||||
|
||||
function getAcpLockDir(): string {
|
||||
const home = process.env.HAPI_HOME?.trim() || join(tmpdir(), 'hapi');
|
||||
return join(home, 'locks', 'agent-acp-active');
|
||||
return getAgentAcpLockDir();
|
||||
}
|
||||
|
||||
function getPidsDir(lockDir: string): string {
|
||||
return join(lockDir, 'pids');
|
||||
}
|
||||
|
||||
function getRegisteringDir(lockDir: string): string {
|
||||
return join(lockDir, REGISTERING_MARKER);
|
||||
}
|
||||
|
||||
function beginRegistering(lockDir: string): void {
|
||||
const dir = getRegisteringDir(lockDir);
|
||||
mkdirSync(dir, { recursive: true });
|
||||
writeFileSync(join(dir, String(process.pid)), String(Date.now()), 'utf8');
|
||||
}
|
||||
|
||||
function endRegistering(lockDir: string): void {
|
||||
try {
|
||||
rmSync(join(getRegisteringDir(lockDir), String(process.pid)), { force: true });
|
||||
} catch {
|
||||
// Best effort.
|
||||
}
|
||||
}
|
||||
|
||||
/** True if any live host still holds a mid-publish reservation marker. */
|
||||
function isRegistering(lockDir: string): boolean {
|
||||
const dir = getRegisteringDir(lockDir);
|
||||
if (!existsSync(dir)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
let anyLive = false;
|
||||
for (const entry of readdirSync(dir)) {
|
||||
const pid = Number(entry);
|
||||
if (!Number.isInteger(pid) || pid <= 0) {
|
||||
try {
|
||||
rmSync(join(dir, entry), { force: true });
|
||||
} catch {
|
||||
// Best effort.
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (isProcessAlive(pid)) {
|
||||
anyLive = true;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
rmSync(join(dir, entry), { force: true });
|
||||
} catch {
|
||||
// Best effort — crash/reboot left a dead registrar marker.
|
||||
}
|
||||
}
|
||||
return anyLive;
|
||||
}
|
||||
|
||||
function isFreshPrespawnReservation(lockDir: string): boolean {
|
||||
try {
|
||||
return Date.now() - statSync(lockDir).mtimeMs < PRESPAWN_RESERVATION_GRACE_MS;
|
||||
} catch {
|
||||
// Fail closed — prefer keeping a disputed lock over list-models SIGTERM.
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
function readLockPid(lockDir: string): number | null {
|
||||
const pidPath = join(lockDir, 'pid');
|
||||
if (!existsSync(pidPath)) {
|
||||
@@ -68,10 +166,38 @@ function writeLockCount(lockDir: string, count: number): void {
|
||||
writeFileSync(join(lockDir, 'count'), String(Math.max(0, count)), 'utf8');
|
||||
}
|
||||
|
||||
function writeChildPidHint(lockDir: string, childPid: number): void {
|
||||
writeFileSync(join(lockDir, 'child-pid'), String(childPid), 'utf8');
|
||||
}
|
||||
|
||||
function clearChildPidHint(lockDir: string): void {
|
||||
try {
|
||||
rmSync(join(lockDir, 'child-pid'), { force: true });
|
||||
} catch {
|
||||
// Best effort.
|
||||
}
|
||||
}
|
||||
|
||||
function addLockPid(lockDir: string, pid: number): void {
|
||||
const pidsDir = getPidsDir(lockDir);
|
||||
mkdirSync(pidsDir, { recursive: true });
|
||||
writeFileSync(join(pidsDir, String(pid)), String(pid), 'utf8');
|
||||
const pidPath = join(pidsDir, String(pid));
|
||||
// Retry once if a concurrent last-unregister deleted the lock mid-publish.
|
||||
for (let attempt = 0; attempt < 2; attempt++) {
|
||||
mkdirSync(lockDir, { recursive: true });
|
||||
mkdirSync(pidsDir, { recursive: true });
|
||||
addLockPidHook?.('after-pids-mkdir');
|
||||
try {
|
||||
writeFileSync(pidPath, String(pid), { encoding: 'utf8', flag: 'w' });
|
||||
addLockPidHook?.('after-pid-write');
|
||||
return;
|
||||
} catch (error) {
|
||||
const code = (error as NodeJS.ErrnoException).code;
|
||||
if (attempt === 0 && (code === 'ENOENT' || code === 'ENOTDIR')) {
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function removeLockPid(lockDir: string, pid: number): void {
|
||||
@@ -112,6 +238,10 @@ function removeAcpLockDir(): void {
|
||||
function reconcileRefcountLock(lockDir: string): boolean {
|
||||
const pidsDir = getPidsDir(lockDir);
|
||||
if (!existsSync(pidsDir)) {
|
||||
// Registrar mid-publish, or grace before `registering` / first pid.
|
||||
if (isRegistering(lockDir) || isFreshPrespawnReservation(lockDir)) {
|
||||
return true;
|
||||
}
|
||||
removeAcpLockDir();
|
||||
return false;
|
||||
}
|
||||
@@ -141,6 +271,34 @@ function reconcileRefcountLock(lockDir: string): boolean {
|
||||
}
|
||||
|
||||
if (liveCount <= 0) {
|
||||
// Re-read: registrar may have published a pid during our scan, or we
|
||||
// are between mkdir(pids) and writeFile (empty dir — fail closed).
|
||||
// A live `registering` marker covers overlap with leftover count>0
|
||||
// from a concurrent last-unregister.
|
||||
let entries: string[] = [];
|
||||
try {
|
||||
entries = readdirSync(pidsDir);
|
||||
} catch {
|
||||
entries = [];
|
||||
}
|
||||
const liveAgain = entries.filter((entry) => {
|
||||
const pid = Number(entry);
|
||||
return Number.isInteger(pid) && pid > 0 && isProcessAlive(pid);
|
||||
});
|
||||
if (liveAgain.length > 0) {
|
||||
writeLockCount(lockDir, liveAgain.length);
|
||||
return true;
|
||||
}
|
||||
if (isRegistering(lockDir)) {
|
||||
return true;
|
||||
}
|
||||
if (
|
||||
entries.length === 0
|
||||
&& readLockCount(lockDir) <= 0
|
||||
&& (isFreshPrespawnReservation(pidsDir) || isFreshPrespawnReservation(lockDir))
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
removeAcpLockDir();
|
||||
return false;
|
||||
}
|
||||
@@ -167,19 +325,59 @@ function clearStaleAcpLockIfNeeded(): void {
|
||||
reconcileRefcountLock(lockDir);
|
||||
}
|
||||
|
||||
export function registerActiveAcpTransport(): void {
|
||||
/**
|
||||
* Reserve / register the ACP lock. Call before spawn (no childPid) so
|
||||
* list-models cannot race the new `agent` process, then call
|
||||
* {@link recordActiveAcpChildPid} once the child PID is known.
|
||||
*
|
||||
* Publish order is fail-closed: `pids/<hostPid>` (and optional child) land
|
||||
* before `count`, so concurrent reconcile never treats the reservation as
|
||||
* a lock with no pids.
|
||||
*/
|
||||
export function registerActiveAcpTransport(options?: AgentAcpGuardPidOptions): void {
|
||||
activeAcpTransportCount += 1;
|
||||
const lockDir = getAcpLockDir();
|
||||
const childPid = normalizePid(options?.childPid);
|
||||
try {
|
||||
mkdirSync(lockDir, { recursive: true });
|
||||
writeLockCount(lockDir, readLockCount(lockDir) + 1);
|
||||
beginRegistering(lockDir);
|
||||
registerPublishHook?.('after-mkdir');
|
||||
// Always keep the HAPI host PID for crash/stale cleanup of the session
|
||||
// process; also record the ACP child when known — before count.
|
||||
addLockPid(lockDir, process.pid);
|
||||
if (childPid !== null) {
|
||||
addLockPid(lockDir, childPid);
|
||||
writeChildPidHint(lockDir, childPid);
|
||||
}
|
||||
registerPublishHook?.('after-host-pid');
|
||||
writeLockCount(lockDir, readLockCount(lockDir) + 1);
|
||||
registerPublishHook?.('after-count');
|
||||
} catch {
|
||||
// Another process may have created the lock; in-process guard still applies.
|
||||
} finally {
|
||||
endRegistering(lockDir);
|
||||
}
|
||||
}
|
||||
|
||||
export function unregisterActiveAcpTransport(): void {
|
||||
/** Upgrade a pre-spawn reservation with the real ACP child PID. */
|
||||
export function recordActiveAcpChildPid(childPid: number): void {
|
||||
const pid = normalizePid(childPid);
|
||||
if (pid === null) {
|
||||
return;
|
||||
}
|
||||
const lockDir = getAcpLockDir();
|
||||
if (!existsSync(lockDir)) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
addLockPid(lockDir, pid);
|
||||
writeChildPidHint(lockDir, pid);
|
||||
} catch {
|
||||
// Best effort.
|
||||
}
|
||||
}
|
||||
|
||||
export function unregisterActiveAcpTransport(options?: AgentAcpGuardPidOptions): void {
|
||||
activeAcpTransportCount = Math.max(0, activeAcpTransportCount - 1);
|
||||
|
||||
const lockDir = getAcpLockDir();
|
||||
@@ -195,8 +393,13 @@ export function unregisterActiveAcpTransport(): void {
|
||||
}
|
||||
|
||||
try {
|
||||
const childPid = normalizePid(options?.childPid);
|
||||
if (childPid !== null) {
|
||||
removeLockPid(lockDir, childPid);
|
||||
}
|
||||
if (activeAcpTransportCount <= 0) {
|
||||
removeLockPid(lockDir, process.pid);
|
||||
clearChildPidHint(lockDir);
|
||||
}
|
||||
reconcileRefcountLock(lockDir);
|
||||
} catch {
|
||||
@@ -219,10 +422,54 @@ export function isAgentAcpTransportActive(): boolean {
|
||||
return pid !== null && isProcessAlive(pid);
|
||||
}
|
||||
|
||||
return readLockCount(lockDir) > 0;
|
||||
if (readLockCount(lockDir) > 0) {
|
||||
return true;
|
||||
}
|
||||
// Mid-publish: registering marker or mtime grace without a count yet.
|
||||
if (isRegistering(lockDir)) {
|
||||
return true;
|
||||
}
|
||||
return isFreshPrespawnReservation(lockDir);
|
||||
}
|
||||
|
||||
/** Debug attribution for exit / list-models races (PID, lock dir, activity). */
|
||||
export function describeAgentAcpGuardState(childPid?: number | null): {
|
||||
lockDir: string;
|
||||
inProcessCount: number;
|
||||
childPid: number | null;
|
||||
childAlive: boolean | null;
|
||||
guardActive: boolean;
|
||||
} {
|
||||
const pid = normalizePid(childPid ?? undefined);
|
||||
return {
|
||||
lockDir: getAgentAcpLockDir(),
|
||||
inProcessCount: activeAcpTransportCount,
|
||||
childPid: pid,
|
||||
childAlive: pid === null ? null : isProcessAlive(pid),
|
||||
guardActive: isAgentAcpTransportActive()
|
||||
};
|
||||
}
|
||||
|
||||
export function _setRegisterPublishHookForTests(
|
||||
hook: ((step: 'after-mkdir' | 'after-host-pid' | 'after-count') => void) | null
|
||||
): void {
|
||||
registerPublishHook = hook;
|
||||
}
|
||||
|
||||
export function _setAddLockPidHookForTests(
|
||||
hook: ((phase: 'after-pids-mkdir' | 'after-pid-write') => void) | null
|
||||
): void {
|
||||
addLockPidHook = hook;
|
||||
}
|
||||
|
||||
/** Simulate a cross-process reader (no in-process reservation). */
|
||||
export function _setActiveAcpTransportCountForTests(count: number): void {
|
||||
activeAcpTransportCount = Math.max(0, count);
|
||||
}
|
||||
|
||||
export function _resetAgentCliGuardForTests(): void {
|
||||
activeAcpTransportCount = 0;
|
||||
registerPublishHook = null;
|
||||
addLockPidHook = null;
|
||||
removeAcpLockDir();
|
||||
}
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
|
||||
import { dirname, join } from 'node:path';
|
||||
import { tmpdir } from 'node:os';
|
||||
import type { CursorModelsResponse } from '@hapi/protocol/apiTypes';
|
||||
import { resolveHapiHomeDir } from '@/configuration';
|
||||
|
||||
function getHapiHomeDir(): string {
|
||||
return process.env.HAPI_HOME?.trim() || join(tmpdir(), 'hapi');
|
||||
return resolveHapiHomeDir();
|
||||
}
|
||||
|
||||
function getSharedCachePath(): string {
|
||||
|
||||
Reference in New Issue
Block a user