From bc54b650d71dc3bbcf0f6258bc8db921374f575d Mon Sep 17 00:00:00 2001 From: Aliou Diallo Date: Sun, 31 May 2026 11:36:32 +0200 Subject: [PATCH] feat: wire process notify actions --- .../processes/notifications/service.test.ts | 146 +++++++++++++-- extensions/processes/notifications/service.ts | 8 +- extensions/processes/tools/index.ts | 11 +- .../processes/tools/start/index.test.ts | 59 ++++++ extensions/processes/tools/start/index.ts | 11 +- extensions/processes/tools/stop/index.test.ts | 174 ++++++++++++++++++ extensions/processes/tools/stop/index.ts | 23 ++- tests/utils/async.ts | 3 + 8 files changed, 415 insertions(+), 20 deletions(-) create mode 100644 extensions/processes/tools/stop/index.test.ts create mode 100644 tests/utils/async.ts diff --git a/extensions/processes/notifications/service.test.ts b/extensions/processes/notifications/service.test.ts index 830ebad..ec83a1c 100644 --- a/extensions/processes/notifications/service.test.ts +++ b/extensions/processes/notifications/service.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import type { ProcessInfo } from "../../../src/types"; +import { flushQueuedMicrotasks } from "../../../tests/utils/async"; import { createNotificationRegistry } from "./registry"; import { createNotificationService } from "./service"; @@ -57,7 +58,7 @@ function createFakePi() { } describe("NotificationService", () => { - it("sends a turn notification for a failed process with default config", () => { + it("sends a turn notification for a failed process with default config", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -79,6 +80,7 @@ describe("NotificationService", () => { }); fakeManager.emit({ type: "process_ended", info }); + await flushQueuedMicrotasks(); expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); const [message, options] = fakePi.sendMessage.mock.calls[0]; @@ -90,7 +92,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("sends a context notification for a successful process with default config", () => { + it("sends a context notification for a successful process with default config", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -113,6 +115,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); const [, options] = fakePi.sendMessage.mock.calls[0]; @@ -122,7 +125,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("suppresses killed notification for intentional stop", () => { + it("suppresses killed notification for intentional stop", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -147,13 +150,14 @@ describe("NotificationService", () => { exitCode: null, }), }); + await flushQueuedMicrotasks(); expect(fakePi.sendMessage).not.toHaveBeenCalled(); service.dispose(); }); - it("sends notification for killed process when not intentional", () => { + it("sends notification for killed process when not intentional", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -179,6 +183,7 @@ describe("NotificationService", () => { exitCode: null, }), }); + await flushQueuedMicrotasks(); // killed with ignore attention and not forced display = no message expect(fakePi.sendMessage).not.toHaveBeenCalled(); @@ -186,7 +191,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("forces display for crash even when attention is ignore", () => { + it("forces display for crash even when attention is ignore", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -210,6 +215,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); // Crash forces display: attention becomes "context" expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); @@ -222,7 +228,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("forces display for timeout even when attention is ignore", () => { + it("forces display for timeout even when attention is ignore", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -246,6 +252,7 @@ describe("NotificationService", () => { exitCode: null, }), }); + await flushQueuedMicrotasks(); expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); const [message] = fakePi.sendMessage.mock.calls[0]; @@ -339,7 +346,7 @@ describe("NotificationService", () => { }); describe("disposal", () => { - it("does not send after dispose()", () => { + it("does not send after dispose()", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -365,6 +372,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); fakeManager.emit({ type: "process_output_changed", @@ -375,7 +383,7 @@ describe("NotificationService", () => { expect(fakePi.sendMessage).not.toHaveBeenCalled(); }); - it("does not send for events emitted during disposal sequence", () => { + it("does not send for events emitted during disposal sequence", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -401,6 +409,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); expect(fakePi.sendMessage).not.toHaveBeenCalled(); }); @@ -425,7 +434,7 @@ describe("NotificationService", () => { }); }); - it("sends default failure notification for unregistered process with no config", () => { + it("sends default failure notification for unregistered process with no config", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -446,6 +455,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); // No config registered, but defaults resolve by kind: failure -> turn expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); @@ -456,7 +466,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("sends context notification for unregistered successful process", () => { + it("sends context notification for unregistered successful process", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -477,6 +487,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); // No config registered, defaults resolve by kind: success -> context expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); @@ -487,7 +498,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("unregisters process from registry after process_ended", () => { + it("unregisters process from registry after process_ended", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -510,6 +521,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); expect(registry.get("proc_1")).toBeNull(); @@ -535,7 +547,7 @@ describe("NotificationService", () => { service.dispose(); }); - it("uses custom onSuccess attention from config", () => { + it("uses custom onSuccess attention from config", async () => { const fakeManager = createFakeManager(); const fakePi = createFakePi(); const registry = createNotificationRegistry(); @@ -558,6 +570,7 @@ describe("NotificationService", () => { endReason: "exit", }), }); + await flushQueuedMicrotasks(); expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); const [, options] = fakePi.sendMessage.mock.calls[0]; @@ -588,4 +601,113 @@ describe("NotificationService", () => { service.dispose(); }); + + describe("deferred process_ended and config registration race", () => { + it("downgrades forced failure display to context when onFailure:ignore is registered before microtask", async () => { + const fakeManager = createFakeManager(); + const fakePi = createFakePi(); + const registry = createNotificationRegistry(); + + const service = createNotificationService({ + pi: fakePi as never, + manager: fakeManager as never, + registry, + getProcess: (id) => processes.get(id) ?? null, + }); + + // Simulate: manager.start() emits process_ended synchronously + // (e.g. missing_pid), then executeStart() registers config. + fakeManager.emit({ + type: "process_ended", + info: makeInfo({ + id: "proc_1", + success: false, + exitCode: 1, + endReason: "exit", + }), + }); + + // Config registered after emit but before microtask runs + registry.register("proc_1", { onFailure: "ignore" }); + + await flushQueuedMicrotasks(); + + // failure is forced display, so ignore is upgraded to context + expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); + const [message, options] = fakePi.sendMessage.mock.calls[0]; + expect(message.display).toBe(true); + expect(options.triggerTurn).toBe(false); + expect(options.deliverAs).toBe("steer"); + + service.dispose(); + }); + + it("applies onKilled:context registered after emit but before microtask", async () => { + const fakeManager = createFakeManager(); + const fakePi = createFakePi(); + const registry = createNotificationRegistry(); + + const service = createNotificationService({ + pi: fakePi as never, + manager: fakeManager as never, + registry, + getProcess: (id) => processes.get(id) ?? null, + }); + + fakeManager.emit({ + type: "process_ended", + info: makeInfo({ + id: "proc_1", + status: "killed", + success: false, + endReason: "signal", + exitCode: null, + }), + }); + + // killed is not forced display, so onKilled:context takes effect + registry.register("proc_1", { onKilled: "context" }); + + await flushQueuedMicrotasks(); + + expect(fakePi.sendMessage).toHaveBeenCalledTimes(1); + const [, options] = fakePi.sendMessage.mock.calls[0]; + expect(options.triggerTurn).toBe(false); + expect(options.deliverAs).toBe("steer"); + + service.dispose(); + }); + + it("does not send if disposed before microtask fires", async () => { + const fakeManager = createFakeManager(); + const fakePi = createFakePi(); + const registry = createNotificationRegistry(); + + const service = createNotificationService({ + pi: fakePi as never, + manager: fakeManager as never, + registry, + getProcess: (id) => processes.get(id) ?? null, + }); + + registry.register("proc_1", {}); + + fakeManager.emit({ + type: "process_ended", + info: makeInfo({ + id: "proc_1", + success: false, + exitCode: 1, + endReason: "exit", + }), + }); + + // Dispose before microtask fires + service.dispose(); + + await flushQueuedMicrotasks(); + + expect(fakePi.sendMessage).not.toHaveBeenCalled(); + }); + }); }); diff --git a/extensions/processes/notifications/service.ts b/extensions/processes/notifications/service.ts index fdaa4bc..36c0083 100644 --- a/extensions/processes/notifications/service.ts +++ b/extensions/processes/notifications/service.ts @@ -54,7 +54,13 @@ export function createNotificationService(deps: NotificationServiceDeps): { if (disposed) return; if (event.type === "process_ended") { - handleProcessEnded(event.info); + // Defer to a microtask so that tool code (e.g. executeStart) can + // register the notify config in the registry before we read it. + // manager.start() may synchronously emit process_ended for missing_pid. + queueMicrotask(() => { + if (disposed) return; + handleProcessEnded(event.info); + }); return; } diff --git a/extensions/processes/tools/index.ts b/extensions/processes/tools/index.ts index 5fc6caa..f6e8607 100644 --- a/extensions/processes/tools/index.ts +++ b/extensions/processes/tools/index.ts @@ -22,7 +22,7 @@ type ProcessDetails = StartDetails | ListDetails | StopDetails; export function registerProcessTool( pi: ExtensionAPI, manager: ProcessManager, - _notifications: NotificationRegistry, + notifications: NotificationRegistry, ): void { pi.registerTool( defineTool({ @@ -35,10 +35,12 @@ export function registerProcessTool( "Use process start to start long-running background commands instead of shell background patterns like &, nohup, disown, or setsid.", "Use process list to inspect running background processes before starting duplicates; do not re-summarize the listed processes to the user because the tool output is already visible to them.", "Use process stop to stop a background process started with process start.", + "By default, process failures trigger an agent turn, successes add context only, and killed processes are ignored. Use notify to override.", + "Use notify.logMatches on start to get notified when output matches a readiness or error pattern. Log matchers control agent attention, not display.", ], parameters: ProcessesParams, async execute(_toolCallId, params, _signal, _onUpdate, ctx) { - const details = await execute(params, manager, ctx); + const details = await execute(params, manager, ctx, notifications); return { content: [{ type: "text", text: formatDetails(details) }], details, @@ -54,14 +56,15 @@ async function execute( params: ProcessesParamsType, manager: ProcessManager, ctx: ExtensionContext, + notifications: NotificationRegistry, ): Promise { switch (params.action) { case "start": - return executeStart(params, manager, ctx); + return executeStart(params, manager, ctx, notifications); case "list": return executeList(manager, params); case "stop": - return executeStop(params, manager); + return executeStop(params, manager, notifications); default: throw new Error(`unsupported process action: ${String(params.action)}`); } diff --git a/extensions/processes/tools/start/index.test.ts b/extensions/processes/tools/start/index.test.ts index 0b9ed87..d768929 100644 --- a/extensions/processes/tools/start/index.test.ts +++ b/extensions/processes/tools/start/index.test.ts @@ -3,6 +3,8 @@ import { describe, expect, it, vi } from "vitest"; import type { ProcessManager } from "../../../../src/manager"; import type { ProcessInfo } from "../../../../src/types"; +import type { NotificationRegistry } from "../../notifications/registry"; +import { createNotificationRegistry } from "../../notifications/registry"; import { executeStart } from "."; const processInfo: ProcessInfo = { @@ -25,10 +27,15 @@ const processInfo: ProcessInfo = { const ctx = { cwd: "/repo" } as ExtensionContext; +function createFakeRegistry(): NotificationRegistry { + return createNotificationRegistry(); +} + describe("executeStart", () => { it("validates notify before starting the process", () => { const start = vi.fn(() => processInfo); const manager = { start } as unknown as ProcessManager; + const registry = createFakeRegistry(); expect(() => executeStart( @@ -40,6 +47,7 @@ describe("executeStart", () => { }, manager, ctx, + registry, ), ).toThrow(/not a valid regular expression/); @@ -49,6 +57,7 @@ describe("executeStart", () => { it("returns normalized notify config in start details", () => { const start = vi.fn(() => processInfo); const manager = { start } as unknown as ProcessManager; + const registry = createFakeRegistry(); const details = executeStart( { @@ -59,6 +68,7 @@ describe("executeStart", () => { }, manager, ctx, + registry, ); expect(start).toHaveBeenCalledWith("dev", "pnpm dev", "/repo"); @@ -77,4 +87,53 @@ describe("executeStart", () => { ], }); }); + + it("registers notify config with the notification registry after successful start", () => { + const start = vi.fn(() => processInfo); + const manager = { start } as unknown as ProcessManager; + const registry = createFakeRegistry(); + + const notify = { logMatches: [{ pattern: "ready" }] }; + executeStart( + { + action: "start", + name: "dev", + command: "pnpm dev", + notify, + }, + manager, + ctx, + registry, + ); + + const registered = registry.get("proc_1"); + expect(registered).not.toBeNull(); + expect(registered?.onSuccess).toBe("context"); + expect(registered?.onFailure).toBe("turn"); + expect(registered?.onKilled).toBe("ignore"); + expect(registered?.logMatches).toHaveLength(1); + expect(registered?.logMatches?.[0]?.pattern).toBe("ready"); + }); + + it("does not register notify config when validation fails", () => { + const start = vi.fn(() => processInfo); + const manager = { start } as unknown as ProcessManager; + const registry = createFakeRegistry(); + + expect(() => + executeStart( + { + action: "start", + name: "dev", + command: "pnpm dev", + notify: { logMatches: [{ pattern: "[", mode: "regex" }] }, + }, + manager, + ctx, + registry, + ), + ).toThrow(/not a valid regular expression/); + + expect(registry.get("proc_1")).toBeNull(); + }); }); diff --git a/extensions/processes/tools/start/index.ts b/extensions/processes/tools/start/index.ts index a09b87c..5ad8a34 100644 --- a/extensions/processes/tools/start/index.ts +++ b/extensions/processes/tools/start/index.ts @@ -2,7 +2,10 @@ import type { ExtensionContext } from "@earendil-works/pi-coding-agent"; import type { ProcessManager } from "../../../../src/manager"; import type { ProcessInfo } from "../../../../src/types"; -import type { NotifyConfig } from "../../notifications/registry"; +import type { + NotificationRegistry, + NotifyConfig, +} from "../../notifications/registry"; import { normalizeNotifyConfig } from "../notify"; import type { ProcessesParamsType } from "../schema"; @@ -16,6 +19,7 @@ export function executeStart( params: ProcessesParamsType, manager: ProcessManager, ctx: ExtensionContext, + notifications: NotificationRegistry, ): StartDetails { if (!params.name) { throw new Error("process start requires name"); @@ -27,9 +31,12 @@ export function executeStart( const notify = normalizeNotifyConfig(params.notify); + const process = manager.start(params.name, params.command, ctx.cwd); + notifications.register(process.id, notify); + return { action: "start", - process: manager.start(params.name, params.command, ctx.cwd), + process, notify, }; } diff --git a/extensions/processes/tools/stop/index.test.ts b/extensions/processes/tools/stop/index.test.ts new file mode 100644 index 0000000..ec8c89e --- /dev/null +++ b/extensions/processes/tools/stop/index.test.ts @@ -0,0 +1,174 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { ProcessManager } from "../../../../src/manager"; +import type { KillResult, ProcessInfo } from "../../../../src/types"; +import { createNotificationRegistry } from "../../notifications/registry"; +import { executeStop } from "."; + +function makeInfo(overrides: Partial = {}): ProcessInfo { + return { + id: "proc_1", + name: "dev", + pid: 123, + command: "pnpm dev", + cwd: "/repo", + startTime: 1000, + endTime: 2000, + status: "killed", + exitCode: null, + success: false, + stdoutFile: "/tmp/stdout.log", + stderrFile: "/tmp/stderr.log", + endReason: "signal", + signal: null, + errorMessage: null, + ...overrides, + }; +} + +describe("executeStop", () => { + it("marks intentional stop before killing", async () => { + const info = makeInfo(); + const kill = vi.fn(async () => ({ ok: true, info }) as KillResult); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + registry.register("proc_1", {}); + + await executeStop({ action: "stop", id: "proc_1" }, manager, registry); + + expect(kill).toHaveBeenCalledWith("proc_1"); + }); + + it("marks intentional stop in registry before manager.kill is called", async () => { + let intentionalStopAtCallTime = false; + + const info = makeInfo(); + const kill = vi.fn(async () => { + // At the time kill is called, the intentional stop should already be set + intentionalStopAtCallTime = registry.consumeIntentionalStop("proc_1"); + return { ok: true, info } as KillResult; + }); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + registry.register("proc_1", {}); + + await executeStop({ action: "stop", id: "proc_1" }, manager, registry); + + expect(intentionalStopAtCallTime).toBe(true); + }); + + it("clears intentional stop marker on not_found failure", async () => { + const info = makeInfo(); + const kill = vi.fn( + async () => + ({ + ok: false, + info, + reason: "not_found", + }) as KillResult, + ); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + await executeStop({ action: "stop", id: "proc_1" }, manager, registry); + + // The intentional stop marker should be cleared since kill returned not_found + expect(registry.consumeIntentionalStop("proc_1")).toBe(false); + }); + + it("clears intentional stop marker on error failure", async () => { + const info = makeInfo(); + const kill = vi.fn( + async () => + ({ + ok: false, + info, + reason: "error", + }) as KillResult, + ); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + await executeStop({ action: "stop", id: "proc_1" }, manager, registry); + + expect(registry.consumeIntentionalStop("proc_1")).toBe(false); + }); + + it("preserves intentional stop marker on timeout failure", async () => { + const info = makeInfo(); + const kill = vi.fn( + async () => + ({ + ok: false, + info, + reason: "timeout", + }) as KillResult, + ); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + await executeStop({ action: "stop", id: "proc_1" }, manager, registry); + + // Timeout is not an immediate failure; the marker should remain + expect(registry.consumeIntentionalStop("proc_1")).toBe(true); + }); + + it("clears intentional stop marker when kill returns ok for already-finished process", async () => { + // manager.kill() returns ok:true with status:"exited" for non-live processes + // No process_ended event will fire, so the marker must be cleared here + const info = makeInfo({ + status: "exited", + endReason: "exit", + exitCode: 0, + success: true, + }); + const kill = vi.fn(async () => ({ ok: true, info }) as KillResult); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + await executeStop({ action: "stop", id: "proc_1" }, manager, registry); + + expect(registry.consumeIntentionalStop("proc_1")).toBe(false); + }); + + it("clears intentional stop marker when manager.kill throws", async () => { + const kill = vi.fn(async () => { + throw new Error("unexpected failure"); + }); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + await expect( + executeStop({ action: "stop", id: "proc_1" }, manager, registry), + ).rejects.toThrow(/process stop failed/); + + expect(registry.consumeIntentionalStop("proc_1")).toBe(false); + }); + + it("preserves existing stop result behavior", async () => { + const info = makeInfo(); + const kill = vi.fn(async () => ({ ok: true, info }) as KillResult); + const manager = { kill } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + const result = await executeStop( + { action: "stop", id: "proc_1" }, + manager, + registry, + ); + + expect(result.action).toBe("stop"); + expect(result.result.ok).toBe(true); + expect(result.result.info.id).toBe("proc_1"); + }); + + it("throws when id is missing", async () => { + const manager = { kill: vi.fn() } as unknown as ProcessManager; + const registry = createNotificationRegistry(); + + await expect( + executeStop({ action: "stop" }, manager, registry), + ).rejects.toThrow(/requires id/); + }); +}); diff --git a/extensions/processes/tools/stop/index.ts b/extensions/processes/tools/stop/index.ts index 42c4a12..1f57fdb 100644 --- a/extensions/processes/tools/stop/index.ts +++ b/extensions/processes/tools/stop/index.ts @@ -1,5 +1,7 @@ import type { ProcessManager } from "../../../../src/manager"; import type { KillResult } from "../../../../src/types"; +import { LIVE_STATUSES } from "../../../../src/types"; +import type { NotificationRegistry } from "../../notifications/registry"; import type { ProcessesParamsType } from "../schema"; export interface StopDetails { @@ -10,14 +12,33 @@ export interface StopDetails { export async function executeStop( params: ProcessesParamsType, manager: ProcessManager, + notifications: NotificationRegistry, ): Promise { if (!params.id) { throw new Error("process stop requires id"); } + notifications.markIntentionalStop(params.id); + + let result: KillResult; + try { + result = await manager.kill(params.id); + } catch { + notifications.consumeIntentionalStop(params.id); + throw new Error(`process stop failed for ${params.id}`); + } + + if (!result.ok) { + if (result.reason === "not_found" || result.reason === "error") { + notifications.consumeIntentionalStop(params.id); + } + } else if (!LIVE_STATUSES.has(result.info.status)) { + notifications.consumeIntentionalStop(params.id); + } + return { action: "stop", - result: await manager.kill(params.id), + result, }; } diff --git a/tests/utils/async.ts b/tests/utils/async.ts new file mode 100644 index 0000000..683c3b0 --- /dev/null +++ b/tests/utils/async.ts @@ -0,0 +1,3 @@ +export function flushQueuedMicrotasks(): Promise { + return new Promise((resolve) => queueMicrotask(resolve)); +} -- 2.51.2