Keep parent agents alive while child work runs (#2458)
* fix(server): keep active agent trees resident Idle collection treated an inactive parent turn as process quiescence even when descendant work was still running. Protect managed and provider-owned subagent trees, carry descendant activity into the idle window, and use a conservative 30-minute fallback. * fix(server): close idle collection races Retain managed ancestry across closed intermediates, refresh descendant state after each awaited close, and invalidate running provider subagents when their runtime is terminated. * fix(server): serialize agent tree collection Keep ancestors resident until managed descendants are collected, serialize descendant registration with idle collection, and preserve provider subagent state across hot reloads. * fix(server): retain closed descendant activity Carry persisted descendant activity through runtime closure, avoid error children pinning parents, and cancel stale provider children when reload replacement aborts. * test(server): make descendant expiry deterministic * fix(server): scope idle child protection * fix(server): tighten idle tree coordination * fix(server): reduce aggressive idle cleanup Keep the mitigation conservative: extend the idle timeout without inferring tree liveness from incomplete lifecycle signals. * fix(server): keep parents alive during child work Idle cleanup now respects running managed and provider-native children while retaining a conservative 30-minute fallback. * fix(server): clear running child state on close * fix(server): drain child events before close
This commit is contained in:
@@ -21,11 +21,12 @@ the agent runs through `ensureAgentLoaded()`, which resumes the durable provider
|
|||||||
same Paseo agent ID. Provider history is not appended again when the canonical timeline is already
|
same Paseo agent ID. Provider history is not appended again when the canonical timeline is already
|
||||||
primed.
|
primed.
|
||||||
|
|
||||||
The daemon collects an eligible idle runtime after two minutes and sweeps every 15 seconds. Only
|
The daemon collects an eligible idle runtime after 30 minutes and sweeps every minute. Only
|
||||||
unarchived, non-internal agents that are exactly `idle`, have no active or pending run, replacement,
|
unarchived, non-internal agents that are exactly `idle`, have no active or pending run, replacement,
|
||||||
or permission, and have not been activated during the idle window are eligible. `running`,
|
or permission, and have not been activated during the idle window are eligible. `running`,
|
||||||
`initializing`, and `error` agents stay resident. Subagents are considered independently; collection
|
`initializing`, and `error` agents stay resident. An idle parent also stays resident while current
|
||||||
does not cascade or change parentage.
|
in-memory state shows a running managed child or provider subagent. Otherwise agents are evaluated
|
||||||
|
independently; collection does not cascade or change parentage.
|
||||||
|
|
||||||
Active schedules targeting an existing agent protect that agent from collection. Paused, completed,
|
Active schedules targeting an existing agent protect that agent from collection. Paused, completed,
|
||||||
and new-agent schedules do not. A pane may remain open after collection; its next prompt resumes the
|
and new-agent schedules do not. A pane may remain open after collection; its next prompt resumes the
|
||||||
|
|||||||
@@ -168,6 +168,16 @@ class TestAgentClient implements AgentClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
class SessionRecordingAgentClient extends TestAgentClient {
|
||||||
|
readonly sessions: TestAgentSession[] = [];
|
||||||
|
|
||||||
|
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||||
|
const session = new TestAgentSession(config);
|
||||||
|
this.sessions.push(session);
|
||||||
|
return session;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
class HeldAgentCreationClient extends TestAgentClient {
|
class HeldAgentCreationClient extends TestAgentClient {
|
||||||
private readonly creationStarted = deferred<void>();
|
private readonly creationStarted = deferred<void>();
|
||||||
private readonly creationAllowed = deferred<void>();
|
private readonly creationAllowed = deferred<void>();
|
||||||
@@ -7590,15 +7600,7 @@ test("a shared agent load upgrades provider history hydration to broadcast", asy
|
|||||||
|
|
||||||
test("collectIdleAgents leaves recent, protected, internal, running, and error agents resident", async () => {
|
test("collectIdleAgents leaves recent, protected, internal, running, and error agents resident", async () => {
|
||||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-eligibility-"));
|
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-eligibility-"));
|
||||||
const client = new (class extends TestAgentClient {
|
const client = new SessionRecordingAgentClient();
|
||||||
readonly sessions: TestAgentSession[] = [];
|
|
||||||
|
|
||||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
|
||||||
const session = new TestAgentSession(config);
|
|
||||||
this.sessions.push(session);
|
|
||||||
return session;
|
|
||||||
}
|
|
||||||
})();
|
|
||||||
const ids = [
|
const ids = [
|
||||||
"00000000-0000-4000-8000-000000000211",
|
"00000000-0000-4000-8000-000000000211",
|
||||||
"00000000-0000-4000-8000-000000000212",
|
"00000000-0000-4000-8000-000000000212",
|
||||||
@@ -7669,6 +7671,160 @@ test("collectIdleAgents leaves recent, protected, internal, running, and error a
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("collectIdleAgents protects an idle parent with a running managed child", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-running-child-"));
|
||||||
|
const client = new SessionRecordingAgentClient();
|
||||||
|
const manager = new AgentManager({ clients: { codex: client }, logger });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const parent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||||
|
workspaceId: undefined,
|
||||||
|
});
|
||||||
|
const child = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||||
|
labels: { [PARENT_AGENT_ID_LABEL]: parent.id },
|
||||||
|
workspaceId: undefined,
|
||||||
|
});
|
||||||
|
const independent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||||
|
workspaceId: undefined,
|
||||||
|
});
|
||||||
|
client.sessions[1]!.pushEvent({
|
||||||
|
type: "turn_started",
|
||||||
|
provider: "codex",
|
||||||
|
turnId: "managed-child-running",
|
||||||
|
});
|
||||||
|
await manager.flush();
|
||||||
|
|
||||||
|
const collection = await manager.collectIdleAgents({
|
||||||
|
cutoff: new Date(Date.now() + 1_000),
|
||||||
|
protectedAgentIds: new Set(),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(collection).toEqual({
|
||||||
|
collected: [expect.objectContaining({ agentId: independent.id })],
|
||||||
|
failures: [],
|
||||||
|
});
|
||||||
|
expect(manager.getAgent(parent.id)?.lifecycle).toBe("idle");
|
||||||
|
expect(manager.getAgent(child.id)?.lifecycle).toBe("running");
|
||||||
|
expect(manager.getAgent(independent.id)).toBeNull();
|
||||||
|
} finally {
|
||||||
|
await Promise.all(manager.listAgents().map((agent) => manager.closeAgent(agent.id))).catch(
|
||||||
|
() => undefined,
|
||||||
|
);
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("collectIdleAgents protects an idle parent with a running provider subagent", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-running-provider-child-"));
|
||||||
|
const client = new SessionRecordingAgentClient();
|
||||||
|
const manager = new AgentManager({ clients: { codex: client }, logger });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const parent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||||
|
workspaceId: undefined,
|
||||||
|
});
|
||||||
|
const independent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||||
|
workspaceId: undefined,
|
||||||
|
});
|
||||||
|
client.sessions[0]!.pushEvent({
|
||||||
|
type: "provider_subagent",
|
||||||
|
provider: "codex",
|
||||||
|
event: {
|
||||||
|
type: "upsert",
|
||||||
|
id: "provider-child-running",
|
||||||
|
title: "Provider child",
|
||||||
|
status: "running",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await manager.flush();
|
||||||
|
|
||||||
|
const collection = await manager.collectIdleAgents({
|
||||||
|
cutoff: new Date(Date.now() + 1_000),
|
||||||
|
protectedAgentIds: new Set(),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(collection).toEqual({
|
||||||
|
collected: [expect.objectContaining({ agentId: independent.id })],
|
||||||
|
failures: [],
|
||||||
|
});
|
||||||
|
expect(manager.getAgent(parent.id)?.lifecycle).toBe("idle");
|
||||||
|
expect(manager.getAgent(independent.id)).toBeNull();
|
||||||
|
} finally {
|
||||||
|
await Promise.all(manager.listAgents().map((agent) => manager.closeAgent(agent.id))).catch(
|
||||||
|
() => undefined,
|
||||||
|
);
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("closed provider subagents do not block collection after resume", async () => {
|
||||||
|
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-closed-provider-child-"));
|
||||||
|
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||||
|
const client = new SessionRecordingAgentClient();
|
||||||
|
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const parent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||||
|
workspaceId: undefined,
|
||||||
|
});
|
||||||
|
client.sessions[0]!.pushEvent({
|
||||||
|
type: "provider_subagent",
|
||||||
|
provider: "codex",
|
||||||
|
event: {
|
||||||
|
type: "upsert",
|
||||||
|
id: "provider-child-running",
|
||||||
|
title: "Provider child",
|
||||||
|
status: "running",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
client.sessions[0]!.pushEvent({
|
||||||
|
type: "provider_subagent",
|
||||||
|
provider: "codex",
|
||||||
|
event: {
|
||||||
|
type: "upsert",
|
||||||
|
id: "provider-child-finishing",
|
||||||
|
title: "Finishing provider child",
|
||||||
|
status: "running",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await manager.flush();
|
||||||
|
|
||||||
|
client.sessions[0]!.pushEvent({
|
||||||
|
type: "provider_subagent",
|
||||||
|
provider: "codex",
|
||||||
|
event: {
|
||||||
|
type: "upsert",
|
||||||
|
id: "provider-child-finishing",
|
||||||
|
status: "completed",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await manager.closeAgent(parent.id);
|
||||||
|
await ensureAgentLoaded(parent.id, {
|
||||||
|
agentManager: manager,
|
||||||
|
agentStorage: storage,
|
||||||
|
logger,
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(manager.getProviderSubagent(parent.id, "provider-child-running")?.status).toBe(
|
||||||
|
"canceled",
|
||||||
|
);
|
||||||
|
expect(manager.getProviderSubagent(parent.id, "provider-child-finishing")?.status).toBe(
|
||||||
|
"completed",
|
||||||
|
);
|
||||||
|
const collection = await manager.collectIdleAgents({
|
||||||
|
cutoff: new Date(Date.now() + 1_000),
|
||||||
|
protectedAgentIds: new Set(),
|
||||||
|
});
|
||||||
|
expect(collection.collected).toEqual([expect.objectContaining({ agentId: parent.id })]);
|
||||||
|
} finally {
|
||||||
|
await Promise.all(manager.listAgents().map((agent) => manager.closeAgent(agent.id))).catch(
|
||||||
|
() => undefined,
|
||||||
|
);
|
||||||
|
await storage.flush().catch(() => undefined);
|
||||||
|
rmSync(workdir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
test("load waits for an in-flight collection close and creates only one resumed runtime", async () => {
|
test("load waits for an in-flight collection close and creates only one resumed runtime", async () => {
|
||||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-close-race-"));
|
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-close-race-"));
|
||||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||||
|
|||||||
@@ -1388,6 +1388,8 @@ export class AgentManager {
|
|||||||
},
|
},
|
||||||
"agent.manager.close.start",
|
"agent.manager.close.start",
|
||||||
);
|
);
|
||||||
|
await this.drainSessionEvents(agentId);
|
||||||
|
this.cancelRunningProviderSubagents(agentId);
|
||||||
const closedAgent = this.prepareAgentForClosure(agent, "agent closed");
|
const closedAgent = this.prepareAgentForClosure(agent, "agent closed");
|
||||||
let closeError: unknown;
|
let closeError: unknown;
|
||||||
try {
|
try {
|
||||||
@@ -1420,6 +1422,20 @@ export class AgentManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private cancelRunningProviderSubagents(parentAgentId: string): void {
|
||||||
|
for (const subagent of this.providerSubagents.list(parentAgentId)) {
|
||||||
|
if (subagent.status !== "running") {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const event = this.providerSubagents.apply(parentAgentId, subagent.provider, {
|
||||||
|
type: "upsert",
|
||||||
|
id: subagent.id,
|
||||||
|
status: "canceled",
|
||||||
|
});
|
||||||
|
this.dispatch({ type: "provider_subagent", event });
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async collectIdleAgents(options: {
|
async collectIdleAgents(options: {
|
||||||
cutoff: Date;
|
cutoff: Date;
|
||||||
protectedAgentIds: ReadonlySet<string>;
|
protectedAgentIds: ReadonlySet<string>;
|
||||||
@@ -1461,10 +1477,25 @@ export class AgentManager {
|
|||||||
!this.runs.hasRun(agent.id) &&
|
!this.runs.hasRun(agent.id) &&
|
||||||
!agent.pendingReplacement &&
|
!agent.pendingReplacement &&
|
||||||
agent.pendingPermissions.size === 0 &&
|
agent.pendingPermissions.size === 0 &&
|
||||||
agent.inFlightPermissionResponses.size === 0
|
agent.inFlightPermissionResponses.size === 0 &&
|
||||||
|
!this.hasRunningChild(agent.id)
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private hasRunningChild(parentAgentId: string): boolean {
|
||||||
|
for (const agent of this.agents.values()) {
|
||||||
|
if (
|
||||||
|
agent.lifecycle === "running" &&
|
||||||
|
getParentAgentIdFromLabels(agent.labels) === parentAgentId
|
||||||
|
) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return this.providerSubagents
|
||||||
|
.list(parentAgentId)
|
||||||
|
.some((subagent) => subagent.status === "running");
|
||||||
|
}
|
||||||
|
|
||||||
async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
|
async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
|
||||||
const agent = this.requireAgent(agentId);
|
const agent = this.requireAgent(agentId);
|
||||||
if (!this.registry) {
|
if (!this.registry) {
|
||||||
|
|||||||
@@ -215,8 +215,8 @@ import { DaemonExecutions } from "./hub/daemon-executions.js";
|
|||||||
|
|
||||||
const MAX_MCP_DEBUG_BATCH_ITEMS = 10;
|
const MAX_MCP_DEBUG_BATCH_ITEMS = 10;
|
||||||
const REDACTED_LOG_VALUE = "[redacted]";
|
const REDACTED_LOG_VALUE = "[redacted]";
|
||||||
const IDLE_AGENT_RUNTIME_TTL_MS = 2 * 60 * 1000;
|
const IDLE_AGENT_RUNTIME_TTL_MS = 30 * 60 * 1000;
|
||||||
const IDLE_AGENT_RUNTIME_SWEEP_INTERVAL_MS = 15 * 1000;
|
const IDLE_AGENT_RUNTIME_SWEEP_INTERVAL_MS = 60 * 1000;
|
||||||
const DOWNLOAD_OPEN_FLAGS =
|
const DOWNLOAD_OPEN_FLAGS =
|
||||||
process.platform === "win32" ? constants.O_RDONLY : constants.O_RDONLY | constants.O_NOFOLLOW;
|
process.platform === "win32" ? constants.O_RDONLY : constants.O_RDONLY | constants.O_NOFOLLOW;
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user