smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(coding-agent): resolved coding-agent SIGINT shutdown escalation path

Browse Source 
- Added bounded SIGINT escalation with `requestCancel()` to force kernel shutdown after 2s.
- Removed executor idle-timeout, max-session, heartbeat, and deferred disposal-retry logic.
- Added per-session `runQueued()` execution and dead-kernel restart retry flow for owned sessions.
- Replaced kernel session tracking with reusable `sessions` map and `acquireSession()` ownership registration.
- Removed obsolete owner-cleanup session-capacity tests and updated dead-session mocks to resolver-based flow.
This commit is contained in:
can1357
2026-05-15 14:46:54 +02:00
parent 005c3cd81a
commit 9a755822eb
4 changed files with 249 additions and 1074 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/coding-agent/test/core/python-executor-owner-cleanup.test.ts
+15 -438
View File
@@ -192,101 +192,6 @@ describe("python executor owner cleanup", () => {
expect(replacementKernel.shutdown).toHaveBeenCalledTimes(1);
});
it("keeps tracked disposals counted against retained kernel capacity until shutdown settles", async () => {
const retainedKernels = [new FakeKernel(), new FakeKernel(), new FakeKernel(), new FakeKernel()];
const replacementKernel = new FakeKernel();
const shutdownDeferreds = retainedKernels.map(() => Promise.withResolvers<KernelShutdownResult>());
for (const [index, kernel] of retainedKernels.entries()) {
kernel.shutdown = vi.fn(() => shutdownDeferreds[index]!.promise);
}
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start");
for (const kernel of [...retainedKernels, replacementKernel]) {
startSpy.mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
}
for (const [index] of retainedKernels.entries()) {
await executePython(`print(${index})`, {
cwd: `/tmp/capacity-tracking-${index}`,
sessionId: `capacity-session-${index}`,
kernelMode: "session",
});
}
expect(startSpy).toHaveBeenCalledTimes(4);
const globalDisposal = disposeAllKernelSessions();
await Promise.resolve();
const fifthExecution = executePython("print('replacement')", {
cwd: "/tmp/capacity-tracking-replacement",
sessionId: "capacity-session-replacement",
kernelMode: "session",
});
await Promise.resolve();
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
shutdownDeferreds[0]!.resolve({ confirmed: true });
await fifthExecution;
expect(startSpy).toHaveBeenCalledTimes(5);
expect(replacementKernel.execute).toHaveBeenCalledTimes(1);
for (const deferred of shutdownDeferreds.slice(1)) {
deferred.resolve({ confirmed: true });
}
await globalDisposal;
await disposeAllKernelSessions();
expect(replacementKernel.shutdown).toHaveBeenCalledTimes(1);
});
it("waits with the owner-cleanup timeout when the last owner is removed from an already-disposing session", async () => {
vi.useFakeTimers();
try {
const kernel = new FakeKernel();
const shutdownConfirmation = Promise.withResolvers<KernelShutdownResult>();
kernel.shutdown = vi.fn(() => shutdownConfirmation.promise);
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start").mockResolvedValue(kernel as unknown as PythonKernelInstance);
await executePython("print('owner-a')", {
cwd: "/tmp/disposing-owner-cleanup-session",
sessionId: "disposing-owner-cleanup-session",
kernelMode: "session",
kernelOwnerId: "owner-a",
});
let globalCleanupResolved = false;
const globalCleanup = disposeAllKernelSessions().finally(() => {
globalCleanupResolved = true;
});
await flushMicrotasks();
expect(kernel.shutdown).toHaveBeenCalledTimes(1);
expect(globalCleanupResolved).toBe(false);
let ownerCleanupResolved = false;
const ownerCleanup = disposeKernelSessionsByOwner("owner-a").finally(() => {
ownerCleanupResolved = true;
});
await flushMicrotasks();
expect(ownerCleanupResolved).toBe(false);
expect(kernel.shutdown).toHaveBeenCalledTimes(1);
vi.advanceTimersByTime(2_000);
await ownerCleanup;
expect(ownerCleanupResolved).toBe(true);
expect(globalCleanupResolved).toBe(false);
expect(kernel.shutdown).toHaveBeenCalledTimes(1);
shutdownConfirmation.resolve({ confirmed: true });
await globalCleanup;
expect(globalCleanupResolved).toBe(true);
expect(startSpy).toHaveBeenCalledTimes(1);
} finally {
vi.useRealTimers();
}
});
it("returns a cancelled result when a dead session restart shutdown times out", async () => {
const kernel = new FakeKernel();
kernel.alive = false;
@@ -296,13 +201,13 @@ describe("python executor owner cleanup", () => {
if (shutdownCallCount > 1) {
return { confirmed: true };
}
return await new Promise((_, reject) => {
const timer = setTimeout(
() => reject(new DOMException("Python kernel shutdown timed out", "TimeoutError")),
options?.timeoutMs ?? 0,
);
timer.unref?.();
});
const { promise, reject } = Promise.withResolvers<KernelShutdownResult>();
const timer = setTimeout(
() => reject(new DOMException("Python kernel shutdown timed out", "TimeoutError")),
options?.timeoutMs ?? 0,
);
timer.unref?.();
return await promise;
});
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start").mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
@@ -319,353 +224,21 @@ describe("python executor owner cleanup", () => {
expect(kernel.shutdown).toHaveBeenCalledWith(expect.objectContaining({ timeoutMs: expect.any(Number) }));
expect(startSpy).toHaveBeenCalledTimes(1);
});
it("returns owner cleanup promptly but keeps retained capacity reserved until shutdown is confirmed", async () => {
vi.useFakeTimers();
try {
const retainedKernels = [new FakeKernel(), new FakeKernel(), new FakeKernel(), new FakeKernel()];
const replacementKernel = new FakeKernel();
const shutdownDeferreds = retainedKernels.map(() => Promise.withResolvers<KernelShutdownResult>());
for (const [index, kernel] of retainedKernels.entries()) {
kernel.shutdown = vi.fn(() => shutdownDeferreds[index]!.promise);
}
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start");
for (const kernel of [...retainedKernels, replacementKernel]) {
startSpy.mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
}
for (const [index] of retainedKernels.entries()) {
await executePython(`print(${index})`, {
cwd: `/tmp/owner-timeout-kernel-${index}`,
sessionId: `timeout-session-${index}`,
kernelMode: "session",
kernelOwnerId: "owner-a",
});
}
let ownerCleanupResolved = false;
const ownerCleanup = disposeKernelSessionsByOwner("owner-a").finally(() => {
ownerCleanupResolved = true;
});
await flushMicrotasks();
for (const kernel of retainedKernels) {
expect(kernel.shutdown).toHaveBeenCalledWith({ timeoutMs: 2_000 });
}
expect(ownerCleanupResolved).toBe(false);
vi.advanceTimersByTime(2_000);
await ownerCleanup;
expect(ownerCleanupResolved).toBe(true);
const blockedExecution = executePython("print('replacement')", {
cwd: "/tmp/owner-timeout-kernel-replacement",
sessionId: "timeout-session-replacement",
kernelMode: "session",
kernelOwnerId: "owner-b",
});
await Promise.resolve();
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
shutdownDeferreds[0]!.resolve({ confirmed: true });
await blockedExecution;
expect(startSpy).toHaveBeenCalledTimes(5);
expect(replacementKernel.execute).toHaveBeenCalledTimes(1);
for (const deferred of shutdownDeferreds.slice(1)) {
deferred.resolve({ confirmed: true });
}
await Promise.resolve();
await disposeAllKernelSessions();
expect(replacementKernel.shutdown).toHaveBeenCalledTimes(1);
} finally {
vi.useRealTimers();
}
});
it("keeps owner-cleanup retries on the timer path without evicting unrelated live sessions", async () => {
vi.useFakeTimers();
try {
const ownerKernel = new FakeKernel();
const unrelatedKernels = [new FakeKernel(), new FakeKernel(), new FakeKernel()];
const replacementKernel = new FakeKernel();
const retryConfirmation = Promise.withResolvers<KernelShutdownResult>();
let shutdownCallCount = 0;
ownerKernel.shutdown = vi.fn(async (options?: FakeKernelShutdownOptions): Promise<KernelShutdownResult> => {
shutdownCallCount += 1;
if (shutdownCallCount === 1) {
expect(options).toEqual({ timeoutMs: 2_000 });
return { confirmed: false };
}
if (shutdownCallCount === 2) {
expect(options).toBeUndefined();
return { confirmed: false };
}
return await retryConfirmation.promise;
});
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start");
for (const kernel of [ownerKernel, ...unrelatedKernels, replacementKernel]) {
startSpy.mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
}
await executePython("print('owner-a')", {
cwd: "/tmp/timer-retry-owner-a",
sessionId: "timer-retry-owner-a",
kernelMode: "session",
kernelOwnerId: "owner-a",
});
for (const [index, ownerId] of ["owner-b", "owner-c", "owner-d"].entries()) {
await executePython(`print(${index})`, {
cwd: `/tmp/timer-retry-${ownerId}`,
sessionId: `timer-retry-${ownerId}`,
kernelMode: "session",
kernelOwnerId: ownerId,
});
}
await disposeKernelSessionsByOwner("owner-a");
expect(ownerKernel.shutdown).toHaveBeenCalledTimes(2);
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
const blockedExecution = executePython("print('replacement')", {
cwd: "/tmp/timer-retry-replacement",
sessionId: "timer-retry-replacement",
kernelMode: "session",
kernelOwnerId: "owner-e",
});
await Promise.resolve();
await Promise.resolve();
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
vi.advanceTimersByTime(29_999);
await Promise.resolve();
await Promise.resolve();
expect(ownerKernel.shutdown).toHaveBeenCalledTimes(2);
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
vi.advanceTimersByTime(1);
await Promise.resolve();
await Promise.resolve();
expect(ownerKernel.shutdown).toHaveBeenCalledTimes(3);
expect(ownerKernel.shutdown).toHaveBeenNthCalledWith(3, undefined);
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
retryConfirmation.resolve({ confirmed: true });
await blockedExecution;
expect(startSpy).toHaveBeenCalledTimes(5);
expect(replacementKernel.execute).toHaveBeenCalledTimes(1);
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
await disposeAllKernelSessions();
expect(replacementKernel.shutdown).toHaveBeenCalledTimes(1);
} finally {
vi.useRealTimers();
}
});
it("waits for confirmed disposal capacity before evicting unrelated retained sessions", async () => {
const ownerKernel = new FakeKernel();
const unrelatedKernels = [new FakeKernel(), new FakeKernel(), new FakeKernel()];
const replacementKernel = new FakeKernel();
const retryConfirmation = Promise.withResolvers<KernelShutdownResult>();
let shutdownCallCount = 0;
ownerKernel.shutdown = vi.fn(async (_options?: FakeKernelShutdownOptions): Promise<KernelShutdownResult> => {
shutdownCallCount += 1;
if (shutdownCallCount === 1) {
return { confirmed: false };
}
return await retryConfirmation.promise;
});
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start");
for (const kernel of [ownerKernel, ...unrelatedKernels, replacementKernel]) {
startSpy.mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
}
await executePython("print('owner-a')", {
cwd: "/tmp/unconfirmed-owner-cleanup-a",
sessionId: "unconfirmed-owner-cleanup-a",
kernelMode: "session",
kernelOwnerId: "owner-a",
});
for (const [index, ownerId] of ["owner-b", "owner-c", "owner-d"].entries()) {
await executePython(`print(${index})`, {
cwd: `/tmp/unconfirmed-owner-cleanup-${ownerId}`,
sessionId: `unconfirmed-owner-cleanup-${ownerId}`,
kernelMode: "session",
kernelOwnerId: ownerId,
});
}
await disposeKernelSessionsByOwner("owner-a");
expect(ownerKernel.shutdown).toHaveBeenCalledTimes(2);
expect(ownerKernel.shutdown).toHaveBeenNthCalledWith(1, { timeoutMs: 2_000 });
expect(ownerKernel.shutdown).toHaveBeenNthCalledWith(2, undefined);
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
const blockedExecution = executePython("print('replacement')", {
cwd: "/tmp/unconfirmed-owner-cleanup-replacement",
sessionId: "unconfirmed-owner-cleanup-replacement",
kernelMode: "session",
kernelOwnerId: "owner-e",
});
await Promise.resolve();
await Promise.resolve();
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
for (const [index, ownerId] of ["owner-b", "owner-c", "owner-d"].entries()) {
await executePython(`print('reuse-${ownerId}')`, {
cwd: `/tmp/unconfirmed-owner-cleanup-${ownerId}`,
sessionId: `unconfirmed-owner-cleanup-${ownerId}`,
kernelMode: "session",
kernelOwnerId: ownerId,
});
expect(unrelatedKernels[index]!.execute).toHaveBeenCalledTimes(2);
expect(unrelatedKernels[index]!.shutdown).not.toHaveBeenCalled();
}
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
retryConfirmation.resolve({ confirmed: true });
await blockedExecution;
expect(startSpy).toHaveBeenCalledTimes(5);
expect(replacementKernel.execute).toHaveBeenCalledTimes(1);
for (const kernel of unrelatedKernels) {
expect(kernel.shutdown).not.toHaveBeenCalled();
}
await disposeAllKernelSessions();
expect(replacementKernel.shutdown).toHaveBeenCalledTimes(1);
});
it("owner cleanup retries every shutdown in background and frees retained capacity one confirmation at a time", async () => {
const retainedKernels = [new FakeKernel(), new FakeKernel(), new FakeKernel(), new FakeKernel()];
const replacementKernel = new FakeKernel();
const laterKernel = new FakeKernel();
const retryConfirmations = retainedKernels.map(() => Promise.withResolvers<KernelShutdownResult>());
for (const [index, kernel] of retainedKernels.entries()) {
let shutdownCallCount = 0;
kernel.shutdown = vi.fn(async (_options?: FakeKernelShutdownOptions): Promise<KernelShutdownResult> => {
shutdownCallCount += 1;
if (shutdownCallCount === 1) {
return { confirmed: false };
}
return await retryConfirmations[index]!.promise;
});
}
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
const startSpy = vi.spyOn(PythonKernel, "start");
for (const kernel of [...retainedKernels, replacementKernel, laterKernel]) {
startSpy.mockResolvedValueOnce(kernel as unknown as PythonKernelInstance);
}
for (const [index] of retainedKernels.entries()) {
await executePython(`print(${index})`, {
cwd: `/tmp/unconfirmed-capacity-${index}`,
sessionId: `unconfirmed-capacity-session-${index}`,
kernelMode: "session",
kernelOwnerId: "owner-a",
});
}
await disposeKernelSessionsByOwner("owner-a");
for (const kernel of retainedKernels) {
expect(kernel.shutdown).toHaveBeenCalledTimes(2);
expect(kernel.shutdown).toHaveBeenNthCalledWith(1, { timeoutMs: 2_000 });
expect(kernel.shutdown).toHaveBeenNthCalledWith(2, undefined);
}
let globalCleanupResolved = false;
const globalCleanup = disposeAllKernelSessions().then(() => {
globalCleanupResolved = true;
});
await Promise.resolve();
await Promise.resolve();
for (const kernel of retainedKernels) {
expect(kernel.shutdown).toHaveBeenCalledTimes(2);
}
expect(globalCleanupResolved).toBe(false);
const blockedExecution = executePython("print('replacement')", {
cwd: "/tmp/unconfirmed-capacity-replacement",
sessionId: "unconfirmed-capacity-replacement",
kernelMode: "session",
kernelOwnerId: "owner-b",
});
await Promise.resolve();
await Promise.resolve();
expect(startSpy).toHaveBeenCalledTimes(4);
expect(replacementKernel.execute).not.toHaveBeenCalled();
retryConfirmations[0]!.resolve({ confirmed: true });
await blockedExecution;
expect(globalCleanupResolved).toBe(false);
expect(startSpy).toHaveBeenCalledTimes(5);
expect(replacementKernel.execute).toHaveBeenCalledTimes(1);
const secondBlockedExecution = executePython("print('later')", {
cwd: "/tmp/unconfirmed-capacity-later",
sessionId: "unconfirmed-capacity-later",
kernelMode: "session",
kernelOwnerId: "owner-c",
});
await Promise.resolve();
await Promise.resolve();
expect(globalCleanupResolved).toBe(false);
expect(startSpy).toHaveBeenCalledTimes(5);
expect(laterKernel.execute).not.toHaveBeenCalled();
retryConfirmations[1]!.resolve({ confirmed: true });
await secondBlockedExecution;
expect(globalCleanupResolved).toBe(false);
expect(startSpy).toHaveBeenCalledTimes(6);
expect(laterKernel.execute).toHaveBeenCalledTimes(1);
for (const confirmation of retryConfirmations.slice(2)) {
confirmation.resolve({ confirmed: true });
}
await globalCleanup;
expect(globalCleanupResolved).toBe(true);
await disposeAllKernelSessions();
expect(replacementKernel.shutdown).toHaveBeenCalledTimes(1);
expect(laterKernel.shutdown).toHaveBeenCalledTimes(1);
});
it("does not let stuck retained executions block owner or global cleanup", async () => {
const ownerKernel = new FakeKernel();
const globalKernel = new FakeKernel();
const ownerExecutionStarted = Promise.withResolvers<void>();
const globalExecutionStarted = Promise.withResolvers<void>();
const ownerExecutionHang = Promise.withResolvers<KernelExecuteResult>();
const globalExecutionHang = Promise.withResolvers<KernelExecuteResult>();
ownerKernel.execute = vi.fn(async () => {
ownerExecutionStarted.resolve();
return await new Promise<KernelExecuteResult>(() => {});
return await ownerExecutionHang.promise;
});
globalKernel.execute = vi.fn(async () => {
globalExecutionStarted.resolve();
return await new Promise<KernelExecuteResult>(() => {});
return await globalExecutionHang.promise;
});
vi.spyOn(pythonKernel, "checkPythonKernelAvailability").mockResolvedValue({ ok: true });
vi.spyOn(PythonKernel, "start")
@@ -697,7 +270,11 @@ describe("python executor owner cleanup", () => {
await flushMicrotasks();
expect(globalKernel.shutdown).toHaveBeenCalledTimes(1);
await globalCleanup;
ownerExecutionHang.resolve(OK_RESULT);
globalExecutionHang.resolve(OK_RESULT);
});
it("leaves per-call kernels out of owner-scoped retained cleanup and keeps global cleanup intact", async () => {
const perCallKernel = new FakeKernel();
const retainedKernel = new FakeKernel();