mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-24 10:12:44 +00:00
feat(server): release idle agent runtimes with provider safety checks
Require provider permission before releasing resumable runtimes so foreground idleness does not discard tracked background work. Preserve the saved conversation and serialize prompt admission with runtime closure; allow hosts to disable collection.
This commit is contained in:
Mohamed Boudra
+35
-2
@@ -25,8 +25,41 @@ 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 agents remain resident indefinitely. Runtime closure happens only through an explicit lifecycle
|
||||
action such as archive, replacement, reload, workspace teardown, or daemon shutdown.
|
||||
Idle resumable runtimes are collected after 30 minutes 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.
|
||||
|
||||
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
|
||||
after closure starts waits for release and resumes the saved session.
|
||||
|
||||
Provider checks follow the signals their runtimes expose:
|
||||
|
||||
- Claude uses its process-local background roster and pending completion notifications. A fresh
|
||||
process starts empty. CLI 2.1.292 or newer is known to support the roster; older binaries must
|
||||
demonstrate support by emitting one. Unknown versions stay resident until then.
|
||||
- Codex queries background terminals for loaded threads, including native children, and protects
|
||||
pending provider requests. App servers without the terminal-list RPC stay resident.
|
||||
- OpenCode checks active executions and descendants. V1 checks PTYs; v2 also checks shell and
|
||||
inbox inventories. The shared server remains leased while a session vetoes collection.
|
||||
- Pi checks streaming, compaction, pending replies, and adapter-owned operations. Known native
|
||||
subagent signals also protect their parent through the daemon's subagent store.
|
||||
- OMP additionally requires `get_state.isSettled: true` and `hasPendingAsyncWork: false`.
|
||||
These protect jobs with unsuppressed completion delivery, pending deliveries, queued result
|
||||
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.
|
||||
|
||||
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).
|
||||
|
||||
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
|
||||
|
||||
@@ -498,6 +498,11 @@ 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
|
||||
[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.
|
||||
|
||||
Agent refresh closes the current provider session and opens it again with current configuration and
|
||||
persistence. Re-read credentials and provider-owned configuration on `session.open`; consume the
|
||||
daemon launch from `connect` and the per-session env and MCP servers from `session.open`. There is
|
||||
|
||||
@@ -9,6 +9,7 @@ export const PROVIDER_CAPABILITIES = [
|
||||
"prompt.image",
|
||||
"prompt.output_schema",
|
||||
"prompt.steer",
|
||||
"session.can_stop_runtime",
|
||||
"session.archive",
|
||||
"session.configure",
|
||||
"session.list",
|
||||
@@ -277,6 +278,7 @@ export type ProviderInput =
|
||||
}
|
||||
| { type: "session.archive"; requestId: string; persistence: ProviderPersistence }
|
||||
| { type: "session.unarchive"; requestId: string; persistence: ProviderPersistence }
|
||||
| { type: "session.can_stop_runtime"; requestId: string; sessionId: string }
|
||||
| { type: "session.close"; requestId: string; sessionId: string };
|
||||
|
||||
export interface ProviderThinkingOption {
|
||||
@@ -552,6 +554,7 @@ export type ProviderTimelineItem =
|
||||
});
|
||||
|
||||
export type ProviderEvent =
|
||||
| { type: "session.can_stop_runtime"; requestId: string; sessionId: string; canStop: boolean }
|
||||
| { type: "catalog"; requestId: string; catalog: ProviderCatalog }
|
||||
| { type: "sessions"; requestId: string; sessions: ProviderSessionSummary[] }
|
||||
| { type: "request.completed"; requestId: string }
|
||||
@@ -634,6 +637,8 @@ function requiredPromptCapabilities(prompt: ProviderPrompt): readonly ProviderCa
|
||||
|
||||
export function requiredProviderCapabilities(input: ProviderInput): readonly ProviderCapability[] {
|
||||
switch (input.type) {
|
||||
case "session.can_stop_runtime":
|
||||
return ["session.can_stop_runtime"];
|
||||
case "catalog":
|
||||
case "session.interrupt":
|
||||
case "session.close":
|
||||
@@ -955,6 +960,13 @@ export const ProviderInputSchema: z.ZodType<ProviderInput> = z.discriminatedUnio
|
||||
persistence: persistenceSchema,
|
||||
})
|
||||
.strict(),
|
||||
z
|
||||
.object({
|
||||
type: z.literal("session.can_stop_runtime"),
|
||||
requestId: idSchema,
|
||||
sessionId: idSchema,
|
||||
})
|
||||
.strict(),
|
||||
z.object({ type: z.literal("session.close"), requestId: idSchema, sessionId: idSchema }).strict(),
|
||||
]);
|
||||
|
||||
@@ -1326,6 +1338,14 @@ export const ProviderEventSchema: z.ZodType<ProviderEvent> = z.discriminatedUnio
|
||||
sessionId: idSchema,
|
||||
})
|
||||
.strip(),
|
||||
z
|
||||
.object({
|
||||
type: z.literal("session.can_stop_runtime"),
|
||||
requestId: idSchema,
|
||||
sessionId: idSchema,
|
||||
canStop: z.boolean(),
|
||||
})
|
||||
.strip(),
|
||||
z
|
||||
.object({
|
||||
type: z.literal("session.closed"),
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { startIdleRuntimeCleanup } from "./idle-runtime-cleanup.js";
|
||||
import {
|
||||
prepareAgentMessage,
|
||||
formatAgentMessage,
|
||||
@@ -10007,6 +10008,46 @@ test("closeAgent persists one final closed snapshot", async () => {
|
||||
}
|
||||
});
|
||||
|
||||
test("idle runtime cleanup requires provider permission 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;
|
||||
const client = new (class extends TestAgentClient {
|
||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
const session = await super.createSession(config);
|
||||
session.canStopRuntime = async () => safe;
|
||||
return session;
|
||||
}
|
||||
})();
|
||||
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
|
||||
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: [] });
|
||||
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([
|
||||
{ agentId: agent.id, provider: "codex", sessionId: agent.persistence?.sessionId },
|
||||
]);
|
||||
expect(manager.getAgent(agent.id)).toBeNull();
|
||||
const resumed = await ensureAgentLoaded(agent.id, {
|
||||
agentManager: manager,
|
||||
agentStorage: storage,
|
||||
logger,
|
||||
});
|
||||
expect(resumed.id).toBe(agent.id);
|
||||
expect(resumed.persistence).toEqual(agent.persistence);
|
||||
await manager.runAgent(agent.id, "Continue the saved conversation");
|
||||
} finally {
|
||||
await manager.closeAgent(manager.listAgents()[0]?.id ?? "missing").catch(() => undefined);
|
||||
await storage.flush();
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("idle agents remain resident until an explicit lifecycle action closes them", async () => {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-residency-"));
|
||||
let closeCount = 0;
|
||||
@@ -10951,12 +10992,12 @@ test("tryRunOutOfBand gives providers the command another agent sent, not its en
|
||||
|
||||
try {
|
||||
const fromAgent = prepareAgentMessage("/handled", { kind: "agent-message", agentId: "sender" });
|
||||
expect(manager.tryRunOutOfBand(agent.id, fromAgent.prompt)).toBe(true);
|
||||
expect(await manager.tryRunOutOfBand(agent.id, fromAgent.prompt)).toBe(true);
|
||||
expect(seenPrompts).toEqual(["/handled"]);
|
||||
|
||||
// A human who pastes an envelope gets it quoted, and it stays a plain prompt.
|
||||
const pasted = prepareAgentMessage(String(fromAgent.prompt));
|
||||
expect(manager.tryRunOutOfBand(agent.id, pasted.prompt)).toBe(false);
|
||||
expect(await manager.tryRunOutOfBand(agent.id, pasted.prompt)).toBe(false);
|
||||
expect(seenPrompts.at(-1)).toBe(pasted.prompt);
|
||||
} finally {
|
||||
await manager.closeAgent(agent.id);
|
||||
@@ -12083,3 +12124,221 @@ test("reads the last durable assistant response without loading a stored-only ch
|
||||
);
|
||||
expect(manager.getAgent("stored-child")).toBeNull();
|
||||
});
|
||||
|
||||
function idleCleanupHarness() {
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-idle-protection-"));
|
||||
const storage = new AgentStorage(join(workdir, "agents"), logger);
|
||||
const commands: AgentPromptInput[] = [];
|
||||
const prompts: AgentPromptInput[] = [];
|
||||
const recordPrompts = (session: AgentSession) => {
|
||||
session.tryHandleOutOfBand = (prompt) =>
|
||||
prompt === "/maintenance"
|
||||
? {
|
||||
run: async () => {
|
||||
commands.push(prompt);
|
||||
},
|
||||
}
|
||||
: null;
|
||||
const start = session.startTurn.bind(session);
|
||||
session.startTurn = (prompt, options) => {
|
||||
prompts.push(prompt);
|
||||
return start(prompt, options);
|
||||
};
|
||||
return session;
|
||||
};
|
||||
const sessions: AgentSession[] = [];
|
||||
const probes: Array<(() => Promise<boolean>) | undefined> = [];
|
||||
const closeEntered = deferred<void>();
|
||||
let closeGate: Promise<void> | undefined;
|
||||
let closes = 0;
|
||||
const client = new (class extends TestAgentClient {
|
||||
override async resumeSession(
|
||||
handle: AgentPersistenceHandle,
|
||||
config?: Partial<AgentSessionConfig>,
|
||||
): Promise<AgentSession> {
|
||||
const session = await super.resumeSession(handle, config);
|
||||
session.describePersistence = () => handle;
|
||||
return recordPrompts(session);
|
||||
}
|
||||
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
|
||||
const index = sessions.length;
|
||||
const session: AgentSession = await super.createSession(config);
|
||||
session.canStopRuntime = async () => probes[index]?.() ?? false;
|
||||
session.close = async () => {
|
||||
closes++;
|
||||
closeEntered.resolve();
|
||||
await closeGate;
|
||||
};
|
||||
sessions.push(session);
|
||||
return recordPrompts(session);
|
||||
}
|
||||
})();
|
||||
const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
|
||||
onTestFinished(async () => {
|
||||
closeGate = undefined;
|
||||
await Promise.all(manager.listAgents().map((agent) => manager.closeAgent(agent.id)));
|
||||
await storage.flush();
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
});
|
||||
return {
|
||||
commands,
|
||||
manager,
|
||||
storage,
|
||||
sessions,
|
||||
probes,
|
||||
prompts,
|
||||
closeEntered,
|
||||
get closes() {
|
||||
return closes;
|
||||
},
|
||||
holdClose: (gate: Promise<void>) => {
|
||||
closeGate = gate;
|
||||
},
|
||||
create: (labels?: Record<string, string>) =>
|
||||
manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
labels,
|
||||
}),
|
||||
later: (minutes: number) => new Date(Date.now() + minutes * 60_000),
|
||||
};
|
||||
}
|
||||
|
||||
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();
|
||||
const entered = deferred<void>();
|
||||
const probe = deferred<boolean>();
|
||||
h.probes[0] = () => {
|
||||
entered.resolve();
|
||||
return probe.promise;
|
||||
};
|
||||
const sweep = h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
await entered.promise;
|
||||
await h.manager.runAgent(agent.id, "arrived during safety query");
|
||||
probe.resolve(true);
|
||||
expect((await sweep).collected).toEqual([]);
|
||||
expect(h.closes).toBe(0);
|
||||
});
|
||||
|
||||
test("idle cleanup retains a parent whose idle managed child owns background work", async () => {
|
||||
const h = idleCleanupHarness();
|
||||
const parent = await h.create();
|
||||
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(h.closes).toBe(0);
|
||||
});
|
||||
|
||||
test("idle cleanup rechecks queued deliveries admitted during the safety probe", async () => {
|
||||
const h = idleCleanupHarness();
|
||||
await h.create();
|
||||
let queued = false;
|
||||
h.probes[0] = async () => {
|
||||
queued = true;
|
||||
return true;
|
||||
};
|
||||
expect(
|
||||
(await h.manager.collectIdleAgents({ now: h.later(31), isProtected: () => queued })).collected,
|
||||
).toEqual([]);
|
||||
expect(h.closes).toBe(0);
|
||||
});
|
||||
|
||||
test("a prompt racing idle cleanup waits for close and resumes exactly once", async () => {
|
||||
const h = idleCleanupHarness();
|
||||
const agent = await h.create();
|
||||
h.probes[0] = async () => true;
|
||||
const gate = deferred<void>();
|
||||
h.holdClose(gate.promise);
|
||||
const sweep = h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
await h.closeEntered.promise;
|
||||
const run = h.manager.runAgent(agent.id, "arrived while closing");
|
||||
gate.resolve();
|
||||
expect((await sweep).collected).toHaveLength(1);
|
||||
const result = await run;
|
||||
expect(result.sessionId).toBe(agent.persistence?.sessionId);
|
||||
expect(h.closes).toBe(1);
|
||||
expect(h.prompts).toEqual(["arrived while closing"]);
|
||||
});
|
||||
|
||||
test("an out-of-band command racing idle cleanup waits for the resumed runtime", async () => {
|
||||
const h = idleCleanupHarness();
|
||||
const agent = await h.create();
|
||||
h.probes[0] = async () => true;
|
||||
const gate = deferred<void>();
|
||||
h.holdClose(gate.promise);
|
||||
const sweep = h.manager.collectIdleAgents({ now: h.later(31) });
|
||||
await h.closeEntered.promise;
|
||||
const command = startAgentRun(h.manager, agent.id, "/maintenance", logger);
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
const deliveredWhileClosing = [...h.commands];
|
||||
gate.resolve();
|
||||
await sweep;
|
||||
expect(deliveredWhileClosing).toEqual([]);
|
||||
expect(await command).toEqual({ disposition: "out_of_band" });
|
||||
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 { ensureAgentLoaded } from "./agent-loading.js";
|
||||
import { projectTimelineRows } from "./timeline-projection.js";
|
||||
import type { PluginLifecycle } from "../plugins/lifecycle/index.js";
|
||||
import { describeHookAgent, publishAgentStream } from "../plugins/lifecycle/index.js";
|
||||
@@ -758,6 +759,8 @@ 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 collectingRuntimes = new Set<string>();
|
||||
private readonly lifecycleMutationTails = new Map<string, Promise<void>>();
|
||||
private readonly agentStreamCoalescer: AgentStreamCoalescer;
|
||||
private mcpBaseUrl: string | null;
|
||||
@@ -1747,6 +1750,134 @@ 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 =>
|
||||
agent.lifecycle === "idle" &&
|
||||
!agent.internal &&
|
||||
agent.activeForegroundTurnId === null &&
|
||||
!this.runs.hasRun(agent.id) &&
|
||||
!agent.pendingReplacement &&
|
||||
agent.pendingPermissions.size === 0 &&
|
||||
agent.inFlightPermissionResponses.size === 0 &&
|
||||
(inCollection || !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;
|
||||
}
|
||||
|
||||
closeAgent(agentId: string): Promise<void> {
|
||||
const existing = this.inFlightAgentCloses.get(agentId);
|
||||
if (existing) {
|
||||
@@ -1767,7 +1898,7 @@ export class AgentManager {
|
||||
return close;
|
||||
}
|
||||
|
||||
private async closeAgentRuntime(agentId: string): Promise<void> {
|
||||
private async closeAgentRuntime(agentId: string, canClose?: () => boolean): Promise<boolean> {
|
||||
const agent = this.requireAgent(agentId);
|
||||
this.logger.trace(
|
||||
{
|
||||
@@ -1782,6 +1913,7 @@ export class AgentManager {
|
||||
"agent.manager.close.start",
|
||||
);
|
||||
await this.drainSessionEvents(agentId);
|
||||
if (canClose && !canClose()) return false;
|
||||
// Retain ownership until shutdown succeeds. A failed close may still own a
|
||||
// native writer, so publishing a resumable closed snapshot would orphan it.
|
||||
await agent.session.close();
|
||||
@@ -1807,6 +1939,7 @@ export class AgentManager {
|
||||
if (persistError !== undefined) {
|
||||
throw persistError;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private cancelRunningProviderSubagents(parentAgentId: string): void {
|
||||
@@ -2459,7 +2592,19 @@ export class AgentManager {
|
||||
* emitted by the handler flow through dispatchStream so they persist and
|
||||
* broadcast like normal timeline events.
|
||||
*/
|
||||
tryRunOutOfBand(agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions): boolean {
|
||||
async tryRunOutOfBand(
|
||||
agentId: string,
|
||||
prompt: AgentPromptInput,
|
||||
options?: AgentRunOptions,
|
||||
): Promise<boolean> {
|
||||
if (this.collectingRuntimes.has(agentId) || !this.agents.has(agentId)) {
|
||||
await this.waitForAgentClose(agentId);
|
||||
await ensureAgentLoaded(agentId, {
|
||||
agentManager: this,
|
||||
agentStorage: this.requireRegistry(),
|
||||
logger: this.logger,
|
||||
});
|
||||
}
|
||||
const agent = this.requireSessionAgent(agentId);
|
||||
// Providers parse slash commands from the raw text, so a command another agent sent
|
||||
// must reach them without its sender envelope.
|
||||
@@ -2470,6 +2615,7 @@ export class AgentManager {
|
||||
if (!handler) {
|
||||
return false;
|
||||
}
|
||||
this.touchUpdatedAt(agent);
|
||||
if (options?.clientMessageId) {
|
||||
this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "accepted");
|
||||
this.emitState(agent);
|
||||
@@ -2489,7 +2635,7 @@ export class AgentManager {
|
||||
}
|
||||
this.dispatchStream(agent.id, event, { timestamp: new Date().toISOString() });
|
||||
};
|
||||
void (async () => {
|
||||
void this.runForegroundMutation(agentId, async () => {
|
||||
try {
|
||||
await handler.run({ emit: dispatch });
|
||||
} catch (error) {
|
||||
@@ -2500,7 +2646,7 @@ export class AgentManager {
|
||||
item: { type: "assistant_message", text: `[Error] ${text}` },
|
||||
});
|
||||
}
|
||||
})();
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -2584,11 +2730,27 @@ export class AgentManager {
|
||||
}
|
||||
}
|
||||
|
||||
private async *streamAfterRuntimeCollection(
|
||||
agentId: string,
|
||||
prompt: AgentPromptInput,
|
||||
options?: AgentRunOptions,
|
||||
): AsyncGenerator<AgentStreamEvent> {
|
||||
await this.waitForAgentClose(agentId);
|
||||
await ensureAgentLoaded(agentId, {
|
||||
agentManager: this,
|
||||
agentStorage: this.requireRegistry(),
|
||||
logger: this.logger,
|
||||
});
|
||||
yield* this.streamAgent(agentId, prompt, options);
|
||||
}
|
||||
|
||||
streamAgent(
|
||||
agentId: string,
|
||||
prompt: AgentPromptInput,
|
||||
options?: AgentRunOptions,
|
||||
): AsyncGenerator<AgentStreamEvent> {
|
||||
if (this.collectingRuntimes.has(agentId) || (!this.agents.has(agentId) && this.registry))
|
||||
return this.streamAfterRuntimeCollection(agentId, prompt, options);
|
||||
const existingAgent = this.requireSessionAgent(agentId);
|
||||
this.logger.trace(
|
||||
{
|
||||
|
||||
@@ -283,3 +283,33 @@ test("batches retain each sender's attachment context and image order", async ()
|
||||
)?.map((message) => message.source?.agentId),
|
||||
).toEqual(["one", "two"]);
|
||||
});
|
||||
|
||||
test("runtime safety sees queued messages until provider dispatch settles", async () => {
|
||||
const clock = createClock();
|
||||
let release!: () => void;
|
||||
const dispatched = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const delivery = createAgentMessageDelivery({
|
||||
schedule: clock.schedule,
|
||||
dispatch: async () => {
|
||||
await dispatched;
|
||||
return "steered";
|
||||
},
|
||||
onError: (error) => {
|
||||
throw error;
|
||||
},
|
||||
});
|
||||
expect(delivery.hasPending("parent")).toBe(false);
|
||||
delivery.enqueue({
|
||||
recipientId: "parent",
|
||||
prompt: "finished",
|
||||
source: { kind: "agent-message", agentId: "child" },
|
||||
});
|
||||
expect(delivery.hasPending("parent")).toBe(true);
|
||||
clock.advance(2000);
|
||||
expect(delivery.hasPending("parent")).toBe(true);
|
||||
release();
|
||||
await delivery.close();
|
||||
expect(delivery.hasPending("parent")).toBe(false);
|
||||
});
|
||||
|
||||
@@ -43,6 +43,8 @@ interface Dependencies {
|
||||
}
|
||||
export interface AgentMessageDelivery {
|
||||
enqueue(message: Message): Receipt;
|
||||
/** Queued messages and provider dispatches still owned by this recipient. */
|
||||
hasPending(recipientId: string): boolean;
|
||||
/** Stop admission, flush pending messages, and await provider dispatch acknowledgments. */
|
||||
close(): Promise<void>;
|
||||
/** Existing completion waits include agent messages admitted before the wait. */
|
||||
@@ -138,6 +140,7 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
|
||||
}
|
||||
|
||||
return {
|
||||
hasPending: (recipientId) => pending.has(recipientId) || dispatching.has(recipientId),
|
||||
enqueue(message) {
|
||||
if (closed) throw new Error("Agent message delivery is closed");
|
||||
let resolve!: Admitted["resolve"];
|
||||
|
||||
@@ -110,7 +110,7 @@ export async function startAgentRun(
|
||||
// Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an
|
||||
// in-flight turn — replaceAgentRun would interrupt the running turn. The
|
||||
// intercept lives at this layer so it covers every prompt entrypoint.
|
||||
if (agentManager.tryRunOutOfBand(agentId, prompt, options?.runOptions)) {
|
||||
if (await agentManager.tryRunOutOfBand(agentId, prompt, options?.runOptions)) {
|
||||
options?.onDispatch?.("out_of_band");
|
||||
return { disposition: "out_of_band" };
|
||||
}
|
||||
|
||||
@@ -694,6 +694,9 @@ export interface AgentSession {
|
||||
* still uncertain.
|
||||
*/
|
||||
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. */
|
||||
canStopRuntime?(): Promise<boolean>;
|
||||
/** Release live runtime resources without archiving or deleting the durable native session. */
|
||||
close(): Promise<void>;
|
||||
listCommands?(): Promise<AgentSlashCommand[]>;
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
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;
|
||||
};
|
||||
}
|
||||
@@ -32,7 +32,7 @@ class FakePermissionAgentManager {
|
||||
return this.permissionResult;
|
||||
}
|
||||
|
||||
tryRunOutOfBand(): boolean {
|
||||
async tryRunOutOfBand(): Promise<boolean> {
|
||||
return this.outOfBandHandled;
|
||||
}
|
||||
|
||||
|
||||
@@ -1234,3 +1234,50 @@ test("config-required plugin membership is independent of enablement and other p
|
||||
await plugins.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
test.each([true, false, "error", "wrong-session", "missing"] as const)(
|
||||
"plugin runtime safety query: %s",
|
||||
async (answer) => {
|
||||
const harness = createProviderHarness({
|
||||
capabilities:
|
||||
answer === "missing" ? CAPABILITIES : [...CAPABILITIES, "session.can_stop_runtime"],
|
||||
handleInput: async (input, emit) => {
|
||||
if (input.type !== "session.can_stop_runtime") return false;
|
||||
if (answer === "error")
|
||||
emit({
|
||||
type: "request.failed",
|
||||
requestId: input.requestId,
|
||||
error: { message: "Probe failed" },
|
||||
});
|
||||
else
|
||||
emit({
|
||||
type: "session.can_stop_runtime",
|
||||
requestId: input.requestId,
|
||||
sessionId: answer === "wrong-session" ? "child-1" : input.sessionId,
|
||||
canStop: answer === true,
|
||||
});
|
||||
return true;
|
||||
},
|
||||
});
|
||||
const registry = new PluginAgentClientRegistry(createTestLogger());
|
||||
registry.replace([harness.registration]);
|
||||
const client = registry.clients()[harness.registration.id]!;
|
||||
const session = await client.createSession({ provider: harness.registration.id, cwd: "/tmp" });
|
||||
try {
|
||||
if (answer === "error" || answer === "wrong-session") {
|
||||
await expect(session.canStopRuntime!()).rejects.toThrow(
|
||||
answer === "error" ? "Probe failed" : "another session",
|
||||
);
|
||||
} 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")
|
||||
expect(requests[0]).toMatchObject({
|
||||
sessionId: harness.inputs.find((input) => input.type === "session.open")?.sessionId,
|
||||
});
|
||||
} finally {
|
||||
await session.close();
|
||||
await registry.shutdown();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
@@ -113,9 +113,15 @@ function providerError(error: ProviderError): Error {
|
||||
|
||||
function isProviderRequestReply(
|
||||
event: ProviderEvent,
|
||||
): event is Extract<ProviderEvent, { type: "request.completed" | "catalog" | "sessions" }> {
|
||||
): event is Extract<
|
||||
ProviderEvent,
|
||||
{ type: "request.completed" | "catalog" | "sessions" | "session.can_stop_runtime" }
|
||||
> {
|
||||
return (
|
||||
event.type === "request.completed" || event.type === "catalog" || event.type === "sessions"
|
||||
event.type === "request.completed" ||
|
||||
event.type === "catalog" ||
|
||||
event.type === "sessions" ||
|
||||
event.type === "session.can_stop_runtime"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -564,10 +570,21 @@ class ProviderRuntime {
|
||||
}
|
||||
|
||||
private finishRequest(
|
||||
event: Extract<ProviderEvent, { type: "request.completed" | "catalog" | "sessions" }>,
|
||||
event: Extract<
|
||||
ProviderEvent,
|
||||
{ type: "request.completed" | "catalog" | "sessions" | "session.can_stop_runtime" }
|
||||
>,
|
||||
): void {
|
||||
const request = this.requests.get(event.requestId);
|
||||
if (!request) return;
|
||||
if (
|
||||
event.type === "session.can_stop_runtime" &&
|
||||
this.providerSessions.get(event.sessionId)?.id !== request.sessionId
|
||||
) {
|
||||
this.requests.delete(event.requestId);
|
||||
request.reject(new Error("Runtime safety response belongs to another session"));
|
||||
return;
|
||||
}
|
||||
this.requests.delete(event.requestId);
|
||||
request.resolve(event);
|
||||
}
|
||||
@@ -635,6 +652,27 @@ class ProviderRuntimeSession {
|
||||
}
|
||||
}
|
||||
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
if (
|
||||
this.terminal ||
|
||||
this.prompts.size ||
|
||||
this.activeTurnIds.size ||
|
||||
!this.capabilities.includes("session.can_stop_runtime")
|
||||
)
|
||||
return false;
|
||||
const event = await this.runtime.complete(
|
||||
{
|
||||
type: "session.can_stop_runtime",
|
||||
requestId: randomUUID(),
|
||||
sessionId: this.providerSessionId,
|
||||
},
|
||||
AbortSignal.timeout(10_000),
|
||||
);
|
||||
if (event.type !== "session.can_stop_runtime")
|
||||
throw new Error("Invalid runtime safety response");
|
||||
return event.canStop;
|
||||
}
|
||||
|
||||
async configure(changes: ProviderConfigChanges): Promise<void> {
|
||||
const input: Extract<ProviderInput, { type: "session.configure" }> = {
|
||||
type: "session.configure",
|
||||
@@ -1201,6 +1239,10 @@ class PluginAgentSession implements AgentSession {
|
||||
for (const event of this.history) yield event;
|
||||
}
|
||||
|
||||
canStopRuntime(): Promise<boolean> {
|
||||
return this.bridge.canStopRuntime();
|
||||
}
|
||||
|
||||
async getRuntimeInfo() {
|
||||
return {
|
||||
provider: this.provider,
|
||||
|
||||
@@ -469,6 +469,7 @@ export function wrapSessionProvider(provider: AgentProvider, inner: AgentSession
|
||||
describePersistence: () => mapPersistenceHandle(provider, inner.describePersistence()),
|
||||
interrupt: () => inner.interrupt(),
|
||||
close: () => inner.close(),
|
||||
canStopRuntime: inner.canStopRuntime?.bind(inner),
|
||||
listCommands: inner.listCommands?.bind(inner),
|
||||
setModel: inner.setModel?.bind(inner),
|
||||
setThinkingOption: inner.setThinkingOption?.bind(inner),
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { ClaudeRuntimeSafety } from "./runtime-safety.js";
|
||||
import type { Query, SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
|
||||
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
@@ -706,3 +708,137 @@ describe("background Claude subagents", () => {
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
test("Claude runtime safety protects silent shells, monitors and pending completion notifications", () => {
|
||||
const safety = new ClaudeRuntimeSafety();
|
||||
const observe = (message: unknown) => safety.observe(message as SDKMessage);
|
||||
observe({ type: "system", subtype: "init", claude_code_version: "2.1.292" });
|
||||
expect(safety.canStopRuntime()).toBe(true);
|
||||
observe({
|
||||
type: "system",
|
||||
subtype: "background_tasks_changed",
|
||||
tasks: [
|
||||
{ task_id: "quiet-shell", task_type: "local_bash", description: "sleep 7200" },
|
||||
{ task_id: "monitor", task_type: "local_bash", description: "watch build" },
|
||||
{ task_id: "native-child", task_type: "local_agent", description: "research" },
|
||||
],
|
||||
});
|
||||
expect(safety.canStopRuntime()).toBe(false);
|
||||
observe({ type: "system", subtype: "background_tasks_changed", tasks: [] });
|
||||
expect(safety.canStopRuntime()).toBe(false);
|
||||
for (const task_id of ["quiet-shell", "monitor"])
|
||||
observe({ type: "system", subtype: "task_notification", task_id });
|
||||
expect(safety.canStopRuntime()).toBe(false);
|
||||
observe({ type: "system", subtype: "task_notification", task_id: "native-child" });
|
||||
expect(safety.canStopRuntime()).toBe(true);
|
||||
safety.reset();
|
||||
observe({ type: "system", subtype: "init", claude_code_version: "2.1.220" });
|
||||
expect(safety.canStopRuntime()).toBe(false);
|
||||
observe({ type: "system", subtype: "background_tasks_changed", tasks: [] });
|
||||
expect(safety.canStopRuntime()).toBe(true);
|
||||
});
|
||||
|
||||
test("Claude session runtime safety observes pumped background work and autonomous replies", async () => {
|
||||
let wake: (() => void) | undefined;
|
||||
const channel = { closed: false };
|
||||
const queue: Array<{ message: SDKMessage; consumed(): void }> = [];
|
||||
const pump = (async function* () {
|
||||
while (!channel.closed) {
|
||||
const entry = queue.shift();
|
||||
if (!entry) {
|
||||
await new Promise<void>((resolve) => {
|
||||
wake = resolve;
|
||||
});
|
||||
continue;
|
||||
}
|
||||
yield entry.message;
|
||||
entry.consumed();
|
||||
}
|
||||
})();
|
||||
const push = (message: unknown) =>
|
||||
new Promise<void>((consumed) => {
|
||||
queue.push({ message: message as SDKMessage, consumed });
|
||||
wake?.();
|
||||
});
|
||||
const end = () => {
|
||||
channel.closed = true;
|
||||
wake?.();
|
||||
};
|
||||
let nativeInput!: AsyncIterator<SDKUserMessage>;
|
||||
let connected!: () => void;
|
||||
const connection = new Promise<void>((resolve) => {
|
||||
connected = resolve;
|
||||
});
|
||||
// The SDK Query is injected through the production adapter's existing launch port.
|
||||
const query = {
|
||||
next: () => pump.next(),
|
||||
return: async () => {
|
||||
end();
|
||||
return pump.return(undefined);
|
||||
},
|
||||
close: end,
|
||||
interrupt: async () => {},
|
||||
setPermissionMode: async () => {},
|
||||
setModel: async () => {},
|
||||
supportedModels: async () => [],
|
||||
supportedCommands: async () => [],
|
||||
[Symbol.asyncIterator]() {
|
||||
return this;
|
||||
},
|
||||
} as unknown as Query;
|
||||
const session = await new ClaudeAgentClient({
|
||||
logger: createTestLogger(),
|
||||
resolveBinary: async () => "/test/claude/bin",
|
||||
queryFactory: ({ prompt }) => {
|
||||
if (typeof prompt === "string") throw new Error("Expected streaming input");
|
||||
nativeInput = prompt[Symbol.asyncIterator]();
|
||||
connected();
|
||||
return query;
|
||||
},
|
||||
}).createSession({ provider: "claude", cwd: process.cwd() });
|
||||
const init = {
|
||||
type: "system",
|
||||
subtype: "init",
|
||||
session_id: "runtime-safety",
|
||||
claude_code_version: "2.1.292",
|
||||
permissionMode: "default",
|
||||
};
|
||||
const result = {
|
||||
type: "result",
|
||||
subtype: "success",
|
||||
usage: { input_tokens: 1, output_tokens: 1 },
|
||||
total_cost_usd: 0,
|
||||
};
|
||||
try {
|
||||
await session.startTurn("start background work");
|
||||
await connection;
|
||||
const input = await nativeInput.next();
|
||||
await push({ type: "command_lifecycle", command_uuid: input.value.uuid, state: "started" });
|
||||
await push(init);
|
||||
await push(result);
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
await push({
|
||||
type: "system",
|
||||
subtype: "background_tasks_changed",
|
||||
tasks: [{ task_id: "monitor", task_type: "local_bash", description: "quiet monitor" }],
|
||||
});
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
await push({ type: "system", subtype: "background_tasks_changed", tasks: [] });
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
await push({
|
||||
type: "system",
|
||||
subtype: "task_notification",
|
||||
task_id: "monitor",
|
||||
status: "completed",
|
||||
output: "ready",
|
||||
summary: "ready",
|
||||
});
|
||||
// Completion can wake Claude autonomously after the foreground turn has ended.
|
||||
await push(init);
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
await push(result);
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ClaudeRuntimeSafety } from "./runtime-safety.js";
|
||||
import { validateProviderOptions } from "../../provider-options.js";
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
@@ -2255,6 +2256,8 @@ class ClaudeAgentSession implements AgentSession {
|
||||
private readonly queryFactory?: ClaudeQueryFactory;
|
||||
private readonly resolveBinary: () => Promise<string>;
|
||||
private query: Query | null = null;
|
||||
private readonly runtimeSafety = new ClaudeRuntimeSafety();
|
||||
private processingSdkMessages = 0;
|
||||
private readonly harnessEnvironment: Record<string, string>;
|
||||
private readonly usageSessionKey = randomUUID();
|
||||
private childProcess: ChildProcess | null = null;
|
||||
@@ -2930,6 +2933,18 @@ class ClaudeAgentSession implements AgentSession {
|
||||
return this.persistence;
|
||||
}
|
||||
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
return (
|
||||
!this.activeForegroundTurnId &&
|
||||
!this.autonomousTurn &&
|
||||
!this.mainTurnInFlight &&
|
||||
this.unstartedMessageUuids.size === 0 &&
|
||||
this.pendingPermissions.size === 0 &&
|
||||
this.processingSdkMessages === 0 &&
|
||||
this.runtimeSafety.canStopRuntime()
|
||||
);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
this.logger.trace(
|
||||
{
|
||||
@@ -3423,6 +3438,7 @@ class ClaudeAgentSession implements AgentSession {
|
||||
const options = await this.buildOptions(launchMode);
|
||||
this.logger.debug({ options: summarizeClaudeOptionsForLog(options) }, "claude query");
|
||||
this.input = input;
|
||||
this.runtimeSafety.reset();
|
||||
// A fresh Claude process has no turn of its own in flight.
|
||||
this.mainTurnInFlight = false;
|
||||
this.query = claudeQuery(
|
||||
@@ -4078,13 +4094,17 @@ class ClaudeAgentSession implements AgentSession {
|
||||
);
|
||||
};
|
||||
const handlePumpedMessage = async (message: SDKMessage): Promise<boolean> => {
|
||||
logRawMessage(message);
|
||||
consecutiveInterruptAbortRecoveries = 0;
|
||||
if (await this.handleMissingResumedConversation(message, activeQuery)) {
|
||||
return true;
|
||||
this.processingSdkMessages++;
|
||||
try {
|
||||
this.runtimeSafety.observe(message);
|
||||
logRawMessage(message);
|
||||
consecutiveInterruptAbortRecoveries = 0;
|
||||
if (await this.handleMissingResumedConversation(message, activeQuery)) return true;
|
||||
await this.routeSdkMessageFromPump(message);
|
||||
return false;
|
||||
} finally {
|
||||
this.processingSdkMessages--;
|
||||
}
|
||||
await this.routeSdkMessageFromPump(message);
|
||||
return false;
|
||||
};
|
||||
const drainActiveQuery = async (): Promise<boolean> => {
|
||||
for await (const message of activeQuery) {
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
|
||||
import { parseClaudeCodeVersion } from "./model-manifest.js";
|
||||
|
||||
/** Process-local background ownership. Completion delivery can follow an empty roster. */
|
||||
export class ClaudeRuntimeSafety {
|
||||
private supported = false;
|
||||
private liveTasks = new Set<string>();
|
||||
private readonly awaitingNotifications = new Set<string>();
|
||||
|
||||
reset(): void {
|
||||
this.supported = false;
|
||||
this.liveTasks.clear();
|
||||
this.awaitingNotifications.clear();
|
||||
}
|
||||
|
||||
observe(message: SDKMessage): void {
|
||||
if (message.type !== "system") return;
|
||||
if (message.subtype === "init") {
|
||||
// 2.1.292 is a known roster-capable release. Older binaries can demonstrate
|
||||
// support by sending a roster; a fresh process starts with an empty roster.
|
||||
const version = parseClaudeCodeVersion(message.claude_code_version ?? "");
|
||||
if (
|
||||
version &&
|
||||
(version[0] > 2 ||
|
||||
(version[0] === 2 && (version[1] > 1 || (version[1] === 1 && version[2] >= 292))))
|
||||
)
|
||||
this.supported = true;
|
||||
}
|
||||
if (message.subtype === "background_tasks_changed") {
|
||||
this.supported = true;
|
||||
this.liveTasks = new Set(message.tasks.map((task) => task.task_id));
|
||||
for (const id of this.liveTasks) this.awaitingNotifications.add(id);
|
||||
}
|
||||
if (message.subtype === "task_notification") {
|
||||
this.awaitingNotifications.delete(message.task_id);
|
||||
}
|
||||
}
|
||||
|
||||
canStopRuntime(): boolean {
|
||||
return this.supported && this.liveTasks.size === 0 && this.awaitingNotifications.size === 0;
|
||||
}
|
||||
}
|
||||
@@ -7167,3 +7167,35 @@ describe("Codex interactive resume startup", () => {
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
test("Codex runtime safety queries live terminals and retains unsupported app servers", async () => {
|
||||
let response: unknown = { data: [], nextCursor: null };
|
||||
const appServer = createFakeCodexAppServer({ "thread/backgroundTerminals/list": () => response });
|
||||
const session = new CodexAppServerAgentSession(
|
||||
createConfig(),
|
||||
null,
|
||||
createTestLogger(),
|
||||
async () => appServer.child,
|
||||
);
|
||||
try {
|
||||
await session.startTurn("hello");
|
||||
await appServer.waitForTurnStart();
|
||||
expect(await session.canStopRuntime()).toBe(false);
|
||||
appServer.completeTurn({ threadId: "thread-1" });
|
||||
await expect.poll(() => session.canStopRuntime()).toBe(true);
|
||||
response = { data: [{ processId: "shell-1" }], nextCursor: null };
|
||||
expect(await session.canStopRuntime()).toBe(false);
|
||||
response = {};
|
||||
await expect(session.canStopRuntime()).rejects.toThrow();
|
||||
response = { __jsonRpcError: { code: -32601, message: "Method not found" } };
|
||||
await expect(session.canStopRuntime()).rejects.toThrow("Method not found");
|
||||
expect(
|
||||
appServer
|
||||
.requests()
|
||||
.findLast((request) => request.method === "thread/backgroundTerminals/list"),
|
||||
).toMatchObject({ params: { threadId: "thread-1" } });
|
||||
appServer.assertNoErrors();
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -5029,6 +5029,34 @@ export class CodexAppServerAgentSession implements AgentSession {
|
||||
}
|
||||
}
|
||||
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
if (
|
||||
!this.client ||
|
||||
this.client.hasPendingRequests() ||
|
||||
!this.currentThreadId ||
|
||||
this.activeForegroundTurnId ||
|
||||
this.currentTurnId ||
|
||||
this.pendingForegroundStart ||
|
||||
this.pendingPermissionHandlers.size ||
|
||||
this.getPendingPermissions().length ||
|
||||
this.pendingSubAgentNotificationsByThreadId.size
|
||||
)
|
||||
return false;
|
||||
// Older app servers reject this request. A failed or malformed probe must retain
|
||||
// the process: command completion alone does not mean its terminal has exited.
|
||||
const loaded = z
|
||||
.object({ data: z.array(z.string()) })
|
||||
.parse(await this.client.request("thread/loaded/list", {}, 10_000));
|
||||
const threadIds = new Set([this.currentThreadId, ...loaded.data]);
|
||||
for (const threadId of threadIds) {
|
||||
const response = z
|
||||
.object({ data: z.array(z.unknown()), nextCursor: z.string().nullable() })
|
||||
.parse(await this.client.request("thread/backgroundTerminals/list", { threadId }, 10_000));
|
||||
if (response.data.length > 0 || response.nextCursor !== null) return false;
|
||||
}
|
||||
return !this.client.hasPendingRequests();
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
this.closed = true;
|
||||
this.clearPendingPermissions();
|
||||
|
||||
@@ -202,6 +202,7 @@ function createNewlineLineReader(
|
||||
|
||||
export class CodexAppServerClient {
|
||||
private readonly stdoutLineReader: { close(): void };
|
||||
private activeProviderRequests = 0;
|
||||
private readonly pending = new Map<number, PendingRequest>();
|
||||
private readonly requestHandlers = new Map<string, RequestHandler>();
|
||||
private notificationHandler: NotificationHandler | null = null;
|
||||
@@ -255,6 +256,10 @@ export class CodexAppServerClient {
|
||||
this.requestHandlers.set(method, handler);
|
||||
}
|
||||
|
||||
hasPendingRequests(): boolean {
|
||||
return this.pending.size > 0 || this.activeProviderRequests > 0;
|
||||
}
|
||||
|
||||
request(method: string, params?: unknown, timeoutMs = DEFAULT_TIMEOUT_MS): Promise<unknown> {
|
||||
if (this.disposed) {
|
||||
return Promise.reject(new Error("Codex app-server client is closed"));
|
||||
@@ -392,6 +397,7 @@ export class CodexAppServerClient {
|
||||
this.traceRawEvent(request);
|
||||
const handler = this.requestHandlers.get(request.method);
|
||||
try {
|
||||
this.activeProviderRequests++;
|
||||
const result = handler ? await handler(request.params, request.id) : {};
|
||||
this.writeJsonRpcResponse({ id: request.id, result });
|
||||
} catch (error) {
|
||||
@@ -399,6 +405,8 @@ export class CodexAppServerClient {
|
||||
id: request.id,
|
||||
error: { message: error instanceof Error ? error.message : String(error) },
|
||||
});
|
||||
} finally {
|
||||
this.activeProviderRequests--;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -10,7 +10,7 @@ import type { OmpAgentMessage } from "./rpc-types.js";
|
||||
import type { OmpNoTurnScheduler, OmpProviderIdleScheduler } from "./agent.js";
|
||||
import type { OmpUsagePollScheduler } from "./usage-poller.js";
|
||||
import { resolveOmpProviderOptions } from "./provider-config.js";
|
||||
import { OmpRuntimeEventSchema } from "./rpc-types.js";
|
||||
import { OmpRuntimeEventSchema, OmpSessionStateSchema } from "./rpc-types.js";
|
||||
import { OmpHarness } from "./test-utils/omp-harness.js";
|
||||
import { OmpAgentClient } from "./agent.js";
|
||||
import { FakeOmp } from "./test-utils/fake-omp.js";
|
||||
@@ -1544,3 +1544,86 @@ describe("OMP agent client and session", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
test("omp runtime safety protects queued replies and runtime-owned activity", async () => {
|
||||
const runtime = new FakeOmp();
|
||||
const session = await new OmpAgentClient({ logger: createTestLogger(), runtime }).createSession({
|
||||
provider: "omp",
|
||||
cwd: process.cwd(),
|
||||
});
|
||||
onTestFinished(() => session.close());
|
||||
const native = runtime.latestSession();
|
||||
native.state.isSettled = true;
|
||||
native.state.hasPendingAsyncWork = false;
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
native.state.queuedMessageCount = 1;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.queuedMessageCount = 0;
|
||||
native.state.isStreaming = true;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.isStreaming = false;
|
||||
native.state.isCompacting = true;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.isCompacting = false;
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
native.getStateError = new Error("Unable to prove idle");
|
||||
await expect(session.canStopRuntime!()).rejects.toThrow("Unable to prove idle");
|
||||
});
|
||||
|
||||
test("OMP runtime safety retains async work after the foreground turn finishes", async () => {
|
||||
const runtime = new FakeOmp();
|
||||
const session = await new OmpAgentClient({ logger: createTestLogger(), runtime }).createSession({
|
||||
provider: "omp",
|
||||
cwd: process.cwd(),
|
||||
});
|
||||
onTestFinished(() => session.close());
|
||||
const native = runtime.latestSession();
|
||||
native.state.hasPendingAsyncWork = true;
|
||||
native.state.isSettled = false;
|
||||
const prompted = native.nextPrompt();
|
||||
const run = session.run("start a background job");
|
||||
await prompted;
|
||||
native.beginTurn();
|
||||
native.acceptPrompt("start a background job", "async-user");
|
||||
native.streamAssistantText("Job started in background");
|
||||
native.finishTurn();
|
||||
await run;
|
||||
expect(native.state.isStreaming).toBe(false);
|
||||
expect(native.state.queuedMessageCount).toBe(0);
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
// The delivery is drained, but an admitted/scheduled continuation still prevents settlement.
|
||||
native.state.hasPendingAsyncWork = false;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.isSettled = true;
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
});
|
||||
|
||||
test.each([
|
||||
{ hasPendingAsyncWork: true, isSettled: true, expected: false },
|
||||
{ hasPendingAsyncWork: false, isSettled: false, expected: false },
|
||||
{ hasPendingAsyncWork: false, isSettled: true, expected: true },
|
||||
{ hasPendingAsyncWork: undefined, isSettled: true, expected: false },
|
||||
{ hasPendingAsyncWork: false, isSettled: undefined, expected: false },
|
||||
{ hasPendingAsyncWork: undefined, isSettled: undefined, expected: false },
|
||||
])(
|
||||
"OMP runtime safety requires explicit settlement: %j",
|
||||
async ({ hasPendingAsyncWork, isSettled, expected }) => {
|
||||
const runtime = new FakeOmp();
|
||||
const session = await new OmpAgentClient({ logger: createTestLogger(), runtime }).createSession(
|
||||
{
|
||||
provider: "omp",
|
||||
cwd: process.cwd(),
|
||||
},
|
||||
);
|
||||
onTestFinished(() => session.close());
|
||||
const native = runtime.latestSession();
|
||||
const response = { ...native.state };
|
||||
if (hasPendingAsyncWork !== undefined) response.hasPendingAsyncWork = hasPendingAsyncWork;
|
||||
if (isSettled !== undefined) response.isSettled = isSettled;
|
||||
// Legacy get_state responses stay valid; missing safety fields never imply settlement.
|
||||
native.state = OmpSessionStateSchema.parse(response);
|
||||
expect(native.state.hasPendingAsyncWork).toBe(hasPendingAsyncWork);
|
||||
expect(native.state.isSettled).toBe(isSettled);
|
||||
expect(await session.canStopRuntime!()).toBe(expected);
|
||||
},
|
||||
);
|
||||
|
||||
@@ -1172,6 +1172,27 @@ export class OmpAgentSession implements AgentSession {
|
||||
this.activeToolCalls.clear();
|
||||
}
|
||||
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
if (
|
||||
this.closed ||
|
||||
this.activeTurnId ||
|
||||
this.activeToolCalls.size ||
|
||||
this.pendingExtensionUiRequests.size ||
|
||||
this.pendingClientMessages.length ||
|
||||
this.pendingNoTurnOutputs.length ||
|
||||
this.pendingNoTurnCompletionAbort
|
||||
)
|
||||
return false;
|
||||
const state = await this.runtimeSession.getState();
|
||||
return (
|
||||
state.isSettled === true &&
|
||||
state.hasPendingAsyncWork === false &&
|
||||
!state.isStreaming &&
|
||||
!state.isCompacting &&
|
||||
state.queuedMessageCount === 0
|
||||
);
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) {
|
||||
return;
|
||||
|
||||
@@ -152,6 +152,8 @@ export const OmpSessionStateSchema = z
|
||||
thinkingLevel: OmpThinkingLevelSchema.optional(),
|
||||
isStreaming: z.boolean(),
|
||||
isCompacting: z.boolean(),
|
||||
hasPendingAsyncWork: z.boolean().optional(),
|
||||
isSettled: z.boolean().optional(),
|
||||
autoCompactionEnabled: z.boolean().optional(),
|
||||
fastModeEnabled: z.boolean().optional(),
|
||||
fastModeActive: z.boolean().optional(),
|
||||
|
||||
@@ -7010,3 +7010,21 @@ describe("OpenCode session permission rules", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
test("OpenCode v1 runtime safety retains terminals and busy sessions on its shared server", async () => {
|
||||
const { parent, openCode } = await createParentSession("runtime-safety");
|
||||
try {
|
||||
expect(await parent.canStopRuntime!()).toBe(true);
|
||||
openCode.ptyListResponse = { data: [{ id: "shell", status: "running" }] };
|
||||
expect(await parent.canStopRuntime!()).toBe(false);
|
||||
openCode.ptyListResponse = { data: [] };
|
||||
openCode.sessionStatusResponse = { data: { "other-native-session": { type: "busy" } } };
|
||||
expect(await parent.canStopRuntime!()).toBe(false);
|
||||
openCode.sessionStatusResponse = { data: {} };
|
||||
expect(await parent.canStopRuntime!()).toBe(true);
|
||||
openCode.ptyListResponse = { error: "Inventory unavailable" };
|
||||
await expect(parent.canStopRuntime!()).rejects.toThrow("Unable to inspect OpenCode terminals");
|
||||
} finally {
|
||||
await parent.close();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -5055,6 +5055,43 @@ class OpenCodeAgentSession implements AgentSession {
|
||||
};
|
||||
}
|
||||
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
if (
|
||||
this.closed ||
|
||||
this.serverExited ||
|
||||
this.reconnection ||
|
||||
this.turnState.status !== "idle" ||
|
||||
this.runningToolCalls.size ||
|
||||
this.pendingPermissions.size ||
|
||||
this.pendingSteerSubmissions.length
|
||||
)
|
||||
return false;
|
||||
const request = { signal: AbortSignal.timeout(10_000) };
|
||||
const family = new Map([[this.sessionId, this.config.cwd]]);
|
||||
const inspected = new Set<string>();
|
||||
for (const [sessionId, directory] of family) {
|
||||
if (!inspected.has(directory)) {
|
||||
const statuses = await this.client.session.status({ directory }, request);
|
||||
if (statuses.error || !statuses.data)
|
||||
throw new Error("Unable to inspect OpenCode executions");
|
||||
if (Object.values(statuses.data).some((status) => status.type !== "idle")) return false;
|
||||
const terminals = await this.client.pty.list({ directory }, request);
|
||||
if (terminals.error || !terminals.data)
|
||||
throw new Error("Unable to inspect OpenCode terminals");
|
||||
if (terminals.data.some((terminal) => terminal.status === "running")) return false;
|
||||
inspected.add(directory);
|
||||
}
|
||||
const response = await this.client.session.children(
|
||||
{ sessionID: sessionId, directory },
|
||||
request,
|
||||
);
|
||||
const children = readOpenCodeChildSessionInfosFromResponse(response);
|
||||
if (!children) throw new Error("Unable to inspect OpenCode child sessions");
|
||||
for (const child of children) family.set(child.id, child.directory ?? directory);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
try {
|
||||
this.closed = true;
|
||||
|
||||
+2
@@ -170,6 +170,7 @@ export class TestOpenCodeClient {
|
||||
null;
|
||||
sessionPromptAsyncEvents: unknown[] = [idleEvent()];
|
||||
sessionPromptAsyncResponse: OpenCodeResponse = {};
|
||||
ptyListResponse: OpenCodeResponse = { data: [] };
|
||||
sessionStatusResponse: OpenCodeResponse = { data: {} };
|
||||
sessionStatusImplementation:
|
||||
| ((parameters: unknown, options: unknown) => Promise<OpenCodeResponse>)
|
||||
@@ -195,6 +196,7 @@ export class TestOpenCodeClient {
|
||||
|
||||
asSdkClient(): OpencodeClient {
|
||||
return {
|
||||
pty: { list: async () => this.ptyListResponse },
|
||||
app: {
|
||||
agents: async (parameters: unknown, options: unknown) => {
|
||||
this.calls.appAgents.push(parameters);
|
||||
|
||||
@@ -90,6 +90,9 @@ export class V2Harness {
|
||||
|
||||
readonly api: V2Api = {
|
||||
server: { info: unexpected },
|
||||
shell: {
|
||||
list: async () => ({ location: { directory: this.info.location.directory }, data: [] }),
|
||||
},
|
||||
plugin: {
|
||||
list: async () => ({
|
||||
location: { directory: this.info.location.directory },
|
||||
@@ -131,6 +134,7 @@ export class V2Harness {
|
||||
},
|
||||
permission: { list: async () => [], reply: unexpected },
|
||||
session: {
|
||||
inbox: { list: async () => [] },
|
||||
form: { list: async () => [], reply: unexpected, cancel: unexpected },
|
||||
create: async (input = {}) => {
|
||||
this.creates.push(input);
|
||||
|
||||
@@ -1074,3 +1074,110 @@ describe("OpenCode v2 session lifecycle", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
test("OpenCode v2 runtime safety retains shells, queued reports and shared executions", async () => {
|
||||
const harness = new V2Harness();
|
||||
const client = new OpenCodeV2AgentClient({
|
||||
logger: createTestLogger(),
|
||||
runtime: harness.runtime,
|
||||
});
|
||||
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
|
||||
try {
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
harness.api.shell.list = async () => ({
|
||||
location: { directory: "/tmp/project" },
|
||||
data: [
|
||||
{
|
||||
id: "quiet-shell",
|
||||
status: "running",
|
||||
command: "sleep 7200",
|
||||
cwd: "/tmp/project",
|
||||
shell: "sh",
|
||||
file: "/tmp/out",
|
||||
metadata: {},
|
||||
time: { started: 1 },
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
harness.api.shell.list = async () => ({ location: { directory: "/tmp/project" }, data: [] });
|
||||
harness.api.session.inbox.list = async () => [
|
||||
{
|
||||
id: "report",
|
||||
sessionID: session.id!,
|
||||
type: "synthetic",
|
||||
delivery: "queue",
|
||||
time: { created: 1 },
|
||||
payload: { text: "Background child finished" },
|
||||
},
|
||||
];
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
harness.api.session.inbox.list = async () => [];
|
||||
harness.api.session.active = async () => ({
|
||||
"unrelated-autonomous-session": { type: "running" },
|
||||
});
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
harness.api.session.active = async () => ({});
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
harness.api.shell.list = async () => {
|
||||
throw new Error("No shell inventory");
|
||||
};
|
||||
await expect(session.canStopRuntime!()).rejects.toThrow("No shell inventory");
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("OpenCode v2 runtime safety follows descendant locations before releasing the last lease", async () => {
|
||||
const harness = new V2Harness();
|
||||
const child = {
|
||||
...harness.info,
|
||||
id: "child",
|
||||
parentID: harness.info.id,
|
||||
location: { directory: "/tmp/child-worktree" },
|
||||
};
|
||||
const locations: string[] = [];
|
||||
const client = new OpenCodeV2AgentClient({
|
||||
logger: createTestLogger(),
|
||||
runtime: harness.runtime,
|
||||
});
|
||||
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
|
||||
// A completed native child can still own a quiet shell in another location.
|
||||
harness.api.session.list = async (input) => ({
|
||||
data: input.parentID === session.id ? [child] : [],
|
||||
cursor: {},
|
||||
});
|
||||
let running = true;
|
||||
harness.api.shell.list = async (input) => {
|
||||
const directory = input?.location?.directory ?? "";
|
||||
locations.push(directory);
|
||||
return {
|
||||
location: { directory },
|
||||
data:
|
||||
directory === child.location.directory && running
|
||||
? [
|
||||
{
|
||||
id: "child-shell",
|
||||
status: "running",
|
||||
command: "sleep 7200",
|
||||
cwd: directory,
|
||||
shell: "sh",
|
||||
file: "/tmp/out",
|
||||
metadata: {},
|
||||
time: { started: 1 },
|
||||
},
|
||||
]
|
||||
: [],
|
||||
};
|
||||
};
|
||||
try {
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
expect(locations).toEqual(["/tmp/project", "/tmp/child-worktree"]);
|
||||
running = false;
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
await session.close();
|
||||
expect(harness.releases).toBe(1);
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -2,6 +2,7 @@ import type { OpenCodeClient } from "@opencode/client";
|
||||
|
||||
export interface V2Api {
|
||||
server: OpenCodeClient["server"];
|
||||
shell: Pick<OpenCodeClient["shell"], "list">;
|
||||
session: Pick<
|
||||
OpenCodeClient["session"],
|
||||
| "create"
|
||||
@@ -20,6 +21,7 @@ export interface V2Api {
|
||||
| "revert"
|
||||
| "log"
|
||||
> & {
|
||||
inbox: Pick<OpenCodeClient["session"]["inbox"], "list">;
|
||||
form: Pick<OpenCodeClient["session"]["form"], "list" | "reply" | "cancel">;
|
||||
};
|
||||
plugin: Pick<OpenCodeClient["plugin"], "list">;
|
||||
|
||||
@@ -516,6 +516,36 @@ export class OpenCodeV2Session implements AgentSession {
|
||||
await delay(100, undefined, { signal }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
if (this.closed || this.exited || this.reconnecting || this.turns.id || this.dirty)
|
||||
return false;
|
||||
const request = { signal: AbortSignal.timeout(10_000) };
|
||||
// Active executions are server-wide. Releasing the last lease must not stop an
|
||||
// autonomous execution merely because it belongs to another session or location.
|
||||
if (Object.keys(await this.client.session.active(request)).length > 0) return false;
|
||||
const family = new Map([[this.info.id, this.info]]);
|
||||
const directories = new Set<string>();
|
||||
for (const info of family.values()) {
|
||||
if ((await this.client.session.inbox.list({ sessionID: info.id }, request)).length)
|
||||
return false;
|
||||
directories.add(info.location.directory);
|
||||
let cursor: string | undefined;
|
||||
do {
|
||||
const page = await this.client.session.list(
|
||||
{ ...(cursor ? { cursor } : { parentID: info.id }), limit: 100 },
|
||||
request,
|
||||
);
|
||||
for (const child of page.data) family.set(child.id, child);
|
||||
cursor = page.cursor.next ?? undefined;
|
||||
} while (cursor);
|
||||
}
|
||||
for (const directory of directories) {
|
||||
const shells = await this.client.shell.list({ location: { directory } }, request);
|
||||
if (shells.data.some((shell) => shell.status === "running")) return false;
|
||||
}
|
||||
return !this.turns.id && !this.dirty;
|
||||
}
|
||||
|
||||
async close() {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
|
||||
@@ -3488,3 +3488,23 @@ describe("transformPiModels", () => {
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
test("pi runtime safety protects queued replies and runtime-owned activity", async () => {
|
||||
const runtime = new FakePi();
|
||||
const session = await createClient(runtime).createSession(createConfig());
|
||||
onTestFinished(() => session.close());
|
||||
const native = runtime.latestSession();
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
native.state.pendingMessageCount = 1;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.pendingMessageCount = 0;
|
||||
native.state.isStreaming = true;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.isStreaming = false;
|
||||
native.state.isCompacting = true;
|
||||
expect(await session.canStopRuntime!()).toBe(false);
|
||||
native.state.isCompacting = false;
|
||||
expect(await session.canStopRuntime!()).toBe(true);
|
||||
native.getStateError = new Error("Unable to prove idle");
|
||||
await expect(session.canStopRuntime!()).rejects.toThrow("Unable to prove idle");
|
||||
});
|
||||
|
||||
@@ -1581,6 +1581,22 @@ export class PiRpcAgentSession implements AgentSession {
|
||||
return await resultPromise;
|
||||
}
|
||||
|
||||
async canStopRuntime(): Promise<boolean> {
|
||||
if (
|
||||
this.closed ||
|
||||
this.activeTurnId ||
|
||||
this.pendingExtensionUiRequests.size ||
|
||||
this.pendingExtensionResults.size ||
|
||||
this.pendingExtensionHydrations.size ||
|
||||
this.pendingSteerSubmissions.length ||
|
||||
this.pendingNoTurnOutputs.length ||
|
||||
this.interruptingTurn
|
||||
)
|
||||
return false;
|
||||
const state = await this.runtimeSession.getState();
|
||||
return !state.isStreaming && !state.isCompacting && state.pendingMessageCount === 0;
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) {
|
||||
return;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { startIdleRuntimeCleanup } from "./agent/idle-runtime-cleanup.js";
|
||||
import {
|
||||
createFinishSubscriptions,
|
||||
createFinishObservation,
|
||||
@@ -453,6 +454,7 @@ export interface PaseoDaemonConfig {
|
||||
dictationFinalTimeoutMs?: number;
|
||||
downloadTokenTtlMs?: number;
|
||||
agentProviderSettings?: AgentProviderRuntimeSettingsMap;
|
||||
idleRuntimeCleanup?: boolean;
|
||||
providerCatalogRefreshTimeoutMs?: number;
|
||||
metadataGeneration?: {
|
||||
providers?: Array<{
|
||||
@@ -1398,6 +1400,15 @@ 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 {
|
||||
await scheduleService.completeForAgent(agentId);
|
||||
@@ -1907,6 +1918,7 @@ export async function createPaseoDaemon(
|
||||
await hubRelationships.stop();
|
||||
workspaceReconciliation.dispose();
|
||||
scriptHealthMonitor.stop();
|
||||
await stopIdleRuntimeCleanup();
|
||||
// Freeze both ingress and registration before taking the agent closure snapshot.
|
||||
wsServer?.prepareForShutdown();
|
||||
await finishSubscriptions.close();
|
||||
|
||||
@@ -27,6 +27,17 @@ describe("server config", () => {
|
||||
expect(standaloneConfig.desktopManaged).toBe(false);
|
||||
});
|
||||
|
||||
test("idle runtime cleanup defaults on and supports a host-wide opt-out", async () => {
|
||||
const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-config-idle-cleanup-"));
|
||||
roots.push(paseoHome);
|
||||
expect(loadConfig(paseoHome, { env: {} }).idleRuntimeCleanup).toBe(true);
|
||||
await writeFile(
|
||||
path.join(paseoHome, "config.json"),
|
||||
JSON.stringify({ agents: { idleRuntimeCleanup: false } }),
|
||||
);
|
||||
expect(loadConfig(paseoHome, { env: {} }).idleRuntimeCleanup).toBe(false);
|
||||
});
|
||||
|
||||
test("loads the provider catalog refresh timeout", async () => {
|
||||
const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-config-provider-timeout-"));
|
||||
roots.push(paseoHome);
|
||||
|
||||
@@ -660,6 +660,7 @@ export function resolveConfigFromPersisted(
|
||||
voiceLlmProviderExplicit: voiceLlm.providerExplicit,
|
||||
voiceLlmModel: voiceLlm.model,
|
||||
agentProviderSettings: extractAgentProviderSettings(providerOverrides),
|
||||
idleRuntimeCleanup: persisted.agents?.idleRuntimeCleanup ?? true,
|
||||
providerCatalogRefreshTimeoutMs: persisted.agents?.catalogRefreshTimeoutMs,
|
||||
metadataGeneration: persisted.agents?.metadataGeneration,
|
||||
providerOverrides,
|
||||
|
||||
@@ -339,6 +339,7 @@ export const PersistedConfigSchema = z
|
||||
agents: z
|
||||
.object({
|
||||
providers: z.preprocess(normalizeAgentProviders, ProviderOverridesSchema).optional(),
|
||||
idleRuntimeCleanup: z.boolean().optional(),
|
||||
catalogRefreshTimeoutMs: z.number().int().positive().max(2_147_483_647).optional(),
|
||||
metadataGeneration: AgentMetadataGenerationSchema.optional(),
|
||||
skills: z.object({ selection: AgentSkillSelectionSchema.optional() }).strict().optional(),
|
||||
|
||||
@@ -552,6 +552,9 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"idleRuntimeCleanup": {
|
||||
"type": "boolean"
|
||||
},
|
||||
"catalogRefreshTimeoutMs": {
|
||||
"type": "integer",
|
||||
"exclusiveMinimum": 0,
|
||||
|
||||
@@ -74,7 +74,7 @@ Runtime-safe settings include relay enablement, MCP settings, browser tools, hos
|
||||
|
||||
New homes keep relay disabled when you remove `daemon.relay.enabled`. A daemon whose config already omitted this field when it started keeps the legacy relay-enabled behavior for compatibility. Set `daemon.relay.enabled` explicitly when editing an older config.
|
||||
|
||||
Listen addresses, authentication, relay endpoints and TLS, worktree allocation, service-proxy addresses, the bundled web UI, logging, speech, voice, credentials, and local model settings require a restart. Reload applies other valid edits in the same file before reporting those paths.
|
||||
Idle runtime cleanup, listen addresses, authentication, relay endpoints and TLS, worktree allocation, service-proxy addresses, the bundled web UI, logging, speech, voice, credentials, and local model settings require a restart. Reload applies other valid edits in the same file before reporting those paths.
|
||||
|
||||
Deployment environment variables and legacy supervisor flags remain authoritative. Reload reports a changed file setting under `overrideControlledPaths` when a launch override prevents it from taking effect. This includes startup settings such as listen addresses, passwords, relay endpoints and TLS, service-proxy and web UI settings, logging, speech, and voice configuration. List settings such as hostnames and CORS origins still append across sources, so values from `config.json` continue to apply. Stop and relaunch the deployment without the override to make the file value authoritative.
|
||||
|
||||
@@ -84,6 +84,19 @@ Agent providers, both the first-class ones Paseo ships with and custom entries y
|
||||
|
||||
See [Providers](/docs/providers) for the mental model and [Supported providers](/docs/supported-providers) for the full list of agents Paseo can launch. For pointing Claude at Anthropic-compatible endpoints (Z.AI, Alibaba/Qwen), multiple profiles, custom binaries, ACP agents, and the `additionalModels` merge behavior, see [Custom providers](/docs/custom-providers). The full field reference lives on GitHub at [docs/custom-providers.md](https://github.com/getpaseo/paseo/blob/main/docs/custom-providers.md).
|
||||
|
||||
## 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
|
||||
for the entire host with:
|
||||
|
||||
```json
|
||||
{ "agents": { "idleRuntimeCleanup": false } }
|
||||
```
|
||||
|
||||
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.
|
||||
|
||||
## Worktrees
|
||||
|
||||
New worktrees are created under `$PASEO_HOME/worktrees` by default. To place new worktrees somewhere else, set `worktrees.root`:
|
||||
|
||||
@@ -306,6 +306,49 @@ turn or finish as a side effect with a `completed` prompt result.
|
||||
Advertise only capabilities you implement. If steering is unsupported, omit `prompt.steer`; Paseo
|
||||
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.
|
||||
|
||||
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:
|
||||
|
||||
```ts
|
||||
case "session.can_stop_runtime": {
|
||||
try {
|
||||
const session = sessions.get(input.sessionId);
|
||||
if (!session) throw new Error("Session not found");
|
||||
emit({
|
||||
type: "session.can_stop_runtime",
|
||||
requestId: input.requestId,
|
||||
sessionId: input.sessionId,
|
||||
canStop: await session.canStopRuntime(),
|
||||
});
|
||||
} catch (error) {
|
||||
emit({
|
||||
type: "request.failed",
|
||||
requestId: input.requestId,
|
||||
error: { message: error instanceof Error ? error.message : String(error) },
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
```
|
||||
|
||||
`canStopRuntime(): Promise<boolean>` belongs to your provider's session implementation. Return
|
||||
`true` only when releasing the runtime loses no background shells, monitors, native children,
|
||||
queued completion reports, or pending autonomous replies. This query must not stop or mutate the
|
||||
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.
|
||||
The host setting `agents.idleRuntimeCleanup: false` disables collection for every provider.
|
||||
|
||||
## Publish timeline items
|
||||
|
||||
Publish complete snapshots through `timeline.item`. Reuse an item `id` when updating streamed text,
|
||||
|
||||
Reference in New Issue
Block a user