From 22547a3c957d36e7b1218a5e695e2f6147a262ae Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 23 Oct 2026 11:25:05 -0700 Subject: [PATCH 1/4] test(e2e): keep an approved mcp call settled across a reload --- e2e/reactor-approval-reload.test.ts | 138 ++++++++++++++++++++++++++-- 1 file changed, 129 insertions(+), 9 deletions(-) diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts index 0ccf547f7..aa24c5f0e 100644 --- a/e2e/reactor-approval-reload.test.ts +++ b/e2e/reactor-approval-reload.test.ts @@ -10,6 +10,8 @@ import { afterEach, describe, expect, spyOn, test } from "bun:test"; import type { ReactorEmittedEvent } from "@intx/inference"; import type { ConversationTurn } from "@intx/types/runtime"; +import type { MCPClient } from "../src/mcp/client.js"; +import { mcpClientTools } from "../src/mcp/plugin.js"; import { createPermissionGate } from "../src/permission/gate.js"; import { createReactorAuthorize } from "../src/permission/reactor-authorize.js"; import { @@ -27,6 +29,50 @@ import { // Ask-tier, but answered locally: nothing listens on the discard port. const CLOSED_PORT_PROBE = "curl -sS --max-time 1 http://127.0.0.1:9/"; +const SAVE_ISSUE = "mcp__linear__save_issue"; +const CREATE_LABEL = "mcp__linear__create_label"; + +function linearClient(calls: Record[]): MCPClient { + return { + serverName: "linear", + tools: [ + { + name: "save_issue", + description: "create or update a linear issue", + inputSchema: { + type: "object", + properties: { title: { type: "string" } }, + }, + }, + ], + async callResult(_toolName, args) { + calls.push(args); + return { blocks: [{ type: "text", text: "SAVED-1" }], isError: false }; + }, + async close() { + return undefined; + }, + }; +} + +function labelClient(): MCPClient { + return { + serverName: "linear", + tools: [ + { + name: "create_label", + description: "create a linear label", + inputSchema: { type: "object", properties: {} }, + }, + ], + async callResult() { + return { blocks: [], isError: false }; + }, + async close() { + return undefined; + }, + }; +} // Reactor gates run on the real clock (setTimeout + Date.now), not the // inference harness clock, so the approval deadline is driven here. Only @@ -155,29 +201,45 @@ interface Reloaded { clock: ReturnType; } +interface ApproveThenReloadOpts { + /** Runs on the opened session before the call is scripted. */ + prepare?: (session: IntegrationSession) => Promise; + /** Runs while the approval prompt is live, before it is approved. */ + afterApprovalOpened?: (session: IntegrationSession) => void; + /** + * Rebuild over the same toolset, as the TUI does for an MCP tool-set + * change. Reopening from the store builds a fresh toolset, which would + * drop tools mounted on the old one. + */ + rebuildInPlace?: boolean; +} + /** - * Parks `command` on an ask, approves it, and once `reloadWhen` holds for - * the approved run, rebuilds the agent on the same store. The model turn - * after the approved result stays open, so the reload always lands before - * the reply terminal commits the cycle. + * Parks `call` on an ask, approves it, and once `reloadWhen` holds for the + * approved run, rebuilds the agent on the same store. The model turn after + * the approved result stays open, so the reload always lands before the + * reply terminal commits the cycle. */ async function approveThenReload( - command: string, + call: { name: string; args: Record }, reloadWhen: ( session: IntegrationSession, events: ReactorEmittedEvent[], ) => boolean, + opts: ApproveThenReloadOpts = {}, ): Promise { const ctx = deferredApprovalGate(); const sessionOpts = { permissionGate: ctx.gate, authorize: createReactorAuthorize(ctx.gate), + pinToolSetAtBuild: true, }; let session = await openIntegrationSession(sessionOpts); cleanups.push(() => closeIntegrationSession(session)); + await opts.prepare?.(session); session.harness.scenario.replyOnce("anthropic", { - toolCalls: [{ name: "run_shell", args: { command } }], + toolCalls: [call], }); session.harness.scenario.stall(); @@ -201,6 +263,7 @@ async function approveThenReload( }); const handling = resume.handle(turn.result); await waitFor(ctx.asked, "approval prompt"); + opts.afterApprovalOpened?.(session); ctx.approve(); expect(await handling).toBe(true); void session.harness @@ -211,7 +274,8 @@ async function approveThenReload( const clock = holdLongTimers(); cleanups.push(() => clock.restore()); - session = await rebuildIntegrationSession(session, sessionOpts); + if (opts.rebuildInPlace === true) await session.rebuild(); + else session = await rebuildIntegrationSession(session, sessionOpts); return { session, events: collect(session), callId, deadline, clock }; } @@ -229,7 +293,7 @@ describe("integration: approving an ask, then reloading the agent", () => { "after the approved result: the rebuilt agent continues, and the old deadline adds no result", async () => { const reloaded = await approveThenReload( - CLOSED_PORT_PROBE, + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, (session) => session.harness.scenario.matchedRequests().length === 2, ); const { session, callId } = reloaded; @@ -252,6 +316,59 @@ describe("integration: approving an ask, then reloading the agent", () => { 20_000, ); + test.serial( + "an approved parked MCP call remains settled across a pending tool-set change, reopen, and original deadline", + async () => { + const calls: Record[] = []; + const reloaded = await approveThenReload( + { name: SAVE_ISSUE, args: { title: "late approval" } }, + (session) => session.harness.scenario.matchedRequests().length === 2, + { + prepare: async (session) => { + session.toolset.dynamicRunner.addTools( + mcpClientTools(linearClient(calls)), + ); + await session.rebuild(); + }, + afterApprovalOpened: (session) => { + // The prompt is live before this MCP change requests its reload. + session.toolset.dynamicRunner.addTools( + mcpClientTools(labelClient()), + ); + expect( + session.toolset.dynamicRunner.agentToolSetDrift().added, + ).toContain(CREATE_LABEL); + }, + rebuildInPlace: true, + }, + ); + const { session, callId } = reloaded; + const rebuiltTools = session.toolset.dynamicRunner + .agentDefinitions() + .map((definition) => definition.name); + expect(rebuiltTools).toContain(SAVE_ISSUE); + expect(rebuiltTools).toContain(CREATE_LABEL); + expect(calls).toEqual([{ title: "late approval" }]); + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + + session.harness.scenario.replyOnce("anthropic", { text: "Saved." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Saved."); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + const results = toolResultsFor(await session.agent.history(), callId); + expect(calls).toEqual([{ title: "late approval" }]); + expect(results).toHaveLength(1); + expect(results.join("")).not.toContain("approval timed out"); + }, + 20_000, + ); + // The reload interrupts the approved call, which the closing reactor // answers itself. What an interrupted call should report is not pinned // here; only that the old deadline cannot add a second result. @@ -259,7 +376,10 @@ describe("integration: approving an ask, then reloading the agent", () => { "during the approved call: the old deadline adds no second result", async () => { const reloaded = await approveThenReload( - `sleep 20; ${CLOSED_PORT_PROBE}`, + { + name: "run_shell", + args: { command: `sleep 20; ${CLOSED_PORT_PROBE}` }, + }, (_session, events) => events.some((e) => e.type === "tool.start"), ); const { session, callId } = reloaded; From 925b7552d857d185af87caf18e35b991ce98b46e Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 23 Oct 2026 11:25:07 -0700 Subject: [PATCH 2/4] test(tui): defer mcp rebuild until an approval-resumed run ends --- src/tui/runner/exit.test.ts | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/tui/runner/exit.test.ts b/src/tui/runner/exit.test.ts index db69ba7a8..462999141 100644 --- a/src/tui/runner/exit.test.ts +++ b/src/tui/runner/exit.test.ts @@ -1031,6 +1031,32 @@ describe("reload-if-idle waits for the reactor's message run", () => { expect(state.fatalBuildError).toBeNull(); }); + test("an MCP change requested while its approval is live defers rebuild until the resumed run ends", async () => { + const { state, services, push, builds } = await lifecycleWithOpenRun(); + const mcp = wireLateMcp(state, services); + // `runWhileAgentBusy` is the approval surface's occupancy bracket: the + // operator has a live ask while the parked reactor run remains open. + await runWhileAgentBusy(state, async () => { + mcp.connectLateMCPServer({ + name: "linear", + url: "https://l.test/mcp", + }); + await drainStream(); + await services.sessionOps.awaitTail(); + expect(state.pendingReload).toBe(true); + expect(builds()).toBe(1); + }); + + // The resumed MCP tool and its next model turn remain inside run-1, so + // reload cannot close the agent until the reactor reports this boundary. + push(runEnded("run-1")); + await drainStream(); + await services.sessionOps.awaitTail(); + expect(builds()).toBe(2); + expect(state.pendingReload).toBe(false); + expect(state.fatalBuildError).toBeNull(); + }); + test("a run that ends into an immediately following run rebuilds only after the successor ends", async () => { const { state, services, push, builds } = await lifecycleWithOpenRun(); state.pendingReload = true; From 4dd4fb469a73296a510acd397b2ef482f6e5dc8e Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 23 Oct 2026 15:41:14 -0700 Subject: [PATCH 3/4] test(agent): widen ceiling-resume margin in stale-resume watchdog test B's ceiling fires at ~240ms and the remaining ~120ms budget expires at ~360ms; the old 370ms check left 10ms of slack and flaked on CI. --- src/agent/tool-execution-watchdog.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent/tool-execution-watchdog.test.ts b/src/agent/tool-execution-watchdog.test.ts index 595d5c9c8..085fba864 100644 --- a/src/agent/tool-execution-watchdog.test.ts +++ b/src/agent/tool-execution-watchdog.test.ts @@ -650,7 +650,7 @@ describe("tool execution watchdog", () => { await new Promise((r) => setTimeout(r, 150)); expect(budget.signal.aborted).toBe(false); // B's ceiling eventually force-resumes it on its own account. - await new Promise((r) => setTimeout(r, 70)); + await new Promise((r) => setTimeout(r, 120)); expect(budget.signal.aborted).toBe(true); budget.resume(tokenB); budget.dispose(); From 9ca3fa366456eaca3b0f67748c115c8402067dd4 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Sat, 24 Oct 2026 13:28:14 -0700 Subject: [PATCH 4/4] test(e2e): keep rejected, timed-out and rebuilt approvals settled --- e2e/reactor-approval-reload.test.ts | 223 ++++++++++++++++++++++++---- 1 file changed, 193 insertions(+), 30 deletions(-) diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts index aa24c5f0e..816330907 100644 --- a/e2e/reactor-approval-reload.test.ts +++ b/e2e/reactor-approval-reload.test.ts @@ -1,6 +1,7 @@ /** - * An approved ask whose agent is rebuilt from the same store before the next - * cycle commit (the TUI's deferred reload) must not leave the parked call's + * A settled ask (approved, rejected, or timed out) whose agent is rebuilt + * from the same store, before the next cycle commit or at the run's end + * where the TUI's deferred reload fires, must not leave the parked call's * pending operation behind: the rebuilt reactor would re-arm its gate and, at * the original deadline, answer an already-answered call with "approval timed * out". @@ -151,6 +152,7 @@ function deferredApprovalGate() { }), }), approve: () => release?.({ allow: true }), + reject: () => release?.({ allow: false }), }; } @@ -201,10 +203,21 @@ interface Reloaded { clock: ReturnType; } -interface ApproveThenReloadOpts { +interface DecideThenReloadOpts { + /** + * How the parked call is settled. `expire` never opens the approval + * prompt; the reactor's own deadline answers the call. + */ + decision?: "approve" | "reject" | "expire"; + /** + * The model's reply after the settled call. Without it the next model + * turn stays open, so the reload lands before the cycle commits; with it + * the run can end, which is where the TUI's deferred reload fires. + */ + replyAfterDecision?: string; /** Runs on the opened session before the call is scripted. */ prepare?: (session: IntegrationSession) => Promise; - /** Runs while the approval prompt is live, before it is approved. */ + /** Runs while the approval prompt is live, before it is decided. */ afterApprovalOpened?: (session: IntegrationSession) => void; /** * Rebuild over the same toolset, as the TUI does for an MCP tool-set @@ -215,19 +228,23 @@ interface ApproveThenReloadOpts { } /** - * Parks `call` on an ask, approves it, and once `reloadWhen` holds for the - * approved run, rebuilds the agent on the same store. The model turn after - * the approved result stays open, so the reload always lands before the - * reply terminal commits the cycle. + * Parks `call` on an ask, settles it per `opts.decision` (approve by + * default), and once `reloadWhen` holds for the settled run, rebuilds the + * agent on the same store. `reloadWhen` sees only events from after the + * decision. */ -async function approveThenReload( +async function decideThenReload( call: { name: string; args: Record }, reloadWhen: ( session: IntegrationSession, events: ReactorEmittedEvent[], ) => boolean, - opts: ApproveThenReloadOpts = {}, + opts: DecideThenReloadOpts = {}, ): Promise { + // Held from before the park so the expire case can fire the original gate + // timer; the approve and reject cases clear it unfired. + const clock = holdLongTimers(); + cleanups.push(() => clock.restore()); const ctx = deferredApprovalGate(); const sessionOpts = { permissionGate: ctx.gate, @@ -241,7 +258,11 @@ async function approveThenReload( session.harness.scenario.replyOnce("anthropic", { toolCalls: [call], }); - session.harness.scenario.stall(); + if (opts.replyAfterDecision === undefined) session.harness.scenario.stall(); + else + session.harness.scenario.replyOnce("anthropic", { + text: opts.replyAfterDecision, + }); const turn = await runUntilSuspended(session, "Probe the local port."); if (turn.result.type !== "suspended") throw new Error("did not park"); @@ -255,33 +276,48 @@ async function approveThenReload( throw new Error("parked call has no persisted call id or deadline"); } - const resume = createApprovalResume({ - getAgent: () => session.agent, - gate: ctx.gate, - resolveParkedCallId: (id) => - resolveParkedCallIdFromStore(session.storage, id), - }); - const handling = resume.handle(turn.result); - await waitFor(ctx.asked, "approval prompt"); - opts.afterApprovalOpened?.(session); - ctx.approve(); - expect(await handling).toBe(true); + const decision = opts.decision ?? "approve"; + const decidedAt = turn.events.length; + if (decision === "expire") { + clock.advanceTo(deadline + 1_000); + } else { + const resume = createApprovalResume({ + getAgent: () => session.agent, + gate: ctx.gate, + resolveParkedCallId: (id) => + resolveParkedCallIdFromStore(session.storage, id), + }); + const handling = resume.handle(turn.result); + await waitFor(ctx.asked, "approval prompt"); + opts.afterApprovalOpened?.(session); + if (decision === "approve") ctx.approve(); + else ctx.reject(); + expect(await handling).toBe(true); + } void session.harness .run({ wallClockBudgetMs: 30_000 }) .catch(() => undefined); const before = session; - await waitFor(() => reloadWhen(before, turn.events), "reload point"); + await waitFor( + () => reloadWhen(before, turn.events.slice(decidedAt)), + "reload point", + ); - const clock = holdLongTimers(); - cleanups.push(() => clock.restore()); if (opts.rebuildInPlace === true) await session.rebuild(); else session = await rebuildIntegrationSession(session, sessionOpts); return { session, events: collect(session), callId, deadline, clock }; } +// Past both the original deadline and a full approval window freshly armed +// at the reload, so either re-arm would have fired. +const FRESH_WINDOW_PASSED_MS = 2 * 60 * 60_000; + /** Moves past the original approval deadline and lets the reactor react. */ -async function passDeadline(reloaded: Reloaded): Promise { - reloaded.clock.advanceTo(reloaded.deadline + 1_000); +async function passDeadline( + reloaded: Reloaded, + at = reloaded.deadline + 1_000, +): Promise { + reloaded.clock.advanceTo(at); await reloaded.session.harness .run({ wallClockBudgetMs: 2_000 }) .catch(() => undefined); @@ -292,7 +328,7 @@ describe("integration: approving an ask, then reloading the agent", () => { test.serial( "after the approved result: the rebuilt agent continues, and the old deadline adds no result", async () => { - const reloaded = await approveThenReload( + const reloaded = await decideThenReload( { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, (session) => session.harness.scenario.matchedRequests().length === 2, ); @@ -320,7 +356,7 @@ describe("integration: approving an ask, then reloading the agent", () => { "an approved parked MCP call remains settled across a pending tool-set change, reopen, and original deadline", async () => { const calls: Record[] = []; - const reloaded = await approveThenReload( + const reloaded = await decideThenReload( { name: SAVE_ISSUE, args: { title: "late approval" } }, (session) => session.harness.scenario.matchedRequests().length === 2, { @@ -375,7 +411,7 @@ describe("integration: approving an ask, then reloading the agent", () => { test.serial( "during the approved call: the old deadline adds no second result", async () => { - const reloaded = await approveThenReload( + const reloaded = await decideThenReload( { name: "run_shell", args: { command: `sleep 20; ${CLOSED_PORT_PROBE}` }, @@ -397,4 +433,131 @@ describe("integration: approving an ask, then reloading the agent", () => { }, 20_000, ); + + // The TUI defers an MCP tool-set rebuild until the approval-resumed run + // ends (reloadIfIdle in src/tui/runner/exit.ts, covered by its runner + // tests). The harness has no TUI runner, so this rebuilds in place at that + // same boundary, gated on an added tool, one of the drifts reloadIfIdle + // rebuilds for. + test.serial( + "an approved MCP call stays settled when the deferred tool-set rebuild fires at the run's end", + async () => { + const calls: Record[] = []; + const reloaded = await decideThenReload( + { name: SAVE_ISSUE, args: { title: "late approval" } }, + (session, events) => + events.some((e) => e.type === "message.run.ended") && + session.toolset.dynamicRunner.agentToolSetDrift().added.length > 0, + { + replyAfterDecision: "Saved.", + prepare: async (session) => { + session.toolset.dynamicRunner.addTools( + mcpClientTools(linearClient(calls)), + ); + await session.rebuild(); + }, + afterApprovalOpened: (session) => { + session.toolset.dynamicRunner.addTools( + mcpClientTools(labelClient()), + ); + }, + rebuildInPlace: true, + }, + ); + const { session, callId } = reloaded; + expect(session.toolset.dynamicRunner.agentToolSetDrift().added).toEqual( + [], + ); + expect( + session.toolset.dynamicRunner + .agentDefinitions() + .map((definition) => definition.name), + ).toContain(CREATE_LABEL); + expect(calls).toEqual([{ title: "late approval" }]); + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + + session.harness.scenario.replyOnce("anthropic", { text: "Done." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Done."); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + const results = toolResultsFor(await session.agent.history(), callId); + expect(calls).toEqual([{ title: "late approval" }]); + expect(results).toHaveLength(1); + expect(results.join("")).not.toContain("approval timed out"); + }, + 20_000, + ); +}); + +// Reloads at the run's end, where the TUI's deferred reload fires. A reload +// after the decision but before the cycle commits is not covered here: the +// rejected or timed-out result is not yet persisted, so the rebuilt agent +// loads the parked call with no result at all. +describe("integration: an ask settled without running, then reloading the agent at the run's end", () => { + const runEnded = ( + _session: IntegrationSession, + events: ReactorEmittedEvent[], + ): boolean => events.some((e) => e.type === "message.run.ended"); + + test.serial( + "after a rejection: the rebuilt agent continues, and the old deadline adds no result", + async () => { + const reloaded = await decideThenReload( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + runEnded, + { decision: "reject", replyAfterDecision: "Skipped." }, + ); + const { session, callId } = reloaded; + const settled = toolResultsFor(await session.agent.history(), callId); + expect(settled).toHaveLength(1); + expect(settled.join("")).toContain("denied by approver"); + + session.harness.scenario.replyOnce("anthropic", { text: "Done." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Done."); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + const results = toolResultsFor(await session.agent.history(), callId); + expect(results).toEqual(settled); + expect(results.join("")).not.toContain("approval timed out"); + }, + 20_000, + ); + + test.serial( + "after the deadline answers it: the rebuilt agent continues, and no later deadline adds a second result", + async () => { + const reloaded = await decideThenReload( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + runEnded, + { decision: "expire", replyAfterDecision: "Timed out." }, + ); + const { session, callId } = reloaded; + const settled = toolResultsFor(await session.agent.history(), callId); + expect(settled).toHaveLength(1); + expect(settled.join("")).toContain("approval timed out"); + + session.harness.scenario.replyOnce("anthropic", { text: "Done." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Done."); + + await passDeadline(reloaded, reloaded.deadline + FRESH_WINDOW_PASSED_MS); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + expect(toolResultsFor(await session.agent.history(), callId)).toEqual( + settled, + ); + }, + 20_000, + ); });