1270 unmodified lines
Expand all
expect(child.repliedToParentSinceTask).toBe(false);
});
it("queues a parent message while the previous parent turn is finalizing", async () => {
const child = createSession({ depth: 1 });
const finalization = deferred<void>();
const internals = child as unknown as InspectableRlmSession;
internals._rlmParentTurnFinalizations.set("previous-turn", finalization.promise);
const preflight: Array<{ accepted: boolean; queued: boolean | undefined }> = [];
const message = createAgentSessionMessage({
id: "agentmsg-parent-during-finalization",
source: "agent_message",
message: "stop and report",
fromRelationship: "parent",
target: { activeSessionId: "child-active", sessionId: child.sessionId },
it("drains a background completion before a queued parent result turn", async () => {
const resultDir = join(tempDir, "parent-during-result-finalization");
mkdirSync(resultDir);
const contract = normalizeRlmChildContract(
{
schema: {
type: "object",
properties: { status: { type: "string" } },
required: ["status"],
additionalProperties: false,
},
required: true,
},
{ result: { fields: ["status"] } },
)!;
let prepared = prepareRlmChildContract(resultDir, "child-result-turn", "result-worker", contract)!;
const completions: Array<{ turnKey: string; resultGeneration?: number }> = [];
let providerRuns = 0;
const child = createSession({
depth: 1,
streamFn: () => {
providerRuns += 1;
if (providerRuns !== 2) {
submitRlmChildResult(prepared, { status: `round-${providerRuns}` });
}
return streamAnswer(`finished round ${providerRuns}`);
},
rlmResultController: {
contract,
currentResultTurn: () => currentRlmResultTurn(prepared),
beginResultTurn: (turnKey) => {
const transition = beginRlmChildResultTurn(prepared, turnKey);
prepared = transition.prepared;
return transition.turn;
},
completeResultTurn: (turnKey, repliedToParent) => {
const transition = completeRlmChildResultTurn(prepared, turnKey, repliedToParent);
prepared = transition.prepared;
return transition.turn;
},
settleResultTurn: (turnKey) => {
const settlement = settleRlmChildResultTurn(prepared, turnKey);
prepared = settlement.prepared;
return { turn: settlement.turn, ...(settlement.error ? { error: settlement.error } : {}) };
},
failResultTurn: (turnKey, error) => {
prepared = failRlmChildResultTurn(prepared, turnKey, error);
return currentRlmResultTurn(prepared)!;
},
markResultTurnNotified: (turnKey) => {
prepared = markRlmChildResultTurnNotified(prepared, turnKey);
return currentRlmResultTurn(prepared)!;
},
resultTurnInstructions: () => formatRlmResultContractInstructions(prepared),
submitResult: async (value) => submitRlmChildResult(prepared, value),
},
agentMessageController: {
listAgents: () => ({ agents: [] }),
rlmParentFollowupCompleted: (completion) => {
completions.push(completion);
},
sendAgentMessage: async () => {
throw new Error("not used");
},
},
});
const descendantDone = deferred<void>();
const descendant = {
id: "result-turn-descendant",
sessionDir: join(tempDir, "result-turn-descendant"),
abort: () => {},
status: "running",
settled: false,
settlement: descendantDone,
} as InspectableRlmRun;
(child as unknown as InspectableRlmSession)._unsettledRlmChildRuns.add(descendant);
const parentPrompt = (id: string) =>
createAgentSessionMessage({
id,
source: "agent_message",
message: id,
fromRelationship: "parent",
target: { activeSessionId: "child-active", sessionId: child.sessionId },
});
const first = parentPrompt("agentmsg-result-round-1");
await child.acceptAgentMessagePrompt(first.content as string, { customMessage: first });
await waitFor(() => currentRlmResultTurn(prepared)?.providerCompleted === true);
expect(currentRlmResultTurn(prepared)).toMatchObject({ generation: 0, status: "active" });
const preflight: Array<{ accepted: boolean; queued: boolean | undefined }> = [];
const second = parentPrompt("agentmsg-result-round-2");
await child.acceptAgentMessagePrompt(message.content as string, {
customMessage: message,
await child.acceptAgentMessagePrompt(second.content as string, {
customMessage: second,
queueIfBusy: true,
streamingBehavior: "steer",
preflightResult: (accepted, queued) => preflight.push({ accepted, queued }),
});
expect(preflight).toEqual([{ accepted: true, queued: true }]);
expect(child.queuedActionCount).toBe(1);
expect(currentRlmResultTurn(prepared)).toMatchObject({ generation: 0, status: "active" });
child.clearQueue();
internals._rlmParentTurnFinalizations.delete("previous-turn");
finalization.resolve();
(child as unknown as InspectableRlmSession)._onIpythonBashHandleEvent({
kind: "completed",
handleId: "result-turn-bash",
pid: 4242,
exitCode: 0,
duration: 0.5,
});
descendant.settled = true;
descendantDone.resolve();
await waitFor(() => completions.length === 2 && child.unfinishedActionCount === 0);
expect(providerRuns).toBe(3);
expect(completions.map(({ resultGeneration }) => resultGeneration)).toEqual([0, 1]);
expect(readRlmChildResult(prepared).result).toEqual({ status: "round-3" });
expect(
child.messages.some((message) => message.role === "custom" && message.customType === "bash_handle_completion"),
).toBe(true);
expect(readRlmChildResult(prepared, undefined, 0).result).toEqual({ status: "round-1" });
});
it("resets replied state when a parent follow-up is queued", async () => {
const child = createSession({ depth: 1 });
4062 unmodified lines
Expand all