Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 131 additions & 0 deletions src/tui/runner/exit.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,14 @@ import { getLogger } from "@intx/log";
import type { InferenceSource } from "@intx/types/runtime";

import * as codexSession from "../../auth/codex/session.js";
import { createChatDirector } from "../../agent/director.js";
import { createSubAgentSessionStore } from "../../subagent/session-store.js";
import type {
ReactorAction,
ReactorCapabilities,
ReactorInboundEvent,
ReactorState,
} from "@intx/types/runtime";
import { LOG_NAMESPACE_ROOT } from "../../branding.js";
import { defined } from "../../../tests/helpers/defined.js";
import {
Expand Down Expand Up @@ -259,3 +267,126 @@ describe("agentProxy.send vs /clear", () => {
}
});
});

const rebuildMockState: ReactorState = {} as unknown as ReactorState;

const rebuildMockCapabilities: ReactorCapabilities = {
infer: (options) =>
({
type: "infer",
...(options !== undefined ? { options } : {}),
}) as ReactorAction,
executeTools: (calls) => ({ type: "execute_tools", calls }),
suspend: (gate) => ({ type: "suspend", gate }),
fork: (mode, forkId) => ({ type: "fork", mode, forkId }),
emit: (eventType, data) => ({ type: "emit", eventType, data }),
reply: (content) => ({ type: "reply", content }),
checkpoint: (message = "") => ({ type: "checkpoint", message }),
compact: (compactor, reason) => ({ type: "compact", compactor, reason }),
wait: () => ({ type: "wait" }),
done: () => ({ type: "done" }),
};

function rebuildManageTasksEvent(): ReactorInboundEvent {
return {
type: "inference.done",
turn: {
role: "assistant",
model: "test",
timestamp: 0,
content: [
{
type: "tool_call",
id: "m",
name: "manage_tasks",
arguments: {
action: "create",
tasks: [{ id: "t1", title: "work", status: "doing" }],
},
},
],
},
usage: { input: 0, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
source: { model: "test-model" },
} as unknown as ReactorInboundEvent;
}

function rebuildTextTurn(): ReactorInboundEvent {
return {
type: "inference.done",
turn: {
role: "assistant",
model: "test",
timestamp: 0,
content: [{ type: "text", text: "all set" }],
},
usage: { input: 10, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 },
source: { model: "test-model" },
} as unknown as ReactorInboundEvent;
}

describe("rebuild re-syncs idle-with-fleet while drained", () => {
test("reload-if-idle and interrupt rebuilds resume the open-task nudge with no fleet transition", async () => {
const store = createSubAgentSessionStore();
const directorHolder: RunnerServices["directorHolder"] = {};
const agent = recordingAgent([]);
const { state, services } = stubSendLifecycle(agent);
services.directorHolder =
directorHolder as unknown as RunnerServices["directorHolder"];
services.subAgentSessions =
store as unknown as RunnerServices["subAgentSessions"];
services.workflowHost = {
reattach: () => undefined,
} as unknown as RunnerServices["workflowHost"];
services.cycleRecorder = {
dispose: async () => "",
reset: () => undefined,
handleEvent: () => undefined,
} as unknown as RunnerServices["cycleRecorder"];
services.buildAgent = (async () => {
// Every rebuild mints a fresh director from the static true seed (fleet
// lanes may appear mid-session), exactly like the TUI session assembly.
directorHolder.instance = createChatDirector("base", [], {
onTasksChange: () => undefined,
allowIdleWithFleet: true,
});
return agent;
}) as unknown as RunnerServices["buildAgent"];
const fleetEvents: unknown[] = [];
services.emitter.on("event", (event: { type: string }) => {
if (event.type === "fleet") fleetEvents.push(event);
});
await createRunLifecycle(state, services);
const expectOpenTaskNudge = async (): Promise<void> => {
const director = defined(
directorHolder.instance,
"directorHolder.instance",
);
await director.decide(
rebuildManageTasksEvent(),
rebuildMockState,
rebuildMockCapabilities,
);
const actions = await director.decide(
rebuildTextTurn(),
rebuildMockState,
rebuildMockCapabilities,
);
const list = Array.isArray(actions) ? actions : [actions];
expect(list.some((action) => action.type === "infer")).toBe(true);
};
// Drained fleet: the idle reload rebuilds onto the static true seed.
state.pendingReload = true;
defined(state.reloadIfIdle, "reloadIfIdle")();
await services.sessionOps.awaitTail();
expect(state.fatalBuildError).toBeNull();
await expectOpenTaskNudge();
// The interrupt rebuild inherits the same seed.
defined(state.interrupt, "interrupt")();
await services.sessionOps.awaitTail();
expect(state.fatalBuildError).toBeNull();
await expectOpenTaskNudge();
expect(store.list()).toEqual([]);
expect(fleetEvents).toEqual([]);
});
});
15 changes: 15 additions & 0 deletions src/tui/runner/exit.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import {
import { getLogger } from "@intx/log";
import type { InferenceSource } from "@intx/types/runtime";
import { consumeStream } from "../../session/stream-consumer.js";
import { liveFleetCount } from "../../subagent/index.js";
import { getTelemetry } from "../../telemetry/singleton.js";
import { onTurnBoundary } from "../../agent/reactor-events.js";
import { setAgentSourceUnlessClosed } from "../agent-source-sync.js";
Expand Down Expand Up @@ -145,6 +146,18 @@ export async function closeAgentForRebuild(
}
}

// Rebuilt directors seed allowIdleWithFleet=true (fleet lanes may appear
// mid-session), so a rebuild while drained must re-sync the new director from
// the live fleet count — otherwise the open-task nudge stays suppressed until
// the next fleet transition, which never comes for an already-drained fleet.
function resyncIdleWithFleetFlag(
services: Pick<RunnerServices, "directorHolder" | "subAgentSessions">,
): void {
services.directorHolder.instance?.setAllowIdleWithFleet(
liveFleetCount(services.subAgentSessions.list()) > 0,
);
}

// Every rebuild site funnels its failure (a lock left held by a failed
// close, or any other buildAgent failure) through here so it surfaces as a
// plain-language, caught error rather than an unhandled rejection.
Expand Down Expand Up @@ -345,6 +358,7 @@ export async function createRunLifecycle(
throw new AgentContextLockError(state.workdir);
}
state.currentAgent = await services.buildAgent();
resyncIdleWithFleetFlag(services);
state.streamPromise = consumeStream(
liveAgent(state).stream(),
streamSink,
Expand Down Expand Up @@ -547,6 +561,7 @@ export async function createRunLifecycle(
throw new AgentContextLockError(state.workdir);
}
state.currentAgent = await services.buildAgent();
resyncIdleWithFleetFlag(services);
services.cycleRecorder.reset();
state.streamPromise = consumeStream(
liveAgent(state).stream(),
Expand Down
Loading