diff --git a/apps/desktop/renderer-architecture.json b/apps/desktop/renderer-architecture.json index ee12ca20f71..c7235b949c0 100644 --- a/apps/desktop/renderer-architecture.json +++ b/apps/desktop/renderer-architecture.json @@ -73,6 +73,7 @@ "src/renderer/locales/settings-web-search-copy.ts", "src/renderer/locales/shell-copy.ts", "src/renderer/locales/shell-remaining-copy.ts", + "src/renderer/locales/storage-retention-copy.ts", "src/renderer/locales/storage-usage-copy.ts", "src/renderer/locales/task-readiness-copy.ts", "src/renderer/locales/work-board-error-copy.ts", @@ -1455,6 +1456,15 @@ "actionFactories": [], "dependencyPaths": {} }, + "src/renderer/locales/storage-retention-copy.ts": { + "bridgePaths": {}, + "environmentCapabilities": {}, + "hookCalls": {}, + "lifecycleMethods": {}, + "unresolvedDependencies": 0, + "actionFactories": [], + "dependencyPaths": {} + }, "src/renderer/locales/storage-usage-copy.ts": { "bridgePaths": {}, "environmentCapabilities": {}, diff --git a/apps/desktop/src/main/runtime-host-renderer-ipc-main.ts b/apps/desktop/src/main/runtime-host-renderer-ipc-main.ts index 54c7f104fea..8f741feed37 100644 --- a/apps/desktop/src/main/runtime-host-renderer-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-renderer-ipc-main.ts @@ -100,6 +100,10 @@ function request( return client.request(operation, HOST_OPERATION_SPECS[operation].decodeInput(value)); case 'scheduled-task.query': return client.request(operation, HOST_OPERATION_SPECS[operation].decodeInput(value)); + case 'storage.retention.query': + return client.request(operation, HOST_OPERATION_SPECS[operation].decodeInput(value)); + case 'storage.retention.set': + return client.request(operation, HOST_OPERATION_SPECS[operation].decodeInput(value)); case 'storage.usage.query': return client.request(operation, HOST_OPERATION_SPECS[operation].decodeInput(value)); case 'storage.usage.sessions.query': diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index f16dde83b26..f4ddf2c9f4f 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -159,6 +159,9 @@ import type { SessionRemovePreviewResult, SessionStorageUsage, StorageUsageQueryResult, + StorageRetentionQueryResult, + StorageRetentionPolicy, + StorageRetentionSetInput, } from '@maka/runtime-host/protocol'; import type { TestProxyInput } from '@maka/core/settings/network-settings'; import type { ExternalSessionImportIpcResult } from './external-session-import-result.js'; @@ -1853,6 +1856,8 @@ export interface MakaBridge { test(input: { provider?: WebSearchProvider; apiKey?: string }, host?: DesktopRuntimeHostRef): Promise; }; storage: { + retentionQuery(host?: DesktopRuntimeHostRef): Promise; + retentionSet(input: StorageRetentionSetInput, host?: DesktopRuntimeHostRef): Promise; /** One Runtime Host's State Root footprint. Read-only; nothing is reclaimed. */ usage(host?: DesktopRuntimeHostRef): Promise; /** diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index f6f70531b6c..a27b37e4a5a 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -260,6 +260,9 @@ import { type SessionRemovePreviewResult, type SessionStorageUsage, type StorageUsageQueryResult, + type StorageRetentionQueryResult, + type StorageRetentionPolicy, + type StorageRetentionSetInput, } from '@maka/runtime-host/protocol'; import type { PlanControlIpcResult } from '../shared/plan-mode-ipc.js'; import { createSessionStorageUsageReader } from './session-storage-usage.js'; @@ -3735,6 +3738,12 @@ const makaBridge = { }, }, storage: { + async retentionQuery(host?: DesktopRuntimeHostRef): Promise { + return scopedRuntimeHost(await selectedRuntimeHostScope(host)).query('storage.retention.query', {}); + }, + async retentionSet(input: StorageRetentionSetInput, host?: DesktopRuntimeHostRef): Promise { + return scopedRuntimeHost(await selectedRuntimeHostScope(host)).command('storage.retention.set', input); + }, async usage(host?: DesktopRuntimeHostRef): Promise { return scopedRuntimeHost(await selectedRuntimeHostScope(host)).query('storage.usage.query', {}); }, diff --git a/apps/desktop/src/preload/runtime-host-renderer-operations.ts b/apps/desktop/src/preload/runtime-host-renderer-operations.ts index d327cb205dd..ce9897b14d5 100644 --- a/apps/desktop/src/preload/runtime-host-renderer-operations.ts +++ b/apps/desktop/src/preload/runtime-host-renderer-operations.ts @@ -28,11 +28,13 @@ export const RENDERER_RUNTIME_HOST_QUERY_OPERATIONS = [ 'execution.inspect.query', 'scheduled-task.query', // Read-only State Root and per-task size measurement; it reclaims nothing. + 'storage.retention.query', 'storage.usage.query', 'storage.usage.sessions.query', ] as const satisfies readonly (keyof OperationSpecMap)[]; export const RENDERER_RUNTIME_HOST_COMMAND_OPERATIONS = [ + 'storage.retention.set', 'daily-review.mutate', 'scheduled-task.mutate', 'web-search.execute', diff --git a/apps/desktop/src/renderer/features/storage-usage/index.ts b/apps/desktop/src/renderer/features/storage-usage/index.ts index f64555059a9..c1dfdce358b 100644 --- a/apps/desktop/src/renderer/features/storage-usage/index.ts +++ b/apps/desktop/src/renderer/features/storage-usage/index.ts @@ -24,3 +24,5 @@ export { StorageUsageSection } from './ui/storage-usage-section.js'; export { TaskStorageSize } from './ui/task-storage-size.js'; export { StorageUsageServicesProvider } from './services-context.js'; export type { StorageUsageHostTarget, StorageUsageServices } from './ports.js'; + +export { StorageRetentionSection } from './ui/storage-retention-section.js'; diff --git a/apps/desktop/src/renderer/features/storage-usage/ports.ts b/apps/desktop/src/renderer/features/storage-usage/ports.ts index dbc4c7235ca..d25990e51bf 100644 --- a/apps/desktop/src/renderer/features/storage-usage/ports.ts +++ b/apps/desktop/src/renderer/features/storage-usage/ports.ts @@ -17,7 +17,7 @@ * under the License. */ -import type { SessionStorageUsage, StorageUsageQueryResult } from '@maka/runtime-host/protocol'; +import type { SessionStorageUsage, StorageUsageQueryResult, StorageRetentionQueryResult, StorageRetentionSetInput, StorageRetentionPolicy } from '@maka/runtime-host/protocol'; /** The Runtime Host a Settings page is pointed at, stated structurally. */ export interface StorageUsageHostTarget { @@ -30,6 +30,8 @@ export interface StorageUsageHostTarget { * measurements. Nothing here reclaims or deletes data. */ export interface StorageUsageServices { + loadRetention?(host: StorageUsageHostTarget): Promise; + setRetention?(host: StorageUsageHostTarget, input: StorageRetentionSetInput): Promise; /** One Runtime Host's State Root footprint. */ loadUsage(host: StorageUsageHostTarget): Promise; /** diff --git a/apps/desktop/src/renderer/features/storage-usage/ui/storage-retention-section.tsx b/apps/desktop/src/renderer/features/storage-usage/ui/storage-retention-section.tsx new file mode 100644 index 00000000000..50d046ac0c4 --- /dev/null +++ b/apps/desktop/src/renderer/features/storage-usage/ui/storage-retention-section.tsx @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { useEffect, useRef, useState } from 'react'; +import { Button, Selector, Switch, formatBytes, useUiLocale } from '@maka/ui'; +import { RETENTION_DAYS, type StorageRetentionQueryResult, type RetentionDays } from '@maka/runtime-host/protocol'; +import { SettingsRow, SettingsSection, useOptionalRuntimeHostSettingsGenerationKey, useOptionalRuntimeHostSettingsTarget } from '../../../application/contracts/settings-presentation/index.js'; +import { getStorageRetentionCopy } from '../../../locales/storage-retention-copy.js'; +import { useOptionalStorageUsageServices } from '../services-context.js'; + +export function StorageRetentionSection() { + const host = useOptionalRuntimeHostSettingsTarget(); + const hostKey = useOptionalRuntimeHostSettingsGenerationKey(); + const services = useOptionalStorageUsageServices(); + const locale = useUiLocale(); const copy = getStorageRetentionCopy(locale); + const [snapshot, setSnapshot] = useState<{ key: string | undefined; result: StorageRetentionQueryResult }>(); + const [failed, setFailed] = useState(false); const [saving, setSaving] = useState(false); const [attempt, setAttempt] = useState(0); + const generation = useRef(0); + useEffect(() => { + const current = ++generation.current; setFailed(false); setSaving(false); + if (!host || !services?.loadRetention) return; + services.loadRetention(host).then(result => { + if (generation.current === current) setSnapshot({ key: hostKey, result }); + }, () => { if (generation.current === current) setFailed(true); }); + return () => { generation.current++; }; + }, [host, hostKey, services, attempt]); + if (!host || !services?.loadRetention || !services.setRetention) return null; + const result = snapshot && snapshot.key === hostKey ? snapshot.result : undefined; + const disabled = !result || saving || failed; + async function save(enabled: boolean, days: RetentionDays) { + if (!host || !result || !services?.setRetention) return; + const current = generation.current; setSaving(true); + try { + const policy = await services.setRetention(host, { enabled, days, expectedRevision: result.policy.revision }); + if (generation.current === current) { + setSnapshot({ key: hostKey, result: { ...result, policy } }); + setAttempt(value => value + 1); + } + } catch { if (generation.current === current) setFailed(true); } + finally { if (generation.current === current) setSaving(false); } + } + const date = (time: number) => new Date(time).toLocaleString(locale); + return setAttempt(value => value + 1)} />}> + void save(enabled, result?.policy.days ?? 30)} />} /> + ({ value: String(days), label: copy.dayOption(days) }))} isDisabled={disabled} onChange={days => void save(result?.policy.enabled ?? false, Number(days) as RetentionDays)} />} /> + + {result?.lastDeletion && } + {result?.lastSweep && } + ; +} diff --git a/apps/desktop/src/renderer/locales/storage-retention-copy.ts b/apps/desktop/src/renderer/locales/storage-retention-copy.ts new file mode 100644 index 00000000000..986b3258905 --- /dev/null +++ b/apps/desktop/src/renderer/locales/storage-retention-copy.ts @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { UiCatalog, UiLocale } from '@maka/core/ui-locale'; +interface StorageRetentionCopy { + title: string; + help: string; + enabled: string; + days: string; + refresh: string; + loading: string; + failed: string; + dayOption: (days: number) => string; + preview: (count: number, date: string) => string; + disabled: string; + cleanup: (count: number, size: string, date: string) => string; + unknownBytes: string; + needsReview: (count: number) => string; +} + +const COPY = { + en: { + title: 'Automatic cleanup on this Host', + help: 'Applies to all archived tasks on this Host. The clock starts when enabled or the duration changes. Pinned tasks are kept.', + enabled: 'Delete archived tasks automatically', days: 'Delete after', refresh: 'Refresh', + loading: 'Loading retention settings…', failed: 'Could not load or save retention settings. Refresh before trying again.', + dayOption: (days: number) => `${days} days`, + preview: (count: number, date: string) => `${count} archived tasks become eligible on ${date}.`, + disabled: 'Automatic deletion is off.', + cleanup: (count: number, size: string, date: string) => `Last automatic cleanup: deleted ${count} tasks (${size}) on ${date}.`, + unknownBytes: 'size unavailable', needsReview: (count: number) => `${count} tasks need manual review.`, + }, + 'zh-CN': { + title: '此 Host 的自动清理', help: '适用于此 Host 的所有已归档任务。启用或修改天数时重新计时，已置顶任务会保留。', + enabled: '自动删除已归档任务', days: '保留时长', refresh: '刷新', loading: '正在读取自动清理设置…', failed: '无法读取或保存设置，请刷新后重试。', + dayOption: (days: number) => `${days} 天`, preview: (count: number, date: string) => `${count} 个已归档任务将在 ${date} 后符合清理条件。`, + disabled: '自动删除已关闭。', cleanup: (count: number, size: string, date: string) => `上次自动清理：${date} 删除了 ${count} 个任务（${size}）。`, + unknownBytes: '大小未知', needsReview: (count: number) => `${count} 个任务需要手动检查。`, + }, + 'zh-TW': { + title: '此 Host 的自動清理', help: '適用於此 Host 的所有已封存任務。啟用或修改天數時重新計時，已置頂任務會保留。', + enabled: '自動刪除已封存任務', days: '保留時長', refresh: '重新整理', loading: '正在讀取自動清理設定…', failed: '無法讀取或儲存設定，請重新整理後重試。', + dayOption: (days: number) => `${days} 天`, preview: (count: number, date: string) => `${count} 個已封存任務將在 ${date} 後符合清理條件。`, + disabled: '自動刪除已關閉。', cleanup: (count: number, size: string, date: string) => `上次自動清理：${date} 刪除了 ${count} 個任務（${size}）。`, + unknownBytes: '大小未知', needsReview: (count: number) => `${count} 個任務需要手動檢查。`, + }, +} satisfies UiCatalog; +export function getStorageRetentionCopy(locale: UiLocale) { return COPY[locale]; } diff --git a/apps/desktop/src/renderer/platform/desktop/create-storage-usage-services.ts b/apps/desktop/src/renderer/platform/desktop/create-storage-usage-services.ts index d38bfff68bb..07e6442f092 100644 --- a/apps/desktop/src/renderer/platform/desktop/create-storage-usage-services.ts +++ b/apps/desktop/src/renderer/platform/desktop/create-storage-usage-services.ts @@ -27,6 +27,8 @@ export function createDesktopStorageUsageServices( bridge: DesktopStorageUsageBridge = window.maka, ): StorageUsageServices { return { + loadRetention: (host) => bridge.storage.retentionQuery(host), + setRetention: (host, input) => bridge.storage.retentionSet(input, host), loadUsage: (host) => bridge.storage.usage(host), // No host argument: each task is measured by the Host that holds it, and // the bridge routes by the projected id for exactly that reason. diff --git a/apps/desktop/src/renderer/settings/settings-nav.ts b/apps/desktop/src/renderer/settings/settings-nav.ts index 47b830b3333..ae53f7a35c7 100644 --- a/apps/desktop/src/renderer/settings/settings-nav.ts +++ b/apps/desktop/src/renderer/settings/settings-nav.ts @@ -113,7 +113,7 @@ const SETTINGS_SECTION_SCOPES: Record< 'bot-chat': 'client', search: 'runtime-host', usage: 'runtime-host', - 'archived-tasks': 'client', + 'archived-tasks': 'mixed', 'import-tasks': 'runtime-host', 'daily-review': 'runtime-host', data: 'mixed', diff --git a/apps/desktop/src/renderer/settings/tasks-settings-page.tsx b/apps/desktop/src/renderer/settings/tasks-settings-page.tsx index a2c868b53a8..ec1cdb2f8fa 100644 --- a/apps/desktop/src/renderer/settings/tasks-settings-page.tsx +++ b/apps/desktop/src/renderer/settings/tasks-settings-page.tsx @@ -31,7 +31,7 @@ import type { DesktopSessionSummary } from '../../preload/bridge-contract.js'; import type { SessionCatalogController } from '../application/contracts/session-catalog/session-catalog-state.js'; import { getSettingsTasksCopy } from '../locales/settings-tasks-copy.js'; import { getStorageUsageCopy } from '../locales/storage-usage-copy.js'; -import { TaskStorageSize } from '../features/storage-usage/index.js'; +import { TaskStorageSize, StorageRetentionSection } from '../features/storage-usage/index.js'; import { SettingsPage, SettingsSection } from './settings-section'; import { isOrphanedSubagentTask } from './task-catalog-rows'; @@ -89,6 +89,7 @@ export function TasksSettingsPage( if (props.scope.rows.length === 0) { return ( + ); @@ -96,6 +97,7 @@ export function TasksSettingsPage( return ( + {props.scope.controls} {visible.length === 0 ? ( diff --git a/apps/desktop/stories/settings/settings-pages.stories.tsx b/apps/desktop/stories/settings/settings-pages.stories.tsx index 473b720c6f2..c6fed257299 100644 --- a/apps/desktop/stories/settings/settings-pages.stories.tsx +++ b/apps/desktop/stories/settings/settings-pages.stories.tsx @@ -55,6 +55,8 @@ import { AppUpdateServicesProvider, type AppUpdateServices, } from '../../src/renderer/features/app-update/index.js'; +import { StorageUsageServicesProvider, type StorageUsageServices } from '../../src/renderer/features/storage-usage/index.js'; +import type { StorageRetentionPolicy } from '@maka/runtime-host/protocol'; import type { SessionSummary } from '@maka/core/session'; import { revisionFamilySessionIds } from '@maka/core/session-revisions'; import type { @@ -1970,11 +1972,27 @@ function fieldChrome(element: HTMLElement) { * reach a provider its own component renders. */ function SettingsStory(props: SettingsStoryProps) { + // IO fixture for the actual production feature, on the existing Settings frame. + const [storageServices] = useState(() => { + let policy: StorageRetentionPolicy = { enabled: false, days: 30, enabledAt: null, revision: 0 }; + return { + loadUsage: async () => ({ measuredAt: Date.now(), totals: [], reclaimableBytes: 0, worktreeCount: 0 }), + loadSessionUsage: async () => ({}), + loadRetention: async () => ({ policy, preview: { count: policy.enabled ? archivedTaskSessions.length : 0, eligibleAt: policy.enabledAt === null ? null : policy.enabledAt + policy.days * 24 * 60 * 60 * 1000 }, lastSweep: null, lastDeletion: null }), + setRetention: async (_host, input) => { + if (input.expectedRevision !== policy.revision) throw new Error('Retention revision changed'); + policy = { enabled: input.enabled, days: input.days, enabledAt: input.enabled ? Date.now() : null, revision: policy.revision + 1 }; + return policy; + }, + }; + }); return ( - + + + diff --git a/docs/astryx-surface-file-inventory.md b/docs/astryx-surface-file-inventory.md index cfdb529ebce..1e0dbc9621d 100644 --- a/docs/astryx-surface-file-inventory.md +++ b/docs/astryx-surface-file-inventory.md @@ -6,7 +6,7 @@ Generated against `@astryxdesign/core@0.6.3` (195 component exports). Wiki bar: Design Conventions · API Use-the-System · Theming · Container Padding. -**Totals:** 322 files — blocker 0, reimplementation 0, polish 4, aligned 318. +**Totals:** 323 files — blocker 0, reimplementation 0, polish 4, aligned 319. ## Exclusions (explicit) @@ -119,6 +119,7 @@ Wiki bar: Design Conventions · API Use-the-System · Theming · Container Paddi | `apps/desktop/src/renderer/features/session-settings/services-context.tsx` | shell-chrome-or-panel | none | aligned — no raw controls; no Astryx JSX usage | aligned | | `apps/desktop/src/renderer/features/session-settings/ui/session-settings-provider.tsx` | shell-chrome-or-panel | none | aligned — no raw controls; no Astryx JSX usage | aligned | | `apps/desktop/src/renderer/features/storage-usage/services-context.tsx` | other | none | aligned — no raw controls; no Astryx JSX usage | aligned | +| `apps/desktop/src/renderer/features/storage-usage/ui/storage-retention-section.tsx` | other | Button, Selector, Switch | aligned — uses Astryx (Button, Selector, Switch) | aligned | | `apps/desktop/src/renderer/features/storage-usage/ui/storage-usage-section.tsx` | other | Banner, Button | aligned — uses Astryx (Banner, Button) | aligned | | `apps/desktop/src/renderer/features/storage-usage/ui/task-storage-size.tsx` | other | Text | aligned — uses Astryx (Text) | aligned | | `apps/desktop/src/renderer/features/task-entry/services-context.tsx` | other | none | aligned — no raw controls; no Astryx JSX usage | aligned | diff --git a/docs/astryx-surface-file-inventory.paths b/docs/astryx-surface-file-inventory.paths index c6ce7a0c1ec..d9f05f25846 100644 --- a/docs/astryx-surface-file-inventory.paths +++ b/docs/astryx-surface-file-inventory.paths @@ -89,6 +89,7 @@ apps/desktop/src/renderer/features/session-navigation/ui/session-navigation-prov apps/desktop/src/renderer/features/session-settings/services-context.tsx apps/desktop/src/renderer/features/session-settings/ui/session-settings-provider.tsx apps/desktop/src/renderer/features/storage-usage/services-context.tsx +apps/desktop/src/renderer/features/storage-usage/ui/storage-retention-section.tsx apps/desktop/src/renderer/features/storage-usage/ui/storage-usage-section.tsx apps/desktop/src/renderer/features/storage-usage/ui/task-storage-size.tsx apps/desktop/src/renderer/features/task-entry/services-context.tsx diff --git a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts index 67c8900143e..d8c34181de6 100644 --- a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts @@ -335,6 +335,8 @@ describe('Runtime Host operator commands', () => { assert.equal(resolved.operationGrants.includes('turn.start'), true); assert.equal(resolved.operationGrants.includes('session.prompt-suggestion.generate'), true); assert.equal(resolved.operationGrants.includes('project.catalog.query'), true); + assert.equal(resolved.operationGrants.includes('storage.retention.query'), true); + assert.equal(resolved.operationGrants.includes('storage.retention.set'), true); } assert.equal(desktop.canPublishClientCapabilities, true); assert.equal(desktop.operationGrants.includes('client.capability.replace'), true); diff --git a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts index ac9272b7cef..15ba71eb8f2 100644 --- a/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-retirement-coordinator.test.ts @@ -57,6 +57,142 @@ const CONNECTION_CONTEXT: ConnectionContext = { }; describe('Host Session retirement coordinator', () => { + test('retention uses family pins, exact deadlines, current archive time, and Host retirement guards', async () => { + await withHarness(async (harness) => { + await harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: harness.rootId, state: 'archived' }, + CONNECTION_CONTEXT, + ); + const archivedAt = (await harness.store.readCatalogRecord(harness.rootId)).summary + .archivedAt!; + const policy = { enabled: true, days: 30 as const, enabledAt: archivedAt, revision: 1 }; + const deadline = archivedAt + 30 * 24 * 60 * 60 * 1000; + const remove = async () => + harness.coordinator.removeForRetention( + { + sessionId: harness.rootId, + expectedRevision: (await harness.store.readHeaderRecordSnapshot(harness.rootId)) + .revision, + }, + policy, + deadline, + ); + harness.now = deadline; + assert.equal(await remove(), 'kept'); + harness.now++; + await harness.store.updateHeader(harness.revisionId, { isFlagged: true }); + assert.equal(await remove(), 'kept'); + assert.deepEqual( + (await harness.store.listRetentionCandidates({ cutoff: deadline, limit: 8 })).sessionIds, + [], + ); + await harness.store.updateHeader(harness.revisionId, { isFlagged: false }); + const candidates = await harness.store.listRetentionCandidates({ + cutoff: deadline, + limit: 8, + }); + assert.equal(candidates.sessionIds.length, 1); + harness.blockers.message.add(harness.rootId); + assert.equal(await remove(), 'busy'); + harness.blockers.message.clear(); + harness.now = deadline - 1; + assert.equal(await remove(), 'kept'); + harness.now = deadline + 1; + assert.equal(await remove(), 'removed'); + assert.equal((await harness.store.probeSessionRemoval(harness.rootId)).kind, 'removed'); + }); + }); + + test('retention never reclaims a bound worktree', async () => { + await withHarness(async (harness) => { + const childId = await createClosedSubagent(harness, harness.rootId, 98, { + schemaVersion: 1, + kind: 'git_worktree', + leaseId: `subagent_worktree_${'a'.repeat(32)}`, + gitCommonDir: '/tmp/project/.git', + worktreePath: '/tmp/maka-subagent-worktree', + branch: `maka/subagent/${'a'.repeat(32)}`, + baseCommit: 'b'.repeat(40), + }); + await harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: childId, state: 'archived' }, + CONNECTION_CONTEXT, + ); + const archivedAt = (await harness.store.readCatalogRecord(childId)).summary.archivedAt!; + harness.now = archivedAt + 31 * 24 * 60 * 60 * 1000; + const target = await harness.store.readHeaderRecordSnapshot(childId); + assert.equal( + await harness.coordinator.removeForRetention( + { sessionId: childId, expectedRevision: target.revision }, + { enabled: true, days: 30, enabledAt: archivedAt, revision: 1 }, + ), + 'needs_review', + ); + assert.equal((await harness.store.probeSessionRemoval(childId)).kind, 'present'); + }); + }); + + test('retention restores cancel deadlines and re-archive starts fresh', async (t) => { + t.mock.timers.enable({ apis: ['Date'], now: 1_000 }); + await withHarness(async (harness) => { + const lifecycle = async (state: 'active' | 'archived') => + harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: harness.rootId, state }, + CONNECTION_CONTEXT, + ); + await lifecycle('archived'); + const policy = { enabled: true, days: 30 as const, enabledAt: 1_000, revision: 1 }; + const remove = async () => + harness.coordinator.removeForRetention( + { + sessionId: harness.rootId, + expectedRevision: (await harness.store.readHeaderRecordSnapshot(harness.rootId)) + .revision, + }, + policy, + ); + t.mock.timers.tick(30 * 24 * 60 * 60 * 1000 + 1); + await lifecycle('active'); + assert.equal(await remove(), 'kept'); + await lifecycle('archived'); + assert.equal(await remove(), 'kept'); + t.mock.timers.tick(30 * 24 * 60 * 60 * 1000 + 1); + assert.equal(await remove(), 'removed'); + }); + }); + test('retention keeps active subtasks for review and excludes parented subtasks from candidates', async () => { + await withHarness(async (harness) => { + await harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: harness.rootId, state: 'archived' }, + CONNECTION_CONTEXT, + ); + const childId = await createClosedSubagent(harness, harness.rootId, 99); + const archivedAt = (await harness.store.readCatalogRecord(harness.rootId)).summary + .archivedAt!; + harness.now = archivedAt + 31 * 24 * 60 * 60 * 1000; + const policy = { enabled: true, days: 30 as const, enabledAt: archivedAt, revision: 1 }; + const target = await harness.store.readHeaderRecordSnapshot(harness.rootId); + assert.equal( + await harness.coordinator.removeForRetention( + { sessionId: harness.rootId, expectedRevision: target.revision }, + policy, + ), + 'needs_review', + ); + assert.equal((await harness.store.readHeaderSnapshot(childId)).isArchived, false); + await harness.coordinator.handlers['session.lifecycle.set']( + { sessionId: childId, state: 'archived' }, + CONNECTION_CONTEXT, + ); + assert.equal( + ( + await harness.store.listRetentionCandidates({ cutoff: harness.now, limit: 8 }) + ).sessionIds.includes(childId), + false, + ); + }); + }); + test('retirement finalizes admitted patches and fences late artifact publication', async (t) => { await withHarness(async (harness) => { const capability = await resolveStorageRoot({ diff --git a/packages/runtime-host/src/__tests__/storage-retention-coordinator.test.ts b/packages/runtime-host/src/__tests__/storage-retention-coordinator.test.ts new file mode 100644 index 00000000000..aa2afa34485 --- /dev/null +++ b/packages/runtime-host/src/__tests__/storage-retention-coordinator.test.ts @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import type { ExecutionSessionWriter } from '@maka/storage/execution-stores'; +import { + HostStorageRetentionPolicy, + RETENTION_DAY_MS, +} from '../server/storage-retention-policy.js'; +import { HostStorageRetentionCoordinator } from '../server/storage-retention-coordinator.js'; + +const context = { + hostEpoch: 'test', + connectionId: 'test', + principal: 'local_os_user' as const, + acquireResidency: () => ({ release() {} }), +}; +test('retention does no candidate SQL before opt-in deadline, pauses rollback across restart, and bounds each tick', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-retention-')); + t.after(() => rm(root, { recursive: true, force: true })); + let now = 1_000; + let queries = 0; + let deleted = 0; + const policy = await HostStorageRetentionPolicy.open(root, () => now); + const options = { + stateRoot: root, + policy, + now: () => now, + stores: { + listRetentionCandidates: async (input: { limit: number; cutoff: number; after?: string }) => { + queries++; + assert.equal(input.limit, 8); + return { + sessionIds: Array.from({ length: 8 }, (_, i) => String(i)), + hasMore: input.cutoff > 1_000, + }; + }, + readCatalogRecord: async () => + ({ revision: 1, summary: {} }) as Awaited< + ReturnType + >, + }, + retirement: { + estimateRetentionBytes: async () => undefined, + removeForRetention: async () => { + deleted++; + return 'removed' as const; + }, + }, + }; + let coordinator = await HostStorageRetentionCoordinator.open(options); + assert.equal(await coordinator.run({ maxFamilies: 8 }), false); + assert.equal(queries, 0); + await policy.set({ enabled: true, days: 30, expectedRevision: 0 }); + now += 30 * RETENTION_DAY_MS; + assert.equal(await coordinator.run({ maxFamilies: 8 }), false); + assert.equal(queries, 0); + now++; + assert.equal(await coordinator.run({ maxFamilies: 8 }), true); + assert.equal(deleted, 8); + now--; + coordinator = await HostStorageRetentionCoordinator.open(options); + assert.equal(await coordinator.run({ maxFamilies: 8 }), false); + assert.equal(queries, 1); + const result = await coordinator.handlers['storage.retention.query']({}, context); + assert.equal(result.ok, true); + if (result.ok) { + assert.equal(result.result.lastSweep?.deleted, 8); + assert.equal(result.result.lastDeletion?.count, 8); + } + coordinator.beginDrain(); + assert.equal(await coordinator.run({ maxFamilies: 8 }), false); +}); + +test('changing a policy while candidates are read rejects the old sweep', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-retention-')); + t.after(() => rm(root, { recursive: true, force: true })); + let now = 1_000; + const policy = await HostStorageRetentionPolicy.open(root, () => now); + await policy.set({ enabled: true, days: 30, expectedRevision: 0 }); + now += 30 * RETENTION_DAY_MS + 1; + const coordinator = await HostStorageRetentionCoordinator.open({ + stateRoot: root, + policy, + now: () => now, + stores: { + listRetentionCandidates: async () => { + await policy.set({ enabled: false, days: 30, expectedRevision: 1 }); + return { sessionIds: ['candidate'], hasMore: false }; + }, + readCatalogRecord: async () => + ({ revision: 1, summary: {} }) as Awaited< + ReturnType + >, + }, + retirement: { + estimateRetentionBytes: async () => undefined, + removeForRetention: async () => assert.fail('stale policy must not delete'), + }, + }); + assert.equal(await coordinator.run({ maxFamilies: 8 }), false); +}); diff --git a/packages/runtime-host/src/__tests__/storage-retention-policy.test.ts b/packages/runtime-host/src/__tests__/storage-retention-policy.test.ts new file mode 100644 index 00000000000..38ea6e795c0 --- /dev/null +++ b/packages/runtime-host/src/__tests__/storage-retention-policy.test.ts @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { + HostStorageRetentionPolicy, + retentionDeadline, + RETENTION_DAY_MS, +} from '../server/storage-retention-policy.js'; + +test('retention defaults off, persists a Host clock, and changes restart aging', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-retention-')); + t.after(() => rm(root, { recursive: true, force: true })); + let now = 10_000; + const store = await HostStorageRetentionPolicy.open(root, () => now); + assert.equal(retentionDeadline(store.snapshot(), 0), null); + const policy = await store.set({ enabled: true, days: 30, expectedRevision: 0 }); + assert.equal(policy?.enabledAt, now); + assert.equal(retentionDeadline(policy!, 0), now + 30 * RETENTION_DAY_MS); + assert.equal(retentionDeadline(policy!), now + 30 * RETENTION_DAY_MS); + assert.equal(retentionDeadline(policy!, now + 100), now + 100 + 30 * RETENTION_DAY_MS); + assert.equal(await store.set({ enabled: false, days: 30, expectedRevision: 0 }), undefined); + now += 1_000; + const changed = await store.set({ enabled: true, days: 60, expectedRevision: 1 }); + assert.equal(changed?.enabledAt, now); + assert.deepEqual((await HostStorageRetentionPolicy.open(root)).snapshot(), changed); + const disabled = await store.set({ enabled: false, days: 60, expectedRevision: 2 }); + assert.equal(disabled?.enabledAt, null); + assert.equal(retentionDeadline(disabled!), null); +}); + +test('policy changes serialize with admitted removal and reject stale sweeps', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-retention-')); + t.after(() => rm(root, { recursive: true, force: true })); + const store = await HostStorageRetentionPolicy.open(root, () => 100); + await store.set({ enabled: true, days: 30, expectedRevision: 0 }); + let finish!: () => void; + const barrier = new Promise((resolve) => { + finish = resolve; + }); + const removing = store.withCurrent(1, async () => { + await barrier; + return 'removed'; + }); + const disabling = store.set({ enabled: false, days: 30, expectedRevision: 1 }); + await Promise.resolve(); + assert.equal(store.snapshot().enabled, true); + finish(); + assert.equal(await removing, 'removed'); + await disabling; + assert.equal(await store.withCurrent(1, async () => assert.fail('stale removal')), undefined); +}); + +test('malformed destructive opt-in never silently enables retention', async (t) => { + const root = await mkdtemp(join(tmpdir(), 'maka-retention-')); + t.after(() => rm(root, { recursive: true, force: true })); + await writeFile(join(root, 'storage-retention.json'), '{"enabled":true}'); + await assert.rejects(HostStorageRetentionPolicy.open(root)); +}); diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 27e6bffdc56..a09fb59133c 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -86,6 +86,7 @@ export * from './external-session.js'; export * from './message.js'; export * from './queue-mutation.js'; export * from './operations.js'; +export * from './storage-retention.js'; export * from './runtime-resource.js'; export * from './session-continuity.js'; export * from './session-catalog-change.js'; @@ -104,7 +105,8 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 202 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 204 as const; +// 203: Host-owned opt-in retention query/set and automatic archived-task cleanup. // 202: `session.remove.preview` takes a bounded list of Sessions and reports the // child tasks, worktrees and optionally the bytes their removal would delete; // `session.remove` takes `requireArchivedForMs` and may answer `too_recent`. diff --git a/packages/runtime-host/src/protocol/operations.ts b/packages/runtime-host/src/protocol/operations.ts index 308ed53848d..4dbc4bec7fa 100644 --- a/packages/runtime-host/src/protocol/operations.ts +++ b/packages/runtime-host/src/protocol/operations.ts @@ -64,6 +64,7 @@ import { SESSION_RETIREMENT_OPERATION_SPECS } from './session-retirement.js'; import { PROMPT_SUGGESTION_OPERATION_SPECS } from './prompt-suggestions.js'; import { SESSION_EFFECT_OPERATION_SPECS } from './session-effects.js'; import { SKILL_CATALOG_OPERATION_SPECS } from './skill-catalog.js'; +import { STORAGE_RETENTION_OPERATION_SPECS } from './storage-retention.js'; import { STORAGE_USAGE_OPERATION_SPECS } from './storage-usage.js'; import { TURN_OPERATION_SPECS } from './turn.js'; import { USAGE_PRICING_OPERATION_SPECS } from './usage-pricing.js'; @@ -234,6 +235,7 @@ export const HOST_OPERATION_SPECS = composeOperationSpecMaps( SKILL_CATALOG_OPERATION_SPECS, USAGE_PRICING_OPERATION_SPECS, STORAGE_USAGE_OPERATION_SPECS, + STORAGE_RETENTION_OPERATION_SPECS, MEMORY_OPERATION_SPECS, OAUTH_OPERATION_SPECS, EXTERNAL_AGENT_SETUP_OPERATION_SPECS, @@ -359,6 +361,8 @@ export const REMOTE_OWNER_OPERATION_GRANTS = Object.freeze([ 'skill.catalog.mutate', 'skill.catalog.preview-update', 'skill.catalog.query', + 'storage.retention.query', + 'storage.retention.set', 'storage.usage.query', 'storage.usage.sessions.query', 'subscription.close', diff --git a/packages/runtime-host/src/protocol/storage-retention.ts b/packages/runtime-host/src/protocol/storage-retention.ts new file mode 100644 index 00000000000..0490a3c4f64 --- /dev/null +++ b/packages/runtime-host/src/protocol/storage-retention.ts @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { requireCount, requireExactRecord } from './codec.js'; +import { invalidProtocolFrame } from './errors.js'; +import { defineOperation } from './operation-spec.js'; +export const RETENTION_DAYS = [30, 60, 90] as const; +export type RetentionDays = (typeof RETENTION_DAYS)[number]; +export interface StorageRetentionPolicy { + readonly enabled: boolean; + readonly days: RetentionDays; + readonly enabledAt: number | null; + readonly revision: number; +} +export interface StorageRetentionSetInput { + readonly enabled: boolean; + readonly days: RetentionDays; + readonly expectedRevision: number; +} +export interface StorageRetentionSweep { + readonly at: number; + readonly deleted: number; + readonly busy: number; + readonly needsReview: number; + readonly failed: number; +} +export interface StorageRetentionQueryResult { + readonly policy: StorageRetentionPolicy; + readonly preview: { readonly count: number; readonly eligibleAt: number | null }; + readonly lastSweep: StorageRetentionSweep | null; + readonly lastDeletion: { + readonly at: number; + readonly count: number; + readonly estimatedBytes: number | null; + } | null; +} +const ERRORS = [ + 'host_not_ready', + 'host_draining', + 'operation_unavailable', + 'persistence_failed', + 'operation_conflict', + 'internal_failure', +] as const; +export const STORAGE_RETENTION_OPERATION_SPECS = { + 'storage.retention.query': defineOperation< + Record, + StorageRetentionQueryResult, + (typeof ERRORS)[number] + >({ + mode: 'query', + availability: 'ready', + errors: ERRORS, + decodeInput: (value) => { + requireExactRecord(value, 'retention query', []); + return {}; + }, + decodeOutput: decodeStorageRetentionQueryResult, + }), + 'storage.retention.set': defineOperation< + StorageRetentionSetInput, + StorageRetentionPolicy, + (typeof ERRORS)[number] + >({ + mode: 'command', + availability: 'ready', + errors: ERRORS, + decodeInput: (value) => { + const item = requireExactRecord(value, 'retention set', [ + 'expectedRevision', + 'enabled', + 'days', + ]); + return { + enabled: bool(item.enabled), + days: days(item.days), + expectedRevision: requireCount(item.expectedRevision, 'retention revision'), + }; + }, + decodeOutput: decodeStorageRetentionPolicy, + }), +} as const; +function bool(value: unknown): boolean { + if (typeof value !== 'boolean') throw invalidProtocolFrame('Invalid retention enabled'); + return value; +} +function days(value: unknown): RetentionDays { + if (!RETENTION_DAYS.includes(value as RetentionDays)) + throw invalidProtocolFrame('Invalid retention days'); + return value as RetentionDays; +} +function nullableCount(value: unknown): number | null { + return value === null ? null : requireCount(value, 'retention count'); +} +export function decodeStorageRetentionPolicy(value: unknown): StorageRetentionPolicy { + const item = requireExactRecord(value, 'retention policy', [ + 'enabled', + 'days', + 'enabledAt', + 'revision', + ]); + const enabled = bool(item.enabled); + const enabledAt = nullableCount(item.enabledAt); + if (enabled !== (enabledAt !== null)) throw invalidProtocolFrame('Invalid retention timestamp'); + return { + enabled, + enabledAt, + days: days(item.days), + revision: requireCount(item.revision, 'retention revision'), + }; +} +export function decodeStorageRetentionQueryResult(value: unknown): StorageRetentionQueryResult { + const item = requireExactRecord(value, 'retention result', [ + 'policy', + 'preview', + 'lastSweep', + 'lastDeletion', + ]); + const preview = requireExactRecord(item.preview, 'retention preview', ['count', 'eligibleAt']); + const sweep = + item.lastSweep === null + ? null + : requireExactRecord(item.lastSweep, 'retention sweep', [ + 'at', + 'deleted', + 'busy', + 'needsReview', + 'failed', + ]); + const deletion = + item.lastDeletion === null + ? null + : requireExactRecord(item.lastDeletion, 'retention deletion', [ + 'at', + 'count', + 'estimatedBytes', + ]); + return { + policy: decodeStorageRetentionPolicy(item.policy), + preview: { + count: requireCount(preview.count, 'preview count'), + eligibleAt: nullableCount(preview.eligibleAt), + }, + lastSweep: + sweep === null + ? null + : { + at: requireCount(sweep.at, 'sweep time'), + deleted: requireCount(sweep.deleted, 'deleted'), + busy: requireCount(sweep.busy, 'busy'), + needsReview: requireCount(sweep.needsReview, 'needs review'), + failed: requireCount(sweep.failed, 'failed'), + }, + lastDeletion: + deletion === null + ? null + : { + at: requireCount(deletion.at, 'deletion time'), + count: requireCount(deletion.count, 'deletion count'), + estimatedBytes: nullableCount(deletion.estimatedBytes), + }, + }; +} diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index 6e0a2ccc164..3e86c710deb 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -241,6 +241,8 @@ import { SessionAdmissionGate } from './session-admission-gate.js'; import { HostSessionCatalogCoordinator } from './session-catalog-coordinator.js'; import { HostWorkspaceResolver } from './workspace-resolver.js'; import { HostSessionRetirementCoordinator } from './session-retirement-coordinator.js'; +import { HostStorageRetentionPolicy } from './storage-retention-policy.js'; +import { HostStorageRetentionCoordinator } from './storage-retention-coordinator.js'; import { HostStorageMaintenance } from './storage-maintenance.js'; import { HostStorageUsageCoordinator } from './storage-usage-coordinator.js'; import { HostSessionRevisionCoordinator } from './session-revision-coordinator.js'; @@ -2845,7 +2847,14 @@ export async function createExecutionRuntimeHostComposition( ); let recoverySessions: Awaited> = []; const storageUsage = new HostStorageUsageCoordinator({ footprint: storage.footprint }); + const storageRetention = await HostStorageRetentionCoordinator.open({ + policy: await HostStorageRetentionPolicy.open(context.owner.capability.canonicalPath), + stores: stores.sessionStore, + retirement: sessionRetirement, + stateRoot: context.owner.capability.canonicalPath, + }); const storageMaintenance = new HostStorageMaintenance({ + retention: storageRetention, artifacts: openedArtifactStore, contextOffload: openedContextOffloadStore, onError: (name, error) => @@ -2857,6 +2866,11 @@ export async function createExecutionRuntimeHostComposition( drain: [() => storageMaintenance.beginDrain()], close: [() => storageMaintenance.close()], }), + createRuntimeHostDomainModule({ + id: 'storage-retention', + handlers: [storageRetention.handlers], + drain: [() => storageRetention.beginDrain()], + }), createRuntimeHostDomainModule({ id: 'storage-usage', handlers: [storageUsage.handlers], diff --git a/packages/runtime-host/src/server/operation-dispatcher.ts b/packages/runtime-host/src/server/operation-dispatcher.ts index c87bc7058b4..f98915b608f 100644 --- a/packages/runtime-host/src/server/operation-dispatcher.ts +++ b/packages/runtime-host/src/server/operation-dispatcher.ts @@ -68,6 +68,7 @@ import { SESSION_TODO_OPERATION_SPECS } from '../protocol/session-todo.js'; import { SESSION_TRANSCRIPT_OPERATION_SPECS } from '../protocol/session-transcript.js'; import { SESSION_TURNS_OPERATION_SPECS } from '../protocol/session-turns.js'; import { SKILL_CATALOG_OPERATION_SPECS } from '../protocol/skill-catalog.js'; +import { STORAGE_RETENTION_OPERATION_SPECS } from '../protocol/storage-retention.js'; import { STORAGE_USAGE_OPERATION_SPECS } from '../protocol/storage-usage.js'; import { TURN_OPERATION_SPECS } from '../protocol/turn.js'; import { USAGE_PRICING_OPERATION_SPECS } from '../protocol/usage-pricing.js'; @@ -417,3 +418,8 @@ async function dispatchTypedOperation( } return response; } + +export type StorageRetentionOperationHandlerMap = Pick< + OperationHandlerMap, + keyof typeof STORAGE_RETENTION_OPERATION_SPECS +>; diff --git a/packages/runtime-host/src/server/session-retirement-coordinator.ts b/packages/runtime-host/src/server/session-retirement-coordinator.ts index 9c23a463040..61031b7e167 100644 --- a/packages/runtime-host/src/server/session-retirement-coordinator.ts +++ b/packages/runtime-host/src/server/session-retirement-coordinator.ts @@ -62,6 +62,7 @@ import { type SessionAdmissionLease, SessionAdmissionGate } from './session-admi import type { SessionContinuityCoordinator } from './session-continuity-coordinator.js'; import type { RootTurnCoordinator } from './root-turn-coordinator.js'; import type { HostRuntimeResourceCoordinator } from './runtime-resource-coordinator.js'; +import { retentionDeadline, type StorageRetentionPolicy } from './storage-retention-policy.js'; import { purgeSessionSidecars } from './session-sidecar-purge.js'; import type { MemoryExtractionSessionLane } from './memory-extraction-session-lane.js'; @@ -316,7 +317,79 @@ export class HostSessionRetirementCoordinator { } } - async #remove(input: SessionRemoveInput): Promise> { + async estimateRetentionBytes(sessionId: string): Promise { + const plan = await this.#readRemovalPlanSessionIds(sessionId); + return this.#measureRemoved(plan.removeSessionIds); + } + + /** Host-internal unattended removal; clients cannot weaken these guards. */ + async removeForRetention( + input: SessionRemoveInput, + policy: StorageRetentionPolicy, + clockFloor = policy.enabledAt ?? 0, + ): Promise<'removed' | 'kept' | 'needs_review' | 'busy' | 'failed'> { + const disposition: { value: 'eligible' | 'kept' | 'needs_review' } = { value: 'eligible' }; + const outcome = await this.#remove(input, { + policy, + clockFloor, + report: (value) => { + disposition.value = value; + }, + }); + if (disposition.value === 'needs_review') return 'needs_review'; + if (!outcome.ok) + return outcome.error.code === 'session_busy' + ? 'busy' + : ['persistence_failed', 'commit_outcome_unknown', 'internal_failure'].includes( + outcome.error.code, + ) + ? 'failed' + : 'kept'; + if (outcome.result.kind === 'removed') return 'removed'; + return 'kept'; + } + + async #retentionDisposition( + plan: StableRemovalPlan, + policy: StorageRetentionPolicy, + clockFloor = policy.enabledAt ?? 0, + ): Promise<'eligible' | 'kept' | 'needs_review'> { + if (!policy.enabled) return 'kept'; + if (plan.archive.sessionIds.length > 0) return 'needs_review'; + for (const id of plan.remove.sessionIds) { + const record = await this.#stores.readCatalogRecord(id); + if (!record.header.isArchived || record.header.isFlagged) return 'kept'; + if (record.header.subagentWorkspace) return 'needs_review'; + const deadline = retentionDeadline(policy, record.summary.archivedAt); + if (deadline === null || this.#now() <= deadline) return 'kept'; + } + return 'eligible'; + } + + async #checkRetention( + plan: StableRemovalPlan, + retention: { + policy: StorageRetentionPolicy; + clockFloor: number; + report: (value: 'eligible' | 'kept' | 'needs_review') => void; + }, + ): Promise<'eligible' | 'kept' | 'needs_review'> { + const disposition = + this.#now() < retention.clockFloor + ? 'kept' + : await this.#retentionDisposition(plan, retention.policy); + retention.report(disposition); + return disposition; + } + + async #remove( + input: SessionRemoveInput, + retention?: { + policy: StorageRetentionPolicy; + clockFloor: number; + report: (value: 'eligible' | 'kept' | 'needs_review') => void; + }, + ): Promise> { let probe; try { probe = await this.#stores.probeSessionRemoval(input.sessionId); @@ -324,6 +397,7 @@ export class HostSessionRetirementCoordinator { return removeFailure('persistence_failed', 'Session removal state is unavailable'); } if (probe.kind === 'removed') { + if (retention) return removeOutcome({ kind: 'too_recent', sessionId: input.sessionId }); try { this.#scheduleCleanup( await this.#stores.listPendingSessionRetirementCleanupIds(input.sessionId), @@ -354,6 +428,9 @@ export class HostSessionRetirementCoordinator { } } + if (retention && (await this.#checkRetention(plan, retention)) !== 'eligible') { + return removeOutcome({ kind: 'too_recent', sessionId: input.sessionId }); + } let removeHandles: RetirementHandles | undefined; let archiveHandles: RetirementHandles | undefined; let committed = false; @@ -367,6 +444,13 @@ export class HostSessionRetirementCoordinator { await this.#disposeBackends(allSessionIds); const committableRemove = await this.#refreshFamilyRecords(plan.remove); const committableArchive = await this.#refreshFamilyRecords(plan.archive); + if (retention && (await this.#checkRetention(plan, retention)) !== 'eligible') { + removeHandles.goal.rollback(); + removeHandles.scheduledTasks.rollback(); + archiveHandles?.goal.rollback(); + archiveHandles?.scheduledTasks.rollback(); + return removeOutcome({ kind: 'too_recent', sessionId: input.sessionId }); + } const removedSessionIds = await this.#stores.removeSessionsVersioned( versionedFamily(committableRemove), versionedFamily(committableArchive), diff --git a/packages/runtime-host/src/server/storage-maintenance.ts b/packages/runtime-host/src/server/storage-maintenance.ts index 629380f5659..56d646d6984 100644 --- a/packages/runtime-host/src/server/storage-maintenance.ts +++ b/packages/runtime-host/src/server/storage-maintenance.ts @@ -31,6 +31,8 @@ interface MaintenanceLane { timer?: ReturnType; pending?: Promise; failures: number; + readonly activeDelayMs?: number; + readonly idleDelayMs?: number; } /** Physical cleanup is optional work, admitted only after the Host publishes Ready. */ @@ -44,6 +46,7 @@ export class HostStorageMaintenance { artifacts: Pick; contextOffload?: Pick; onError: (name: string, error: unknown) => void; + retention?: { run: (input: { maxFamilies: number }) => Promise }; }) { this.#onError = input.onError; let after: string | undefined; @@ -67,6 +70,16 @@ export class HostStorageMaintenance { }, }, ]; + if (input.retention) { + const retention = input.retention; + this.#lanes.push({ + name: 'archived task retention', + failures: 0, + activeDelayMs: 1_000, + idleDelayMs: 15 * 60_000, + run: () => retention.run({ maxFamilies: 8 }), + }); + } const context = input.contextOffload; if (context) this.#lanes.push({ @@ -86,7 +99,7 @@ export class HostStorageMaintenance { start(): void { if (this.#started || this.#draining) return; this.#started = true; - for (const lane of this.#lanes) this.#schedule(lane, ACTIVE_DELAY_MS); + for (const lane of this.#lanes) this.#schedule(lane, lane.activeDelayMs ?? ACTIVE_DELAY_MS); } beginDrain(): void { @@ -113,7 +126,7 @@ export class HostStorageMaintenance { try { const more = await lane.run(); lane.failures = 0; - delay = more ? ACTIVE_DELAY_MS : IDLE_DELAY_MS; + delay = more ? (lane.activeDelayMs ?? ACTIVE_DELAY_MS) : (lane.idleDelayMs ?? IDLE_DELAY_MS); } catch (error) { lane.failures = Math.min(lane.failures + 1, 7); delay = Math.min(IDLE_DELAY_MS, 1000 * 2 ** (lane.failures - 1)); diff --git a/packages/runtime-host/src/server/storage-retention-coordinator.ts b/packages/runtime-host/src/server/storage-retention-coordinator.ts new file mode 100644 index 00000000000..67138334184 --- /dev/null +++ b/packages/runtime-host/src/server/storage-retention-coordinator.ts @@ -0,0 +1,230 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { readFile } from 'node:fs/promises'; +import { join } from 'node:path'; +import type { ExecutionSessionWriter } from '@maka/storage/execution-stores'; +import { + decodeStorageRetentionQueryResult, + type StorageRetentionQueryResult, + type StorageRetentionSweep, + type StorageRetentionPolicy, +} from '../protocol/storage-retention.js'; +import type { StorageRetentionOperationHandlerMap } from './operation-dispatcher.js'; +import type { HostSessionRetirementCoordinator } from './session-retirement-coordinator.js'; +import { + HostStorageRetentionPolicy, + RETENTION_DAY_MS, + retentionDeadline, + writeRetentionDocument, +} from './storage-retention-policy.js'; + +export class HostStorageRetentionCoordinator { + readonly handlers: StorageRetentionOperationHandlerMap = { + 'storage.retention.query': async () => { + if (this.#draining) + return { ok: false, error: { code: 'host_draining', message: 'Host draining' } }; + try { + const policy = this.#policy.snapshot(); + const preview = await this.#preview(policy); + return { + ok: true, + result: { policy, preview, lastSweep: this.#lastSweep, lastDeletion: this.#lastDeletion }, + }; + } catch { + return { + ok: false, + error: { code: 'persistence_failed', message: 'Retention preview unavailable' }, + }; + } + }, + 'storage.retention.set': async (input) => { + if (this.#draining) + return { ok: false, error: { code: 'host_draining', message: 'Host draining' } }; + try { + const policy = await this.#policy.set(input); + if (!policy) + return { + ok: false, + error: { + code: 'operation_conflict', + message: 'Retention setting changed; refresh before saving', + }, + }; + this.#after = undefined; + return { ok: true, result: policy }; + } catch { + return { + ok: false, + error: { code: 'persistence_failed', message: 'Retention setting could not be saved' }, + }; + } + }, + }; + readonly #policy: HostStorageRetentionPolicy; + readonly #stores: Pick; + readonly #retirement: Pick< + HostSessionRetirementCoordinator, + 'removeForRetention' | 'estimateRetentionBytes' + >; + readonly #path: string; + readonly #now: () => number; + #latestTime = 0; + #lastSweep: StorageRetentionSweep | null = null; + #lastDeletion: StorageRetentionQueryResult['lastDeletion'] = null; + #after: string | undefined; + #revision = -1; + #draining = false; + private constructor(input: { + policy: HostStorageRetentionPolicy; + stores: Pick; + retirement: Pick< + HostSessionRetirementCoordinator, + 'removeForRetention' | 'estimateRetentionBytes' + >; + stateRoot: string; + now?: () => number; + }) { + this.#policy = input.policy; + this.#stores = input.stores; + this.#retirement = input.retirement; + this.#path = join(input.stateRoot, 'storage-retention-results.json'); + this.#now = input.now ?? Date.now; + } + static async open( + input: Parameters[0], + ): Promise { + const coordinator = HostStorageRetentionCoordinator.create(input); + try { + const state = JSON.parse(await readFile(coordinator.#path, 'utf8')); + if (!Number.isSafeInteger(state.latestTime) || state.latestTime < 0) + throw new Error('Invalid retention clock'); + const decoded = decodeStorageRetentionQueryResult({ + policy: input.policy.snapshot(), + preview: { count: 0, eligibleAt: null }, + lastSweep: state.lastSweep, + lastDeletion: state.lastDeletion, + }); + coordinator.#latestTime = state.latestTime; + coordinator.#lastSweep = decoded.lastSweep; + coordinator.#lastDeletion = decoded.lastDeletion; + } catch (error) { + if (!(error instanceof Error && 'code' in error && error.code === 'ENOENT')) throw error; + } + return coordinator; + } + static create(input: { + policy: HostStorageRetentionPolicy; + stores: Pick; + retirement: Pick< + HostSessionRetirementCoordinator, + 'removeForRetention' | 'estimateRetentionBytes' + >; + stateRoot: string; + now?: () => number; + }): HostStorageRetentionCoordinator { + return new HostStorageRetentionCoordinator(input); + } + beginDrain(): void { + this.#draining = true; + } + async run(input: { maxFamilies: number }): Promise { + if (this.#draining) return false; + const policy = this.#policy.snapshot(); + if (!policy.enabled || policy.enabledAt === null) return false; + const at = this.#now(); + if (at < Math.max(this.#latestTime, policy.enabledAt)) return false; + this.#latestTime = at; + await this.#save(); + if (at <= policy.enabledAt + policy.days * RETENTION_DAY_MS) return false; + if (this.#revision !== policy.revision) { + this.#after = undefined; + this.#revision = policy.revision; + } + const page = await this.#stores.listRetentionCandidates({ + cutoff: at - policy.days * RETENTION_DAY_MS, + after: this.#after, + limit: Math.min(input.maxFamilies, 8), + }); + const sweep = { at, deleted: 0, busy: 0, needsReview: 0, failed: 0 }; + let bytes = 0; + let bytesKnown = true; + for (const id of page.sessionIds) { + if (this.#draining) break; + try { + const record = await this.#stores.readCatalogRecord(id); + const deadline = retentionDeadline(policy, record.summary.archivedAt); + if (deadline === null || at <= deadline) continue; + let estimated: number | undefined; + try { + estimated = await this.#retirement.estimateRetentionBytes(id); + } catch { + /* Optional estimate remains unknown. */ + } + const result = await this.#policy.withCurrent(policy.revision, async (current) => { + if (this.#now() < this.#latestTime) return 'kept'; + return this.#retirement.removeForRetention( + { sessionId: id, expectedRevision: record.revision }, + current, + at, + ); + }); + if (result === 'removed') { + sweep.deleted++; + if (estimated === undefined) bytesKnown = false; + else bytes += estimated; + } else if (result === 'busy') sweep.busy++; + else if (result === 'needs_review') sweep.needsReview++; + else if (result === 'failed') sweep.failed++; + } catch { + sweep.failed++; + } + } + this.#after = page.hasMore ? page.sessionIds.at(-1) : undefined; + this.#lastSweep = sweep; + if (sweep.deleted) + this.#lastDeletion = { at, count: sweep.deleted, estimatedBytes: bytesKnown ? bytes : null }; + await this.#save(); + return page.hasMore && !this.#draining; + } + async #preview(policy: StorageRetentionPolicy): Promise { + const { enabledAt } = policy; + if (enabledAt === null) return { count: 0, eligibleAt: null }; + let count = 0; + let after: string | undefined; + do { + const page = await this.#stores.listRetentionCandidates({ + cutoff: enabledAt, + after, + limit: 8, + }); + count += page.sessionIds.length; + after = page.hasMore ? page.sessionIds.at(-1) : undefined; + if (after) await new Promise((resolve) => setImmediate(resolve)); + } while (after && !this.#draining); + return { count, eligibleAt: enabledAt + policy.days * RETENTION_DAY_MS }; + } + async #save(): Promise { + await writeRetentionDocument(this.#path, { + latestTime: this.#latestTime, + lastSweep: this.#lastSweep, + lastDeletion: this.#lastDeletion, + }); + } +} diff --git a/packages/runtime-host/src/server/storage-retention-policy.ts b/packages/runtime-host/src/server/storage-retention-policy.ts new file mode 100644 index 00000000000..de76998bb3b --- /dev/null +++ b/packages/runtime-host/src/server/storage-retention-policy.ts @@ -0,0 +1,147 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { randomUUID } from 'node:crypto'; +import { open, readFile, rename, rm } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; + +import { + RETENTION_DAYS, + decodeStorageRetentionPolicy, + type RetentionDays, + type StorageRetentionPolicy, +} from '../protocol/storage-retention.js'; +export { + RETENTION_DAYS, + decodeStorageRetentionPolicy, + type RetentionDays, + type StorageRetentionPolicy, +}; +const DEFAULT_POLICY: StorageRetentionPolicy = { + enabled: false, + days: 30, + enabledAt: null, + revision: 0, +}; +export const RETENTION_DAY_MS = 24 * 60 * 60 * 1000; + +/** The destructive opt-in is deliberately outside runtime policy and its import/export. */ +export class HostStorageRetentionPolicy { + readonly #path: string; + readonly #now: () => number; + #policy: StorageRetentionPolicy = DEFAULT_POLICY; + #pending: Promise = Promise.resolve(); + + private constructor(stateRoot: string, now: () => number) { + this.#path = join(stateRoot, 'storage-retention.json'); + this.#now = now; + } + + static async open( + stateRoot: string, + now: () => number = Date.now, + ): Promise { + const store = new HostStorageRetentionPolicy(stateRoot, now); + try { + store.#policy = decodeStorageRetentionPolicy(JSON.parse(await readFile(store.#path, 'utf8'))); + } catch (error) { + if (!(error instanceof Error && 'code' in error && error.code === 'ENOENT')) throw error; + } + return store; + } + + snapshot(): StorageRetentionPolicy { + return { ...this.#policy }; + } + + /** Serialize policy writes with final retirement admission, including its asynchronous commit. */ + withCurrent( + revision: number, + operation: (policy: StorageRetentionPolicy) => Promise, + ): Promise { + return this.#serialize(async () => { + if (!this.#policy.enabled || this.#policy.revision !== revision) return undefined; + return operation(this.snapshot()); + }); + } + + set(input: { + enabled: boolean; + days: RetentionDays; + expectedRevision: number; + }): Promise { + return this.#serialize(async () => { + if (input.expectedRevision !== this.#policy.revision) return undefined; + if (typeof input.enabled !== 'boolean' || !RETENTION_DAYS.includes(input.days)) { + throw new Error('Invalid storage retention policy'); + } + if (input.enabled === this.#policy.enabled && input.days === this.#policy.days) + return this.snapshot(); + const next = decodeStorageRetentionPolicy({ + enabled: input.enabled, + days: input.days, + enabledAt: input.enabled ? this.#now() : null, + revision: this.#policy.revision + 1, + }); + await writeRetentionDocument(this.#path, next); + this.#policy = next; + return this.snapshot(); + }); + } + + #serialize(operation: () => Promise): Promise { + const result = this.#pending.then(operation); + this.#pending = result.catch(() => undefined); + return result; + } +} + +/** A legacy archived task starts aging only when this Host opts in. Equality is not expired. */ +export function retentionDeadline( + policy: StorageRetentionPolicy, + archivedAt?: number, +): number | null { + if (!policy.enabled || policy.enabledAt === null) return null; + return ( + Math.max(archivedAt ?? policy.enabledAt, policy.enabledAt) + policy.days * RETENTION_DAY_MS + ); +} + +/** Opt-in, disable and clock checkpoints must survive a process or machine restart. */ +export async function writeRetentionDocument(path: string, value: unknown): Promise { + const temporary = `${path}.${randomUUID()}.tmp`; + try { + const file = await open(temporary, 'wx', 0o600); + try { + await file.writeFile(`${JSON.stringify(value)}\n`); + await file.sync(); + } finally { + await file.close(); + } + await rename(temporary, path); + const directory = await open(dirname(path), 'r'); + try { + await directory.sync(); + } finally { + await directory.close(); + } + } finally { + await rm(temporary, { force: true }); + } +} diff --git a/packages/storage/src/execution-stores.ts b/packages/storage/src/execution-stores.ts index a11ea096288..a19ccca89cd 100644 --- a/packages/storage/src/execution-stores.ts +++ b/packages/storage/src/execution-stores.ts @@ -616,6 +616,7 @@ async function createExecutionStoresForWrite( listCatalogPage: (filter, cursor, limit, expectedRevision) => run(() => sessionStore.listCatalogPage(filter, cursor, limit, expectedRevision)), listHeaders: () => run(() => sessionStore.listHeaders()), + listRetentionCandidates: (input) => run(() => sessionStore.listRetentionCandidates(input)), listForRecovery: () => run(() => sessionStore.listForRecovery()), readHeaderSnapshot: (sessionId) => run(() => sessionStore.readHeaderSnapshot(sessionId)), readHeaderRecordSnapshot: (sessionId) => diff --git a/packages/storage/src/session-store-contract.ts b/packages/storage/src/session-store-contract.ts index 1becfa40b6c..11835094acf 100644 --- a/packages/storage/src/session-store-contract.ts +++ b/packages/storage/src/session-store-contract.ts @@ -394,6 +394,12 @@ export interface CoordinationTranscriptIndexState { } export interface SessionAuthorityStore extends SessionStore, MessageAdmissionStore { + listRetentionCandidates(input: { + cutoff: number; + after?: string; + limit: number; + }): Promise<{ sessionIds: string[]; hasMore: boolean }>; + readCoordinationTranscriptIndexState(): Promise; appendCoordinationTranscriptIndex( records: readonly CoordinationTranscriptReference[], diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 615f1b320b3..afc589bc3cd 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -611,6 +611,15 @@ class SqliteSessionStore implements SessionAuthorityStore { .map(toCatalogRecordSummary); } + async listRetentionCandidates(input: { + cutoff: number; + after?: string; + limit: number; + }): Promise<{ sessionIds: string[]; hasMore: boolean }> { + await this.ensureCatalogProjectionReadable(); + return this.metadata.listRetentionCandidates(input); + } + async listCatalogPage( filter: SessionListFilter | undefined, cursor: SessionCatalogPageCursor | undefined, diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index b4715adf328..dcd4094cde2 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -1498,6 +1498,52 @@ export class SqliteSessionMetadataStore { }); } + async listRetentionCandidates(input: { + cutoff: number; + after?: string; + limit: number; + }): Promise<{ sessionIds: string[]; hasMore: boolean }> { + this.assertOpen(); + if ( + !Number.isSafeInteger(input.cutoff) || + input.cutoff < 0 || + !Number.isSafeInteger(input.limit) || + input.limit < 1 || + input.limit > 8 + ) { + throw new Error('Invalid retention candidate bounds'); + } + const role = sqliteOrdinarySessionRolePredicate(); + const rows = this.db + .prepare(` + SELECT MIN(metadata.session_id) AS session_id + FROM session_metadata metadata + WHERE metadata.is_flagged = 0 AND metadata.is_archived = 1 + AND (metadata.archived_at IS NULL OR metadata.archived_at <= ?) + AND ${role.sql} + AND COALESCE(json_extract(metadata.payload_json, '$.conversationCopy.state'), '') != 'preparing' + AND COALESCE(json_extract(metadata.payload_json, '$.revisionState'), '') != 'preparing' + AND (metadata.subagent_parent_session_id IS NULL OR NOT EXISTS ( + SELECT 1 FROM session_metadata parent WHERE parent.session_id = metadata.subagent_parent_session_id + )) + AND NOT EXISTS ( + SELECT 1 FROM session_metadata member + WHERE COALESCE(member.revision_root_session_id, member.session_id) = COALESCE(metadata.revision_root_session_id, metadata.session_id) + AND (member.is_flagged = 1 OR member.is_archived = 0) + ) + GROUP BY COALESCE(metadata.revision_root_session_id, metadata.session_id) + HAVING MIN(metadata.session_id) > ? + ORDER BY session_id LIMIT ? + `) + .all(input.cutoff, ...role.parameters, input.after ?? '', input.limit + 1) as unknown as { + session_id: string; + }[]; + return { + sessionIds: rows.slice(0, input.limit).map((row) => row.session_id), + hasMore: rows.length > input.limit, + }; + } + async readCatalogRevision(): Promise { this.assertOpen(); return this.readCatalogRevisionSync(); diff --git a/packages/storage/src/test-only/memory-execution-session.ts b/packages/storage/src/test-only/memory-execution-session.ts index 3df960d17af..5f8ae1791d1 100644 --- a/packages/storage/src/test-only/memory-execution-session.ts +++ b/packages/storage/src/test-only/memory-execution-session.ts @@ -435,6 +435,9 @@ export function createMemorySessionStore( const read = (fn: (s: MemoryState) => T) => a.read(fn); const write = (name: string, fn: (s: MemoryState) => T) => a.write(name, fn); const store: SessionAuthorityStore = { + listRetentionCandidates: async () => { + throw new Error('Retention SQL is unavailable in memory execution store'); + }, ready: async () => {}, close: async () => {}, create: async (input, initial) =>