From adbaae68daee7577b2dc93bd12507bfb67a2ca84 Mon Sep 17 00:00:00 2001 From: Dillon Mulroy Date: Wed, 8 Jul 2026 18:55:14 -0400 Subject: fix: reanchor pi status after session replacement (#1189) * fix: reanchor pi status after session replacement refs #943 * fix: guard active pi sessions from nested startup refs #943 * fix: serialize pi session replacement reports refs #943 --- src/integration/assets/herdr-agent-state.test.ts | 213 +++++++++++++++++------ 1 file changed, 161 insertions(+), 52 deletions(-) (limited to 'src/integration/assets/herdr-agent-state.test.ts') diff --git a/src/integration/assets/herdr-agent-state.test.ts b/src/integration/assets/herdr-agent-state.test.ts index 69cb8894..077d1894 100644 --- a/src/integration/assets/herdr-agent-state.test.ts +++ b/src/integration/assets/herdr-agent-state.test.ts @@ -48,42 +48,13 @@ function importFresh(modulePath: string) { return import(`${modulePath}?test=${importCounter}`); } -for (const integration of integrations) { - test(`${integration.name} reload preserves working state when the agent is active`, async () => { - const recordingSocketPath = join( - tmpdir(), - `herdr-${integration.name.toLowerCase().replaceAll(" ", "-")}-${process.pid}.sock`, - ); - socketPath = recordingSocketPath; - await rm(recordingSocketPath, { force: true }); - - const requests: unknown[] = []; - const recordingServer = createServer((socket) => { - let input = ""; - socket.setEncoding("utf8"); - socket.on("data", (chunk) => { - input += chunk; - const newline = input.indexOf("\n"); - if (newline === -1) { - return; - } - requests.push(JSON.parse(input.slice(0, newline))); - socket.end("{}\n"); - }); - }); - server = recordingServer; - await new Promise((resolve, reject) => { - recordingServer.once("error", reject); - recordingServer.listen(recordingSocketPath, resolve); - }); +type Handler = (event: unknown, context: unknown) => unknown; - process.env.HERDR_ENV = "1"; - process.env.HERDR_SOCKET_PATH = recordingSocketPath; - process.env.HERDR_PANE_ID = "test:p1"; - - type Handler = (event: unknown, context: unknown) => unknown; - const handlers = new Map(); - const pi = { +function createExtensionHarness() { + const handlers = new Map(); + return { + handlers, + pi: { on(event: string, handler: Handler) { handlers.set(event, handler); }, @@ -92,7 +63,50 @@ for (const integration of integrations) { return () => {}; }, }, - }; + }, + }; +} + +function configureIntegrationEnvironment(recordingSocketPath: string) { + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = recordingSocketPath; + process.env.HERDR_PANE_ID = "test:p1"; +} + +async function startRecordingServer(name: string): Promise { + const recordingSocketPath = join(tmpdir(), `herdr-${name}-${process.pid}.sock`); + socketPath = recordingSocketPath; + await rm(recordingSocketPath, { force: true }); + + const requests: unknown[] = []; + const recordingServer = createServer((socket) => { + let input = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + input += chunk; + const newline = input.indexOf("\n"); + if (newline === -1) { + return; + } + requests.push(JSON.parse(input.slice(0, newline))); + socket.end("{}\n"); + }); + }); + server = recordingServer; + await new Promise((resolve, reject) => { + recordingServer.once("error", reject); + recordingServer.listen(recordingSocketPath, resolve); + }); + configureIntegrationEnvironment(recordingSocketPath); + return requests; +} + +for (const integration of integrations) { + test(`${integration.name} reload preserves working state when the agent is active`, async () => { + const requests = await startRecordingServer( + integration.name.toLowerCase().replaceAll(" ", "-"), + ); + const { handlers, pi } = createExtensionHarness(); const { default: install } = await importFresh(integration.modulePath); install(pi); @@ -133,6 +147,115 @@ for (const integration of integrations) { }); } +test("Pi reports the session replacement source", async () => { + const requests = await startRecordingServer("pi-session-source"); + const { handlers, pi } = createExtensionHarness(); + + const { default: install } = await importFresh("./pi/herdr-agent-state.ts"); + install(pi); + + const sessionStart = handlers.get("session_start"); + expect(sessionStart).toBeDefined(); + await sessionStart?.( + { reason: "new" }, + { + hasUI: true, + isIdle: () => true, + sessionManager: { + getSessionFile: () => "/tmp/pi-new.jsonl", + getSessionId: () => "pi-new", + }, + }, + ); + + const reportedSession = () => + requests.find((request) => isRecord(request) && request.method === "pane.report_agent_session"); + const deadline = Date.now() + 1_000; + while (Date.now() < deadline && reportedSession() === undefined) { + await Bun.sleep(5); + } + + const request = reportedSession(); + expect(request).toBeDefined(); + expect(isRecord(request) && isRecord(request.params) ? request.params.session_start_source : null) + .toBe("new"); +}); + +test("Pi waits for a replacement session report before publishing state", async () => { + const recordingSocketPath = join(tmpdir(), `herdr-pi-session-order-${process.pid}.sock`); + socketPath = recordingSocketPath; + await rm(recordingSocketPath, { force: true }); + + const requests: unknown[] = []; + let acknowledgeSessionReport: (() => void) | undefined; + const recordingServer = createServer((socket) => { + let input = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + input += chunk; + const newline = input.indexOf("\n"); + if (newline === -1) { + return; + } + const request = JSON.parse(input.slice(0, newline)); + requests.push(request); + if (isRecord(request) && request.method === "pane.report_agent_session") { + acknowledgeSessionReport = () => socket.end("{}\n"); + return; + } + socket.end("{}\n"); + }); + }); + server = recordingServer; + await new Promise((resolve, reject) => { + recordingServer.once("error", reject); + recordingServer.listen(recordingSocketPath, resolve); + }); + + configureIntegrationEnvironment(recordingSocketPath); + const { handlers, pi } = createExtensionHarness(); + const { default: install } = await importFresh("./pi/herdr-agent-state.ts"); + install(pi); + + const sessionStart = handlers.get("session_start"); + expect(sessionStart).toBeDefined(); + const sessionStartResult = sessionStart?.( + { reason: "new" }, + { + hasUI: true, + isIdle: () => false, + sessionManager: { + getSessionFile: () => "/tmp/pi-new.jsonl", + getSessionId: () => "pi-new", + }, + }, + ); + + const deadline = Date.now() + 1_000; + while (Date.now() < deadline && acknowledgeSessionReport === undefined) { + await Bun.sleep(5); + } + expect(acknowledgeSessionReport).toBeDefined(); + expect( + requests.some((request) => isRecord(request) && request.method === "pane.report_agent"), + ).toBe(false); + + acknowledgeSessionReport?.(); + await sessionStartResult; + + const stateDeadline = Date.now() + 1_000; + while ( + Date.now() < stateDeadline && + !requests.some((request) => isRecord(request) && request.method === "pane.report_agent") + ) { + await Bun.sleep(5); + } + expect(requests.map((request) => (isRecord(request) ? request.method : undefined))).toEqual([ + "pane.report_agent_session", + "pane.report_agent", + ]); +}); + test("Pi retries working state after an unanswered socket attempt", async () => { const recordingSocketPath = join(tmpdir(), `herdr-pi-retry-${process.pid}.sock`); socketPath = recordingSocketPath; @@ -167,22 +290,8 @@ test("Pi retries working state after an unanswered socket attempt", async () => recordingServer.listen(recordingSocketPath, resolve); }); - process.env.HERDR_ENV = "1"; - process.env.HERDR_SOCKET_PATH = recordingSocketPath; - process.env.HERDR_PANE_ID = "test:p1"; - - type Handler = (event: unknown, context: unknown) => unknown; - const handlers = new Map(); - const pi = { - on(event: string, handler: Handler) { - handlers.set(event, handler); - }, - events: { - on() { - return () => {}; - }, - }, - }; + configureIntegrationEnvironment(recordingSocketPath); + const { handlers, pi } = createExtensionHarness(); const { default: install } = await importFresh("./pi/herdr-agent-state.ts"); install(pi); -- cgit v1.2.3