From 8e5f5367666951a7f5933cbb9f66c8047ba8a399 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 26 Oct 2026 12:38:15 +0700 Subject: [PATCH] Release idle agent runtimes with work guards and live host controls (#7782) * 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. * docs: translate idle runtime cleanup configuration and provider contract * 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. * feat(agents): control idle runtime release live from host settings Let hosts retain runtimes without restarting the daemon. Disabling invalidates pending collection decisions while preserving idle clocks for re-enabling. ACP-owned terminals and pending adapter work veto collection until they finish. * fix(settings): clarify idle agent release with an info tooltip --- docs/agent-lifecycle.md | 50 ++- docs/plugins.md | 5 + .../e2e/browser/settings-host-page.spec.ts | 100 ++++++ packages/app/e2e/support/helpers/settings.ts | 19 ++ packages/app/src/i18n/resources/ar.ts | 5 + packages/app/src/i18n/resources/en.ts | 5 + packages/app/src/i18n/resources/es.ts | 5 + packages/app/src/i18n/resources/fr.ts | 5 + packages/app/src/i18n/resources/ja.ts | 5 + packages/app/src/i18n/resources/ko.ts | 5 + packages/app/src/i18n/resources/pt-BR.ts | 5 + packages/app/src/i18n/resources/ru.ts | 5 + packages/app/src/i18n/resources/zh-CN.ts | 5 + .../app/src/screens/settings/host-page.tsx | 2 + .../settings/idle-runtime-cleanup-card.tsx | 78 +++++ packages/plugin/src/server/provider.ts | 20 ++ packages/protocol/src/messages.ts | 6 + .../src/server/agent/agent-manager.test.ts | 219 ++++++++++++- .../server/src/server/agent/agent-manager.ts | 101 +++++- .../agent/agent-messages/delivery.test.ts | 30 ++ .../server/agent/agent-messages/delivery.ts | 3 + .../server/src/server/agent/agent-prompt.ts | 2 +- .../src/server/agent/agent-sdk-types.ts | 4 + .../finish-subscriptions.test.ts | 85 +++++ .../agent/finish-subscriptions/index.ts | 26 +- .../idle-runtime-cleanup.test.ts | 294 ++++++++++++++++++ .../agent/idle-runtime-cleanup/index.ts | 167 ++++++++++ .../server/agent/permission-response.test.ts | 2 +- .../src/server/agent/plugin-provider.test.ts | 87 ++++++ .../src/server/agent/plugin-provider.ts | 49 ++- .../src/server/agent/provider-registry.ts | 1 + .../server/agent/providers/acp-agent.test.ts | 156 ++++++++++ .../src/server/agent/providers/acp-agent.ts | 32 +- .../claude/agent.background-subagent.test.ts | 136 ++++++++ .../server/agent/providers/claude/agent.ts | 32 +- .../agent/providers/claude/runtime-safety.ts | 42 +++ .../providers/codex-app-server-agent.test.ts | 32 ++ .../agent/providers/codex-app-server-agent.ts | 28 ++ .../providers/codex/app-server-transport.ts | 8 + .../server/agent/providers/omp/agent.test.ts | 85 ++++- .../src/server/agent/providers/omp/agent.ts | 21 ++ .../server/agent/providers/omp/rpc-types.ts | 2 + .../agent/providers/opencode-agent.test.ts | 18 ++ .../server/agent/providers/opencode-agent.ts | 37 +++ .../test-utils/test-opencode-harness.ts | 2 + .../opencode/test-utils/v2-harness.ts | 4 + .../agent/providers/opencode/v2/agent.test.ts | 107 +++++++ .../server/agent/providers/opencode/v2/api.ts | 2 + .../agent/providers/opencode/v2/session.ts | 30 ++ .../server/agent/providers/pi/agent.test.ts | 20 ++ .../src/server/agent/providers/pi/agent.ts | 16 + packages/server/src/server/bootstrap.test.ts | 144 ++++++++- packages/server/src/server/bootstrap.ts | 28 ++ packages/server/src/server/config.test.ts | 11 + packages/server/src/server/config.ts | 1 + .../src/server/daemon-config-store.test.ts | 62 ++++ .../server/src/server/daemon-config-store.ts | 11 + .../server/src/server/persisted-config.ts | 1 + .../server/src/server/websocket-server.ts | 2 + .../public/schemas/paseo.config.v1.json | 3 + public-docs/configuration.md | 18 +- public-docs/configuration.zh-CN.md | 12 +- public-docs/plugins/providers.md | 46 +++ public-docs/plugins/providers.zh-CN.md | 32 ++ 64 files changed, 2540 insertions(+), 36 deletions(-) create mode 100644 packages/app/src/screens/settings/idle-runtime-cleanup-card.tsx create mode 100644 packages/server/src/server/agent/idle-runtime-cleanup/idle-runtime-cleanup.test.ts create mode 100644 packages/server/src/server/agent/idle-runtime-cleanup/index.ts create mode 100644 packages/server/src/server/agent/providers/claude/runtime-safety.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index e9ca2211c2..719b0b5468 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -25,8 +25,54 @@ 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 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. The host setting applies immediately; config-file edits +apply on reload. Disabling invalidates pending safety checks and close admission. A close already +started finishes. Re-enabling keeps the existing idle timestamps. + +Foreground runs, permissions and responses, queued deliveries, lifecycle operations, and working +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 +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 checks adapter-owned foreground turns, permissions, buffered user messages, and running + ACP terminals, including quiet terminals. Exited or failed-to-spawn terminal records do not + prevent collection. This does not inspect processes launched internally by the provider. + +These checks do not discover arbitrary detached processes or work hidden by third-party +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 diff --git a/docs/plugins.md b/docs/plugins.md index 2a223bfe85..21a924c4e8 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -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 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. + 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 diff --git a/packages/app/e2e/browser/settings-host-page.spec.ts b/packages/app/e2e/browser/settings-host-page.spec.ts index b531232615..76e5b331e6 100644 --- a/packages/app/e2e/browser/settings-host-page.spec.ts +++ b/packages/app/e2e/browser/settings-host-page.spec.ts @@ -1,3 +1,7 @@ +import { readFile, mkdir, rename, rm } from "node:fs/promises"; +import path from "node:path"; +import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { connectDaemonClient } from "../support/helpers/daemon-client-loader"; import { expect, test } from "../support/fixtures"; import { gotoAppShell, openSettings } from "../support/helpers/app"; import { getE2EDaemonPort } from "../support/helpers/daemon-port"; @@ -25,6 +29,8 @@ import { removeHostFromHostPage, addDirectHostFromSettings, expectHostOnlineWithoutError, + expectIdleAgentRelease, + setIdleAgentRelease, } from "../support/helpers/settings"; test.describe("Settings host page", () => { @@ -137,6 +143,100 @@ test.describe("Settings host page", () => { } }); + test("idle agent release changes live on one host and stays saved without restarting", async ({ + page, + }) => { + test.setTimeout(120_000); + const primaryServerId = getServerId(); + const daemon = await startIsolatedHostDaemon("srv_e2e_idle_agent_release"); + const client = await connectDaemonClient({ + clientIdPrefix: "idle-release-settings", + port: daemon.port, + }); + const originalPid = (await client.getDaemonStatus()).pid; + const expectLiveSavedSetting = async (enabled: boolean) => { + const saved = JSON.parse(await readFile(path.join(daemon.paseoHome, "config.json"), "utf8")); + expect(saved.agents.idleRuntimeCleanup).toBe(enabled); + expect((await client.getDaemonConfig()).config.idleRuntimeCleanup).toBe(enabled); + expect((await client.reloadDaemonConfig()).restartRequiredPaths).toEqual([]); + expect((await client.getDaemonStatus()).pid).toBe(originalPid); + }; + const hosts = [ + { + serverId: primaryServerId, + label: TEST_HOST_LABEL, + endpoint: `127.0.0.1:${getE2EDaemonPort()}`, + }, + { + serverId: daemon.serverId, + label: "Idle release host", + endpoint: `127.0.0.1:${daemon.port}`, + }, + ]; + const openHostAgentsInFreshApp = async (serverId: string) => { + await seedSavedSettingsHosts(page, hosts); + await page.reload(); + await openSettings(page); + await selectSettingsHost(page, serverId); + await openHostSection(page, serverId, "agents"); + }; + try { + await test.step("idle agent release starts on and turns off", async () => { + await openHostAgentsInFreshApp(daemon.serverId); + await expectIdleAgentRelease(page, true); + await setIdleAgentRelease(page, false); + await expectLiveSavedSetting(false); + }); + + await test.step("a fresh app reads the saved setting from the same running host", async () => { + await openHostAgentsInFreshApp(daemon.serverId); + await expectIdleAgentRelease(page, false); + await expectLiveSavedSetting(false); + await setIdleAgentRelease(page, true); + await expectLiveSavedSetting(true); + await setIdleAgentRelease(page, false); + }); + + await test.step("other hosts keep their own setting", async () => { + await selectSettingsHost(page, primaryServerId); + await openHostSection(page, primaryServerId, "agents"); + await expectIdleAgentRelease(page, true); + await expectLiveSavedSetting(false); + }); + } finally { + await client.close(); + await daemon.close(); + } + }); + + test("a failed idle release save stays visible and retry applies it live", async ({ page }) => { + const daemon = await startIsolatedHostDaemon("srv_e2e_idle_release_failure"); + const configPath = path.join(daemon.paseoHome, "config.json"); + try { + await seedSavedSettingsHosts(page, [ + { serverId: daemon.serverId, label: "Failure host", endpoint: `127.0.0.1:${daemon.port}` }, + ]); + await page.reload(); + await openSettings(page); + await openHostSection(page, daemon.serverId, "agents"); + await expectIdleAgentRelease(page, true); + // A real filesystem failure, with the daemon and RPC path intact. + await rename(configPath, `${configPath}.saved`); + await mkdir(configPath); + await page.getByRole("switch", { name: "Release idle agents" }).click(); + await expect(page.getByTestId("host-page-idle-runtime-cleanup-error")).toContainText( + "Failed to read", + ); + await expectIdleAgentRelease(page, true); + await rm(configPath, { recursive: true }); + await rename(`${configPath}.saved`, configPath); + await setIdleAgentRelease(page, false); + await expect(page.getByTestId("host-page-idle-runtime-cleanup-error")).toHaveCount(0); + } finally { + await daemon.close(); + } + }); + test("navigating to /settings/hosts/[serverId] redirects to the connections section", async ({ page, }) => { diff --git a/packages/app/e2e/support/helpers/settings.ts b/packages/app/e2e/support/helpers/settings.ts index 952cc5b69d..e68f554f34 100644 --- a/packages/app/e2e/support/helpers/settings.ts +++ b/packages/app/e2e/support/helpers/settings.ts @@ -326,6 +326,25 @@ export async function expectHostInjectMcpCard(page: Page): Promise { await expect(card.getByRole("switch", { name: "Inject Paseo tools" })).toBeVisible(); } +function idleAgentReleaseSwitch(page: Page) { + return page + .getByTestId("host-page-idle-runtime-cleanup-card") + .getByRole("switch", { name: "Release idle agents" }); +} + +export async function expectIdleAgentRelease(page: Page, enabled: boolean): Promise { + const toggle = idleAgentReleaseSwitch(page); + await expect(toggle).toBeEnabled(); + await expect(toggle).toHaveAttribute("aria-checked", String(enabled)); +} + +export async function setIdleAgentRelease(page: Page, enabled: boolean): Promise { + const toggle = idleAgentReleaseSwitch(page); + await expect(toggle).toHaveAttribute("aria-checked", String(!enabled)); + await toggle.click(); + await expectIdleAgentRelease(page, enabled); +} + export async function openHostSection( page: Page, serverId: string, diff --git a/packages/app/src/i18n/resources/ar.ts b/packages/app/src/i18n/resources/ar.ts index 4fc0ff5277..a5743acad3 100644 --- a/packages/app/src/i18n/resources/ar.ts +++ b/packages/app/src/i18n/resources/ar.ts @@ -2698,6 +2698,11 @@ export const ar: TranslationResources = { }, agents: { unavailable: "Connect to this host to manage agents", + idleRuntimeCleanup: { + title: "تحرير الوكلاء الخاملين", + hint: "تحرير الذاكرة بعد ساعة من الخمول.", + info: "يوقف عملية الوكيل مع الاحتفاظ بمحادثته. أرسل رسالة لاستئنافها.\n\nيمنع العمل النشط والإشعارات المعلّقة وفحوصات أمان المزوّد عملية التنظيف.", + }, }, workspaces: { unavailable: "Connect to this host to manage workspaces", diff --git a/packages/app/src/i18n/resources/en.ts b/packages/app/src/i18n/resources/en.ts index 2e40707121..10564eab19 100644 --- a/packages/app/src/i18n/resources/en.ts +++ b/packages/app/src/i18n/resources/en.ts @@ -2789,6 +2789,11 @@ export const en = { }, agents: { unavailable: "Connect to this host to manage agents", + idleRuntimeCleanup: { + title: "Release idle agents", + hint: "Free up memory after an hour of inactivity.", + info: "Stops the agent's process while keeping its conversation. Send a message to resume.\n\nActive work, pending notifications, and provider safety checks prevent cleanup.", + }, }, workspaces: { unavailable: "Connect to this host to manage workspaces", diff --git a/packages/app/src/i18n/resources/es.ts b/packages/app/src/i18n/resources/es.ts index b62913d268..7f0e2b6529 100644 --- a/packages/app/src/i18n/resources/es.ts +++ b/packages/app/src/i18n/resources/es.ts @@ -2718,6 +2718,11 @@ export const es: TranslationResources = { }, agents: { unavailable: "Connect to this host to manage agents", + idleRuntimeCleanup: { + title: "Liberar agentes inactivos", + hint: "Libera memoria tras una hora de inactividad.", + info: "Detiene el proceso del agente y conserva la conversación. Envía un mensaje para reanudarla.\n\nEl trabajo activo, las notificaciones pendientes y las comprobaciones de seguridad del proveedor impiden la limpieza.", + }, }, workspaces: { unavailable: "Connect to this host to manage workspaces", diff --git a/packages/app/src/i18n/resources/fr.ts b/packages/app/src/i18n/resources/fr.ts index cc3742abad..8fbf03bb1f 100644 --- a/packages/app/src/i18n/resources/fr.ts +++ b/packages/app/src/i18n/resources/fr.ts @@ -2721,6 +2721,11 @@ export const fr: TranslationResources = { }, agents: { unavailable: "Connectez-vous à cet hôte pour gérer les agents", + idleRuntimeCleanup: { + title: "Libérer les agents inactifs", + hint: "Libère de la mémoire après une heure d’inactivité.", + info: "Arrête le processus de l’agent tout en conservant sa conversation. Envoyez un message pour la reprendre.\n\nLe travail en cours, les notifications en attente et les vérifications de sécurité du fournisseur empêchent le nettoyage.", + }, }, workspaces: { unavailable: "Connectez-vous à cet hôte pour gérer les espaces de travail", diff --git a/packages/app/src/i18n/resources/ja.ts b/packages/app/src/i18n/resources/ja.ts index 67d46fe471..b6f218adef 100644 --- a/packages/app/src/i18n/resources/ja.ts +++ b/packages/app/src/i18n/resources/ja.ts @@ -2687,6 +2687,11 @@ export const ja: TranslationResources = { }, agents: { unavailable: "エージェントを管理するにはこのホストに接続してください", + idleRuntimeCleanup: { + title: "アイドル状態のエージェントを解放", + hint: "1 時間のアイドル状態の後にメモリを解放します。", + info: "会話を保持したままエージェントのプロセスを停止します。メッセージを送信すると再開します。\n\n実行中の作業、保留中の通知、プロバイダーの安全性チェックがある場合は解放しません。", + }, }, workspaces: { unavailable: "ワークスペースを管理するにはこのホストに接続してください", diff --git a/packages/app/src/i18n/resources/ko.ts b/packages/app/src/i18n/resources/ko.ts index d95899ec35..7bd4bc1b63 100644 --- a/packages/app/src/i18n/resources/ko.ts +++ b/packages/app/src/i18n/resources/ko.ts @@ -2673,6 +2673,11 @@ export const ko: TranslationResources = { }, agents: { unavailable: "에이전트를 관리하려면 이 호스트에 연결하세요", + idleRuntimeCleanup: { + title: "유휴 에이전트 해제", + hint: "1시간 동안 활동이 없으면 메모리를 확보합니다.", + info: "대화를 유지하면서 에이전트 프로세스를 중지합니다. 메시지를 보내면 다시 이어집니다.\n\n진행 중인 작업, 대기 중인 알림, 프로바이더 안전 검사가 있으면 정리하지 않습니다.", + }, }, workspaces: { unavailable: "워크스페이스를 관리하려면 이 호스트에 연결하세요", diff --git a/packages/app/src/i18n/resources/pt-BR.ts b/packages/app/src/i18n/resources/pt-BR.ts index fd41468395..21293e9c44 100644 --- a/packages/app/src/i18n/resources/pt-BR.ts +++ b/packages/app/src/i18n/resources/pt-BR.ts @@ -2700,6 +2700,11 @@ export const ptBR: TranslationResources = { }, agents: { unavailable: "Conecte-se a este host para gerenciar agentes", + idleRuntimeCleanup: { + title: "Liberar agentes ociosos", + hint: "Libera memória após uma hora de inatividade.", + info: "Encerra o processo do agente e mantém a conversa. Envie uma mensagem para retomar.\n\nTrabalho ativo, notificações pendentes e verificações de segurança do provedor impedem a limpeza.", + }, }, workspaces: { unavailable: "Conecte-se a este host para gerenciar workspaces", diff --git a/packages/app/src/i18n/resources/ru.ts b/packages/app/src/i18n/resources/ru.ts index 093fc60a51..8f51e5567e 100644 --- a/packages/app/src/i18n/resources/ru.ts +++ b/packages/app/src/i18n/resources/ru.ts @@ -2709,6 +2709,11 @@ export const ru: TranslationResources = { }, agents: { unavailable: "Подключитесь к этому хосту, чтобы управлять агентами", + idleRuntimeCleanup: { + title: "Освобождать неактивных агентов", + hint: "Освобождает память после часа бездействия.", + info: "Останавливает процесс агента, сохраняя разговор. Отправьте сообщение, чтобы продолжить.\n\nАктивная работа, ожидающие уведомления и проверки безопасности провайдера предотвращают очистку.", + }, }, workspaces: { unavailable: "Подключитесь к этому хосту, чтобы управлять рабочими пространствами", diff --git a/packages/app/src/i18n/resources/zh-CN.ts b/packages/app/src/i18n/resources/zh-CN.ts index d899a10fe8..f3626c933a 100644 --- a/packages/app/src/i18n/resources/zh-CN.ts +++ b/packages/app/src/i18n/resources/zh-CN.ts @@ -2634,6 +2634,11 @@ export const zhCN: TranslationResources = { }, agents: { unavailable: "连接到这个 Host 以管理 Agent", + idleRuntimeCleanup: { + title: "释放空闲 Agent", + hint: "空闲一小时后释放内存。", + info: "停止 Agent 进程，但保留对话。发送消息即可继续。\n\n进行中的工作、待处理的通知和 Provider 安全检查会阻止清理。", + }, }, workspaces: { unavailable: "连接到这个 Host 以管理 Workspace", diff --git a/packages/app/src/screens/settings/host-page.tsx b/packages/app/src/screens/settings/host-page.tsx index c70c4941cb..0fb6fb5aa9 100644 --- a/packages/app/src/screens/settings/host-page.tsx +++ b/packages/app/src/screens/settings/host-page.tsx @@ -65,6 +65,7 @@ import { ICON_SIZE } from "@/styles/theme"; import type { Theme } from "@/styles/theme"; import { useProviderIcon } from "@/components/provider-icons"; import { BrowserToolsOptInCard } from "./browser-tools-card"; +import { IdleRuntimeCleanupCard } from "./idle-runtime-cleanup-card"; import { restartDaemonFromSettings, updateDaemonFromSettings } from "./daemon-lifecycle"; const ThemedRestart = withUnistyles(RotateCw); @@ -291,6 +292,7 @@ export function HostAgentsPage({ serverId }: { serverId: string }) { + ) : ( diff --git a/packages/app/src/screens/settings/idle-runtime-cleanup-card.tsx b/packages/app/src/screens/settings/idle-runtime-cleanup-card.tsx new file mode 100644 index 0000000000..4d6fc355e8 --- /dev/null +++ b/packages/app/src/screens/settings/idle-runtime-cleanup-card.tsx @@ -0,0 +1,78 @@ +import { useCallback } from "react"; +import { Text, View } from "react-native"; +import { useMutation } from "@tanstack/react-query"; +import { useTranslation } from "react-i18next"; +import { StyleSheet } from "react-native-unistyles"; +import { SettingsInfoTip } from "@/components/settings/headings/settings-info-tip"; +import { Switch } from "@/components/ui/switch"; +import { useDaemonConfig } from "@/hooks/use-daemon-config"; +import { useHostFeature } from "@/runtime/host-features"; +import { useHostRuntimeIsConnected } from "@/runtime/host-runtime"; +import { settingsStyles } from "@/styles/settings"; + +// Host-wide, persisted, and applied live by the daemon. +export function IdleRuntimeCleanupCard({ serverId }: { serverId: string }) { + const { t } = useTranslation(); + const isConnected = useHostRuntimeIsConnected(serverId); + const isSupported = useHostFeature(serverId, "idleRuntimeCleanupSetting"); + const { config, patchConfig } = useDaemonConfig(serverId); + const mutation = useMutation({ + mutationFn: async (next: boolean) => { + const result = await patchConfig({ idleRuntimeCleanup: next }); + if (!result) { + throw new Error(t("workspace.terminal.hostDisconnected")); + } + return result; + }, + }); + + const handleValueChange = useCallback( + (next: boolean) => { + mutation.mutate(next); + }, + [mutation], + ); + + if (!isConnected || !isSupported || !config) return null; + + return ( + + + + + + {t("settings.host.agents.idleRuntimeCleanup.title")} + + + + + {t("settings.host.agents.idleRuntimeCleanup.hint")} + + {mutation.error ? ( + + {mutation.error.message} + + ) : null} + + + + + ); +} + +const styles = StyleSheet.create((theme) => ({ + titleRow: { + flexDirection: "row", + alignItems: "center", + gap: theme.spacing[2], + }, +})); diff --git a/packages/plugin/src/server/provider.ts b/packages/plugin/src/server/provider.ts index 1646e8473b..ee9fbc21c2 100644 --- a/packages/plugin/src/server/provider.ts +++ b/packages/plugin/src/server/provider.ts @@ -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 = 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 = 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"), diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 4beaf0a0b2..bc158a6322 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -224,6 +224,9 @@ export const MutableDaemonConfigSchema = z metadataGeneration: MutableMetadataGenerationConfigSchema.default({ providers: [] }), autoArchiveAfterMerge: z.boolean().default(false), enableTerminalAgentHooks: z.boolean().default(false), + // COMPAT(idleRuntimeCleanupSetting): added in v0.13.0, keep optional while old daemons are supported. + // Persisted host-wide value, applied live by the daemon. + idleRuntimeCleanup: z.boolean().optional(), appendSystemPrompt: z.string().default(""), terminalProfiles: z.array(TerminalProfileSchema).optional(), agentProfiles: z.array(AgentProfileSchema).optional(), @@ -245,6 +248,7 @@ export const MutableDaemonConfigPatchSchema = z metadataGeneration: MutableMetadataGenerationConfigSchema.partial().optional(), autoArchiveAfterMerge: z.boolean().optional(), enableTerminalAgentHooks: z.boolean().optional(), + idleRuntimeCleanup: z.boolean().optional(), appendSystemPrompt: z.string().optional(), terminalProfiles: z.array(TerminalProfileSchema).optional(), agentProfiles: z.array(AgentProfileSchema).optional(), @@ -3694,6 +3698,8 @@ export const ServerInfoStatusPayloadSchema = z pluginTimelineItems: z.boolean().optional(), // COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16. skillManagement: z.boolean().optional(), + // COMPAT(idleRuntimeCleanupSetting): added in v0.13.0, remove gate after 2027-04-26. + idleRuntimeCleanupSetting: z.boolean().optional(), // COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23. "terminal-restore-modes": z.boolean().optional(), // COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02. diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index c01d477004..ae8989c003 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1,3 +1,4 @@ +import { createIdleRuntimeCleanup } from "./idle-runtime-cleanup/index.js"; import { prepareAgentMessage, formatAgentMessage, @@ -10007,6 +10008,47 @@ test("closeAgent persists one final closed snapshot", 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; + const client = new (class extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + const session = await super.createSession(config); + session.canStopRuntime = async () => safe; + return session; + } + })(); + 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() + 61 * 60_000); + expect(await collect({ now })).toEqual({ collected: [], failures: [] }); + safe = true; + 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(); + 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 +10993,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 +12125,176 @@ 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); + 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) | undefined> = []; + const closeEntered = deferred(); + let closeGate: Promise | undefined; + let closes = 0; + const client = new (class extends TestAgentClient { + override async resumeSession( + handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + const session = await super.resumeSession(handle, config); + session.describePersistence = () => handle; + return recordPrompts(session); + } + override async createSession(config: AgentSessionConfig): Promise { + 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, + sweep: runtimeCleanup(manager), + manager, + storage, + sessions, + probes, + prompts, + closeEntered, + get closes() { + return closes; + }, + holdClose: (gate: Promise) => { + closeGate = gate; + }, + create: (labels?: Record) => + manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + labels, + }), + later: (minutes: number) => new Date(Date.now() + minutes * 60_000), + }; +} + +test("idle cleanup rechecks new prompt activity after an asynchronous safety probe", async () => { + const h = idleCleanupHarness(); + const agent = await h.create(); + const entered = deferred(); + const probe = deferred(); + h.probes[0] = () => { + entered.resolve(); + return probe.promise; + }; + const sweep = h.sweep({ now: h.later(61) }); + 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.sweep({ now: h.later(61) })).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.sweep({ now: h.later(61), 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(); + h.holdClose(gate.promise); + const sweep = h.sweep({ now: h.later(61) }); + 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(); + h.holdClose(gate.promise); + const sweep = h.sweep({ now: h.later(61) }); + await h.closeEntered.promise; + const command = startAgentRun(h.manager, agent.id, "/maintenance", logger); + await new Promise((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([]); +}); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index b39fe782b8..de823660da 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1,3 +1,5 @@ +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"; import { describeHookAgent, publishAgentStream } from "../plugins/lifecycle/index.js"; @@ -758,6 +760,8 @@ export class AgentManager { private readonly agentRegistrationTasks = new Set>(); private readonly inFlightAgentCloses = new Map>(); private readonly reloadedSessionCloses = new WeakMap>(); + private readonly runtimeCleanupIdentities = new WeakMap(); + private readonly collectingRuntimes = new Set(); private readonly lifecycleMutationTails = new Map>(); private readonly agentStreamCoalescer: AgentStreamCoalescer; private mcpBaseUrl: string | null; @@ -1747,6 +1751,64 @@ export class AgentManager { }); } + /** 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 && + !this.runs.hasRun(agent.id) && + !agent.pendingReplacement && + agent.pendingPermissions.size === 0 && + agent.inFlightPermissionResponses.size === 0 && + (this.collectingRuntimes.has(agent.id) || !this.lifecycleMutationTails.has(agent.id)) && + !this.foregroundMutationTails.has(agent.id) && + !this.sessionEventTails.has(agent.id) && + !this.providerSubagents.list(agent.id).some((child) => child.status === "running") + ); + } + closeAgent(agentId: string): Promise { const existing = this.inFlightAgentCloses.get(agentId); if (existing) { @@ -1767,7 +1829,7 @@ export class AgentManager { return close; } - private async closeAgentRuntime(agentId: string): Promise { + private async closeAgentRuntime(agentId: string, canClose?: () => boolean): Promise { const agent = this.requireAgent(agentId); this.logger.trace( { @@ -1782,6 +1844,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 +1870,7 @@ export class AgentManager { if (persistError !== undefined) { throw persistError; } + return true; } private cancelRunningProviderSubagents(parentAgentId: string): void { @@ -2459,7 +2523,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 { + 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 +2546,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 +2566,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 +2577,7 @@ export class AgentManager { item: { type: "assistant_message", text: `[Error] ${text}` }, }); } - })(); + }); return true; } @@ -2584,11 +2661,27 @@ export class AgentManager { } } + private async *streamAfterRuntimeCollection( + agentId: string, + prompt: AgentPromptInput, + options?: AgentRunOptions, + ): AsyncGenerator { + 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 { + 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( { diff --git a/packages/server/src/server/agent/agent-messages/delivery.test.ts b/packages/server/src/server/agent/agent-messages/delivery.test.ts index 8cd480c1fa..bbe3d4fb3d 100644 --- a/packages/server/src/server/agent/agent-messages/delivery.test.ts +++ b/packages/server/src/server/agent/agent-messages/delivery.test.ts @@ -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((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); +}); diff --git a/packages/server/src/server/agent/agent-messages/delivery.ts b/packages/server/src/server/agent/agent-messages/delivery.ts index e9313903d2..f26d83ec50 100644 --- a/packages/server/src/server/agent/agent-messages/delivery.ts +++ b/packages/server/src/server/agent/agent-messages/delivery.ts @@ -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; /** 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"]; diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index a8fd2ebd4b..08aca8e626 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -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" }; } diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index 59fb6022e9..6818e65445 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/packages/server/src/server/agent/agent-sdk-types.ts @@ -694,6 +694,10 @@ export interface AgentSession { * still uncertain. */ interrupt(): Promise; + /** True only when releasing this runtime loses no background work or pending replies. + * An optional additional veto: missing methods use daemon baseline eligibility; + * implemented checks that reject or return false keep the runtime resident. */ + canStopRuntime?(): Promise; /** Release live runtime resources without archiving or deleting the durable native session. */ close(): Promise; listCommands?(): Promise; diff --git a/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts b/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts index 872961111b..a4b4514add 100644 --- a/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts +++ b/packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts @@ -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(); + 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(); + 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); +}); diff --git a/packages/server/src/server/agent/finish-subscriptions/index.ts b/packages/server/src/server/agent/finish-subscriptions/index.ts index 58c3c8eff1..22324f0ecc 100644 --- a/packages/server/src/server/agent/finish-subscriptions/index.ts +++ b/packages/server/src/server/agent/finish-subscriptions/index.ts @@ -49,6 +49,8 @@ export interface FinishRegistration { } export interface FinishSubscriptions { start(): Promise; + /** Pending notification intent, including registration, recovery and unacknowledged delivery. */ + hasPendingForCaller(agentId: string): boolean; register( input: Omit, pendingDispatch?: boolean, @@ -72,6 +74,8 @@ export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptio const tasks = new Set>(); const watches = new Map>(); let closed = false; + let recovering = true; + const registering = new Map(); let mutations = Promise.resolve(); function serialize(action: () => Promise): Promise { 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(); @@ -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; diff --git a/packages/server/src/server/agent/idle-runtime-cleanup/idle-runtime-cleanup.test.ts b/packages/server/src/server/agent/idle-runtime-cleanup/idle-runtime-cleanup.test.ts new file mode 100644 index 0000000000..5ed1fa9b79 --- /dev/null +++ b/packages/server/src/server/agent/idle-runtime-cleanup/idle-runtime-cleanup.test.ts @@ -0,0 +1,294 @@ +import { expect, onTestFinished, test } from "vitest"; +import { createTestLogger } from "../../../test-utils/test-logger.js"; +import { createIdleRuntimeCleanup, type RuntimeCleanupCandidate } from "./index.js"; + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((yes) => { + resolve = yes; + }); + return { promise, resolve }; +} +class Runtime { + readonly identity = {}; + busy = false; + closed = false; + resumable = true; + activity = 0; + parentAgentId: string | undefined; + probe: (() => Promise) | undefined; + admitClose: (() => Promise) | undefined; + finishClose: (() => Promise) | 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) => { + await this.admitClose?.(); + if (!eligible()) return false; + this.closed = true; + await this.finishClose?.(); + 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(); + 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(); + 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); +}); + +test("the live host switch retains eligible runtimes and re-enables with their existing idle clocks", async () => { + const s = scenario(); + s.cleanup.setEnabled(false); + expect((await s.sweep(61)).collected).toEqual([]); + expect(s.runtime.closed).toBe(false); + s.cleanup.setEnabled(true); + s.tick(); + expect((await s.cleanup.sweep()).collected).toHaveLength(1); +}); + +test.each([false, true])( + "disabling during a probe invalidates the sweep (re-enable: %s)", + async (reEnable) => { + const s = scenario(); + const probe = deferred(); + s.runtime.probe = () => probe.promise; + const collecting = s.sweep(61); + s.cleanup.setEnabled(false); + if (reEnable) s.cleanup.setEnabled(true); + probe.resolve(true); + expect((await collecting).collected).toEqual([]); + expect(s.runtime.closed).toBe(false); + s.cleanup.setEnabled(true); + expect((await s.sweep(62)).collected).toHaveLength(1); + }, +); + +test.each([false, true])( + "disabling while waiting for lifecycle admission prevents closure (re-enable: %s)", + async (reEnable) => { + const s = scenario(); + const admitted = deferred(); + const waiting = deferred(); + s.runtime.admitClose = () => { + waiting.resolve(); + return admitted.promise; + }; + const collecting = s.sweep(61); + await waiting.promise; + s.cleanup.setEnabled(false); + if (reEnable) s.cleanup.setEnabled(true); + admitted.resolve(); + expect((await collecting).collected).toEqual([]); + expect(s.runtime.closed).toBe(false); + s.cleanup.setEnabled(true); + expect((await s.sweep(62)).collected).toHaveLength(1); + }, +); + +test.each(["veto", "error"])( + "an invalidated provider %s does not reset the idle clock", + async (outcome) => { + const s = scenario(); + const probe = deferred(); + s.runtime.probe = async () => { + await probe.promise; + if (outcome === "error") throw new Error("Old probe failed"); + return false; + }; + const collecting = s.sweep(61); + s.cleanup.setEnabled(false); + s.cleanup.setEnabled(true); + probe.resolve(false); + expect(await collecting).toEqual({ collected: [], failures: [] }); + s.runtime.probe = async () => true; + expect((await s.sweep(62)).collected).toHaveLength(1); + }, +); + +test("disabling lets a started close finish but prevents the next closure in that sweep", async () => { + const s = scenario(); + const sibling = new Runtime("sibling"); + s.runtimes.push(sibling); + const closing = deferred(); + const finished = deferred(); + s.runtime.finishClose = () => { + closing.resolve(); + return finished.promise; + }; + const collecting = s.sweep(61); + await closing.promise; + expect(s.runtime.closed).toBe(true); + s.cleanup.setEnabled(false); + finished.resolve(); + expect((await collecting).collected).toEqual([ + { agentId: "agent", provider: "test", sessionId: "agent" }, + ]); + expect(sibling.closed).toBe(false); + s.cleanup.setEnabled(true); + expect((await s.sweep(62)).collected).toEqual([ + { agentId: "sibling", provider: "test", sessionId: "sibling" }, + ]); +}); diff --git a/packages/server/src/server/agent/idle-runtime-cleanup/index.ts b/packages/server/src/server/agent/idle-runtime-cleanup/index.ts new file mode 100644 index 0000000000..fddea0d25b --- /dev/null +++ b/packages/server/src/server/agent/idle-runtime-cleanup/index.ts @@ -0,0 +1,167 @@ +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; + closeIfEligible(revalidate: () => boolean): Promise; +} +interface Options { + enabled: boolean; + runtimes(): readonly RuntimeCleanupCandidate[]; + isProtected?(agentId: string): boolean; + logger: Pick; + 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(); + let collection: Promise | null = null; + let stopped = false; + let enabled = options.enabled; + let generation = 0; + 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[], + isCurrent: () => boolean, + ) { + for (const runtime of family) { + if (!isCurrent()) return false; + 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 { + const result: CollectionResult = { collected: [], failures: [] }; + const sweepGeneration = generation; + const isCurrent = () => enabled && !stopped && generation === sweepGeneration; + if (!isCurrent()) return result; + const timestamp = now(); + const runtimes = options.runtimes(); + for (const runtime of runtimes) { + if (!isCurrent()) break; + if (!runtime.sessionId) continue; + const family = familyOf(runtime, runtimes); + try { + const allowed = baselineAllows(family) && (await providersAllow(family, isCurrent)); + if (!isCurrent()) break; + if (!allowed) { + 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 ( + isCurrent() && + 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 if (isCurrent()) { + protectedAt.set(runtime.identity, { at: timestamp, protected: true }); + } + } catch (error) { + if (!isCurrent()) break; + 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 { + 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, + setEnabled(value: boolean) { + if (enabled === value) return; + enabled = value; + generation++; + }, + async close() { + stopped = true; + cancel(); + await collection; + }, + }; +} diff --git a/packages/server/src/server/agent/permission-response.test.ts b/packages/server/src/server/agent/permission-response.test.ts index a5cf477e68..5ac7374e88 100644 --- a/packages/server/src/server/agent/permission-response.test.ts +++ b/packages/server/src/server/agent/permission-response.test.ts @@ -32,7 +32,7 @@ class FakePermissionAgentManager { return this.permissionResult; } - tryRunOutOfBand(): boolean { + async tryRunOutOfBand(): Promise { return this.outOfBandHandled; } diff --git a/packages/server/src/server/agent/plugin-provider.test.ts b/packages/server/src/server/agent/plugin-provider.test.ts index aa0aa1a924..3131447098 100644 --- a/packages/server/src/server/agent/plugin-provider.test.ts +++ b/packages/server/src/server/agent/plugin-provider.test.ts @@ -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, @@ -1234,3 +1239,85 @@ 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 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") + expect(requests[0]).toMatchObject({ + sessionId: harness.inputs.find((input) => input.type === "session.open")?.sessionId, + }); + } finally { + await session.close(); + await registry.shutdown(); + } + }, +); + +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 }); + } +}); diff --git a/packages/server/src/server/agent/plugin-provider.ts b/packages/server/src/server/agent/plugin-provider.ts index db5d7651a2..68e9778e43 100644 --- a/packages/server/src/server/agent/plugin-provider.ts +++ b/packages/server/src/server/agent/plugin-provider.ts @@ -113,9 +113,15 @@ function providerError(error: ProviderError): Error { function isProviderRequestReply( event: ProviderEvent, -): event is Extract { +): 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, + 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 { + 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 { const input: Extract = { type: "session.configure", @@ -1201,6 +1239,11 @@ class PluginAgentSession implements AgentSession { for (const event of this.history) yield event; } + get canStopRuntime(): (() => Promise) | undefined { + if (!this.bridge.negotiatedCapabilities.includes("session.can_stop_runtime")) return undefined; + return () => this.bridge.canStopRuntime(); + } + async getRuntimeInfo() { return { provider: this.provider, diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index 9f85e4d37f..0f3c39868a 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -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), diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index a050aa9077..3deaa60c11 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -35,6 +35,7 @@ import { resolveACPModelSelection, summarizeACPRequestError, } from "./acp-agent.js"; +import { createIdleRuntimeCleanup } from "../idle-runtime-cleanup/index.js"; import type { ProcessTerminator, TreeKillTarget } from "../../../utils/tree-kill.js"; import { COPILOT_AGENT_FEATURE_OPTION, @@ -4424,6 +4425,161 @@ new AgentSideConnection( ); `; +describe("ACP idle runtime cleanup", () => { + test("pending permissions veto cleanup until answered", async () => { + const session = createSession(); + try { + expect(await session.canStopRuntime()).toBe(true); + const permission = session.requestPermission({ + sessionId: "session-1", + toolCall: { toolCallId: "tool-1", title: "Run command", status: "pending" }, + options: [{ optionId: "allow", name: "Allow", kind: "allow_once" }], + }); + expect(await session.canStopRuntime()).toBe(false); + await session.respondToPermission(session.getPendingPermissions()[0]!.id, { + behavior: "allow", + }); + await permission; + expect(await session.canStopRuntime()).toBe(true); + } finally { + await session.close(); + } + }); + + test("a failed terminal spawn does not pin the runtime or discard its diagnostic", async () => { + const session = createSession(); + try { + const terminal = await session.createTerminal({ + sessionId: "session-1", + command: path.join(tmpdir(), "paseo-no-such-command", "missing"), + }); + await expect( + session.waitForTerminalExit({ sessionId: "session-1", ...terminal }), + ).rejects.toThrow("ENOENT"); + expect(await session.canStopRuntime()).toBe(true); + expect( + (await session.terminalOutput({ sessionId: "session-1", ...terminal })).output, + ).toContain("ENOENT"); + await session.releaseTerminal({ sessionId: "session-1", ...terminal }); + expect(await session.canStopRuntime()).toBe(true); + } finally { + await session.close(); + } + }); + + test("retains a quiet terminal after a finished turn, then waits a fresh idle hour after exit", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "paseo-acp-idle-")); + const agentScript = path.join(dir, "agent.mjs"); + await writeFile(agentScript, PERMISSIVE_MODEL_ACP_AGENT); + const session = new ACPAgentSession( + { provider: "test-acp", cwd: dir }, + { + provider: "test-acp", + logger: createTestLogger(), + defaultCommand: [process.execPath, agentScript], + defaultModes: [], + capabilities: { supportsStreaming: true, supportsSessionPersistence: true }, + launchEnv: { + ACP_SDK_URL: pathToFileURL( + createRequire(import.meta.url).resolve("@agentclientprotocol/sdk"), + ).href, + }, + }, + ); + const hour = 60 * 60_000; + let now = 0; + let closed = false; + const candidate = { + identity: session, + agentId: "test-agent", + parentAgentId: undefined, + provider: session.provider, + sessionId: "permissive-session", + lastActivityAt: 0, + isEligible: () => !closed, + canStopRuntime: (session as AgentSession).canStopRuntime?.bind(session), + async closeIfEligible(revalidate: () => boolean) { + if (!revalidate()) return false; + await session.close(); + closed = true; + return true; + }, + }; + const cleanup = createIdleRuntimeCleanup({ + enabled: true, + now: () => now, + runtimes: () => (closed ? [] : [candidate]), + schedule: () => () => {}, + logger: createTestLogger(), + }); + try { + await session.initializeNewSession(); + const completed = new Promise((resolve) => + session.subscribe((event) => { + if (event.type === "turn_completed") resolve(); + }), + ); + let activeTurnSafety: Promise | undefined; + session.subscribe((event) => { + if (event.type === "turn_started") activeTurnSafety = session.canStopRuntime(); + }); + await session.startTurn("Finish this foreground turn"); + await expect(activeTurnSafety).resolves.toBe(false); + await completed; + expect(await session.canStopRuntime()).toBe(true); + await session.sessionUpdate({ + sessionId: candidate.sessionId, + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "buffered" }, + }, + }); + expect(await session.canStopRuntime()).toBe(false); + await session.sessionUpdate({ + sessionId: candidate.sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "flush" }, + }, + }); + expect(await session.canStopRuntime()).toBe(true); + const terminal = await session.createTerminal({ + sessionId: candidate.sessionId, + command: process.execPath, + args: ["-e", "setInterval(() => {}, 1000)"], + }); + now = 2 * hour; + expect((await cleanup.sweep()).collected).toEqual([]); + expect(closed).toBe(false); + expect(await session.terminalOutput({ sessionId: candidate.sessionId, ...terminal })).toEqual( + { + output: "", + truncated: false, + exitStatus: undefined, + }, + ); + await session.killTerminal({ sessionId: candidate.sessionId, ...terminal }); + await session.waitForTerminalExit({ sessionId: candidate.sessionId, ...terminal }); + expect((await cleanup.sweep()).collected).toEqual([]); + now += hour - 1; + expect((await cleanup.sweep()).collected).toEqual([]); + now += 1; + expect((await cleanup.sweep()).collected).toEqual([ + { + agentId: candidate.agentId, + provider: candidate.provider, + sessionId: candidate.sessionId, + }, + ]); + expect(closed).toBe(true); + } finally { + await cleanup.close(); + await session.close(); + await rm(dir, { recursive: true, force: true }); + } + }); +}); + describe("custom ACP provider with configured models", () => { let dir: string; let agentScript: string; diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 823f941de1..ae765ecd2e 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -649,6 +649,7 @@ interface TerminalExit { } interface TerminalEntry { + running: boolean; id: string; child: ChildProcess; output: string; @@ -2554,6 +2555,15 @@ export class ACPAgentSession implements AgentSession, ACPClient { } } + async canStopRuntime(): Promise { + return ( + this.activeForegroundTurnId === null && + this.pendingPermissions.size === 0 && + this.pendingUserMessage === null && + !Array.from(this.terminalEntries.values()).some((terminal) => terminal.running) + ); + } + async close(): Promise { if (this.closed) { return; @@ -2589,12 +2599,14 @@ export class ACPAgentSession implements AgentSession, ACPClient { } } - const terminalTerminations = Array.from(this.terminalEntries.values(), (terminal) => - this.terminateProcess(terminal.child, { - gracefulTimeoutMs: 2_000, - forceTimeoutMs: 2_000, - }), - ); + const terminalTerminations = Array.from(this.terminalEntries.values()) + .filter((terminal) => terminal.running) + .map((terminal) => + this.terminateProcess(terminal.child, { + gracefulTimeoutMs: 2_000, + forceTimeoutMs: 2_000, + }), + ); await Promise.all(terminalTerminations); this.terminalEntries.clear(); @@ -2796,6 +2808,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { waitForExit.catch(() => undefined); const entry: TerminalEntry = { + running: true, id: terminalId, child, output: "", @@ -2816,10 +2829,13 @@ export class ACPAgentSession implements AgentSession, ACPClient { child.once("error", (error) => { const spawnError = error instanceof Error ? error : new Error(String(error)); appendTerminalOutput(entry, `${spawnError.message}\n`); + // A failed spawn has no exit event. Other process errors do not prove exit. + if (child.pid === undefined) entry.running = false; rejectExit(spawnError); }); child.once("exit", (code, signal) => { const exit = { exitCode: code, signal }; + entry.running = false; entry.exit = exit; resolveExit(exit); }); @@ -2844,7 +2860,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { async releaseTerminal(params: { sessionId: string; terminalId: string }): Promise { const entry = this.getTerminalEntry(params.terminalId); - if (!entry.exit) { + if (entry.running) { await this.terminateProcess(entry.child, { gracefulTimeoutMs: 2_000, forceTimeoutMs: 2_000 }); } this.terminalEntries.delete(params.terminalId); @@ -2852,7 +2868,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { async killTerminal(params: KillTerminalRequest): Promise> { const entry = this.getTerminalEntry(params.terminalId); - if (!entry.exit) { + if (entry.running) { await this.terminateProcess(entry.child, { gracefulTimeoutMs: 2_000, forceTimeoutMs: 2_000 }); } return {}; diff --git a/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts b/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts index 4cab1a5005..8530e76844 100644 --- a/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts @@ -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((resolve) => { + wake = resolve; + }); + continue; + } + yield entry.message; + entry.consumed(); + } + })(); + const push = (message: unknown) => + new Promise((consumed) => { + queue.push({ message: message as SDKMessage, consumed }); + wake?.(); + }); + const end = () => { + channel.closed = true; + wake?.(); + }; + let nativeInput!: AsyncIterator; + let connected!: () => void; + const connection = new Promise((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(); + } +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 71a279834d..201fa349eb 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -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; private query: Query | null = null; + private readonly runtimeSafety = new ClaudeRuntimeSafety(); + private processingSdkMessages = 0; private readonly harnessEnvironment: Record; private readonly usageSessionKey = randomUUID(); private childProcess: ChildProcess | null = null; @@ -2931,6 +2934,18 @@ class ClaudeAgentSession implements AgentSession { return this.persistence; } + async canStopRuntime(): Promise { + return ( + !this.activeForegroundTurnId && + !this.autonomousTurn && + !this.mainTurnInFlight && + this.unstartedMessageUuids.size === 0 && + this.pendingPermissions.size === 0 && + this.processingSdkMessages === 0 && + this.runtimeSafety.canStopRuntime() + ); + } + async close(): Promise { this.logger.trace( { @@ -3424,6 +3439,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( @@ -4079,13 +4095,17 @@ class ClaudeAgentSession implements AgentSession { ); }; const handlePumpedMessage = async (message: SDKMessage): Promise => { - 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 => { for await (const message of activeQuery) { diff --git a/packages/server/src/server/agent/providers/claude/runtime-safety.ts b/packages/server/src/server/agent/providers/claude/runtime-safety.ts new file mode 100644 index 0000000000..0c32804a57 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/runtime-safety.ts @@ -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(); + private readonly awaitingNotifications = new Set(); + + 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; + } +} diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 8fca5f24c0..8c61635d6f 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -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(); + } +}); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 62ceb14c70..f438c98ec4 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -5029,6 +5029,34 @@ export class CodexAppServerAgentSession implements AgentSession { } } + async canStopRuntime(): Promise { + 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 { this.closed = true; this.clearPendingPermissions(); diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.ts index acb5e76341..29b8cf8880 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.ts @@ -202,6 +202,7 @@ function createNewlineLineReader( export class CodexAppServerClient { private readonly stdoutLineReader: { close(): void }; + private activeProviderRequests = 0; private readonly pending = new Map(); private readonly requestHandlers = new Map(); 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 { 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; } diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index c9911a64b8..983f9cf55f 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -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); + }, +); diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index ab82df9475..434a49d89e 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -1172,6 +1172,27 @@ export class OmpAgentSession implements AgentSession { this.activeToolCalls.clear(); } + async canStopRuntime(): Promise { + 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 { if (this.closed) { return; diff --git a/packages/server/src/server/agent/providers/omp/rpc-types.ts b/packages/server/src/server/agent/providers/omp/rpc-types.ts index 733dedf571..cfe0b4cbd6 100644 --- a/packages/server/src/server/agent/providers/omp/rpc-types.ts +++ b/packages/server/src/server/agent/providers/omp/rpc-types.ts @@ -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(), diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index d6b6c5513a..8beed24be0 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -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(); + } +}); diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 7e49bc50d3..81288f3e4f 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -5055,6 +5055,43 @@ class OpenCodeAgentSession implements AgentSession { }; } + async canStopRuntime(): Promise { + 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(); + 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 { try { this.closed = true; diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts index 7bad72b8b1..f9c1683a48 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/test-opencode-harness.ts @@ -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) @@ -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); diff --git a/packages/server/src/server/agent/providers/opencode/test-utils/v2-harness.ts b/packages/server/src/server/agent/providers/opencode/test-utils/v2-harness.ts index e2dc4f68a6..505955f64c 100644 --- a/packages/server/src/server/agent/providers/opencode/test-utils/v2-harness.ts +++ b/packages/server/src/server/agent/providers/opencode/test-utils/v2-harness.ts @@ -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); diff --git a/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts b/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts index 26cb65885d..564e1f46c5 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/agent.test.ts @@ -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(); + } +}); diff --git a/packages/server/src/server/agent/providers/opencode/v2/api.ts b/packages/server/src/server/agent/providers/opencode/v2/api.ts index fe225cade2..bec40ffbcb 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/api.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/api.ts @@ -2,6 +2,7 @@ import type { OpenCodeClient } from "@opencode/client"; export interface V2Api { server: OpenCodeClient["server"]; + shell: Pick; session: Pick< OpenCodeClient["session"], | "create" @@ -20,6 +21,7 @@ export interface V2Api { | "revert" | "log" > & { + inbox: Pick; form: Pick; }; plugin: Pick; diff --git a/packages/server/src/server/agent/providers/opencode/v2/session.ts b/packages/server/src/server/agent/providers/opencode/v2/session.ts index 5cd45b1ff7..9358a14040 100644 --- a/packages/server/src/server/agent/providers/opencode/v2/session.ts +++ b/packages/server/src/server/agent/providers/opencode/v2/session.ts @@ -516,6 +516,36 @@ export class OpenCodeV2Session implements AgentSession { await delay(100, undefined, { signal }).catch(() => undefined); } } + async canStopRuntime(): Promise { + 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(); + 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; diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index 927a9defd2..1bad16ba30 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -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"); +}); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 146b56b36f..cf193e213b 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -1581,6 +1581,22 @@ export class PiRpcAgentSession implements AgentSession { return await resultPromise; } + async canStopRuntime(): Promise { + 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 { if (this.closed) { return; diff --git a/packages/server/src/server/bootstrap.test.ts b/packages/server/src/server/bootstrap.test.ts index 6f705ec559..0881e828d0 100644 --- a/packages/server/src/server/bootstrap.test.ts +++ b/packages/server/src/server/bootstrap.test.ts @@ -1,6 +1,17 @@ -import { expect, test } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { expect, test, onTestFinished } from "vitest"; +import { MutableDaemonConfigSchema } from "@getpaseo/protocol/messages"; +import { DaemonConfigStore } from "./daemon-config-store.js"; +import { loadPersistedConfig } from "./persisted-config.js"; +import { + createIdleRuntimeCleanup, + type RuntimeCleanupCandidate, +} from "./agent/idle-runtime-cleanup/index.js"; +import { createTestLogger } from "../test-utils/test-logger.js"; -import { fanOutReconciledWorkspaceUpdates } from "./bootstrap.js"; +import { fanOutReconciledWorkspaceUpdates, connectIdleRuntimeCleanupConfig } from "./bootstrap.js"; test("reconciliation emits workspace updates when observer sync fails", async () => { const emittedWorkspaceIds: string[][] = []; @@ -54,3 +65,132 @@ test("reconciliation isolates workspace update failures between sessions", async expect(emittedWorkspaceIds).toEqual([["ws-reclassified"]]); expect(warnings).toHaveLength(1); }); + +// Exercise bootstrap's actual live-config binding with the real collector and store. + +function liveCleanupHost() { + const home = mkdtempSync(path.join(tmpdir(), "paseo-live-cleanup-")); + writeFileSync(path.join(home, "config.json"), JSON.stringify({ version: 1 })); + const store = new DaemonConfigStore( + home, + MutableDaemonConfigSchema.parse({ + relay: { enabled: true }, + mcp: { injectIntoAgents: false }, + idleRuntimeCleanup: true, + }), + undefined, + { + reloadSource: { + resolve: (persisted) => ({ + mutable: MutableDaemonConfigSchema.parse({ + relay: { enabled: true }, + mcp: { injectIntoAgents: false }, + idleRuntimeCleanup: persisted.agents?.idleRuntimeCleanup ?? true, + }), + overrideControlledPaths: [], + }), + }, + }, + ); + let minutes = 61; + let closed = false; + let protectedCaller = false; + const runtime: RuntimeCleanupCandidate = { + identity: {}, + agentId: "agent", + provider: "test", + sessionId: "session", + parentAgentId: undefined, + lastActivityAt: 0, + isEligible: () => !closed, + closeIfEligible: async (eligible) => { + if (!eligible()) return false; + closed = true; + return true; + }, + }; + const cleanup = createIdleRuntimeCleanup({ + enabled: true, + now: () => minutes * 60_000, + runtimes: () => [runtime], + isProtected: () => protectedCaller, + logger: createTestLogger(), + schedule: () => () => {}, + }); + const unsubscribe = connectIdleRuntimeCleanupConfig(store, cleanup); + onTestFinished(async () => { + unsubscribe(); + await cleanup.close(); + rmSync(home, { recursive: true, force: true }); + }); + return { + home, + store, + cleanup, + unsubscribe, + protect: (value: boolean) => { + protectedCaller = value; + }, + sweep: (at = minutes) => { + minutes = at; + return cleanup.sweep(); + }, + }; +} + +test("bootstrap's host toggle prevents collection live, persists, and leaves other hosts enabled", async () => { + const host = liveCleanupHost(); + const other = liveCleanupHost(); + host.store.patch({ idleRuntimeCleanup: false }); + expect(loadPersistedConfig(host.home).agents?.idleRuntimeCleanup).toBe(false); + expect((await host.sweep()).collected).toEqual([]); + expect((await other.sweep()).collected).toHaveLength(1); + host.store.patch({ idleRuntimeCleanup: true }); + host.protect(true); + expect((await host.sweep()).collected).toEqual([]); + host.protect(false); + expect((await host.sweep(62)).collected).toEqual([]); + expect((await host.sweep(122)).collected).toHaveLength(1); +}); + +test("bootstrap's config-file reload binding disables and enables eligible collection live", async () => { + const host = liveCleanupHost(); + for (const enabled of [false, true]) { + writeFileSync( + path.join(host.home, "config.json"), + JSON.stringify({ version: 1, agents: { idleRuntimeCleanup: enabled } }), + ); + expect(host.store.reload()).toEqual({ + appliedPaths: ["agents.idleRuntimeCleanup"], + restartRequiredPaths: [], + overrideControlledPaths: [], + }); + expect((await host.sweep()).collected).toHaveLength(enabled ? 1 : 0); + } +}); + +test.each([false, true])( + "failed config apply restores prior idle cleanup behavior (previous: %s)", + async (previous) => { + const host = liveCleanupHost(); + host.store.patch({ idleRuntimeCleanup: previous }); + const persistedBefore = loadPersistedConfig(host.home); + const refuse = host.store.onApply(() => { + throw new Error("Another live owner refused"); + }); + expect(() => host.store.patch({ idleRuntimeCleanup: !previous })).toThrow( + "Another live owner refused", + ); + expect(host.store.get().idleRuntimeCleanup).toBe(previous); + expect(loadPersistedConfig(host.home)).toEqual(persistedBefore); + expect((await host.sweep()).collected).toHaveLength(previous ? 1 : 0); + refuse(); + }, +); + +test("disposing bootstrap's binding stops configuration delivery", async () => { + const host = liveCleanupHost(); + host.unsubscribe(); + host.store.patch({ idleRuntimeCleanup: false }); + expect((await host.sweep()).collected).toHaveLength(1); +}); diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 03decb5f67..9419694043 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -1,3 +1,4 @@ +import { createIdleRuntimeCleanup } from "./agent/idle-runtime-cleanup/index.js"; import { createFinishSubscriptions, createFinishObservation, @@ -453,6 +454,7 @@ export interface PaseoDaemonConfig { dictationFinalTimeoutMs?: number; downloadTokenTtlMs?: number; agentProviderSettings?: AgentProviderRuntimeSettingsMap; + idleRuntimeCleanup?: boolean; providerCatalogRefreshTimeoutMs?: number; metadataGeneration?: { providers?: Array<{ @@ -577,6 +579,7 @@ function createInitialMutableDaemonConfig(config: PaseoDaemonConfig): MutableDae }, autoArchiveAfterMerge: config.autoArchiveAfterMerge ?? false, enableTerminalAgentHooks: config.enableTerminalAgentHooks ?? false, + idleRuntimeCleanup: config.idleRuntimeCleanup !== false, appendSystemPrompt: config.appendSystemPrompt ?? "", pluginsEnabled: config.pluginsEnabled ?? false, plugins: config.plugins ?? {}, @@ -1398,6 +1401,7 @@ export async function createPaseoDaemon( archiveWorkspace: archiveScheduleWorkspaceExternal, }); await scheduleService.start(); + agentManager.setAgentArchivedCallback(async (agentId) => { try { await scheduleService.completeForAgent(agentId); @@ -1445,6 +1449,20 @@ export async function createPaseoDaemon( }), }); + const idleRuntimeCleanup = createIdleRuntimeCleanup({ + enabled: daemonConfigStore.get().idleRuntimeCleanup !== false, + runtimes: () => agentManager.getRuntimeCleanupCandidates(), + isProtected: (agentId) => + finishSubscriptions.hasPendingForCaller(agentId) || + (wsServer?.agentMessages.hasPending(agentId) ?? false), + logger, + }); + + const unsubscribeIdleRuntimeCleanup = connectIdleRuntimeCleanupConfig( + daemonConfigStore, + idleRuntimeCleanup, + ); + const createAgentToolHostDependencies = ( runtime: PaseoToolRuntimeContext, ): PaseoToolHostDependencies => ({ @@ -1907,6 +1925,8 @@ export async function createPaseoDaemon( await hubRelationships.stop(); workspaceReconciliation.dispose(); scriptHealthMonitor.stop(); + unsubscribeIdleRuntimeCleanup(); + await idleRuntimeCleanup.close(); // Freeze both ingress and registration before taking the agent closure snapshot. wsServer?.prepareForShutdown(); await finishSubscriptions.close(); @@ -1997,3 +2017,11 @@ async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promi }), ); } + +export function connectIdleRuntimeCleanupConfig( + store: DaemonConfigStore, + cleanup: Pick, "setEnabled">, +): () => void { + cleanup.setEnabled(store.get().idleRuntimeCleanup !== false); + return store.onFieldChange("idleRuntimeCleanup", (value) => cleanup.setEnabled(value !== false)); +} diff --git a/packages/server/src/server/config.test.ts b/packages/server/src/server/config.test.ts index a71b5621e8..864ad14100 100644 --- a/packages/server/src/server/config.test.ts +++ b/packages/server/src/server/config.test.ts @@ -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); diff --git a/packages/server/src/server/config.ts b/packages/server/src/server/config.ts index cc4eee4fec..8187ce03c6 100644 --- a/packages/server/src/server/config.ts +++ b/packages/server/src/server/config.ts @@ -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, diff --git a/packages/server/src/server/daemon-config-store.test.ts b/packages/server/src/server/daemon-config-store.test.ts index c68cda02d2..095ae3b14b 100644 --- a/packages/server/src/server/daemon-config-store.test.ts +++ b/packages/server/src/server/daemon-config-store.test.ts @@ -4,6 +4,7 @@ import path from "node:path"; import { afterEach, describe, expect, test } from "vitest"; import { DaemonConfigStore, applyMutableProviderConfigToOverrides } from "./daemon-config-store.js"; +import { loadConfig } from "./config.js"; import { loadPersistedConfig } from "./persisted-config.js"; import type { PersistedConfig } from "./persisted-config.js"; import type { MutableDaemonConfig } from "@getpaseo/protocol/messages"; @@ -26,6 +27,7 @@ function reloadableConfig( metadataGeneration: { providers: agents.metadataGeneration?.providers ?? [] }, autoArchiveAfterMerge: daemon.autoArchiveAfterMerge ?? false, enableTerminalAgentHooks: daemon.enableTerminalAgentHooks ?? false, + idleRuntimeCleanup: agents.idleRuntimeCleanup !== false, appendSystemPrompt: daemon.appendSystemPrompt ?? "", terminalProfiles: daemon.terminalProfiles, agentProfiles: daemon.agentProfiles, @@ -902,6 +904,43 @@ describe("DaemonConfigStore", () => { expect(persisted.agents?.metadataGeneration).toEqual({ providers: [] }); }); + test("patch persists the host idle runtime cleanup opt-out for the next daemon start", () => { + const paseoHome = mkdtempSync(path.join(tmpdir(), "paseo-daemon-config-store-")); + tempDirs.push(paseoHome); + writeFileSync( + path.join(paseoHome, "config.json"), + `${JSON.stringify({ + version: 1, + agents: { metadataGeneration: { providers: [{ provider: "claude" }] } }, + })}\n`, + ); + const store = new DaemonConfigStore( + paseoHome, + { + mcp: { injectIntoAgents: false }, + browserTools: { enabled: false }, + providers: {}, + autoArchiveAfterMerge: false, + enableTerminalAgentHooks: false, + idleRuntimeCleanup: true, + appendSystemPrompt: "", + metadataGeneration: { providers: [{ provider: "claude" }] }, + }, + undefined, + ); + + expect(store.patch({ idleRuntimeCleanup: false }).idleRuntimeCleanup).toBe(false); + + expect(loadPersistedConfig(paseoHome).agents).toEqual({ + metadataGeneration: { providers: [{ provider: "claude" }] }, + idleRuntimeCleanup: false, + }); + expect(loadConfig(paseoHome, { env: {} }).idleRuntimeCleanup).toBe(false); + + store.patch({ idleRuntimeCleanup: true }); + expect(loadConfig(paseoHome, { env: {} }).idleRuntimeCleanup).toBe(true); + }); + test("patch persists custom ACP provider overrides into config.json", () => { const paseoHome = mkdtempSync(path.join(tmpdir(), "paseo-daemon-config-store-")); tempDirs.push(paseoHome); @@ -1013,6 +1052,29 @@ describe("DaemonConfigStore reload", () => { expect(store.get().git).toEqual({ maxProcessesPerSecond: 12, maxProcessConcurrency: 3 }); }); + test("reloads idle runtime cleanup in both directions without requiring a restart", () => { + const { paseoHome, store, persisted } = createReloadableStore(); + const changes: unknown[] = []; + store.onFieldChange("idleRuntimeCleanup", (value) => changes.push(value)); + for (const enabled of [false, true]) { + writeConfig(paseoHome, { ...persisted, agents: { idleRuntimeCleanup: enabled } }); + expect(store.reload()).toEqual({ + appliedPaths: ["agents.idleRuntimeCleanup"], + restartRequiredPaths: [], + overrideControlledPaths: [], + }); + expect(store.get().idleRuntimeCleanup).toBe(enabled); + } + expect(changes).toEqual([false, true]); + store.patch({ idleRuntimeCleanup: false }); + expect(store.reload()).toEqual({ + appliedPaths: [], + restartRequiredPaths: [], + overrideControlledPaths: [], + }); + expect(store.get().idleRuntimeCleanup).toBe(false); + }); + test("applies the global plugin switch in both directions", () => { const { paseoHome, store, persisted } = createReloadableStore({ initialPersisted: { version: 1, pluginsEnabled: false }, diff --git a/packages/server/src/server/daemon-config-store.ts b/packages/server/src/server/daemon-config-store.ts index 489f68ed5d..2241193e50 100644 --- a/packages/server/src/server/daemon-config-store.ts +++ b/packages/server/src/server/daemon-config-store.ts @@ -25,6 +25,7 @@ interface SupportedMutableConfigPatch { metadataGeneration?: MutableDaemonConfig["metadataGeneration"]; autoArchiveAfterMerge?: boolean; enableTerminalAgentHooks?: boolean; + idleRuntimeCleanup?: boolean; appendSystemPrompt?: string; terminalProfiles?: MutableDaemonConfig["terminalProfiles"]; agentProfiles?: MutableDaemonConfig["agentProfiles"]; @@ -185,6 +186,7 @@ const RELOADABLE_PATHS = [ "daemon.terminalProfiles", "daemon.agentProfiles", "app.baseUrl", + "agents.idleRuntimeCleanup", "agents.providers", "agents.catalogRefreshTimeoutMs", "agents.metadataGeneration", @@ -209,6 +211,7 @@ const PERSISTED_TO_MUTABLE_PATH = new Map([ ["daemon.terminalProfiles", "terminalProfiles"], ["daemon.agentProfiles", "agentProfiles"], ["app.baseUrl", "app.baseUrl"], + ["agents.idleRuntimeCleanup", "idleRuntimeCleanup"], ["agents.providers", "providers"], ["agents.catalogRefreshTimeoutMs", "catalogRefreshTimeoutMs"], ["agents.metadataGeneration", "metadataGeneration"], @@ -283,6 +286,9 @@ function pickSupportedPatchFields(patch: MutableDaemonConfigPatch): SupportedMut ...(patch.enableTerminalAgentHooks !== undefined ? { enableTerminalAgentHooks: patch.enableTerminalAgentHooks } : {}), + ...(patch.idleRuntimeCleanup !== undefined + ? { idleRuntimeCleanup: patch.idleRuntimeCleanup } + : {}), ...(patch.appendSystemPrompt !== undefined ? { appendSystemPrompt: patch.appendSystemPrompt } : {}), @@ -616,6 +622,7 @@ function mergeMutableAgentPatch( patch.providers === undefined && patch.metadataGeneration === undefined && patch.skills === undefined && + patch.idleRuntimeCleanup === undefined && removeProviders.length === 0 ) { return persistedAgents; @@ -648,6 +655,10 @@ function mergeMutableAgentPatch( next["skills"] = { selection: patch.skills.selection }; } + if (patch.idleRuntimeCleanup !== undefined) { + next["idleRuntimeCleanup"] = patch.idleRuntimeCleanup; + } + return Object.keys(next).length > 0 ? (next as PersistedConfig["agents"]) : undefined; } diff --git a/packages/server/src/server/persisted-config.ts b/packages/server/src/server/persisted-config.ts index cccc0228d8..8b99a1ac45 100644 --- a/packages/server/src/server/persisted-config.ts +++ b/packages/server/src/server/persisted-config.ts @@ -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(), diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 88fb2c327d..69f116f25b 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -2008,6 +2008,8 @@ export class VoiceAssistantWebSocketServer { pluginTimelineItems: true, // COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16. skillManagement: true, + // COMPAT(idleRuntimeCleanupSetting): added in v0.13.0, remove gate after 2027-04-26. + idleRuntimeCleanupSetting: true, // COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23. "terminal-restore-modes": true, // COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02. diff --git a/packages/website/public/schemas/paseo.config.v1.json b/packages/website/public/schemas/paseo.config.v1.json index 7e34e35e5c..a6d0ed4541 100644 --- a/packages/website/public/schemas/paseo.config.v1.json +++ b/packages/website/public/schemas/paseo.config.v1.json @@ -552,6 +552,9 @@ } } }, + "idleRuntimeCleanup": { + "type": "boolean" + }, "catalogRefreshTimeoutMs": { "type": "integer", "exclusiveMinimum": 0, diff --git a/public-docs/configuration.md b/public-docs/configuration.md index a11c4c69ba..d9b3d59f8d 100644 --- a/public-docs/configuration.md +++ b/public-docs/configuration.md @@ -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,22 @@ 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 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 by turning off **Release idle agents** in **Settings → your host → Agents**, or with: + +```json +{ "agents": { "idleRuntimeCleanup": false } } +``` + +This setting defaults to `true`. Changes in Settings apply immediately; config-file edits apply +on config reload. Re-enabling keeps existing idle timestamps. There is no per-agent 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. ACP adapters +also retain runtimes with running ACP terminals, including silent ones; exited terminals do not +prevent cleanup. This check does not inspect processes launched internally by the provider. + ## Worktrees New worktrees are created under `$PASEO_HOME/worktrees` by default. To place new worktrees somewhere else, set `worktrees.root`: diff --git a/public-docs/configuration.zh-CN.md b/public-docs/configuration.zh-CN.md index 4f904db8d8..532b763907 100644 --- a/public-docs/configuration.zh-CN.md +++ b/public-docs/configuration.zh-CN.md @@ -72,7 +72,7 @@ paseo daemon restart 当你删除 `daemon.relay.enabled` 时，新的主目录会保持中继关闭。如果 daemon 启动时其配置就已经省略了这个字段，它会为了兼容而保持旧版的中继开启行为。编辑较旧的配置时，请显式设置 `daemon.relay.enabled`。 -监听地址、身份验证、中继端点和 TLS、worktree 分配、服务代理地址、内置的 Web UI、日志、语音合成与识别、语音、凭据以及本地模型设置都需要重启。重新加载会先应用同一文件中的其他有效编辑，再报告这些路径。 +空闲运行时清理、监听地址、身份验证、中继端点和 TLS、worktree 分配、服务代理地址、内置的 Web UI、日志、语音合成与识别、语音、凭据以及本地模型设置都需要重启。重新加载会先应用同一文件中的其他有效编辑，再报告这些路径。 部署环境变量和旧版进程管理器参数仍然具有最终决定权。当某个启动覆盖项阻止了文件中已更改的设置生效时，重新加载会在 `overrideControlledPaths` 下报告它。这包括启动设置，例如监听地址、密码、中继端点和 TLS、服务代理和 Web UI 设置、日志、语音合成与识别以及语音配置。主机名和 CORS 来源等列表设置仍然在各个来源之间追加，因此 `config.json` 中的值继续生效。要让文件中的值具有决定权，请停止该部署并在不带覆盖项的情况下重新启动。 @@ -82,6 +82,16 @@ Agent Provider，包括 Paseo 自带的一等 Provider 和你在 `agents.provide 心智模型见 [Provider](/docs/providers)，Paseo 可以启动的全部 Agent 见[支持的 Provider](/docs/supported-providers)。要让 Claude 指向兼容 Anthropic 的端点（Z.AI、Alibaba/Qwen）、使用多个配置、自定义二进制文件、ACP Agent，以及了解 `additionalModels` 的合并行为，见[自定义 Provider](/docs/custom-providers)。完整的字段参考在 GitHub 上的 [docs/custom-providers.md](https://github.com/getpaseo/paseo/blob/main/docs/custom-providers.md)。 +## 空闲运行时 {#idle-runtimes} + +当 daemon 的基础条件允许清理，且 Provider 已实现的安全检查也允许时，Paseo 会在可恢复的 Agent 空闲一小时后释放其进程。你的下一条提示词会恢复同一个会话。要为整个主机禁用此功能，请在 **设置 → 你的主机 → Agents** 中关闭 **释放空闲 Agent**，或设置： + +```json +{ "agents": { "idleRuntimeCleanup": false } } +``` + +此设置默认为 `true`。在设置中更改后立即生效；配置文件中的更改在重新加载配置后生效。重新启用会保留现有的空闲计时。没有单个 Agent 的设置。不提供运行时安全检查的 Provider 使用相同的 daemon 基础条件和一小时空闲期。活动中的工作、后代 Agent 和待发送的完成通知都会阻止清理。ACP 适配器还会在 ACP 终端仍运行时保留运行时，即使终端没有输出；已退出的终端不会阻止清理。此检查不检查 Provider 内部启动的进程。 + ## worktree {#worktrees} 新的 worktree 默认创建在 `$PASEO_HOME/worktrees` 下。要把新的 worktree 放到其他位置，请设置 `worktrees.root`： diff --git a/public-docs/plugins/providers.md b/public-docs/plugins/providers.md index 2f4951669c..6fbe766c61 100644 --- a/public-docs/plugins/providers.md +++ b/public-docs/plugins/providers.md @@ -306,6 +306,52 @@ 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 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: + +```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` 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 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 Publish complete snapshots through `timeline.item`. Reuse an item `id` when updating streamed text, diff --git a/public-docs/plugins/providers.zh-CN.md b/public-docs/plugins/providers.zh-CN.md index d49e559d49..6cc0a2cb14 100644 --- a/public-docs/plugins/providers.zh-CN.md +++ b/public-docs/plugins/providers.zh-CN.md @@ -269,6 +269,38 @@ Paseo 刷新 Agent 的方式是关闭它当前的 Provider 会话，然后用当 只声明你实现了的能力。如果不支持引导，请省略 `prompt.steer`；Paseo 可以改为替换当前轮次。 +## 允许清理空闲运行时 {#permit-idle-runtime-cleanup} + +当 daemon 的基础条件允许清理时，Paseo 可以在可恢复的会话空闲一小时后释放其运行时。在连接和会话的 `session.opened` capabilities 中声明 `session.can_stop_runtime`，即可添加 Provider 自己的否决检查。没有此能力时，会话仍使用相同的基础条件和一小时空闲期，插件不会收到安全查询。 + +在现有的 `send(input)` 处理函数中处理安全查询。请求使用 Provider 自己的会话 ID，包括同一连接服务多个会话的情况： + +```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` 由你的 Provider 会话实现。只有释放运行时不会丢失后台 shell、监控任务、原生子 Agent、排队的完成报告或待处理的自主回复时，才返回 `true`。此查询不得停止或修改会话。仍有工作时返回 `false`。查询拒绝、响应无效或超时也会保留运行时。不要用 `request.completed` 响应此查询。 + +Paseo 始终检查前台工作、权限、排队投递、生命周期操作、工作中的托管或原生后代 Agent，以及调用方尚未完成的本地或远程完成通知订阅。订阅保护覆盖注册、恢复，直到通知投递获确认。Provider 返回 `true` 也不能绕过这些阻止条件。Paseo 会在调用现有的 `session.close` 操作之前重新检查活动。受保护的工作完成后，会重新开始一小时的空闲期。保存的 Provider 持久化数据必须能恢复同一个会话。主机设置 `agents.idleRuntimeCleanup: false` 会禁用所有 Provider 的清理。 + ## 发布时间线条目 {#publish-timeline-items} 通过 `timeline.item` 发布完整的快照。更新流式文本、运行中的工具或待办列表时，复用条目的 `id`。Paseo 会推导出实时增量，并保持正常的时间线。