mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-24 18:22:38 +00:00
feat(server): collect runtimes after one idle hour with optional provider vetoes
Keep callers resident while local or remote finish notification intent remains outstanding. Daemon-owned work determines baseline eligibility; provider checks add optional vetoes. Own the idle clock and sweep lifecycle in one cleanup module while the manager retains guarded closure.
This commit is contained in:
+16
-7
@@ -25,16 +25,25 @@ Reload releases the old runtime before resuming its durable session: an idle pro
|
||||
still own an exclusive writer. A close failure retains that runtime for cleanup and blocks the
|
||||
replacement. Once closure succeeds, a failed resume leaves the durable agent closed and retryable.
|
||||
|
||||
Idle resumable runtimes are collected after 30 minutes without activity, checked once a minute.
|
||||
Idle resumable runtimes are collected after one hour without activity, checked once a minute.
|
||||
Collection releases the process and preserves the agent record, conversation, relationships, and
|
||||
provider session. The next prompt resumes that session. Set `agents.idleRuntimeCleanup: false` in
|
||||
host config to disable automatic collection; this setting takes effect at daemon startup.
|
||||
|
||||
Foreground runs, permissions and responses, queued deliveries, lifecycle operations, and working
|
||||
managed or native descendants prevent collection. An idle managed child can still own background
|
||||
work. Each provider must positively permit shutdown through `canStopRuntime()`; missing methods,
|
||||
failed probes, and probe timeouts retain ownership. Once protected work finishes, a fresh idle
|
||||
period starts. There is no deadline that overrides a provider veto.
|
||||
managed or native descendants prevent collection. Outstanding local or remote finish subscriptions
|
||||
also protect their caller, including registration, recovery and notification dispatch acknowledgment.
|
||||
An idle managed child can still own background work.
|
||||
|
||||
The daemon baseline always applies. An optional provider `canStopRuntime()` adds a veto: false,
|
||||
errors and timeouts retain ownership, while true cannot override daemon blockers. Without a provider
|
||||
check, the baseline decides collection after the same one-hour idle period. Once protected work
|
||||
finishes, a fresh idle hour starts. There is no deadline that overrides a veto.
|
||||
|
||||
The `idle-runtime-cleanup/` module owns the clocks, eligibility policy and sweep lifetime.
|
||||
AgentManager supplies captured runtime capabilities and owns atomic revalidation and shutdown.
|
||||
Finish subscriptions expose caller protection through their public capability; the collector never
|
||||
reads their watches or durable records.
|
||||
|
||||
The collector rechecks daemon state after asynchronous probes and event draining, in the existing
|
||||
close/load lifecycle lane. A prompt admitted before closure prevents collection. A prompt arriving
|
||||
@@ -56,10 +65,10 @@ Provider checks follow the signals their runtimes expose:
|
||||
injection, and admitted or scheduled continuations that can re-wake the session. Older OMP
|
||||
runtimes without either field stay resident. These signals are not an inventory of arbitrary
|
||||
processes.
|
||||
- ACP has no complete signal for agent-owned background work, so its adapters stay resident.
|
||||
- ACP has no provider safety check, so its adapters use baseline-only collection.
|
||||
|
||||
These checks do not discover arbitrary detached processes or work hidden by third-party
|
||||
extensions. A provider plugin opts in through the [runtime safety contract](../public-docs/plugins/providers.md#permit-idle-runtime-cleanup).
|
||||
extensions. A provider plugin adds its own veto through the [runtime safety contract](../public-docs/plugins/providers.md#permit-idle-runtime-cleanup).
|
||||
|
||||
A provider runtime can still die on its own — crash, OOM kill, host suspend. Work the agent parked
|
||||
inside that process dies with it: Claude Code's background Bash shells, `Monitor` watches, and
|
||||
|
||||
+1
-1
@@ -498,7 +498,7 @@ Validate and apply it inside the provider; core does not know your option shape.
|
||||
See [provider options](custom-providers.md#provider-options) for configuration and
|
||||
merge semantics.
|
||||
|
||||
Providers opt in to idle runtime collection through the per-session
|
||||
Providers add an optional veto to daemon baseline cleanup through the per-session
|
||||
[runtime safety query](../public-docs/plugins/providers.md#permit-idle-runtime-cleanup).
|
||||
The daemon calls the same `AgentSession.canStopRuntime()` interface for built-ins and plugins;
|
||||
the plugin bridge transports that query through the existing provider request/event protocol.
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { startIdleRuntimeCleanup } from "./idle-runtime-cleanup.js";
|
||||
import { createIdleRuntimeCleanup } from "./idle-runtime-cleanup/index.js";
|
||||
import {
|
||||
prepareAgentMessage,
|
||||
formatAgentMessage,
|
||||
@@ -10008,7 +10008,7 @@ test("closeAgent persists one final closed snapshot", async () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("idle runtime cleanup requires provider permission and resumes the saved conversation", async () => {
|
||||
test("idle runtime cleanup honors a provider veto and resumes the saved conversation", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-cleanup-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
let safe = false;
|
||||
@@ -10020,16 +10020,17 @@ test("idle runtime cleanup requires provider permission and resumes the saved co
|
||||
}
|
||||
})();
|
||||
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
|
||||
const collect = runtimeCleanup(manager);
|
||||
try {
|
||||
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
const now = new Date(Date.now() + 31 * 60_000);
|
||||
expect(await manager.collectIdleAgents({ now })).toEqual({ collected: [], failures: [] });
|
||||
const now = new Date(Date.now() + 61 * 60_000);
|
||||
expect(await collect({ now })).toEqual({ collected: [], failures: [] });
|
||||
safe = true;
|
||||
expect(await manager.collectIdleAgents({ now })).toEqual({ collected: [], failures: [] });
|
||||
const later = new Date(now.getTime() + 31 * 60_000);
|
||||
expect((await manager.collectIdleAgents({ now: later })).collected).toEqual([
|
||||
expect(await collect({ now })).toEqual({ collected: [], failures: [] });
|
||||
const later = new Date(now.getTime() + 61 * 60_000);
|
||||
expect((await collect({ now: later })).collected).toEqual([
|
||||
{ agentId: agent.id, provider: "codex", sessionId: agent.persistence?.sessionId },
|
||||
]);
|
||||
expect(manager.getAgent(agent.id)).toBeNull();
|
||||
@@ -12125,6 +12126,25 @@ test("reads the last durable assistant response without loading a stored-only ch
|
||||
expect(manager.getAgent("stored-child")).toBeNull();
|
||||
});
|
||||
|
||||
function runtimeCleanup(manager: AgentManager) {
|
||||
let now = Date.now();
|
||||
let protectedCaller = (_agentId: string) => false;
|
||||
const cleanup = createIdleRuntimeCleanup({
|
||||
enabled: true,
|
||||
runtimes: () => manager.getRuntimeCleanupCandidates(),
|
||||
logger,
|
||||
now: () => now,
|
||||
isProtected: (agentId) => protectedCaller(agentId),
|
||||
schedule: () => () => {},
|
||||
});
|
||||
onTestFinished(() => cleanup.close());
|
||||
return (options: { now: Date; isProtected?: (agentId: string) => boolean }) => {
|
||||
now = options.now.getTime();
|
||||
protectedCaller = options.isProtected ?? (() => false);
|
||||
return cleanup.sweep();
|
||||
};
|
||||
}
|
||||
|
||||
function idleCleanupHarness() {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-protection-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
@@ -12182,6 +12202,7 @@ function idleCleanupHarness() {
|
||||
});
|
||||
return {
|
||||
commands,
|
||||
sweep: runtimeCleanup(manager),
|
||||
manager,
|
||||
storage,
|
||||
sessions,
|
||||
@@ -12203,35 +12224,6 @@ function idleCleanupHarness() {
|
||||
};
|
||||
}
|
||||
|
||||
test.each(["missing", "error"])("idle cleanup retains %s safety probes", async (kind) => {
|
||||
const h = idleCleanupHarness();
|
||||
const agent = await h.create();
|
||||
if (kind === "missing") delete h.sessions[0]!.canStopRuntime;
|
||||
else
|
||||
h.probes[0] = async () => {
|
||||
throw new Error("probe failed");
|
||||
};
|
||||
const result = await h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
expect(result.collected).toEqual([]);
|
||||
expect(result.failures).toHaveLength(kind === "error" ? 1 : 0);
|
||||
expect(h.closes).toBe(0);
|
||||
expect(h.manager.getAgent(agent.id)?.lifecycle).toBe("idle");
|
||||
});
|
||||
|
||||
test("idle cleanup starts a fresh idle period when background work ends between sweeps", async () => {
|
||||
const h = idleCleanupHarness();
|
||||
await h.create();
|
||||
h.probes[0] = async () => false;
|
||||
await h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
await h.manager.collectIdleAgents({ now: h.later(59) });
|
||||
h.probes[0] = async () => true;
|
||||
await h.manager.collectIdleAgents({ now: h.later(60) });
|
||||
expect(h.closes).toBe(0);
|
||||
await h.manager.collectIdleAgents({ now: h.later(89) });
|
||||
expect(h.closes).toBe(0);
|
||||
expect((await h.manager.collectIdleAgents({ now: h.later(91) })).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("idle cleanup rechecks new prompt activity after an asynchronous safety probe", async () => {
|
||||
const h = idleCleanupHarness();
|
||||
const agent = await h.create();
|
||||
@@ -12241,7 +12233,7 @@ test("idle cleanup rechecks new prompt activity after an asynchronous safety pro
|
||||
entered.resolve();
|
||||
return probe.promise;
|
||||
};
|
||||
const sweep = h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
const sweep = h.sweep({ now: h.later(61) });
|
||||
await entered.promise;
|
||||
await h.manager.runAgent(agent.id, "arrived during safety query");
|
||||
probe.resolve(true);
|
||||
@@ -12255,7 +12247,7 @@ test("idle cleanup retains a parent whose idle managed child owns background wor
|
||||
await h.create({ [PARENT_AGENT_ID_LABEL]: parent.id });
|
||||
h.probes[0] = async () => true;
|
||||
h.probes[1] = async () => false;
|
||||
expect((await h.manager.collectIdleAgents({ now: h.later(31) })).collected).toEqual([]);
|
||||
expect((await h.sweep({ now: h.later(61) })).collected).toEqual([]);
|
||||
expect(h.closes).toBe(0);
|
||||
});
|
||||
|
||||
@@ -12267,9 +12259,7 @@ test("idle cleanup rechecks queued deliveries admitted during the safety probe",
|
||||
queued = true;
|
||||
return true;
|
||||
};
|
||||
expect(
|
||||
(await h.manager.collectIdleAgents({ now: h.later(31), isProtected: () => queued })).collected,
|
||||
).toEqual([]);
|
||||
expect((await h.sweep({ now: h.later(61), isProtected: () => queued })).collected).toEqual([]);
|
||||
expect(h.closes).toBe(0);
|
||||
});
|
||||
|
||||
@@ -12279,7 +12269,7 @@ test("a prompt racing idle cleanup waits for close and resumes exactly once", as
|
||||
h.probes[0] = async () => true;
|
||||
const gate = deferred<void>();
|
||||
h.holdClose(gate.promise);
|
||||
const sweep = h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
const sweep = h.sweep({ now: h.later(61) });
|
||||
await h.closeEntered.promise;
|
||||
const run = h.manager.runAgent(agent.id, "arrived while closing");
|
||||
gate.resolve();
|
||||
@@ -12296,7 +12286,7 @@ test("an out-of-band command racing idle cleanup waits for the resumed runtime",
|
||||
h.probes[0] = async () => true;
|
||||
const gate = deferred<void>();
|
||||
h.holdClose(gate.promise);
|
||||
const sweep = h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
const sweep = h.sweep({ now: h.later(61) });
|
||||
await h.closeEntered.promise;
|
||||
const command = startAgentRun(h.manager, agent.id, "/maintenance", logger);
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
@@ -12308,37 +12298,3 @@ test("an out-of-band command racing idle cleanup waits for the resumed runtime",
|
||||
expect(h.commands).toEqual(["/maintenance"]);
|
||||
expect(h.prompts).toEqual([]);
|
||||
});
|
||||
|
||||
test.each([false, true])(
|
||||
"host idleRuntimeCleanup=%s controls actual sweep invocation",
|
||||
async (enabled) => {
|
||||
const h = idleCleanupHarness();
|
||||
await h.create();
|
||||
let probes = 0;
|
||||
h.probes[0] = async () => {
|
||||
probes++;
|
||||
return true;
|
||||
};
|
||||
let tick!: () => void;
|
||||
let canceled = false;
|
||||
const stop = startIdleRuntimeCleanup({
|
||||
enabled,
|
||||
collect: () => h.manager.collectIdleAgents({ now: h.later(31) }),
|
||||
onError: (error) => {
|
||||
throw error;
|
||||
},
|
||||
schedule: (callback, intervalMs) => {
|
||||
expect(intervalMs).toBe(60_000);
|
||||
tick = callback;
|
||||
return () => {
|
||||
canceled = true;
|
||||
};
|
||||
},
|
||||
});
|
||||
tick();
|
||||
await stop();
|
||||
expect(probes).toBe(enabled ? 1 : 0);
|
||||
expect(h.closes).toBe(enabled ? 1 : 0);
|
||||
expect(canceled).toBe(true);
|
||||
},
|
||||
);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import type { RuntimeCleanupCandidate } from "./idle-runtime-cleanup/index.js";
|
||||
import { ensureAgentLoaded } from "./agent-loading.js";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import type { PluginLifecycle } from "../plugins/lifecycle/index.js";
|
||||
@@ -759,7 +760,7 @@ export class AgentManager {
|
||||
private readonly agentRegistrationTasks = new Set<Promise<void>>();
|
||||
private readonly inFlightAgentCloses = new Map<string, Promise<void>>();
|
||||
private readonly reloadedSessionCloses = new WeakMap<AgentSession, Promise<void>>();
|
||||
private readonly runtimeProtectedAt = new WeakMap<AgentSession, { at: number; safe: boolean }>();
|
||||
private readonly runtimeCleanupIdentities = new WeakMap<AgentSession, object>();
|
||||
private readonly collectingRuntimes = new Set<string>();
|
||||
private readonly lifecycleMutationTails = new Map<string, Promise<void>>();
|
||||
private readonly agentStreamCoalescer: AgentStreamCoalescer;
|
||||
@@ -1750,24 +1751,50 @@ export class AgentManager {
|
||||
});
|
||||
}
|
||||
|
||||
/** Release only resumable runtimes whose providers positively permit shutdown. */
|
||||
async collectIdleAgents(
|
||||
options: {
|
||||
now?: Date;
|
||||
isProtected?: (agentId: string) => boolean;
|
||||
} = {},
|
||||
): Promise<{
|
||||
collected: Array<{ agentId: string; provider: AgentProvider; sessionId: string }>;
|
||||
failures: Array<{ agentId: string; error: unknown }>;
|
||||
}> {
|
||||
const result: Awaited<ReturnType<AgentManager["collectIdleAgents"]>> = {
|
||||
collected: [],
|
||||
failures: [],
|
||||
};
|
||||
if (!this.registry) return result;
|
||||
const now = options.now?.getTime() ?? Date.now();
|
||||
const cutoff = now - 30 * 60_000;
|
||||
const daemonAllows = (agent: LiveManagedAgent, inCollection = false): boolean =>
|
||||
/** Capture runtime capabilities without handing lifecycle ownership to the collector. */
|
||||
getRuntimeCleanupCandidates(): RuntimeCleanupCandidate[] {
|
||||
if (!this.registry) return [];
|
||||
const candidates: RuntimeCleanupCandidate[] = [];
|
||||
for (const agent of this.agents.values()) {
|
||||
const updatedAt = agent.updatedAt;
|
||||
let identity = this.runtimeCleanupIdentities.get(agent.session);
|
||||
if (!identity) {
|
||||
identity = Object.freeze({});
|
||||
this.runtimeCleanupIdentities.set(agent.session, identity);
|
||||
}
|
||||
const isEligible = () =>
|
||||
this.agents.get(agent.id) === agent &&
|
||||
agent.updatedAt === updatedAt &&
|
||||
this.runtimeHasNoPendingWork(agent);
|
||||
candidates.push({
|
||||
identity,
|
||||
agentId: agent.id,
|
||||
parentAgentId: getParentAgentIdFromLabels(agent.labels) ?? undefined,
|
||||
provider: agent.provider,
|
||||
sessionId: agent.persistence?.sessionId ?? null,
|
||||
lastActivityAt: updatedAt.getTime(),
|
||||
isEligible,
|
||||
canStopRuntime: agent.session.canStopRuntime?.bind(agent.session),
|
||||
closeIfEligible: async (revalidate) => {
|
||||
if (!agent.persistence?.sessionId) return false;
|
||||
// Block prompt admission before entering the close lane. A skipped close also
|
||||
// releases this barrier, so the waiting prompt keeps the existing runtime.
|
||||
this.collectingRuntimes.add(agent.id);
|
||||
try {
|
||||
return await this.runLifecycleMutation(agent.id, () =>
|
||||
this.closeAgentRuntime(agent.id, () => isEligible() && revalidate()),
|
||||
);
|
||||
} finally {
|
||||
this.collectingRuntimes.delete(agent.id);
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
return candidates;
|
||||
}
|
||||
|
||||
private runtimeHasNoPendingWork(agent: LiveManagedAgent): boolean {
|
||||
return (
|
||||
agent.lifecycle === "idle" &&
|
||||
!agent.internal &&
|
||||
agent.activeForegroundTurnId === null &&
|
||||
@@ -1775,107 +1802,11 @@ export class AgentManager {
|
||||
!agent.pendingReplacement &&
|
||||
agent.pendingPermissions.size === 0 &&
|
||||
agent.inFlightPermissionResponses.size === 0 &&
|
||||
(inCollection || !this.lifecycleMutationTails.has(agent.id)) &&
|
||||
(this.collectingRuntimes.has(agent.id) || !this.lifecycleMutationTails.has(agent.id)) &&
|
||||
!this.foregroundMutationTails.has(agent.id) &&
|
||||
!this.sessionEventTails.has(agent.id) &&
|
||||
!options.isProtected?.(agent.id) &&
|
||||
!this.providerSubagents.list(agent.id).some((child) => child.status === "running");
|
||||
const descendants = (root: LiveManagedAgent): LiveManagedAgent[] => {
|
||||
const found = new Map([[root.id, root]]);
|
||||
for (const parent of found.values()) {
|
||||
for (const child of this.agents.values()) {
|
||||
if (getParentAgentIdFromLabels(child.labels) === parent.id) found.set(child.id, child);
|
||||
}
|
||||
}
|
||||
return [...found.values()];
|
||||
};
|
||||
const candidates = [...this.agents.values()];
|
||||
for (const agent of candidates) {
|
||||
if (!agent.persistence?.sessionId) continue;
|
||||
if (!daemonAllows(agent)) {
|
||||
this.runtimeProtectedAt.set(agent.session, { at: now, safe: false });
|
||||
continue;
|
||||
}
|
||||
const family = descendants(agent);
|
||||
const versions = family.map((member) => ({ member, updatedAt: member.updatedAt }));
|
||||
try {
|
||||
const safe = await this.runtimeFamilyCanStop(family, daemonAllows);
|
||||
if (!safe) {
|
||||
this.runtimeProtectedAt.set(agent.session, { at: now, safe: false });
|
||||
this.logger.debug(
|
||||
{ agentId: agent.id },
|
||||
"Idle runtime retained: provider or descendant owns work",
|
||||
);
|
||||
continue;
|
||||
}
|
||||
const lastProbe = this.runtimeProtectedAt.get(agent.session);
|
||||
if (lastProbe && !lastProbe.safe)
|
||||
this.runtimeProtectedAt.set(agent.session, { at: now, safe: true });
|
||||
if (
|
||||
Math.max(agent.updatedAt.getTime(), this.runtimeProtectedAt.get(agent.session)?.at ?? 0) >
|
||||
cutoff
|
||||
)
|
||||
continue;
|
||||
const stillEligible = () => {
|
||||
const currentFamily = descendants(agent);
|
||||
return (
|
||||
currentFamily.length === versions.length &&
|
||||
versions.every(
|
||||
({ member, updatedAt }) =>
|
||||
this.agents.get(member.id) === member &&
|
||||
member.updatedAt === updatedAt &&
|
||||
daemonAllows(member, member === agent),
|
||||
)
|
||||
);
|
||||
};
|
||||
// The lifecycle lane is the loader's close barrier. Recheck after draining queued
|
||||
// events, immediately before calling close; prompt admission below uses the same barrier.
|
||||
const closed = await this.runLifecycleMutation(agent.id, () =>
|
||||
this.closeAgentRuntime(agent.id, () => {
|
||||
if (!stillEligible()) return false;
|
||||
this.collectingRuntimes.add(agent.id);
|
||||
return true;
|
||||
}),
|
||||
);
|
||||
if (closed) {
|
||||
const entry = {
|
||||
agentId: agent.id,
|
||||
provider: agent.provider,
|
||||
sessionId: agent.persistence.sessionId,
|
||||
};
|
||||
result.collected.push(entry);
|
||||
this.logger.info(entry, "Collected idle agent runtime");
|
||||
}
|
||||
} catch (error) {
|
||||
this.runtimeProtectedAt.set(agent.session, { at: now, safe: false });
|
||||
result.failures.push({ agentId: agent.id, error });
|
||||
this.logger.warn(
|
||||
{ agentId: agent.id, err: error },
|
||||
"Idle runtime cleanup failed; retaining ownership",
|
||||
);
|
||||
} finally {
|
||||
this.collectingRuntimes.delete(agent.id);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private async runtimeFamilyCanStop(
|
||||
family: LiveManagedAgent[],
|
||||
daemonAllows: (agent: LiveManagedAgent) => boolean,
|
||||
): Promise<boolean> {
|
||||
for (const member of family) {
|
||||
if (!daemonAllows(member) || !member.session.canStopRuntime) return false;
|
||||
if (
|
||||
!(await withTimeout(
|
||||
member.session.canStopRuntime(),
|
||||
10_000,
|
||||
`Runtime safety probe timed out for ${member.id}`,
|
||||
))
|
||||
)
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
!this.providerSubagents.list(agent.id).some((child) => child.status === "running")
|
||||
);
|
||||
}
|
||||
|
||||
closeAgent(agentId: string): Promise<void> {
|
||||
|
||||
@@ -695,7 +695,8 @@ export interface AgentSession {
|
||||
*/
|
||||
interrupt(): Promise<void>;
|
||||
/** True only when releasing this runtime loses no background work or pending replies.
|
||||
* Missing methods and rejected probes keep the runtime resident. */
|
||||
* An optional additional veto: missing methods use daemon baseline eligibility;
|
||||
* implemented checks that reject or return false keep the runtime resident. */
|
||||
canStopRuntime?(): Promise<boolean>;
|
||||
/** Release live runtime resources without archiving or deleting the durable native session. */
|
||||
close(): Promise<void>;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { createIdleRuntimeCleanup } from "../idle-runtime-cleanup/index.js";
|
||||
import { HostObservationError } from "../host-operations/index.js";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
@@ -412,3 +413,87 @@ test("shutdown during observation setup owns rejection of the already-canceled n
|
||||
expect(await state.store.list()).toHaveLength(1);
|
||||
expect(state.delivery.prompts).toEqual([]);
|
||||
});
|
||||
|
||||
test("remote finish intent protects its caller through registration, observation and delivery", async () => {
|
||||
const s = await scenario();
|
||||
let time = 61 * 60_000;
|
||||
let collected = false;
|
||||
const runtime = {
|
||||
identity: {},
|
||||
agentId: "parent",
|
||||
parentAgentId: undefined,
|
||||
provider: "test",
|
||||
sessionId: "parent",
|
||||
lastActivityAt: 0,
|
||||
isEligible: () => !collected,
|
||||
closeIfEligible: async (revalidate: () => boolean) => {
|
||||
if (!revalidate()) return false;
|
||||
collected = true;
|
||||
return true;
|
||||
},
|
||||
};
|
||||
const cleanup = createIdleRuntimeCleanup({
|
||||
enabled: true,
|
||||
logger: createTestLogger(),
|
||||
now: () => time,
|
||||
runtimes: () => (collected ? [] : [runtime]),
|
||||
isProtected: (agentId) => s.service.hasPendingForCaller(agentId),
|
||||
schedule: () => () => {},
|
||||
});
|
||||
const savedPut = s.store.put.bind(s.store);
|
||||
const writing = deferred<void>();
|
||||
s.store.put = async (record) => {
|
||||
await writing.promise;
|
||||
await savedPut(record);
|
||||
};
|
||||
const registration = s.service.register({ ...intent, serverId: "remote" }, true);
|
||||
const protectedDuringRegistration = s.service.hasPendingForCaller?.("parent") ?? false;
|
||||
writing.resolve();
|
||||
const accepted = await registration;
|
||||
expect(protectedDuringRegistration).toBe(true);
|
||||
expect((await cleanup.sweep()).collected).toEqual([]);
|
||||
expect(s.service.hasPendingForCaller("parent")).toBe(true);
|
||||
accepted.accept();
|
||||
await until(() => s.observation.waits.length === 1);
|
||||
expect(s.service.hasPendingForCaller("parent")).toBe(true);
|
||||
s.observation.waits[0]!.result.resolve(outcome());
|
||||
await until(() => s.delivery.prompts.length === 1);
|
||||
expect(s.service.hasPendingForCaller("parent")).toBe(true);
|
||||
s.delivery.acknowledgments[0]!.resolve("steered");
|
||||
await until(() => !s.service.hasPendingForCaller("parent"));
|
||||
expect(await s.store.list()).toEqual([]);
|
||||
time = 62 * 60_000;
|
||||
expect((await cleanup.sweep()).collected).toEqual([]);
|
||||
time = 122 * 60_000;
|
||||
expect((await cleanup.sweep()).collected).toHaveLength(1);
|
||||
await cleanup.close();
|
||||
});
|
||||
|
||||
test("finish recovery protects callers before durable intent has loaded", async () => {
|
||||
const s = await scenario();
|
||||
await s.service.register({ ...intent, serverId: "remote" });
|
||||
await s.service.close();
|
||||
const loading = deferred<void>();
|
||||
const recovered = createFinishSubscriptions({
|
||||
store: {
|
||||
...s.store,
|
||||
list: async () => {
|
||||
await loading.promise;
|
||||
return s.store.list();
|
||||
},
|
||||
},
|
||||
messages: () => s.delivery,
|
||||
open: s.observation.open.bind(s.observation),
|
||||
callerIsActive: async () => true,
|
||||
serverId: "origin",
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
services.push(recovered);
|
||||
expect(recovered.hasPendingForCaller("parent")).toBe(true);
|
||||
const starting = recovered.start();
|
||||
expect(recovered.hasPendingForCaller("parent")).toBe(true);
|
||||
loading.resolve();
|
||||
await starting;
|
||||
expect(recovered.hasPendingForCaller("parent")).toBe(true);
|
||||
expect(recovered.hasPendingForCaller("unrelated")).toBe(false);
|
||||
});
|
||||
|
||||
@@ -49,6 +49,8 @@ export interface FinishRegistration {
|
||||
}
|
||||
export interface FinishSubscriptions {
|
||||
start(): Promise<void>;
|
||||
/** Pending notification intent, including registration, recovery and unacknowledged delivery. */
|
||||
hasPendingForCaller(agentId: string): boolean;
|
||||
register(
|
||||
input: Omit<FinishSubscription, "id">,
|
||||
pendingDispatch?: boolean,
|
||||
@@ -72,6 +74,8 @@ export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptio
|
||||
const tasks = new Set<Promise<void>>();
|
||||
const watches = new Map<string, Set<Watch>>();
|
||||
let closed = false;
|
||||
let recovering = true;
|
||||
const registering = new Map<string, number>();
|
||||
let mutations = Promise.resolve();
|
||||
function serialize<T>(action: () => Promise<T>): Promise<T> {
|
||||
const result = mutations.then(action);
|
||||
@@ -300,6 +304,13 @@ export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptio
|
||||
}
|
||||
|
||||
return {
|
||||
hasPendingForCaller(agentId) {
|
||||
if (recovering || registering.has(agentId)) return true;
|
||||
for (const group of watches.values()) {
|
||||
for (const watch of group) if (watch.records[0]!.callerAgentId === agentId) return true;
|
||||
}
|
||||
return false;
|
||||
},
|
||||
start: () =>
|
||||
serialize(async () => {
|
||||
const groups = new Map<string, FinishSubscription[]>();
|
||||
@@ -309,9 +320,11 @@ export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptio
|
||||
groups.set(key(record), records);
|
||||
}
|
||||
for (const records of groups.values()) attach(records, false);
|
||||
recovering = false;
|
||||
}),
|
||||
register: (input, pendingDispatch = false) =>
|
||||
serialize(async () => {
|
||||
register: (input, pendingDispatch = false) => {
|
||||
registering.set(input.callerAgentId, (registering.get(input.callerAgentId) ?? 0) + 1);
|
||||
return serialize(async () => {
|
||||
if (closed) throw new Error("Finish subscriptions are closed");
|
||||
const record: FinishSubscription = {
|
||||
...input,
|
||||
@@ -337,13 +350,18 @@ export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptio
|
||||
},
|
||||
reject: () =>
|
||||
serialize(async () => {
|
||||
stop(watch, actualGroup);
|
||||
await remove(watch);
|
||||
stop(watch, actualGroup);
|
||||
}),
|
||||
};
|
||||
if (!pendingDispatch) registration.accept();
|
||||
return registration;
|
||||
}),
|
||||
}).finally(() => {
|
||||
const remaining = registering.get(input.callerAgentId)! - 1;
|
||||
if (remaining) registering.set(input.callerAgentId, remaining);
|
||||
else registering.delete(input.callerAgentId);
|
||||
});
|
||||
},
|
||||
async close() {
|
||||
closed = true;
|
||||
await mutations;
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
interface IdleRuntimeCleanupOptions {
|
||||
enabled: boolean;
|
||||
collect(): Promise<unknown>;
|
||||
onError(error: unknown): void;
|
||||
schedule?: (tick: () => void, intervalMs: number) => () => void;
|
||||
}
|
||||
|
||||
/** Host-owned sweep lifetime; shutdown waits for an already admitted sweep. */
|
||||
export function startIdleRuntimeCleanup(options: IdleRuntimeCleanupOptions): () => Promise<void> {
|
||||
let collection: Promise<unknown> | null = null;
|
||||
let stopped = false;
|
||||
const schedule =
|
||||
options.schedule ??
|
||||
((tick, intervalMs) => {
|
||||
const timer = setInterval(tick, intervalMs);
|
||||
timer.unref();
|
||||
return () => clearInterval(timer);
|
||||
});
|
||||
const cancel = schedule(() => {
|
||||
if (!options.enabled || stopped || collection) return;
|
||||
collection = options
|
||||
.collect()
|
||||
.catch(options.onError)
|
||||
.finally(() => {
|
||||
collection = null;
|
||||
});
|
||||
}, 60_000);
|
||||
return async () => {
|
||||
stopped = true;
|
||||
cancel();
|
||||
await collection;
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
import { expect, onTestFinished, test } from "vitest";
|
||||
import { createTestLogger } from "../../../test-utils/test-logger.js";
|
||||
import { createIdleRuntimeCleanup, type RuntimeCleanupCandidate } from "./index.js";
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((yes) => {
|
||||
resolve = yes;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
class Runtime {
|
||||
readonly identity = {};
|
||||
busy = false;
|
||||
closed = false;
|
||||
resumable = true;
|
||||
activity = 0;
|
||||
parentAgentId: string | undefined;
|
||||
probe: (() => Promise<boolean>) | undefined;
|
||||
constructor(readonly id = "agent") {}
|
||||
capture(): RuntimeCleanupCandidate {
|
||||
const activity = this.activity;
|
||||
return {
|
||||
identity: this.identity,
|
||||
agentId: this.id,
|
||||
parentAgentId: this.parentAgentId,
|
||||
provider: "test",
|
||||
sessionId: this.resumable ? this.id : null,
|
||||
lastActivityAt: activity,
|
||||
isEligible: () => !this.closed && !this.busy && this.activity === activity,
|
||||
canStopRuntime: this.probe,
|
||||
closeIfEligible: async (eligible) => {
|
||||
if (!eligible()) return false;
|
||||
this.closed = true;
|
||||
return true;
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
function scenario(enabled = true) {
|
||||
const runtime = new Runtime();
|
||||
const runtimes = [runtime];
|
||||
let time = 0;
|
||||
let protectedCaller = false;
|
||||
let tick!: () => void;
|
||||
const cleanup = createIdleRuntimeCleanup({
|
||||
enabled,
|
||||
logger: createTestLogger(),
|
||||
now: () => time,
|
||||
runtimes: () => runtimes.filter((item) => !item.closed).map((item) => item.capture()),
|
||||
isProtected: () => protectedCaller,
|
||||
schedule: (callback, interval) => {
|
||||
expect(interval).toBe(60_000);
|
||||
tick = callback;
|
||||
return () => {};
|
||||
},
|
||||
});
|
||||
onTestFinished(() => cleanup.close());
|
||||
return {
|
||||
runtime,
|
||||
runtimes,
|
||||
cleanup,
|
||||
tick: () => tick(),
|
||||
protect: (value: boolean) => {
|
||||
protectedCaller = value;
|
||||
},
|
||||
sweep: (minutes: number) => {
|
||||
time = minutes * 60_000;
|
||||
return cleanup.sweep();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
test("a runtime without a provider check is collected at exactly one hour", async () => {
|
||||
const s = scenario();
|
||||
expect((await s.sweep(59)).collected).toEqual([]);
|
||||
expect((await s.sweep(60)).collected).toEqual([
|
||||
{ agentId: "agent", provider: "test", sessionId: "agent" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("activity restarts the single one-hour idle period", async () => {
|
||||
const s = scenario();
|
||||
await s.sweep(59);
|
||||
s.runtime.activity = 59 * 60_000;
|
||||
expect((await s.sweep(118)).collected).toEqual([]);
|
||||
expect((await s.sweep(119)).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test.each(["foreground", "finish subscription"])(
|
||||
"a positive provider check cannot override %s",
|
||||
async (blocker) => {
|
||||
const s = scenario();
|
||||
let calls = 0;
|
||||
s.runtime.probe = async () => {
|
||||
calls++;
|
||||
return true;
|
||||
};
|
||||
if (blocker === "foreground") s.runtime.busy = true;
|
||||
else s.protect(true);
|
||||
expect((await s.sweep(61)).collected).toEqual([]);
|
||||
expect(calls).toBe(0);
|
||||
s.runtime.busy = false;
|
||||
s.protect(false);
|
||||
expect((await s.sweep(62)).collected).toEqual([]);
|
||||
expect((await s.sweep(121)).collected).toEqual([]);
|
||||
expect((await s.sweep(122)).collected).toHaveLength(1);
|
||||
},
|
||||
);
|
||||
|
||||
test.each(["veto", "error"])(
|
||||
"a supported provider %s retains the runtime until a fresh idle hour passes",
|
||||
async (decision) => {
|
||||
const s = scenario();
|
||||
s.runtime.probe = async () => {
|
||||
if (decision === "error") throw new Error("Probe failed");
|
||||
return false;
|
||||
};
|
||||
const blocked = await s.sweep(61);
|
||||
expect(blocked.collected).toEqual([]);
|
||||
expect(blocked.failures).toHaveLength(decision === "error" ? 1 : 0);
|
||||
await s.sweep(119);
|
||||
s.runtime.probe = async () => true;
|
||||
expect((await s.sweep(120)).collected).toEqual([]);
|
||||
expect((await s.sweep(179)).collected).toEqual([]);
|
||||
expect((await s.sweep(180)).collected).toHaveLength(1);
|
||||
},
|
||||
);
|
||||
|
||||
test("an idle managed descendant with provider-held work vetoes its parent", async () => {
|
||||
const s = scenario();
|
||||
const child = new Runtime("child");
|
||||
child.parentAgentId = s.runtime.id;
|
||||
child.probe = async () => false;
|
||||
s.runtimes.push(child);
|
||||
expect((await s.sweep(61)).collected).toEqual([]);
|
||||
child.probe = async () => true;
|
||||
expect((await s.sweep(62)).collected).toEqual([]);
|
||||
expect((await s.sweep(122)).collected).toHaveLength(2);
|
||||
});
|
||||
|
||||
test("new caller protection during an async provider probe prevents stale collection", async () => {
|
||||
const s = scenario();
|
||||
const probe = deferred<boolean>();
|
||||
s.runtime.probe = () => probe.promise;
|
||||
const collecting = s.sweep(61);
|
||||
s.protect(true);
|
||||
probe.resolve(true);
|
||||
expect((await collecting).collected).toEqual([]);
|
||||
expect(s.runtime.closed).toBe(false);
|
||||
s.protect(false);
|
||||
expect((await s.sweep(62)).collected).toEqual([]);
|
||||
expect((await s.sweep(122)).collected).toHaveLength(1);
|
||||
});
|
||||
|
||||
test("a supported provider timeout retains ownership even if it later returns true", async () => {
|
||||
const s = scenario();
|
||||
const probe = deferred<boolean>();
|
||||
s.runtime.probe = () => probe.promise;
|
||||
const result = await s.sweep(61);
|
||||
expect(result.collected).toEqual([]);
|
||||
expect(result.failures[0]?.error).toEqual(new Error("Runtime safety probe timed out for agent"));
|
||||
probe.resolve(true);
|
||||
expect((await s.sweep(62)).collected).toEqual([]);
|
||||
expect(s.runtime.closed).toBe(false);
|
||||
}, 15_000);
|
||||
|
||||
test("the host opt-out disables explicit sweeps and scheduled ticks", async () => {
|
||||
const s = scenario(false);
|
||||
let calls = 0;
|
||||
s.runtime.probe = async () => {
|
||||
calls++;
|
||||
return true;
|
||||
};
|
||||
expect((await s.sweep(61)).collected).toEqual([]);
|
||||
s.tick();
|
||||
await s.cleanup.close();
|
||||
expect(calls).toBe(0);
|
||||
expect(s.runtime.closed).toBe(false);
|
||||
});
|
||||
|
||||
test("a managed descendant without persistence still protects its parent while working", async () => {
|
||||
const s = scenario();
|
||||
const child = new Runtime("initializing-child");
|
||||
child.parentAgentId = s.runtime.id;
|
||||
child.resumable = false;
|
||||
child.busy = true;
|
||||
s.runtimes.push(child);
|
||||
expect((await s.sweep(61)).collected).toEqual([]);
|
||||
child.busy = false;
|
||||
expect((await s.sweep(62)).collected).toEqual([]);
|
||||
expect((await s.sweep(122)).collected).toEqual([
|
||||
{ agentId: "agent", provider: "test", sessionId: "agent" },
|
||||
]);
|
||||
expect(child.closed).toBe(false);
|
||||
});
|
||||
@@ -0,0 +1,149 @@
|
||||
import type { Logger } from "pino";
|
||||
import { withTimeout } from "../../../utils/promise-timeout.js";
|
||||
|
||||
/** A captured runtime capability. Its lifecycle owner revalidates before releasing it. */
|
||||
export interface RuntimeCleanupCandidate {
|
||||
readonly identity: object;
|
||||
readonly agentId: string;
|
||||
readonly parentAgentId: string | undefined;
|
||||
readonly provider: string;
|
||||
readonly sessionId: string | null;
|
||||
readonly lastActivityAt: number;
|
||||
isEligible(): boolean;
|
||||
canStopRuntime?(): Promise<boolean>;
|
||||
closeIfEligible(revalidate: () => boolean): Promise<boolean>;
|
||||
}
|
||||
interface Options {
|
||||
enabled: boolean;
|
||||
runtimes(): readonly RuntimeCleanupCandidate[];
|
||||
isProtected?(agentId: string): boolean;
|
||||
logger: Pick<Logger, "debug" | "info" | "warn">;
|
||||
now?: () => number;
|
||||
schedule?: (tick: () => void, intervalMs: number) => () => void;
|
||||
}
|
||||
interface CollectionEntry {
|
||||
agentId: string;
|
||||
provider: string;
|
||||
sessionId: string;
|
||||
}
|
||||
interface CollectionResult {
|
||||
collected: CollectionEntry[];
|
||||
failures: Array<{ agentId: string; error: unknown }>;
|
||||
}
|
||||
const IDLE_PERIOD_MS = 60 * 60_000;
|
||||
|
||||
/** Owns idle clocks, baseline eligibility, optional provider vetoes and the sweep lifetime. */
|
||||
export function createIdleRuntimeCleanup(options: Options) {
|
||||
const protectedAt = new WeakMap<object, { at: number; protected: boolean }>();
|
||||
let collection: Promise<CollectionResult> | null = null;
|
||||
let stopped = false;
|
||||
const isProtected = options.isProtected ?? (() => false);
|
||||
const now = options.now ?? Date.now;
|
||||
|
||||
function familyOf(root: RuntimeCleanupCandidate, runtimes: readonly RuntimeCleanupCandidate[]) {
|
||||
const family = new Map([[root.agentId, root]]);
|
||||
for (const parent of family.values()) {
|
||||
for (const child of runtimes) {
|
||||
if (child.parentAgentId === parent.agentId) family.set(child.agentId, child);
|
||||
}
|
||||
}
|
||||
return [...family.values()];
|
||||
}
|
||||
function baselineAllows(family: readonly RuntimeCleanupCandidate[]) {
|
||||
return family.every((runtime) => runtime.isEligible() && !isProtected(runtime.agentId));
|
||||
}
|
||||
async function providersAllow(family: readonly RuntimeCleanupCandidate[]) {
|
||||
for (const runtime of family) {
|
||||
if (!runtime.canStopRuntime) continue;
|
||||
const allowed = await withTimeout(
|
||||
runtime.canStopRuntime(),
|
||||
10_000,
|
||||
`Runtime safety probe timed out for ${runtime.agentId}`,
|
||||
);
|
||||
if (allowed !== true) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
async function collect(): Promise<CollectionResult> {
|
||||
const result: CollectionResult = { collected: [], failures: [] };
|
||||
if (!options.enabled || stopped) return result;
|
||||
const timestamp = now();
|
||||
const runtimes = options.runtimes();
|
||||
for (const runtime of runtimes) {
|
||||
if (!runtime.sessionId) continue;
|
||||
const family = familyOf(runtime, runtimes);
|
||||
try {
|
||||
if (!baselineAllows(family) || !(await providersAllow(family))) {
|
||||
protectedAt.set(runtime.identity, { at: timestamp, protected: true });
|
||||
options.logger.debug(
|
||||
{ agentId: runtime.agentId },
|
||||
"Idle runtime retained by a baseline or provider veto",
|
||||
);
|
||||
continue;
|
||||
}
|
||||
const previous = protectedAt.get(runtime.identity);
|
||||
if (previous?.protected)
|
||||
protectedAt.set(runtime.identity, { at: timestamp, protected: false });
|
||||
const idleSince = Math.max(
|
||||
runtime.lastActivityAt,
|
||||
protectedAt.get(runtime.identity)?.at ?? 0,
|
||||
);
|
||||
if (timestamp - idleSince < IDLE_PERIOD_MS) continue;
|
||||
const revalidate = () => {
|
||||
const currentFamily = familyOf(runtime, options.runtimes());
|
||||
return (
|
||||
currentFamily.length === family.length &&
|
||||
currentFamily.every((current) =>
|
||||
family.some((captured) => captured.identity === current.identity),
|
||||
) &&
|
||||
baselineAllows(family)
|
||||
);
|
||||
};
|
||||
if (await runtime.closeIfEligible(revalidate)) {
|
||||
const entry = {
|
||||
agentId: runtime.agentId,
|
||||
provider: runtime.provider,
|
||||
sessionId: runtime.sessionId,
|
||||
};
|
||||
result.collected.push(entry);
|
||||
options.logger.info(entry, "Collected idle agent runtime");
|
||||
} else {
|
||||
protectedAt.set(runtime.identity, { at: timestamp, protected: true });
|
||||
}
|
||||
} catch (error) {
|
||||
protectedAt.set(runtime.identity, { at: timestamp, protected: true });
|
||||
result.failures.push({ agentId: runtime.agentId, error });
|
||||
options.logger.warn(
|
||||
{ agentId: runtime.agentId, err: error },
|
||||
"Idle runtime cleanup failed",
|
||||
);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
function sweep(): Promise<CollectionResult> {
|
||||
if (collection) return collection;
|
||||
collection = collect().finally(() => {
|
||||
collection = null;
|
||||
});
|
||||
return collection;
|
||||
}
|
||||
const schedule =
|
||||
options.schedule ??
|
||||
((tick, intervalMs) => {
|
||||
const timer = setInterval(tick, intervalMs);
|
||||
timer.unref();
|
||||
return () => clearInterval(timer);
|
||||
});
|
||||
const cancel = schedule(() => {
|
||||
void sweep().catch((error) => options.logger.warn({ err: error }, "Idle runtime sweep failed"));
|
||||
}, 60_000);
|
||||
return {
|
||||
sweep,
|
||||
async close() {
|
||||
stopped = true;
|
||||
cancel();
|
||||
await collection;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,3 +1,8 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { AgentStorage } from "./agent-storage.js";
|
||||
import { createIdleRuntimeCleanup } from "./idle-runtime-cleanup/index.js";
|
||||
import { relative } from "node:path";
|
||||
import type {
|
||||
ProviderConnection,
|
||||
@@ -1268,7 +1273,8 @@ test.each([true, false, "error", "wrong-session", "missing"] as const)(
|
||||
await expect(session.canStopRuntime!()).rejects.toThrow(
|
||||
answer === "error" ? "Probe failed" : "another session",
|
||||
);
|
||||
} else expect(await session.canStopRuntime!()).toBe(answer === true);
|
||||
} else if (answer === "missing") expect(session.canStopRuntime).toBeUndefined();
|
||||
else expect(await session.canStopRuntime!()).toBe(answer === true);
|
||||
const requests = harness.inputs.filter((input) => input.type === "session.can_stop_runtime");
|
||||
expect(requests).toHaveLength(answer === "missing" ? 0 : 1);
|
||||
if (answer !== "missing")
|
||||
@@ -1281,3 +1287,37 @@ test.each([true, false, "error", "wrong-session", "missing"] as const)(
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
test("plugin without a runtime safety capability uses baseline collection without a query", async () => {
|
||||
const home = await mkdtemp(join(tmpdir(), "plugin-baseline-cleanup-"));
|
||||
const logger = createTestLogger();
|
||||
const storage = new AgentStorage(join(home, "agents"), logger);
|
||||
const harness = createProviderHarness();
|
||||
const registry = new PluginAgentClientRegistry(logger);
|
||||
registry.replace([harness.registration]);
|
||||
const manager = new AgentManager({ clients: registry.clients(), registry: storage, logger });
|
||||
const agent = await manager.createAgent(
|
||||
{ provider: harness.registration.id, cwd: home },
|
||||
undefined,
|
||||
{ workspaceId: undefined },
|
||||
);
|
||||
const cleanup = createIdleRuntimeCleanup({
|
||||
enabled: true,
|
||||
logger,
|
||||
runtimes: () => manager.getRuntimeCleanupCandidates(),
|
||||
now: () => agent.updatedAt.getTime() + 60 * 60_000,
|
||||
schedule: () => () => {},
|
||||
});
|
||||
try {
|
||||
expect(agent.session?.canStopRuntime).toBeUndefined();
|
||||
expect((await cleanup.sweep()).collected).toHaveLength(1);
|
||||
expect(manager.getAgent(agent.id)).toBeNull();
|
||||
expect(harness.inputs.filter((input) => input.type === "session.can_stop_runtime")).toEqual([]);
|
||||
} finally {
|
||||
await cleanup.close();
|
||||
await Promise.all(manager.listAgents().map((item) => manager.closeAgent(item.id)));
|
||||
await registry.shutdown();
|
||||
await storage.flush();
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
@@ -1239,8 +1239,9 @@ class PluginAgentSession implements AgentSession {
|
||||
for (const event of this.history) yield event;
|
||||
}
|
||||
|
||||
canStopRuntime(): Promise<boolean> {
|
||||
return this.bridge.canStopRuntime();
|
||||
get canStopRuntime(): (() => Promise<boolean>) | undefined {
|
||||
if (!this.bridge.negotiatedCapabilities.includes("session.can_stop_runtime")) return undefined;
|
||||
return () => this.bridge.canStopRuntime();
|
||||
}
|
||||
|
||||
async getRuntimeInfo() {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { startIdleRuntimeCleanup } from "./agent/idle-runtime-cleanup.js";
|
||||
import { createIdleRuntimeCleanup } from "./agent/idle-runtime-cleanup/index.js";
|
||||
import {
|
||||
createFinishSubscriptions,
|
||||
createFinishObservation,
|
||||
@@ -1400,14 +1400,6 @@ export async function createPaseoDaemon(
|
||||
archiveWorkspace: archiveScheduleWorkspaceExternal,
|
||||
});
|
||||
await scheduleService.start();
|
||||
const stopIdleRuntimeCleanup = startIdleRuntimeCleanup({
|
||||
enabled: config.idleRuntimeCleanup !== false,
|
||||
collect: () =>
|
||||
agentManager.collectIdleAgents({
|
||||
isProtected: (agentId) => wsServer?.agentMessages.hasPending(agentId) ?? false,
|
||||
}),
|
||||
onError: (error) => logger.warn({ err: error }, "Idle runtime sweep failed"),
|
||||
});
|
||||
|
||||
agentManager.setAgentArchivedCallback(async (agentId) => {
|
||||
try {
|
||||
@@ -1456,6 +1448,15 @@ export async function createPaseoDaemon(
|
||||
}),
|
||||
});
|
||||
|
||||
const idleRuntimeCleanup = createIdleRuntimeCleanup({
|
||||
enabled: config.idleRuntimeCleanup !== false,
|
||||
runtimes: () => agentManager.getRuntimeCleanupCandidates(),
|
||||
isProtected: (agentId) =>
|
||||
finishSubscriptions.hasPendingForCaller(agentId) ||
|
||||
(wsServer?.agentMessages.hasPending(agentId) ?? false),
|
||||
logger,
|
||||
});
|
||||
|
||||
const createAgentToolHostDependencies = (
|
||||
runtime: PaseoToolRuntimeContext,
|
||||
): PaseoToolHostDependencies => ({
|
||||
@@ -1918,7 +1919,7 @@ export async function createPaseoDaemon(
|
||||
await hubRelationships.stop();
|
||||
workspaceReconciliation.dispose();
|
||||
scriptHealthMonitor.stop();
|
||||
await stopIdleRuntimeCleanup();
|
||||
await idleRuntimeCleanup.close();
|
||||
// Freeze both ingress and registration before taking the agent closure snapshot.
|
||||
wsServer?.prepareForShutdown();
|
||||
await finishSubscriptions.close();
|
||||
|
||||
@@ -86,8 +86,8 @@ See [Providers](/docs/providers) for the mental model and [Supported providers](
|
||||
|
||||
## Idle runtimes
|
||||
|
||||
Paseo releases resumable idle agent processes after 30 minutes when the provider confirms that no
|
||||
background work needs the runtime. Your next prompt resumes the same conversation. Disable this
|
||||
Paseo releases resumable idle agent processes after one hour when its daemon baseline allows
|
||||
cleanup and any implemented provider check agrees. Your next prompt resumes the same conversation. Disable this
|
||||
for the entire host with:
|
||||
|
||||
```json
|
||||
@@ -95,7 +95,8 @@ for the entire host with:
|
||||
```
|
||||
|
||||
This setting defaults to `true` and requires a daemon restart to change. There is no per-agent
|
||||
setting. Providers without a runtime safety check stay resident.
|
||||
setting. Providers without a runtime safety check use the same daemon baseline and one-hour idle
|
||||
period. Active work, descendants and pending finish notifications prevent cleanup.
|
||||
|
||||
## Worktrees
|
||||
|
||||
|
||||
@@ -84,13 +84,13 @@ Agent Provider,包括 Paseo 自带的一等 Provider 和你在 `agents.provide
|
||||
|
||||
## 空闲运行时 {#idle-runtimes}
|
||||
|
||||
当 Provider 确认没有后台工作需要运行时时,Paseo 会在可恢复的 Agent 空闲 30 分钟后释放其进程。你的下一条提示词会恢复同一个会话。要为整个主机禁用此功能,请设置:
|
||||
当 daemon 的基础条件允许清理,且 Provider 已实现的安全检查也允许时,Paseo 会在可恢复的 Agent 空闲一小时后释放其进程。你的下一条提示词会恢复同一个会话。要为整个主机禁用此功能,请设置:
|
||||
|
||||
```json
|
||||
{ "agents": { "idleRuntimeCleanup": false } }
|
||||
```
|
||||
|
||||
此设置默认为 `true`,更改后需要重启 daemon。没有单个 Agent 的设置。不提供运行时安全检查的 Provider 会保持驻留。
|
||||
此设置默认为 `true`,更改后需要重启 daemon。没有单个 Agent 的设置。不提供运行时安全检查的 Provider 使用相同的 daemon 基础条件和一小时空闲期。活动中的工作、后代 Agent 和待发送的完成通知都会阻止清理。
|
||||
|
||||
## worktree {#worktrees}
|
||||
|
||||
|
||||
@@ -308,9 +308,10 @@ can replace the active turn instead.
|
||||
|
||||
## Permit idle runtime cleanup
|
||||
|
||||
Paseo can release an idle resumable session after 30 minutes. Opt in by advertising
|
||||
`session.can_stop_runtime` on the connection and the session's `session.opened` capabilities.
|
||||
Without this capability your sessions stay resident. Older plugins receive no new request.
|
||||
Paseo can release an idle resumable session after one hour when its daemon baseline allows cleanup.
|
||||
Add a provider-owned veto by advertising `session.can_stop_runtime` on the connection and the
|
||||
session's `session.opened` capabilities. Without this capability, the same baseline and one-hour
|
||||
idle period apply, and the plugin receives no safety request.
|
||||
|
||||
Handle the safety query in your existing `send(input)` handler. It addresses the provider's own
|
||||
session ID, including when one connection serves several sessions:
|
||||
@@ -343,10 +344,12 @@ queued completion reports, or pending autonomous replies. This query must not st
|
||||
session. Return `false` while work remains. A rejected query, invalid response, or timeout also
|
||||
keeps the runtime alive. Never respond with `request.completed` for this query.
|
||||
|
||||
Paseo checks its own foreground turns, permissions, queued messages, and managed descendants,
|
||||
then rechecks activity before calling the existing `session.close` operation. A positive safety
|
||||
response permits collection; it does not request it. After protected work finishes, a fresh
|
||||
30-minute idle period starts. The saved provider persistence must resume the same conversation.
|
||||
Paseo always checks foreground work, permissions, queued deliveries, lifecycle operations,
|
||||
working managed/native descendants, and outstanding local or remote finish subscriptions for the
|
||||
caller. Subscription protection lasts through registration, recovery and notification dispatch
|
||||
acknowledgment. A positive provider response never overrides these blockers. Paseo rechecks
|
||||
activity before the existing `session.close` operation. After protected work finishes, a fresh
|
||||
one-hour idle period starts. The saved provider persistence must resume the same conversation.
|
||||
The host setting `agents.idleRuntimeCleanup: false` disables collection for every provider.
|
||||
|
||||
## Publish timeline items
|
||||
|
||||
@@ -271,7 +271,7 @@ Paseo 刷新 Agent 的方式是关闭它当前的 Provider 会话,然后用当
|
||||
|
||||
## 允许清理空闲运行时 {#permit-idle-runtime-cleanup}
|
||||
|
||||
Paseo 可以在可恢复的会话空闲 30 分钟后释放其运行时。要启用此功能,请在连接和会话的 `session.opened` capabilities 中声明 `session.can_stop_runtime`。没有此能力的会话会保持驻留。旧版插件不会收到新请求。
|
||||
当 daemon 的基础条件允许清理时,Paseo 可以在可恢复的会话空闲一小时后释放其运行时。在连接和会话的 `session.opened` capabilities 中声明 `session.can_stop_runtime`,即可添加 Provider 自己的否决检查。没有此能力时,会话仍使用相同的基础条件和一小时空闲期,插件不会收到安全查询。
|
||||
|
||||
在现有的 `send(input)` 处理函数中处理安全查询。请求使用 Provider 自己的会话 ID,包括同一连接服务多个会话的情况:
|
||||
|
||||
@@ -299,7 +299,7 @@ case "session.can_stop_runtime": {
|
||||
|
||||
`canStopRuntime(): Promise<boolean>` 由你的 Provider 会话实现。只有释放运行时不会丢失后台 shell、监控任务、原生子 Agent、排队的完成报告或待处理的自主回复时,才返回 `true`。此查询不得停止或修改会话。仍有工作时返回 `false`。查询拒绝、响应无效或超时也会保留运行时。不要用 `request.completed` 响应此查询。
|
||||
|
||||
Paseo 会检查自己管理的前台轮次、权限、排队消息和托管后代 Agent,然后在调用现有的 `session.close` 操作之前重新检查活动。肯定的安全响应允许清理,但不会请求清理。受保护的工作完成后,会重新开始 30 分钟的空闲期。保存的 Provider 持久化数据必须能恢复同一个会话。主机设置 `agents.idleRuntimeCleanup: false` 会禁用所有 Provider 的清理。
|
||||
Paseo 始终检查前台工作、权限、排队投递、生命周期操作、工作中的托管或原生后代 Agent,以及调用方尚未完成的本地或远程完成通知订阅。订阅保护覆盖注册、恢复,直到通知投递获确认。Provider 返回 `true` 也不能绕过这些阻止条件。Paseo 会在调用现有的 `session.close` 操作之前重新检查活动。受保护的工作完成后,会重新开始一小时的空闲期。保存的 Provider 持久化数据必须能恢复同一个会话。主机设置 `agents.idleRuntimeCleanup: false` 会禁用所有 Provider 的清理。
|
||||
|
||||
## 发布时间线条目 {#publish-timeline-items}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user