Skip to content

Commit 19ce192

Browse files
committed
test(e2e): cover reloading the agent after an approved ask
1 parent aad53cf commit 19ce192

2 files changed

Lines changed: 306 additions & 2 deletions

File tree

‎e2e/integration-harness.ts‎

Lines changed: 26 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -149,6 +149,8 @@ export interface OpenIntegrationSessionOpts {
149149
retryPolicy?: RetryPolicy;
150150
/** Fields merged over harness.deps, e.g. a real scheduler when retry delays must elapse. */
151151
depsOverrides?: Partial<Dependencies>;
152+
/** Reuse an existing session cwd, so the agent loads that session's store. */
153+
cwd?: string;
152154
}
153155

154156
export async function openIntegrationSession(
@@ -170,7 +172,8 @@ export async function openIntegrationSession(
170172
throw new Error("openIntegrationSession requires at least one source");
171173
}
172174
const defaultSource = opts.defaultSourceId ?? primarySource.id;
173-
const cwd = mkdtempSync(join(tmpdir(), "corbits-integration-cwd-"));
175+
const cwd =
176+
opts.cwd ?? mkdtempSync(join(tmpdir(), "corbits-integration-cwd-"));
174177
const workdir = join(cwd, ".agent-state", "integration-session");
175178
const evidenceArchiveHolder: { current: CompactionArchive | undefined } = {
176179
current: undefined,
@@ -380,18 +383,39 @@ export async function openIntegrationSession(
380383
};
381384
}
382385

383-
export async function closeIntegrationSession(
386+
async function releaseIntegrationAgent(
384387
session: IntegrationSession,
385388
): Promise<void> {
386389
try {
387390
await session.agent.close();
388391
} finally {
389392
await session.toolset.dispose();
390393
session.harness.dispose();
394+
}
395+
}
396+
397+
export async function closeIntegrationSession(
398+
session: IntegrationSession,
399+
): Promise<void> {
400+
try {
401+
await releaseIntegrationAgent(session);
402+
} finally {
391403
rmSync(session.cwd, { recursive: true, force: true });
392404
}
393405
}
394406

407+
/**
408+
* The TUI reload path: close the agent without a final commit, then build a
409+
* fresh one on the same session store. The new session owns the cwd.
410+
*/
411+
export async function rebuildIntegrationSession(
412+
session: IntegrationSession,
413+
opts: OpenIntegrationSessionOpts,
414+
): Promise<IntegrationSession> {
415+
await releaseIntegrationAgent(session);
416+
return openIntegrationSession({ ...opts, cwd: session.cwd });
417+
}
418+
395419
export interface TurnResult {
396420
events: ReactorEmittedEvent[];
397421
reply: string;
Lines changed: 280 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,280 @@
1+
/**
2+
* An approved ask whose agent is rebuilt from the same store before the next
3+
* cycle commit (the TUI's deferred reload) must not leave the parked call's
4+
* pending operation behind: the rebuilt reactor would re-arm its gate and, at
5+
* the original deadline, answer an already-answered call with "approval timed
6+
* out".
7+
*/
8+
import { afterEach, describe, expect, spyOn, test } from "bun:test";
9+
10+
import type { ReactorEmittedEvent } from "@intx/inference";
11+
import type { ConversationTurn } from "@intx/types/runtime";
12+
13+
import { createPermissionGate } from "../src/permission/gate.js";
14+
import { createReactorAuthorize } from "../src/permission/reactor-authorize.js";
15+
import {
16+
createApprovalResume,
17+
resolveParkedCallIdFromStore,
18+
} from "../src/session/approval-resume.js";
19+
import {
20+
closeIntegrationSession,
21+
openIntegrationSession,
22+
rebuildIntegrationSession,
23+
runUntilDone,
24+
runUntilSuspended,
25+
type IntegrationSession,
26+
} from "./integration-harness.js";
27+
28+
// Ask-tier, but answered locally: nothing listens on the discard port.
29+
const CLOSED_PORT_PROBE = "curl -sS --max-time 1 http://127.0.0.1:9/";
30+
31+
// Reactor gates run on the real clock (setTimeout + Date.now), not the
32+
// inference harness clock, so the approval deadline is driven here. Only
33+
// timers at least this long are held; everything shorter keeps real timing.
34+
const HELD_TIMER_MIN_MS = 30 * 60_000;
35+
36+
interface HeldTimer {
37+
dueAt: number;
38+
fire: () => void;
39+
cleared: boolean;
40+
}
41+
42+
function holdLongTimers() {
43+
const realNow = Date.now.bind(Date);
44+
const realSetTimeout = globalThis.setTimeout;
45+
const realClearTimeout = globalThis.clearTimeout;
46+
let offset = 0;
47+
const held = new Map<object, HeldTimer>();
48+
const now = spyOn(Date, "now").mockImplementation(() => realNow() + offset);
49+
const set = spyOn(globalThis, "setTimeout").mockImplementation(((
50+
fire: () => void,
51+
delay?: number,
52+
...args: unknown[]
53+
) => {
54+
if (delay === undefined || delay < HELD_TIMER_MIN_MS) {
55+
return realSetTimeout(fire, delay, ...args);
56+
}
57+
const handle = {};
58+
held.set(handle, { dueAt: Date.now() + delay, fire, cleared: false });
59+
return handle;
60+
}) as typeof setTimeout);
61+
const clear = spyOn(globalThis, "clearTimeout").mockImplementation(((
62+
handle: Parameters<typeof clearTimeout>[0],
63+
) => {
64+
const timer =
65+
typeof handle === "object" && handle !== null
66+
? held.get(handle)
67+
: undefined;
68+
if (timer !== undefined) {
69+
timer.cleared = true;
70+
return;
71+
}
72+
realClearTimeout(handle);
73+
}) as typeof clearTimeout);
74+
return {
75+
/** Move the clock to `at` and fire every held timer due by then. */
76+
advanceTo(at: number): void {
77+
offset = at - realNow();
78+
for (const timer of held.values()) {
79+
if (timer.cleared || timer.dueAt > at) continue;
80+
timer.cleared = true;
81+
timer.fire();
82+
}
83+
},
84+
restore(): void {
85+
clear.mockRestore();
86+
set.mockRestore();
87+
now.mockRestore();
88+
},
89+
};
90+
}
91+
92+
function deferredApprovalGate() {
93+
let release: ((outcome: { allow: boolean }) => void) | undefined;
94+
return {
95+
asked: () => release !== undefined,
96+
gate: createPermissionGate({
97+
approvals: [],
98+
interactive: true,
99+
skipPermissions: false,
100+
auto: false,
101+
reactorGated: true,
102+
requestApproval: () =>
103+
new Promise((resolve) => {
104+
release = resolve;
105+
}),
106+
}),
107+
approve: () => release?.({ allow: true }),
108+
};
109+
}
110+
111+
async function waitFor(
112+
predicate: () => boolean,
113+
what: string,
114+
ms = 5_000,
115+
): Promise<void> {
116+
const deadline = performance.now() + ms;
117+
while (!predicate()) {
118+
if (performance.now() > deadline) throw new Error(`timed out: ${what}`);
119+
await new Promise((resolve) => setTimeout(resolve, 5));
120+
}
121+
}
122+
123+
function toolResultsFor(
124+
turns: readonly ConversationTurn[],
125+
callId: string,
126+
): string[] {
127+
return turns
128+
.flatMap((turn) => turn.content)
129+
.flatMap((block) =>
130+
block.type === "tool_result" && block.callId === callId
131+
? [JSON.stringify(block.content)]
132+
: [],
133+
);
134+
}
135+
136+
function collect(session: IntegrationSession): ReactorEmittedEvent[] {
137+
const events: ReactorEmittedEvent[] = [];
138+
void (async () => {
139+
for await (const event of session.agent.stream()) events.push(event);
140+
})().catch(() => undefined);
141+
return events;
142+
}
143+
144+
const cleanups: (() => Promise<void> | void)[] = [];
145+
afterEach(async () => {
146+
for (const cleanup of cleanups.splice(0).reverse()) await cleanup();
147+
});
148+
149+
interface Reloaded {
150+
session: IntegrationSession;
151+
/** Events from the rebuilt agent, collected from its start. */
152+
events: ReactorEmittedEvent[];
153+
callId: string;
154+
deadline: number;
155+
clock: ReturnType<typeof holdLongTimers>;
156+
}
157+
158+
/**
159+
* Parks `command` on an ask, approves it, and once `reloadWhen` holds for
160+
* the approved run, rebuilds the agent on the same store. The model turn
161+
* after the approved result stays open, so the reload always lands before
162+
* the reply terminal commits the cycle.
163+
*/
164+
async function approveThenReload(
165+
command: string,
166+
reloadWhen: (
167+
session: IntegrationSession,
168+
events: ReactorEmittedEvent[],
169+
) => boolean,
170+
): Promise<Reloaded> {
171+
const ctx = deferredApprovalGate();
172+
const sessionOpts = {
173+
permissionGate: ctx.gate,
174+
authorize: createReactorAuthorize(ctx.gate),
175+
};
176+
let session = await openIntegrationSession(sessionOpts);
177+
cleanups.push(() => closeIntegrationSession(session));
178+
179+
session.harness.scenario.replyOnce("anthropic", {
180+
toolCalls: [{ name: "run_shell", args: { command } }],
181+
});
182+
session.harness.scenario.stall();
183+
184+
const turn = await runUntilSuspended(session, "Probe the local port.");
185+
if (turn.result.type !== "suspended") throw new Error("did not park");
186+
const { correlationId } = turn.result;
187+
const parked = (await session.storage.load()).pendingOperations.find(
188+
(op) => op.correlationId === correlationId,
189+
);
190+
const callId = parked?.suspendedCall?.id;
191+
const deadline = parked?.timeoutAt;
192+
if (callId === undefined || deadline === undefined) {
193+
throw new Error("parked call has no persisted call id or deadline");
194+
}
195+
196+
const resume = createApprovalResume({
197+
getAgent: () => session.agent,
198+
gate: ctx.gate,
199+
resolveParkedCallId: (id) =>
200+
resolveParkedCallIdFromStore(session.storage, id),
201+
});
202+
const handling = resume.handle(turn.result);
203+
await waitFor(ctx.asked, "approval prompt");
204+
ctx.approve();
205+
expect(await handling).toBe(true);
206+
void session.harness
207+
.run({ wallClockBudgetMs: 30_000 })
208+
.catch(() => undefined);
209+
const before = session;
210+
await waitFor(() => reloadWhen(before, turn.events), "reload point");
211+
212+
const clock = holdLongTimers();
213+
cleanups.push(() => clock.restore());
214+
session = await rebuildIntegrationSession(session, sessionOpts);
215+
return { session, events: collect(session), callId, deadline, clock };
216+
}
217+
218+
/** Moves past the original approval deadline and lets the reactor react. */
219+
async function passDeadline(reloaded: Reloaded): Promise<void> {
220+
reloaded.clock.advanceTo(reloaded.deadline + 1_000);
221+
await reloaded.session.harness
222+
.run({ wallClockBudgetMs: 2_000 })
223+
.catch(() => undefined);
224+
await new Promise((resolve) => setTimeout(resolve, 50));
225+
}
226+
227+
describe("integration: approving an ask, then reloading the agent", () => {
228+
test.serial(
229+
"after the approved result: the rebuilt agent continues, and the old deadline adds no result",
230+
async () => {
231+
const reloaded = await approveThenReload(
232+
CLOSED_PORT_PROBE,
233+
(session) => session.harness.scenario.matchedRequests().length === 2,
234+
);
235+
const { session, callId } = reloaded;
236+
expect(
237+
toolResultsFor(await session.agent.history(), callId),
238+
).toHaveLength(1);
239+
240+
session.harness.scenario.replyOnce("anthropic", { text: "Port closed." });
241+
const followUp = await runUntilDone(session, "Continue.");
242+
expect(followUp.reply).toBe("Port closed.");
243+
244+
await passDeadline(reloaded);
245+
expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual(
246+
[],
247+
);
248+
const results = toolResultsFor(await session.agent.history(), callId);
249+
expect(results).toHaveLength(1);
250+
expect(results.join("")).not.toContain("approval timed out");
251+
},
252+
20_000,
253+
);
254+
255+
// The reload interrupts the approved call, which the closing reactor
256+
// answers itself. What an interrupted call should report is not pinned
257+
// here; only that the old deadline cannot add a second result.
258+
test.serial(
259+
"during the approved call: the old deadline adds no second result",
260+
async () => {
261+
const reloaded = await approveThenReload(
262+
`sleep 20; ${CLOSED_PORT_PROBE}`,
263+
(_session, events) => events.some((e) => e.type === "tool.start"),
264+
);
265+
const { session, callId } = reloaded;
266+
expect(
267+
toolResultsFor(await session.agent.history(), callId),
268+
).toHaveLength(1);
269+
270+
await passDeadline(reloaded);
271+
expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual(
272+
[],
273+
);
274+
expect(
275+
toolResultsFor(await session.agent.history(), callId),
276+
).toHaveLength(1);
277+
},
278+
20_000,
279+
);
280+
});

0 commit comments

Comments
 (0)