mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-08-31 10:16:57 +00:00
fix(cli): harden agent view liveness recovery
This commit is contained in:
俊良
parent
53dc88564c
commit
6d8e44666a
2 changed files with 221 additions and 13 deletions
|
|
@ -2219,6 +2219,46 @@ describe('Agent View supervisor process helpers', () => {
|
|||
await fs.rm(globalDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('leaves the stream open when attach setup fails', async () => {
|
||||
const globalDir = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), 'qwen-agent-view-store-'),
|
||||
);
|
||||
const handler = createAgentViewSupervisorHandler({
|
||||
globalDir,
|
||||
platform: 'linux',
|
||||
launchPtyHost: async () => fakePtyHost(),
|
||||
});
|
||||
const result = (await handler.dispatch?.({
|
||||
prompt: 'write tests',
|
||||
cwd: globalDir,
|
||||
})) as { sessionId: string };
|
||||
const originalPatch = supervisorStore.patchAgentViewSessionState;
|
||||
const patchSpy = vi
|
||||
.spyOn(supervisorStore, 'patchAgentViewSessionState')
|
||||
.mockImplementation(async (sessionId, patch, options) => {
|
||||
if (patch.attachState === 'attached') {
|
||||
throw new Error('attach write failed');
|
||||
}
|
||||
return originalPatch(sessionId, patch, options);
|
||||
});
|
||||
const socket = new FakeAttachSocket();
|
||||
|
||||
try {
|
||||
await expect(
|
||||
handler.attachStream?.(
|
||||
{ sessionId: result.sessionId },
|
||||
socket as unknown as Socket,
|
||||
'request-1',
|
||||
),
|
||||
).rejects.toThrow('attach write failed');
|
||||
expect(socket.writableEnded).toBe(false);
|
||||
} finally {
|
||||
socket.destroy();
|
||||
patchSpy.mockRestore();
|
||||
await fs.rm(globalDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('rejects send and answer while a live attach is open', async () => {
|
||||
const globalDir = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), 'qwen-agent-view-store-'),
|
||||
|
|
@ -3162,6 +3202,90 @@ describe('Agent View supervisor process helpers', () => {
|
|||
await fs.rm(globalDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('fails closed when worker liveness data is unreadable', async () => {
|
||||
const globalDir = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), 'qwen-agent-view-store-'),
|
||||
);
|
||||
const seedHandler = createAgentViewSupervisorHandler({
|
||||
globalDir,
|
||||
platform: 'linux',
|
||||
launchPtyHost: async () => fakePtyHost(999_999_002, 999_999_001),
|
||||
});
|
||||
const result = (await seedHandler.dispatch?.({
|
||||
prompt: 'write tests',
|
||||
cwd: globalDir,
|
||||
})) as { sessionId: string };
|
||||
const paths = getAgentViewSessionPaths(result.sessionId, { globalDir });
|
||||
await fs.writeFile(paths.workerPath, '{ invalid json', 'utf8');
|
||||
const recoveredHandler = createAgentViewSupervisorHandler({
|
||||
globalDir,
|
||||
platform: 'linux',
|
||||
launchPtyHost: async () => fakePtyHost(999_999_003, 999_999_004),
|
||||
});
|
||||
|
||||
await expect(
|
||||
recoveredHandler.stop?.({ sessionId: result.sessionId }),
|
||||
).rejects.toThrow('temporarily unreadable');
|
||||
await expect(
|
||||
readAgentViewSessionState(result.sessionId, { globalDir }),
|
||||
).resolves.toMatchObject({
|
||||
sessionState: 'starting',
|
||||
processState: 'starting',
|
||||
});
|
||||
|
||||
await fs.rm(globalDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('clears stale worker pids with the terminal heal', async () => {
|
||||
const globalDir = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), 'qwen-agent-view-store-'),
|
||||
);
|
||||
const seedHandler = createAgentViewSupervisorHandler({
|
||||
globalDir,
|
||||
platform: 'linux',
|
||||
launchPtyHost: async () => fakePtyHost(999_999_002, 999_999_001),
|
||||
});
|
||||
const result = (await seedHandler.dispatch?.({
|
||||
prompt: 'write tests',
|
||||
cwd: globalDir,
|
||||
})) as { sessionId: string };
|
||||
await patchSessionStateForTest(result.sessionId, globalDir, {
|
||||
sessionState: 'working',
|
||||
processState: 'alive',
|
||||
});
|
||||
const recoveredHandler = createAgentViewSupervisorHandler({
|
||||
globalDir,
|
||||
platform: 'linux',
|
||||
launchPtyHost: async () => fakePtyHost(999_999_003, 999_999_004),
|
||||
});
|
||||
const killSpy = vi.spyOn(process, 'kill').mockImplementation(() => {
|
||||
const error = new Error('not running') as NodeJS.ErrnoException;
|
||||
error.code = 'ESRCH';
|
||||
throw error;
|
||||
});
|
||||
|
||||
try {
|
||||
await recoveredHandler.list();
|
||||
const worker = JSON.parse(
|
||||
await fs.readFile(
|
||||
getAgentViewSessionPaths(result.sessionId, { globalDir }).workerPath,
|
||||
'utf8',
|
||||
),
|
||||
) as Record<string, unknown>;
|
||||
expect(worker).not.toHaveProperty('hostPid');
|
||||
expect(worker).not.toHaveProperty('workerPid');
|
||||
await expect(
|
||||
readAgentViewSessionState(result.sessionId, { globalDir }),
|
||||
).resolves.toMatchObject({
|
||||
sessionState: 'failed',
|
||||
processState: 'exited',
|
||||
});
|
||||
} finally {
|
||||
killSpy.mockRestore();
|
||||
await fs.rm(globalDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('fails closed instead of signaling unauthenticated stored pids', async () => {
|
||||
const globalDir = await fs.mkdtemp(
|
||||
path.join(os.tmpdir(), 'qwen-agent-view-store-'),
|
||||
|
|
@ -3316,16 +3440,24 @@ describe('Agent View supervisor process helpers', () => {
|
|||
return snapshots;
|
||||
});
|
||||
const list = handler.list();
|
||||
await read;
|
||||
await Promise.resolve();
|
||||
snapshotSpy.mockRestore();
|
||||
releaseRetire();
|
||||
let listError: unknown;
|
||||
try {
|
||||
await read;
|
||||
await expect(list).resolves.toHaveLength(1);
|
||||
} catch (error) {
|
||||
listError = error;
|
||||
} finally {
|
||||
snapshotSpy.mockRestore();
|
||||
releaseRetire();
|
||||
}
|
||||
|
||||
await expect(respawn).resolves.toEqual({
|
||||
sessionId: result.sessionId,
|
||||
respawned: true,
|
||||
});
|
||||
await list;
|
||||
if (listError) {
|
||||
throw listError;
|
||||
}
|
||||
const token = await readWorkerTokenForTest(result.sessionId, globalDir);
|
||||
await expect(
|
||||
handler.workerControl?.({ sessionId: result.sessionId, token }),
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import type {
|
|||
AgentViewSessionStateFile,
|
||||
AgentViewWorkerControlEvent,
|
||||
AgentViewWorkerEvent,
|
||||
AgentViewWorkerFile,
|
||||
} from './protocol.js';
|
||||
import type {
|
||||
AgentViewPtyHostExit,
|
||||
|
|
@ -515,7 +516,10 @@ class AgentViewSupervisorProcessHandler
|
|||
const connected =
|
||||
this.workers.has(adoption.sessionId) ||
|
||||
(await this.workers.reconnectSessionHostLocked(adoption.sessionId));
|
||||
const worker = await readAgentViewWorker(adoption.sessionId, store);
|
||||
const worker = await readAgentViewWorkerForLiveness(
|
||||
adoption.sessionId,
|
||||
store,
|
||||
);
|
||||
const pidAlive =
|
||||
isPidRunning(worker?.hostPid) || isPidRunning(worker?.workerPid);
|
||||
if (connected || pidAlive || worker?.hostEndpoint) {
|
||||
|
|
@ -1653,6 +1657,7 @@ class AgentViewSupervisorProcessHandler
|
|||
}, DEFAULT_ATTACH_LEASE_HEARTBEAT_MS);
|
||||
heartbeat.unref?.();
|
||||
let bridged = false;
|
||||
let propagatingError = false;
|
||||
try {
|
||||
if (
|
||||
controller.signal.aborted ||
|
||||
|
|
@ -1678,6 +1683,9 @@ class AgentViewSupervisorProcessHandler
|
|||
pty: host,
|
||||
detachSignal: controller.signal,
|
||||
});
|
||||
} catch (error) {
|
||||
propagatingError = !bridged;
|
||||
throw error;
|
||||
} finally {
|
||||
clearInterval(heartbeat);
|
||||
const wasCurrent = this.attachSockets.get(sessionId) === socket;
|
||||
|
|
@ -1695,7 +1703,9 @@ class AgentViewSupervisorProcessHandler
|
|||
// Best-effort: a store error during detach must not mask
|
||||
// the original error from the try block.
|
||||
}
|
||||
socket.end();
|
||||
if (bridged || !propagatingError) {
|
||||
socket.end();
|
||||
}
|
||||
}
|
||||
// Pre-bridge failure: leave the socket open so the RPC layer can
|
||||
// deliver the structured error envelope instead of a bare EOF.
|
||||
|
|
@ -2069,7 +2079,10 @@ class WorkerRegistry {
|
|||
await markStoppedSession(sessionId, this.store, 'alive');
|
||||
return;
|
||||
}
|
||||
const storedWorker = await readAgentViewWorker(sessionId, this.store);
|
||||
const storedWorker = await readAgentViewWorkerForLiveness(
|
||||
sessionId,
|
||||
this.store,
|
||||
);
|
||||
if (
|
||||
[storedWorker?.hostPid, storedWorker?.workerPid].some(
|
||||
(pid) => pid !== undefined && isPidRunning(pid),
|
||||
|
|
@ -2644,7 +2657,7 @@ class WorkerRegistry {
|
|||
}
|
||||
|
||||
private async assertNoStoredWorkerProcess(sessionId: string): Promise<void> {
|
||||
const worker = await readAgentViewWorker(sessionId, this.store);
|
||||
const worker = await readAgentViewWorkerForLiveness(sessionId, this.store);
|
||||
for (const pid of [worker?.hostPid, worker?.workerPid]) {
|
||||
if (!pid) continue;
|
||||
if (isPidRunning(pid)) {
|
||||
|
|
@ -2907,6 +2920,9 @@ class WorkerRegistry {
|
|||
state.sessionState === 'stopped' &&
|
||||
state.processState === 'alive'
|
||||
) {
|
||||
if (this.hostSetupQueues.has(state.sessionId)) {
|
||||
return state;
|
||||
}
|
||||
return this.withHostSetupLock(state.sessionId, async () => {
|
||||
const latest = await readAgentViewSessionState(
|
||||
state.sessionId,
|
||||
|
|
@ -2933,7 +2949,10 @@ class WorkerRegistry {
|
|||
}
|
||||
return state;
|
||||
}
|
||||
const worker = await readAgentViewWorker(state.sessionId, this.store);
|
||||
const worker = await readAgentViewWorkerForLiveness(
|
||||
state.sessionId,
|
||||
this.store,
|
||||
);
|
||||
if (isPidRunning(worker?.hostPid) || isPidRunning(worker?.workerPid)) {
|
||||
return state;
|
||||
}
|
||||
|
|
@ -2949,7 +2968,10 @@ class WorkerRegistry {
|
|||
const connected =
|
||||
this.ptyHosts.has(state.sessionId) || (await reconnectHost());
|
||||
if (!connected) {
|
||||
const worker = await readAgentViewWorker(state.sessionId, this.store);
|
||||
const worker = await readAgentViewWorkerForLiveness(
|
||||
state.sessionId,
|
||||
this.store,
|
||||
);
|
||||
if (isPidRunning(worker?.hostPid) || isPidRunning(worker?.workerPid)) {
|
||||
return state;
|
||||
}
|
||||
|
|
@ -3023,7 +3045,10 @@ class WorkerRegistry {
|
|||
: ((await readAgentViewSessionState(state.sessionId, this.store)) ??
|
||||
state);
|
||||
}
|
||||
const worker = await readAgentViewWorker(state.sessionId, this.store);
|
||||
const worker = await readAgentViewWorkerForLiveness(
|
||||
state.sessionId,
|
||||
this.store,
|
||||
);
|
||||
if (isPidRunning(worker?.hostPid) || isPidRunning(worker?.workerPid)) {
|
||||
return state;
|
||||
}
|
||||
|
|
@ -3072,7 +3097,10 @@ class WorkerRegistry {
|
|||
return state;
|
||||
}
|
||||
|
||||
const worker = await readAgentViewWorker(state.sessionId, this.store);
|
||||
const worker = await readAgentViewWorkerForLiveness(
|
||||
state.sessionId,
|
||||
this.store,
|
||||
);
|
||||
if (isPidRunning(worker?.hostPid) || isPidRunning(worker?.workerPid)) {
|
||||
return state;
|
||||
}
|
||||
|
|
@ -3115,6 +3143,9 @@ class WorkerRegistry {
|
|||
},
|
||||
this.store,
|
||||
);
|
||||
if (appliedPatch) {
|
||||
await clearAgentViewWorkerPids(state.sessionId, this.store);
|
||||
}
|
||||
return appliedPatch ? { ...state, ...appliedPatch } : state;
|
||||
}
|
||||
}
|
||||
|
|
@ -3678,6 +3709,51 @@ async function readOrThrowIfAbsent<T>(
|
|||
);
|
||||
}
|
||||
|
||||
async function readAgentViewWorkerForLiveness(
|
||||
sessionId: string,
|
||||
options: { globalDir?: string },
|
||||
): Promise<AgentViewWorkerFile | undefined> {
|
||||
const workerPath = getAgentViewSessionPaths(sessionId, options).workerPath;
|
||||
const filePresent = async () => {
|
||||
try {
|
||||
await fs.promises.access(workerPath);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (isEnoent(error)) {
|
||||
return false;
|
||||
}
|
||||
throw new Error(
|
||||
`Agent View worker record at ${workerPath} is temporarily unreadable. Retry the operation.`,
|
||||
{ cause: error },
|
||||
);
|
||||
}
|
||||
};
|
||||
let worker = await readAgentViewWorker(sessionId, options);
|
||||
if (worker) {
|
||||
return worker;
|
||||
}
|
||||
if (!(await filePresent())) {
|
||||
return undefined;
|
||||
}
|
||||
for (let attempt = 1; attempt <= 3; attempt++) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 50 * attempt));
|
||||
worker = await readAgentViewWorker(sessionId, options);
|
||||
if (worker) {
|
||||
return worker;
|
||||
}
|
||||
if (!(await filePresent())) {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
throw new Error(
|
||||
`Agent View worker record at ${workerPath} is temporarily unreadable. Retry the operation.`,
|
||||
);
|
||||
}
|
||||
|
||||
function isEnoent(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException).code === 'ENOENT';
|
||||
}
|
||||
|
||||
async function applyWorkerEvent(
|
||||
event: AgentViewWorkerEvent,
|
||||
options: { globalDir?: string },
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue