diff --git a/.claude/skills/env-reference/SKILL.md b/.claude/skills/env-reference/SKILL.md index dd32ab2e4..f3f6b4b80 100644 --- a/.claude/skills/env-reference/SKILL.md +++ b/.claude/skills/env-reference/SKILL.md @@ -169,6 +169,14 @@ See `apps/api/.env.example` for the full list. Key variables: - `PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO` — Target usage ratio for explicit superadmin ProjectData emergency purge calls (default: `0.9`) - `PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS` — Oldest `activity_events` and `acp_session_events` rows deleted per table per emergency purge batch (default: `500`) - `PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES` — Maximum emergency purge batches per explicit call (default: `4`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED` — Enables automatic ProjectData cleanup that strips expandable `tool_metadata.content` payloads from old terminal-session tool messages under storage pressure (default: enabled) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO` — ProjectData storage usage ratio that starts automatic terminal-session tool payload cleanup (default: `0.8`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO` — ProjectData storage usage ratio below which automatic tool payload cleanup stops (default: `0.75`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS` — Maximum tool-message rows inspected by one automatic cleanup alarm batch (default: `500`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES` — Maximum legacy `tool_metadata` bytes read into JS by one automatic cleanup alarm batch (default: `1048576`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS` — Minimum terminal-session age before automatic cleanup may strip stored tool payload content (default: `7`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS` — Delay before the next automatic cleanup alarm batch when more candidates remain (default: `60000`) +- `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM` — Maximum terminal sessions scanned by one automatic cleanup alarm batch (default: `25`) Absent operational brake keys and KV read errors mean enabled. This fail-open behavior preserves availability and intentionally differs from the fail-closed diff --git a/.env.example b/.env.example index 2caded198..aea01aff5 100644 --- a/.env.example +++ b/.env.example @@ -81,6 +81,14 @@ PROJECT_DATA_STORAGE_DEGRADED_RATIO=0.95 PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO=0.9 PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS=500 PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES=4 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED=true +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO=0.8 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO=0.75 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS=500 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES=1048576 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS=7 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS=60000 +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM=25 # Development NODE_ENV=development diff --git a/apps/api/.env.example b/apps/api/.env.example index c124439df..315820c16 100644 --- a/apps/api/.env.example +++ b/apps/api/.env.example @@ -801,6 +801,15 @@ INFOMANIAK_IP_POLL_INTERVAL_MS=3000 # PROJECT_COMMENT_LIST_MAX=300 # Max page size for the project-wide comment inbox # PROJECT_COMMENT_LIST_MAX_BYTES=4000000 # Max estimated content bytes for the project-wide comment inbox # DOCUMENT_CARD_RAW_OUTPUT_MAX_BYTES=16384 +# PROJECT_DATA_TOOL_METADATA_MAX_BYTES=131072 # Max stored tool_metadata bytes for new messages before structured tool content is stripped +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED=true # Auto-strip old terminal-session tool_metadata.content when ProjectData storage is under pressure +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO=0.8 # Start cleanup at this databaseSize / PROJECT_DATA_STORAGE_LIMIT_BYTES ratio +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO=0.75 # Continue cleanup until storage is below this ratio or candidates are exhausted +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS=500 # Max tool-message rows inspected per cleanup alarm batch +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES=1048576 # Max legacy tool_metadata bytes read into JS per cleanup alarm batch +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS=7 # Preserve newer terminal sessions from automated tool payload cleanup +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS=60000 # Delay before the next cleanup batch when more candidates remain +# PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM=25 # Max terminal sessions scanned per cleanup alarm # MESSAGE_SIZE_THRESHOLD=102400 # ACTIVITY_RETENTION_DAYS=90 # SESSION_IDLE_TIMEOUT_MINUTES=60 diff --git a/apps/api/src/durable-objects/project-data/index.ts b/apps/api/src/durable-objects/project-data/index.ts index 907ad0abe..7c930ba59 100644 --- a/apps/api/src/durable-objects/project-data/index.ts +++ b/apps/api/src/durable-objects/project-data/index.ts @@ -1018,12 +1018,7 @@ export class ProjectData extends DurableObject { if (await deferAlarmWhenDisabled(this.env, this.ctx.storage, 'ProjectData')) return; try { - await storageSafety.measureAndPersistProjectDataStorage( - this.sql, - this.env, - this.getProjectId(), - 'alarm' - ); + await storageSafety.runProjectDataStorageSafetyAlarm(this.sql, this.env, this.getProjectId()); } catch (err) { log.error('alarm.storage_safety_failed', { error: err instanceof Error ? err.message : String(err), diff --git a/apps/api/src/durable-objects/project-data/storage-safety-meta.ts b/apps/api/src/durable-objects/project-data/storage-safety-meta.ts new file mode 100644 index 000000000..aecd57a42 --- /dev/null +++ b/apps/api/src/durable-objects/project-data/storage-safety-meta.ts @@ -0,0 +1,37 @@ +import { isJsonRecord } from '@simple-agent-manager/shared'; + +export const META_LAST_MEASURED_AT = 'storageSafetyLastMeasuredAt'; +export const META_LAST_STATUS = 'storageSafetyLastStatus'; +export const META_LAST_ERROR = 'storageSafetyLastError'; + +export function readStorageSafetyMeta(sql: SqlStorage, key: string): string | null { + const row = sql.exec('SELECT value FROM do_meta WHERE key = ?', key).toArray()[0]; + if (!isJsonRecord(row)) return null; + const value = (row as Record).value; + return typeof value === 'string' ? value : null; +} + +export function readStorageSafetyMetaNumber(sql: SqlStorage, key: string): number | null { + const raw = readStorageSafetyMeta(sql, key); + if (!raw) return null; + const parsed = Number.parseInt(raw, 10); + return Number.isSafeInteger(parsed) ? parsed : null; +} + +export function writeStorageSafetyMeta(sql: SqlStorage, key: string, value: string): void { + sql.exec( + `INSERT INTO do_meta (key, value) + VALUES (?, ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + key, + value + ); +} + +export function deleteStorageSafetyMeta(sql: SqlStorage, key: string): void { + sql.exec('DELETE FROM do_meta WHERE key = ?', key); +} + +export function truncateStorageSafetyMetaValue(value: string, maxLength: number): string { + return value.length <= maxLength ? value : value.slice(0, maxLength); +} diff --git a/apps/api/src/durable-objects/project-data/storage-safety.ts b/apps/api/src/durable-objects/project-data/storage-safety.ts index dcfad7f06..2625add0d 100644 --- a/apps/api/src/durable-objects/project-data/storage-safety.ts +++ b/apps/api/src/durable-objects/project-data/storage-safety.ts @@ -10,6 +10,20 @@ import { isJsonRecord } from '@simple-agent-manager/shared'; import { createModuleLogger, serializeError } from '../../lib/logger'; import { persistError } from '../../services/observability'; +import { + META_LAST_ERROR, + META_LAST_MEASURED_AT, + META_LAST_STATUS, + readStorageSafetyMeta as readMeta, + readStorageSafetyMetaNumber as readMetaNumber, + truncateStorageSafetyMetaValue as truncate, + writeStorageSafetyMeta as writeMeta, +} from './storage-safety-meta'; +import { + type ProjectDataToolPayloadCleanupResult, + readProjectDataToolPayloadCleanupRecheckAt, + runProjectDataToolPayloadCleanup, +} from './tool-payload-cleanup'; import type { Env } from './types'; const log = createModuleLogger('project_data.storage_safety'); @@ -34,13 +48,16 @@ export const DEFAULT_PROJECT_DATA_STORAGE_DEGRADED_RATIO = 0.95; export const DEFAULT_PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO = 0.9; export const DEFAULT_PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS = 500; export const DEFAULT_PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES = 4; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO = 0.8; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO = 0.75; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS = 500; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES = 1024 * 1024; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS = 7; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS = 60 * 1000; +export const DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM = 25; -const META_LAST_MEASURED_AT = 'storageSafetyLastMeasuredAt'; -const META_LAST_STATUS = 'storageSafetyLastStatus'; const META_LAST_ALERT_AT = 'storageSafetyLastAlertAt'; const META_LAST_ALERT_STATUS = 'storageSafetyLastAlertStatus'; -const META_LAST_ERROR = 'storageSafetyLastError'; - export interface ProjectDataStorageTelemetry { projectId: string; measuredAt: number; @@ -76,7 +93,12 @@ export interface ProjectDataStorageEmergencyPurgeResult { exhaustedCandidates: boolean; } -interface StorageSafetyConfig { +export interface ProjectDataStorageAlarmResult { + measurement: ProjectDataStorageTelemetry | null; + cleanup: ProjectDataToolPayloadCleanupResult | null; +} + +export interface StorageSafetyConfig { enabled: boolean; limitBytes: number; measureIntervalMs: number; @@ -88,6 +110,14 @@ interface StorageSafetyConfig { emergencyTargetRatio: number; emergencyBatchRows: number; emergencyMaxBatches: number; + toolPayloadCleanupEnabled: boolean; + toolPayloadCleanupTriggerRatio: number; + toolPayloadCleanupTargetRatio: number; + toolPayloadCleanupBatchRows: number; + toolPayloadCleanupBatchBytes: number; + toolPayloadCleanupMinSessionAgeMs: number; + toolPayloadCleanupRecheckMs: number; + toolPayloadCleanupMaxSessionsPerAlarm: number; } function parsePositiveInteger(value: string | undefined, fallback: number): number { @@ -96,6 +126,12 @@ function parsePositiveInteger(value: string | undefined, fallback: number): numb return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; } +function parseNonNegativeInteger(value: string | undefined, fallback: number): number { + if (!value) return fallback; + const parsed = Number.parseInt(value, 10); + return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : fallback; +} + function parseBoundedRatio(value: string | undefined, fallback: number): number { if (!value) return fallback; const parsed = Number.parseFloat(value); @@ -127,6 +163,19 @@ export function resolveStorageSafetyConfig(env: Env): StorageSafetyConfig { const thresholdsAreOrdered = noticeRatio < warningRatio && warningRatio < criticalRatio && criticalRatio < degradedRatio; + const cleanupTriggerRatio = parseBoundedRatio( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO + ); + const cleanupTargetRatio = parseBoundedRatio( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO + ); + const cleanupRatiosAreOrdered = cleanupTargetRatio < cleanupTriggerRatio; + const cleanupMinSessionAgeDays = parseNonNegativeInteger( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS + ); return { enabled: envFlagEnabled(env.PROJECT_DATA_STORAGE_TELEMETRY_ENABLED), @@ -162,6 +211,30 @@ export function resolveStorageSafetyConfig(env: Env): StorageSafetyConfig { env.PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES, DEFAULT_PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES ), + toolPayloadCleanupEnabled: envFlagEnabled(env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED), + toolPayloadCleanupTriggerRatio: cleanupRatiosAreOrdered + ? cleanupTriggerRatio + : DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO, + toolPayloadCleanupTargetRatio: cleanupRatiosAreOrdered + ? cleanupTargetRatio + : DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO, + toolPayloadCleanupBatchRows: parsePositiveInteger( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS + ), + toolPayloadCleanupBatchBytes: parsePositiveInteger( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES + ), + toolPayloadCleanupMinSessionAgeMs: cleanupMinSessionAgeDays * 24 * 60 * 60 * 1000, + toolPayloadCleanupRecheckMs: parsePositiveInteger( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS + ), + toolPayloadCleanupMaxSessionsPerAlarm: parsePositiveInteger( + env.PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM, + DEFAULT_PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM + ), }; } @@ -180,33 +253,6 @@ export function classifyStorageUsage( return 'ok'; } -function readMeta(sql: SqlStorage, key: string): string | null { - const row = sql.exec('SELECT value FROM do_meta WHERE key = ?', key).toArray()[0]; - if (!isJsonRecord(row)) return null; - return typeof row.value === 'string' ? row.value : null; -} - -function readMetaNumber(sql: SqlStorage, key: string): number | null { - const raw = readMeta(sql, key); - if (!raw) return null; - const parsed = Number.parseInt(raw, 10); - return Number.isSafeInteger(parsed) ? parsed : null; -} - -function writeMeta(sql: SqlStorage, key: string, value: string): void { - sql.exec( - `INSERT INTO do_meta (key, value) - VALUES (?, ?) - ON CONFLICT(key) DO UPDATE SET value = excluded.value`, - key, - value - ); -} - -function truncate(value: string, maxLength: number): string { - return value.length <= maxLength ? value : value.slice(0, maxLength); -} - function buildTelemetry( sql: SqlStorage, env: Env, @@ -358,7 +404,24 @@ export function computeStorageSafetyAlarmTime( if (!config.enabled) return null; if (!readMeta(sql, 'projectId')) return null; const lastMeasuredAt = readMetaNumber(sql, META_LAST_MEASURED_AT); - return lastMeasuredAt === null ? now : lastMeasuredAt + config.measureIntervalMs; + const measureAt = lastMeasuredAt === null ? now : lastMeasuredAt + config.measureIntervalMs; + const cleanupRecheckAt = config.toolPayloadCleanupEnabled + ? readProjectDataToolPayloadCleanupRecheckAt(sql) + : null; + if (cleanupRecheckAt === null) return measureAt; + return Math.min(measureAt, cleanupRecheckAt); +} + +export function shouldMeasureProjectDataStorage( + sql: SqlStorage, + env: Env, + now: number = Date.now() +): boolean { + const config = resolveStorageSafetyConfig(env); + if (!config.enabled) return false; + if (!readMeta(sql, 'projectId')) return false; + const lastMeasuredAt = readMetaNumber(sql, META_LAST_MEASURED_AT); + return lastMeasuredAt === null || now - lastMeasuredAt >= config.measureIntervalMs; } export async function measureAndPersistProjectDataStorage( @@ -406,9 +469,29 @@ export async function measureAndPersistProjectDataStorage( return telemetry; } +export async function runProjectDataStorageSafetyAlarm( + sql: SqlStorage, + env: Env, + projectId: string | null +): Promise { + const now = Date.now(); + let measurement: ProjectDataStorageTelemetry | null = null; + if (shouldMeasureProjectDataStorage(sql, env, now)) { + measurement = await measureAndPersistProjectDataStorage(sql, env, projectId, 'alarm'); + } + const config = resolveStorageSafetyConfig(env); + const cleanup = await runProjectDataToolPayloadCleanup(sql, env, projectId, config, { + allowStart: measurement !== null, + now, + classifyStatus: (databaseSizeBytes) => classifyStorageUsage(databaseSizeBytes, config), + recordTelemetry: (telemetry, fields) => upsertTelemetry(env, telemetry, fields), + }); + return { measurement, cleanup }; +} + function normalizeCount(row: unknown): number { if (!isJsonRecord(row)) return 0; - const count = row.count; + const count = (row as Record).count; return typeof count === 'number' && Number.isFinite(count) ? count : 0; } diff --git a/apps/api/src/durable-objects/project-data/tool-metadata-storage.ts b/apps/api/src/durable-objects/project-data/tool-metadata-storage.ts index 6dc62be5b..db7dd6d9c 100644 --- a/apps/api/src/durable-objects/project-data/tool-metadata-storage.ts +++ b/apps/api/src/durable-objects/project-data/tool-metadata-storage.ts @@ -139,3 +139,69 @@ export function boundToolMetadataForStorage( truncated: true, }; } + +export function stripToolMetadataPayloadForStorage( + toolMetadata: string | null, + env: Env +): { + value: string | null; + originalBytes: number; + storedBytes: number; + stripped: boolean; + failed: boolean; +} { + if (toolMetadata === null) { + return { value: null, originalBytes: 0, storedBytes: 0, stripped: false, failed: false }; + } + + const originalBytes = utf8Bytes(toolMetadata); + let parsed: unknown; + try { + parsed = JSON.parse(toolMetadata); + } catch { + return { + value: toolMetadata, + originalBytes, + storedBytes: originalBytes, + stripped: false, + failed: true, + }; + } + + let compact: unknown; + try { + compact = stripToolMetadataContent(parsed, resolveCompactMessageOptions(env)); + } catch { + return { + value: toolMetadata, + originalBytes, + storedBytes: originalBytes, + stripped: false, + failed: true, + }; + } + const compactJson = JSON.stringify(compact); + const compactBytes = utf8Bytes(compactJson); + if (compactBytes >= originalBytes) { + return { + value: toolMetadata, + originalBytes, + storedBytes: originalBytes, + stripped: false, + failed: false, + }; + } + + const maxBytes = resolveToolMetadataMaxBytes(env); + const value = compactBytes <= maxBytes + ? compactJson + : serializeWithinLimit(buildMinimalToolMetadata(parsed, originalBytes), maxBytes); + + return { + value, + originalBytes, + storedBytes: utf8Bytes(value), + stripped: true, + failed: false, + }; +} diff --git a/apps/api/src/durable-objects/project-data/tool-payload-cleanup-candidates.ts b/apps/api/src/durable-objects/project-data/tool-payload-cleanup-candidates.ts new file mode 100644 index 000000000..710cc996d --- /dev/null +++ b/apps/api/src/durable-objects/project-data/tool-payload-cleanup-candidates.ts @@ -0,0 +1,358 @@ +import { stripToolMetadataPayloadForStorage } from './tool-metadata-storage'; +import type { Env } from './types'; + +const textEncoder = new TextEncoder(); + +export type ToolPayloadCleanupCursor = { + sessionId: string; + createdAt: number; + sequence: number; + messageId: string; +}; + +export type ToolPayloadCandidate = ToolPayloadCleanupCursor & { + toolMetadataBytes: number; +}; + +export type ToolPayloadCandidateScanResult = { + rowsScanned: number; + rowsUpdated: number; + rowsFailed: number; + toolMetadataBytesScanned: number; + toolMetadataBytesRead: number; + originalToolMetadataBytes: number; + storedToolMetadataBytes: number; + lastCursor: ToolPayloadCleanupCursor | null; + errorMessages: string[]; +}; + +type ToolPayloadCandidateUpdate = { + rowsUpdated: number; + rowsFailed: number; + toolMetadataBytesRead: number; + originalToolMetadataBytes: number; + storedToolMetadataBytes: number; + errorMessage: string | null; +}; + +function rawNumber(value: unknown): number | null { + if (typeof value === 'number' && Number.isSafeInteger(value)) return value; + if (typeof value === 'bigint' && value <= BigInt(Number.MAX_SAFE_INTEGER)) return Number(value); + if (typeof value === 'string') { + const parsed = Number.parseInt(value, 10); + return Number.isSafeInteger(parsed) ? parsed : null; + } + return null; +} + +export function selectToolPayloadCandidates( + sql: SqlStorage, + sessionId: string, + cursor: ToolPayloadCleanupCursor | null, + limit: number, + maxMetadataBytes: number, + allowOversizedFirst: boolean +): ToolPayloadCandidate[] { + const messageCursor = cursor?.sessionId === sessionId ? cursor : null; + const cursorCreatedAt = messageCursor?.createdAt ?? null; + const cursorSequence = messageCursor?.sequence ?? null; + const cursorMessageId = messageCursor?.messageId ?? null; + const rows = sql + .exec( + `WITH limited AS ( + SELECT + id, + created_at, + COALESCE(sequence, 0) AS sequence, + length(CAST(tool_metadata AS BLOB)) AS tool_metadata_bytes + FROM chat_messages + WHERE session_id = ? + AND role = 'tool' + AND tool_metadata IS NOT NULL + AND tool_metadata LIKE '%"content"%' + AND ( + ? IS NULL + OR created_at > ? + OR (created_at = ? AND COALESCE(sequence, 0) > ?) + OR (created_at = ? AND COALESCE(sequence, 0) = ? AND id > ?) + ) + ORDER BY created_at ASC, COALESCE(sequence, 0) ASC, id ASC + LIMIT ? + ), + bounded AS ( + SELECT + id, + created_at, + sequence, + tool_metadata_bytes, + ROW_NUMBER() OVER (ORDER BY created_at ASC, sequence ASC, id ASC) AS row_number, + SUM(tool_metadata_bytes) OVER ( + ORDER BY created_at ASC, sequence ASC, id ASC + ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW + ) AS cumulative_tool_metadata_bytes + FROM limited + ) + SELECT id, created_at, sequence, tool_metadata_bytes + FROM bounded + WHERE cumulative_tool_metadata_bytes <= ? + OR (? = 1 AND row_number = 1) + ORDER BY created_at ASC, sequence ASC, id ASC`, + sessionId, + cursorCreatedAt, + cursorCreatedAt ?? 0, + cursorCreatedAt ?? 0, + cursorSequence ?? 0, + cursorCreatedAt ?? 0, + cursorSequence ?? 0, + cursorMessageId ?? '', + limit, + maxMetadataBytes, + allowOversizedFirst ? 1 : 0 + ) + .raw(); + return parseToolPayloadCandidateRows(sessionId, rows); +} + +export function hasToolPayloadCandidatesAfter( + sql: SqlStorage, + sessionId: string, + cursor: ToolPayloadCleanupCursor | null +): boolean { + const messageCursor = cursor?.sessionId === sessionId ? cursor : null; + const cursorCreatedAt = messageCursor?.createdAt ?? null; + const cursorSequence = messageCursor?.sequence ?? null; + const cursorMessageId = messageCursor?.messageId ?? null; + const rows = sql + .exec( + `SELECT id + FROM chat_messages + WHERE session_id = ? + AND role = 'tool' + AND tool_metadata IS NOT NULL + AND tool_metadata LIKE '%"content"%' + AND ( + ? IS NULL + OR created_at > ? + OR (created_at = ? AND COALESCE(sequence, 0) > ?) + OR (created_at = ? AND COALESCE(sequence, 0) = ? AND id > ?) + ) + ORDER BY created_at ASC, COALESCE(sequence, 0) ASC, id ASC + LIMIT ?`, + sessionId, + cursorCreatedAt, + cursorCreatedAt ?? 0, + cursorCreatedAt ?? 0, + cursorSequence ?? 0, + cursorCreatedAt ?? 0, + cursorSequence ?? 0, + cursorMessageId ?? '', + 1 + ) + .raw(); + let found = false; + for (const row of rows) { + found = found || typeof row[0] === 'string'; + } + return found; +} + +function parseToolPayloadCandidateRows( + sessionId: string, + rows: IterableIterator +): ToolPayloadCandidate[] { + const candidates: ToolPayloadCandidate[] = []; + for (const row of rows) { + const id = row[0]; + const createdAt = rawNumber(row[1]); + const sequence = rawNumber(row[2]); + const toolMetadataBytes = rawNumber(row[3]); + if ( + typeof id === 'string' && + createdAt !== null && + sequence !== null && + toolMetadataBytes !== null && + toolMetadataBytes > 0 + ) { + candidates.push({ sessionId, createdAt, sequence, messageId: id, toolMetadataBytes }); + } + } + return candidates; +} + +function readBoundedToolMetadata( + sql: SqlStorage, + messageId: string, + maxMetadataBytes: number +): string | null { + const rows = sql + .exec( + `SELECT tool_metadata + FROM chat_messages + WHERE id = ? + AND tool_metadata IS NOT NULL + AND length(CAST(tool_metadata AS BLOB)) <= ? + LIMIT 1`, + messageId, + maxMetadataBytes + ) + .raw(); + let toolMetadata: string | null = null; + for (const row of rows) { + if (typeof row[0] === 'string') toolMetadata = row[0]; + } + return toolMetadata; +} + +function updateToolMetadata(sql: SqlStorage, messageId: string, toolMetadata: string | null): void { + sql.exec('UPDATE chat_messages SET tool_metadata = ? WHERE id = ?', toolMetadata, messageId); +} + +function utf8Bytes(value: string): number { + return textEncoder.encode(value).byteLength; +} + +function buildFailClosedToolMetadata( + candidate: ToolPayloadCandidate, + reason: 'oversized_legacy_payload' | 'poison_legacy_payload' +): string { + return JSON.stringify({ + storageSafetyTruncated: true, + contentTruncated: true, + storageSafetyCleanupReason: reason, + originalSizeBytes: candidate.toolMetadataBytes, + }); +} + +function failClosedToolMetadataCandidate( + sql: SqlStorage, + candidate: ToolPayloadCandidate, + reason: 'oversized_legacy_payload' | 'poison_legacy_payload' +): ToolPayloadCandidateUpdate { + const value = buildFailClosedToolMetadata(candidate, reason); + updateToolMetadata(sql, candidate.messageId, value); + return { + rowsUpdated: 1, + rowsFailed: reason === 'poison_legacy_payload' ? 1 : 0, + toolMetadataBytesRead: 0, + originalToolMetadataBytes: candidate.toolMetadataBytes, + storedToolMetadataBytes: utf8Bytes(value), + errorMessage: null, + }; +} + +function processToolPayloadCandidate( + sql: SqlStorage, + env: Env, + candidate: ToolPayloadCandidate, + remainingReadBytes: number +): ToolPayloadCandidateUpdate { + try { + if (candidate.toolMetadataBytes > remainingReadBytes) { + return failClosedToolMetadataCandidate(sql, candidate, 'oversized_legacy_payload'); + } + + const toolMetadata = readBoundedToolMetadata(sql, candidate.messageId, remainingReadBytes); + if (toolMetadata === null) return emptyCandidateUpdate(); + + const stripped = stripToolMetadataPayloadForStorage(toolMetadata, env); + if (stripped.failed) { + return failClosedToolMetadataCandidate(sql, candidate, 'poison_legacy_payload'); + } + if (!stripped.stripped) { + return { + rowsUpdated: 0, + rowsFailed: 0, + toolMetadataBytesRead: stripped.originalBytes, + originalToolMetadataBytes: 0, + storedToolMetadataBytes: 0, + errorMessage: null, + }; + } + + updateToolMetadata(sql, candidate.messageId, stripped.value); + return { + rowsUpdated: 1, + rowsFailed: 0, + toolMetadataBytesRead: stripped.originalBytes, + originalToolMetadataBytes: stripped.originalBytes, + storedToolMetadataBytes: stripped.storedBytes, + errorMessage: null, + }; + } catch (error) { + return failClosedAfterError(sql, candidate, error); + } +} + +function emptyCandidateUpdate(): ToolPayloadCandidateUpdate { + return { + rowsUpdated: 0, + rowsFailed: 0, + toolMetadataBytesRead: 0, + originalToolMetadataBytes: 0, + storedToolMetadataBytes: 0, + errorMessage: null, + }; +} + +function failClosedAfterError( + sql: SqlStorage, + candidate: ToolPayloadCandidate, + error: unknown +): ToolPayloadCandidateUpdate { + try { + const failClosed = failClosedToolMetadataCandidate(sql, candidate, 'poison_legacy_payload'); + return { + ...failClosed, + errorMessage: error instanceof Error ? error.message : String(error), + }; + } catch (failClosedError) { + const originalMessage = error instanceof Error ? error.message : String(error); + const failClosedMessage = + failClosedError instanceof Error ? failClosedError.message : String(failClosedError); + return { + rowsUpdated: 0, + rowsFailed: 1, + toolMetadataBytesRead: 0, + originalToolMetadataBytes: 0, + storedToolMetadataBytes: 0, + errorMessage: `${originalMessage}; fail-closed update failed: ${failClosedMessage}`, + }; + } +} + +export function scanToolPayloadCandidates( + sql: SqlStorage, + env: Env, + batchBytes: number, + candidates: ToolPayloadCandidate[] +): ToolPayloadCandidateScanResult { + const result = createEmptyCandidateScanResult(); + for (const candidate of candidates) { + result.lastCursor = candidate; + result.toolMetadataBytesScanned += candidate.toolMetadataBytes; + const remainingReadBytes = Math.max(batchBytes - result.toolMetadataBytesRead, 0); + const updated = processToolPayloadCandidate(sql, env, candidate, remainingReadBytes); + result.rowsUpdated += updated.rowsUpdated; + result.rowsFailed += updated.rowsFailed; + result.toolMetadataBytesRead += updated.toolMetadataBytesRead; + result.originalToolMetadataBytes += updated.originalToolMetadataBytes; + result.storedToolMetadataBytes += updated.storedToolMetadataBytes; + if (updated.errorMessage) result.errorMessages.push(updated.errorMessage); + } + result.rowsScanned = candidates.length; + return result; +} + +function createEmptyCandidateScanResult(): ToolPayloadCandidateScanResult { + return { + rowsScanned: 0, + rowsUpdated: 0, + rowsFailed: 0, + toolMetadataBytesScanned: 0, + toolMetadataBytesRead: 0, + originalToolMetadataBytes: 0, + storedToolMetadataBytes: 0, + lastCursor: null, + errorMessages: [], + }; +} diff --git a/apps/api/src/durable-objects/project-data/tool-payload-cleanup.ts b/apps/api/src/durable-objects/project-data/tool-payload-cleanup.ts new file mode 100644 index 000000000..bd4558229 --- /dev/null +++ b/apps/api/src/durable-objects/project-data/tool-payload-cleanup.ts @@ -0,0 +1,627 @@ +import { createModuleLogger, serializeError } from '../../lib/logger'; +import type { + ProjectDataStorageStatus, + ProjectDataStorageTelemetry, + StorageSafetyConfig, +} from './storage-safety'; +import { + deleteStorageSafetyMeta as deleteMeta, + META_LAST_ERROR, + META_LAST_MEASURED_AT, + META_LAST_STATUS, + readStorageSafetyMeta as readMeta, + readStorageSafetyMetaNumber as readMetaNumber, + truncateStorageSafetyMetaValue as truncate, + writeStorageSafetyMeta as writeMeta, +} from './storage-safety-meta'; +import { + hasToolPayloadCandidatesAfter, + scanToolPayloadCandidates, + selectToolPayloadCandidates, + type ToolPayloadCleanupCursor, +} from './tool-payload-cleanup-candidates'; +import type { Env } from './types'; + +const log = createModuleLogger('project_data.tool_payload_cleanup'); + +const META_TOOL_CLEANUP_CURSOR_SESSION_ID = 'storageSafetyToolCleanupCursorSessionId'; +const META_TOOL_CLEANUP_CURSOR_CREATED_AT = 'storageSafetyToolCleanupCursorCreatedAt'; +const META_TOOL_CLEANUP_CURSOR_SEQUENCE = 'storageSafetyToolCleanupCursorSequence'; +const META_TOOL_CLEANUP_CURSOR_MESSAGE_ID = 'storageSafetyToolCleanupCursorMessageId'; +const META_TOOL_CLEANUP_RECHECK_AT = 'storageSafetyToolCleanupRecheckAt'; +const TOOL_PAYLOAD_SESSION_EXHAUSTED_MESSAGE_ID = '__session_exhausted__'; + +export interface ProjectDataToolPayloadCleanupResult { + projectId: string; + beforeBytes: number; + afterBytes: number; + limitBytes: number; + triggerBytes: number; + targetBytes: number; + batchRows: number; + batchBytes: number; + sessionsScanned: number; + rowsScanned: number; + rowsUpdated: number; + rowsFailed: number; + toolMetadataBytesScanned: number; + toolMetadataBytesRead: number; + originalToolMetadataBytes: number; + storedToolMetadataBytes: number; + cursor: + | { + sessionId: string; + createdAt: number; + sequence: number; + messageId: string; + } + | null; + exhaustedCandidates: boolean; + recheckAt: number | null; +} + +export interface ProjectDataToolPayloadCleanupOptions { + allowStart?: boolean; + now?: number; + classifyStatus: (databaseSizeBytes: number) => ProjectDataStorageStatus; + recordTelemetry: ( + telemetry: ProjectDataStorageTelemetry, + fields: { + lastPurgeAt?: number | null; + lastPurgeReason?: string | null; + lastPurgeRows?: number | null; + lastPurgeDatabaseSizeBytes?: number | null; + lastError?: string | null; + } + ) => Promise; +} + +export function readProjectDataToolPayloadCleanupRecheckAt(sql: SqlStorage): number | null { + return readMetaNumber(sql, META_TOOL_CLEANUP_RECHECK_AT); +} + +function readToolPayloadCleanupCursor(sql: SqlStorage): ToolPayloadCleanupCursor | null { + const sessionId = readMeta(sql, META_TOOL_CLEANUP_CURSOR_SESSION_ID); + const createdAt = readMetaNumber(sql, META_TOOL_CLEANUP_CURSOR_CREATED_AT); + const sequence = readMetaNumber(sql, META_TOOL_CLEANUP_CURSOR_SEQUENCE); + const messageId = readMeta(sql, META_TOOL_CLEANUP_CURSOR_MESSAGE_ID); + if (!sessionId || createdAt === null || sequence === null || !messageId) return null; + return { sessionId, createdAt, sequence, messageId }; +} + +function writeToolPayloadCleanupCursor( + sql: SqlStorage, + cursor: ToolPayloadCleanupCursor, + recheckAt: number +): void { + writeMeta(sql, META_TOOL_CLEANUP_CURSOR_SESSION_ID, cursor.sessionId); + writeMeta(sql, META_TOOL_CLEANUP_CURSOR_CREATED_AT, String(cursor.createdAt)); + writeMeta(sql, META_TOOL_CLEANUP_CURSOR_SEQUENCE, String(cursor.sequence)); + writeMeta(sql, META_TOOL_CLEANUP_CURSOR_MESSAGE_ID, cursor.messageId); + writeMeta(sql, META_TOOL_CLEANUP_RECHECK_AT, String(recheckAt)); +} + +function writeToolPayloadCleanupRecheckAt(sql: SqlStorage, recheckAt: number): void { + writeMeta(sql, META_TOOL_CLEANUP_RECHECK_AT, String(recheckAt)); +} + +function clearToolPayloadCleanupState(sql: SqlStorage): void { + deleteMeta(sql, META_TOOL_CLEANUP_CURSOR_SESSION_ID); + deleteMeta(sql, META_TOOL_CLEANUP_CURSOR_CREATED_AT); + deleteMeta(sql, META_TOOL_CLEANUP_CURSOR_SEQUENCE); + deleteMeta(sql, META_TOOL_CLEANUP_CURSOR_MESSAGE_ID); + deleteMeta(sql, META_TOOL_CLEANUP_RECHECK_AT); +} + +function isSessionExhaustedCursor(cursor: ToolPayloadCleanupCursor): boolean { + return ( + cursor.createdAt === Number.MAX_SAFE_INTEGER && + cursor.sequence === Number.MAX_SAFE_INTEGER && + cursor.messageId === TOOL_PAYLOAD_SESSION_EXHAUSTED_MESSAGE_ID + ); +} + +function buildSessionExhaustedCursor(sessionId: string): ToolPayloadCleanupCursor { + return { + sessionId, + createdAt: Number.MAX_SAFE_INTEGER, + sequence: Number.MAX_SAFE_INTEGER, + messageId: TOOL_PAYLOAD_SESSION_EXHAUSTED_MESSAGE_ID, + }; +} + +function publicToolPayloadCleanupCursor( + cursor: ToolPayloadCleanupCursor | null +): ProjectDataToolPayloadCleanupResult['cursor'] { + if (!cursor) return null; + return { + sessionId: cursor.sessionId, + createdAt: cursor.createdAt, + sequence: cursor.sequence, + messageId: cursor.messageId, + }; +} + +function selectNextTerminalSessionId( + sql: SqlStorage, + cutoffUpdatedAt: number, + afterSessionId: string +): string | null { + const cursor = sql + .exec( + `SELECT id + FROM chat_sessions + WHERE status IN ('stopped', 'failed') + AND updated_at <= ? + AND id > ? + ORDER BY id ASC + LIMIT 1`, + cutoffUpdatedAt, + afterSessionId + ) + .raw(); + + let sessionId: string | null = null; + for (const row of cursor) { + const id = row[0]; + if (typeof id === 'string') sessionId = id; + } + return sessionId; +} + +type ToolPayloadCleanupPlan = { + projectId: string; + now: number; + beforeBytes: number; + limitBytes: number; + triggerBytes: number; + targetBytes: number; + batchRows: number; + batchBytes: number; + cutoffUpdatedAt: number; + pendingCursor: ToolPayloadCleanupCursor | null; +}; + +type ToolPayloadCleanupBatch = { + sessionsScanned: number; + rowsScanned: number; + rowsUpdated: number; + rowsFailed: number; + toolMetadataBytesScanned: number; + toolMetadataBytesRead: number; + originalToolMetadataBytes: number; + storedToolMetadataBytes: number; + errorMessages: string[]; + lastCursor: ToolPayloadCleanupCursor | null; + lastScannedSessionId: string | null; + pauseCursor: ToolPayloadCleanupCursor | null; + hasMoreCandidates: boolean; + finalSessionId: string | null; +}; + +function createToolPayloadCleanupPlan( + sql: SqlStorage, + projectId: string | null, + config: StorageSafetyConfig, + options: ProjectDataToolPayloadCleanupOptions +): ToolPayloadCleanupPlan | null { + if (!config.enabled || !config.toolPayloadCleanupEnabled || !projectId) return null; + const now = options.now ?? Date.now(); + + const beforeBytes = sql.databaseSize; + const triggerBytes = Math.floor(config.limitBytes * config.toolPayloadCleanupTriggerRatio); + const targetBytes = Math.floor(config.limitBytes * config.toolPayloadCleanupTargetRatio); + const pendingCursor = readToolPayloadCleanupCursor(sql); + const pendingRecheckAt = readMetaNumber(sql, META_TOOL_CLEANUP_RECHECK_AT); + const hasPendingCleanup = pendingCursor !== null || pendingRecheckAt !== null; + + if (beforeBytes <= targetBytes) { + clearToolPayloadCleanupState(sql); + return null; + } + if (pendingRecheckAt !== null && pendingRecheckAt > now) { + return null; + } + if (!hasPendingCleanup && !options.allowStart) { + return null; + } + if (beforeBytes < triggerBytes && !hasPendingCleanup) { + return null; + } + + return { + projectId, + now, + beforeBytes, + limitBytes: config.limitBytes, + triggerBytes, + targetBytes, + batchRows: config.toolPayloadCleanupBatchRows, + batchBytes: config.toolPayloadCleanupBatchBytes, + cutoffUpdatedAt: now - config.toolPayloadCleanupMinSessionAgeMs, + pendingCursor, + }; +} + +function selectInitialCleanupSessionId( + sql: SqlStorage, + cutoffUpdatedAt: number, + cursor: ToolPayloadCleanupCursor | null +): string | null { + if (!cursor) return selectNextTerminalSessionId(sql, cutoffUpdatedAt, ''); + if (isSessionExhaustedCursor(cursor)) { + return selectNextTerminalSessionId(sql, cutoffUpdatedAt, cursor.sessionId); + } + return cursor.sessionId; +} + +function createEmptyToolPayloadCleanupBatch(): ToolPayloadCleanupBatch { + return { + sessionsScanned: 0, + rowsScanned: 0, + rowsUpdated: 0, + rowsFailed: 0, + toolMetadataBytesScanned: 0, + toolMetadataBytesRead: 0, + originalToolMetadataBytes: 0, + storedToolMetadataBytes: 0, + errorMessages: [], + lastCursor: null, + lastScannedSessionId: null, + pauseCursor: null, + hasMoreCandidates: false, + finalSessionId: null, + }; +} + +function scanToolPayloadCleanupBatch( + sql: SqlStorage, + env: Env, + config: StorageSafetyConfig, + plan: ToolPayloadCleanupPlan +): ToolPayloadCleanupBatch { + const batch = createEmptyToolPayloadCleanupBatch(); + let cursor = plan.pendingCursor; + let sessionId = selectInitialCleanupSessionId(sql, plan.cutoffUpdatedAt, cursor); + + while ( + sessionId && + batch.rowsScanned < plan.batchRows && + batch.sessionsScanned < config.toolPayloadCleanupMaxSessionsPerAlarm && + (batch.rowsScanned === 0 || batch.toolMetadataBytesScanned < plan.batchBytes) + ) { + const remainingRows = plan.batchRows - batch.rowsScanned; + const remainingBytes = Math.max(plan.batchBytes - batch.toolMetadataBytesScanned, 0); + const messageCursor = cursor?.sessionId === sessionId ? cursor : null; + const allowOversizedFirst = batch.rowsScanned === 0; + const candidates = selectToolPayloadCandidates( + sql, + sessionId, + messageCursor, + remainingRows, + remainingBytes, + allowOversizedFirst + ); + + if (candidates.length === 0) { + const previousScannedSessionId = batch.lastScannedSessionId; + batch.lastScannedSessionId = sessionId; + batch.sessionsScanned++; + if (hasToolPayloadCandidatesAfter(sql, sessionId, messageCursor)) { + batch.hasMoreCandidates = true; + if (messageCursor) { + batch.pauseCursor = messageCursor; + } else if (previousScannedSessionId) { + batch.pauseCursor = buildSessionExhaustedCursor(previousScannedSessionId); + } else { + batch.pauseCursor = null; + } + batch.finalSessionId = sessionId; + break; + } + + sessionId = selectNextTerminalSessionId(sql, plan.cutoffUpdatedAt, sessionId); + cursor = null; + continue; + } + + const scanned = scanToolPayloadCandidates(sql, env, plan.batchBytes, candidates); + + batch.lastScannedSessionId = sessionId; + batch.sessionsScanned++; + batch.rowsScanned += scanned.rowsScanned; + batch.rowsUpdated += scanned.rowsUpdated; + batch.rowsFailed += scanned.rowsFailed; + batch.toolMetadataBytesScanned += scanned.toolMetadataBytesScanned; + batch.toolMetadataBytesRead += scanned.toolMetadataBytesRead; + batch.originalToolMetadataBytes += scanned.originalToolMetadataBytes; + batch.storedToolMetadataBytes += scanned.storedToolMetadataBytes; + batch.errorMessages.push(...scanned.errorMessages); + batch.lastCursor = scanned.lastCursor ?? batch.lastCursor; + + const lastCandidate = candidates[candidates.length - 1] ?? null; + const moreInSession = + lastCandidate !== null && hasToolPayloadCandidatesAfter(sql, sessionId, lastCandidate); + if (moreInSession) { + batch.hasMoreCandidates = true; + batch.pauseCursor = lastCandidate; + break; + } + + if (batch.toolMetadataBytesScanned >= plan.batchBytes) { + const nextSessionId = selectNextTerminalSessionId(sql, plan.cutoffUpdatedAt, sessionId); + if (nextSessionId) { + batch.hasMoreCandidates = true; + batch.pauseCursor = buildSessionExhaustedCursor(sessionId); + batch.finalSessionId = nextSessionId; + break; + } + } + + sessionId = selectNextTerminalSessionId(sql, plan.cutoffUpdatedAt, sessionId); + cursor = null; + } + + batch.finalSessionId = sessionId; + return batch; +} + +function resolveContinuationCursor( + batch: ToolPayloadCleanupBatch, + config: StorageSafetyConfig, + afterBytes: number, + targetBytes: number +): ToolPayloadCleanupCursor | null { + if (batch.hasMoreCandidates) return batch.pauseCursor ?? batch.lastCursor; + const pausedForSessionScanBudget = + afterBytes > targetBytes && + batch.finalSessionId !== null && + batch.sessionsScanned >= config.toolPayloadCleanupMaxSessionsPerAlarm && + batch.lastScannedSessionId !== null; + if (!pausedForSessionScanBudget || !batch.lastScannedSessionId) return null; + return buildSessionExhaustedCursor(batch.lastScannedSessionId); +} + +function persistToolPayloadCleanupState( + sql: SqlStorage, + continuationCursor: ToolPayloadCleanupCursor | null, + recheckAt: number | null +): void { + if (continuationCursor && recheckAt !== null) { + writeToolPayloadCleanupCursor(sql, continuationCursor, recheckAt); + } else { + clearToolPayloadCleanupState(sql); + } +} + +function buildToolPayloadCleanupResult( + plan: ToolPayloadCleanupPlan, + batch: ToolPayloadCleanupBatch, + afterBytes: number, + shouldContinue: boolean, + continuationCursor: ToolPayloadCleanupCursor | null, + exhaustedCandidates: boolean, + recheckAt: number | null +): ProjectDataToolPayloadCleanupResult { + return { + projectId: plan.projectId, + beforeBytes: plan.beforeBytes, + afterBytes, + limitBytes: plan.limitBytes, + triggerBytes: plan.triggerBytes, + targetBytes: plan.targetBytes, + batchRows: plan.batchRows, + batchBytes: plan.batchBytes, + sessionsScanned: batch.sessionsScanned, + rowsScanned: batch.rowsScanned, + rowsUpdated: batch.rowsUpdated, + rowsFailed: batch.rowsFailed, + toolMetadataBytesScanned: batch.toolMetadataBytesScanned, + toolMetadataBytesRead: batch.toolMetadataBytesRead, + originalToolMetadataBytes: batch.originalToolMetadataBytes, + storedToolMetadataBytes: batch.storedToolMetadataBytes, + cursor: shouldContinue ? publicToolPayloadCleanupCursor(continuationCursor) : null, + exhaustedCandidates, + recheckAt, + }; +} + +function summarizeToolPayloadCleanupFailures(batch: ToolPayloadCleanupBatch): string | null { + if (batch.rowsFailed <= 0 && batch.errorMessages.length === 0) return null; + const details = batch.errorMessages.length > 0 ? `: ${batch.errorMessages[0]}` : ''; + return truncate( + `auto tool payload cleanup failed closed ${batch.rowsFailed} candidate(s)${details}`, + 500 + ); +} + +function recordToolPayloadCleanupFailureMeta( + sql: SqlStorage, + projectId: string, + batch: ToolPayloadCleanupBatch +): string | null { + const message = summarizeToolPayloadCleanupFailures(batch); + if (!message) return null; + writeMeta(sql, META_LAST_ERROR, message); + log.warn('candidate_failed_closed', { + projectId, + rowsFailed: batch.rowsFailed, + errors: batch.errorMessages.slice(0, 3), + }); + return message; +} + +async function recordToolPayloadCleanupTelemetry( + sql: SqlStorage, + config: StorageSafetyConfig, + options: ProjectDataToolPayloadCleanupOptions, + projectId: string, + afterBytes: number, + rowsUpdated: number, + lastError: string | null +): Promise { + if (rowsUpdated <= 0 && !lastError) return; + + const measuredAt = Date.now(); + writeMeta(sql, META_LAST_MEASURED_AT, String(measuredAt)); + const statusAfter = options.classifyStatus(afterBytes); + writeMeta(sql, META_LAST_STATUS, statusAfter); + const telemetry: ProjectDataStorageTelemetry = { + projectId, + measuredAt, + databaseSizeBytes: afterBytes, + limitBytes: config.limitBytes, + usageRatio: afterBytes / config.limitBytes, + status: statusAfter, + }; + + try { + await options.recordTelemetry(telemetry, { + lastPurgeAt: rowsUpdated > 0 ? measuredAt : null, + lastPurgeReason: rowsUpdated > 0 ? 'auto_tool_payload_cleanup' : null, + lastPurgeRows: rowsUpdated > 0 ? rowsUpdated : null, + lastPurgeDatabaseSizeBytes: rowsUpdated > 0 ? afterBytes : null, + lastError, + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + writeMeta(sql, META_LAST_ERROR, truncate(message, 500)); + log.warn('telemetry_upsert_failed', { + projectId, + ...serializeError(error), + }); + } +} + +function shouldReturnToolPayloadCleanupResult( + batch: ToolPayloadCleanupBatch, + exhaustedCandidates: boolean, + shouldContinue: boolean +): boolean { + return ( + batch.rowsScanned > 0 || + batch.rowsUpdated > 0 || + batch.rowsFailed > 0 || + exhaustedCandidates || + shouldContinue + ); +} + +function buildFailedToolPayloadCleanupResult( + plan: ToolPayloadCleanupPlan, + recheckAt: number +): ProjectDataToolPayloadCleanupResult { + return { + projectId: plan.projectId, + beforeBytes: plan.beforeBytes, + afterBytes: plan.beforeBytes, + limitBytes: plan.limitBytes, + triggerBytes: plan.triggerBytes, + targetBytes: plan.targetBytes, + batchRows: plan.batchRows, + batchBytes: plan.batchBytes, + sessionsScanned: 0, + rowsScanned: 0, + rowsUpdated: 0, + rowsFailed: 1, + toolMetadataBytesScanned: 0, + toolMetadataBytesRead: 0, + originalToolMetadataBytes: 0, + storedToolMetadataBytes: 0, + cursor: publicToolPayloadCleanupCursor(plan.pendingCursor), + exhaustedCandidates: false, + recheckAt, + }; +} + +async function handleToolPayloadCleanupFailure( + sql: SqlStorage, + config: StorageSafetyConfig, + options: ProjectDataToolPayloadCleanupOptions, + plan: ToolPayloadCleanupPlan, + error: unknown +): Promise { + const recheckAt = plan.now + config.toolPayloadCleanupRecheckMs; + if (plan.pendingCursor) { + writeToolPayloadCleanupCursor(sql, plan.pendingCursor, recheckAt); + } else { + writeToolPayloadCleanupRecheckAt(sql, recheckAt); + } + + const message = truncate(error instanceof Error ? error.message : String(error), 500); + writeMeta(sql, META_LAST_ERROR, message); + log.warn('failed_retry_scheduled', { + projectId: plan.projectId, + recheckAt, + ...serializeError(error), + }); + + await recordToolPayloadCleanupTelemetry( + sql, + config, + options, + plan.projectId, + plan.beforeBytes, + 0, + message + ); + return buildFailedToolPayloadCleanupResult(plan, recheckAt); +} + +export async function runProjectDataToolPayloadCleanup( + sql: SqlStorage, + env: Env, + projectId: string | null, + config: StorageSafetyConfig, + options: ProjectDataToolPayloadCleanupOptions +): Promise { + const plan = createToolPayloadCleanupPlan(sql, projectId, config, options); + if (!plan) return null; + + let batch: ToolPayloadCleanupBatch; + try { + batch = scanToolPayloadCleanupBatch(sql, env, config, plan); + } catch (error) { + return handleToolPayloadCleanupFailure(sql, config, options, plan, error); + } + const afterBytes = sql.databaseSize; + const continuationCursor = resolveContinuationCursor( + batch, + config, + afterBytes, + plan.targetBytes + ); + const shouldContinue = afterBytes > plan.targetBytes && continuationCursor !== null; + const recheckAt = shouldContinue ? plan.now + config.toolPayloadCleanupRecheckMs : null; + persistToolPayloadCleanupState(sql, continuationCursor, recheckAt); + + const exhaustedCandidates = + afterBytes > plan.targetBytes && !shouldContinue && batch.finalSessionId === null; + const result = buildToolPayloadCleanupResult( + plan, + batch, + afterBytes, + shouldContinue, + continuationCursor, + exhaustedCandidates, + recheckAt + ); + const failureMessage = recordToolPayloadCleanupFailureMeta(sql, plan.projectId, batch); + await recordToolPayloadCleanupTelemetry( + sql, + config, + options, + plan.projectId, + afterBytes, + batch.rowsUpdated, + failureMessage + ); + + if (batch.rowsUpdated > 0 || batch.rowsFailed > 0 || exhaustedCandidates) { + log.warn('completed', { ...result }); + } + + return shouldReturnToolPayloadCleanupResult(batch, exhaustedCandidates, shouldContinue) + ? result + : null; +} diff --git a/apps/api/src/durable-objects/project-data/types.ts b/apps/api/src/durable-objects/project-data/types.ts index 98c57612d..a1c8079df 100644 --- a/apps/api/src/durable-objects/project-data/types.ts +++ b/apps/api/src/durable-objects/project-data/types.ts @@ -45,6 +45,14 @@ export type Env = { PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO?: string; PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS?: string; PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM?: string; ACTIVITY_RETENTION_DAYS?: string; SESSION_IDLE_TIMEOUT_MINUTES?: string; IDLE_CLEANUP_RETRY_DELAY_MS?: string; diff --git a/apps/api/src/env.ts b/apps/api/src/env.ts index af9d319af..e5ba4bc3b 100644 --- a/apps/api/src/env.ts +++ b/apps/api/src/env.ts @@ -533,6 +533,14 @@ export interface Env extends WebhookTriggerEnv, TaskRecoveryEnv { PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO?: string; PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS?: string; PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS?: string; + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM?: string; MESSAGE_SIZE_THRESHOLD?: string; ACTIVITY_RETENTION_DAYS?: string; SESSION_IDLE_TIMEOUT_MINUTES?: string; diff --git a/apps/api/src/services/durable-object-retry.ts b/apps/api/src/services/durable-object-retry.ts index bdb45a2de..834c543a7 100644 --- a/apps/api/src/services/durable-object-retry.ts +++ b/apps/api/src/services/durable-object-retry.ts @@ -14,6 +14,7 @@ const TRANSIENT_DURABLE_OBJECT_PATTERNS = [ const DURABLE_OBJECT_STORAGE_FULL_PATTERNS = [ /\bSQLITE_FULL\b/i, /database or disk is full/i, + /exceeded the maximum database size/i, /durable object.*storage.*full/i, /sqlite.*full/i, ]; diff --git a/apps/api/src/services/project-data-storage-errors.ts b/apps/api/src/services/project-data-storage-errors.ts index 777fd3cac..7436ee750 100644 --- a/apps/api/src/services/project-data-storage-errors.ts +++ b/apps/api/src/services/project-data-storage-errors.ts @@ -7,7 +7,7 @@ export class ProjectDataStorageFullError extends AppError { super( 507, PROJECT_DATA_STORAGE_FULL, - 'ProjectData storage is full; writes are paused until an administrator runs storage recovery.', + 'ProjectData storage is full; storage recovery is required before this write can complete.', { projectId, operation, diff --git a/apps/api/tests/unit/services/durable-object-retry.test.ts b/apps/api/tests/unit/services/durable-object-retry.test.ts index c6310f368..693d74d94 100644 --- a/apps/api/tests/unit/services/durable-object-retry.test.ts +++ b/apps/api/tests/unit/services/durable-object-retry.test.ts @@ -49,6 +49,9 @@ describe('isDurableObjectStorageFullError', () => { it('matches Cloudflare/SQLite full-storage variants', () => { expect(isDurableObjectStorageFullError(new Error('SQLITE_FULL'))).toBe(true); expect(isDurableObjectStorageFullError(new Error('database or disk is full'))).toBe(true); + expect(isDurableObjectStorageFullError(new Error('Exceeded the maximum database size.'))).toBe( + true + ); expect(isDurableObjectStorageFullError(new Error('sqlite full while inserting'))).toBe(true); }); diff --git a/apps/api/tests/workers/project-data-storage-safety.test.ts b/apps/api/tests/workers/project-data-storage-safety.test.ts index de44eb964..6d7fe8ce6 100644 --- a/apps/api/tests/workers/project-data-storage-safety.test.ts +++ b/apps/api/tests/workers/project-data-storage-safety.test.ts @@ -66,7 +66,8 @@ async function readTelemetry(projectId: string) { usage_ratio, status, last_alarm_at, - last_purge_rows + last_purge_rows, + last_error FROM project_data_storage_telemetry WHERE project_id = ?` ) @@ -80,9 +81,37 @@ async function readTelemetry(projectId: string) { status: ProjectDataStorageStatus; last_alarm_at: number | null; last_purge_rows: number | null; + last_error: string | null; }>(); } +function makeToolMetadata(label: string): string { + return JSON.stringify({ + toolCallId: `tool-${label}`, + title: `Tool ${label}`, + status: 'completed', + content: [{ type: 'text', text: `${label}:${'x'.repeat(64 * 1024)}` }], + }); +} + +function makeLegacyToolMetadata(label: string, payloadBytes: number): string { + return JSON.stringify({ + toolCallId: `tool-${label}`, + title: `Tool ${label}`, + status: 'completed', + content: [{ type: 'text', text: `${label}:${'x'.repeat(payloadBytes)}` }], + }); +} + +function makePoisonToolMetadata(label: string): string { + return JSON.stringify([ + { + toolCallId: `tool-${label}`, + content: [{ type: 'text', text: `${label}:poison` }], + }, + ]); +} + describe('ProjectData storage safety firebreak', () => { it('databaseSize drops after deleting rows in the workerd SQLite DO runtime', async () => { const projectId = `storage-size-reclaim-${crypto.randomUUID()}`; @@ -211,6 +240,503 @@ describe('ProjectData storage safety firebreak', () => { ); }); + it('honors the storage measurement interval when unrelated alarms fire', async () => { + const projectId = `storage-measurement-due-${crypto.randomUUID()}`; + await seedProjectGraph(projectId); + const stub = getStub(projectId); + await stub.ensureProjectId(projectId); + await stub.createSession(null, 'Storage measurement cadence'); + + await withProjectDataStorageEnv( + { PROJECT_DATA_STORAGE_MEASURE_INTERVAL_MS: '86400000' }, + async () => { + await runInDurableObject(stub, async (instance) => instance.alarm()); + const first = await readTelemetry(projectId); + expect(first?.measured_at).toBeGreaterThan(0); + + await runInDurableObject(stub, async (instance) => instance.alarm()); + const second = await readTelemetry(projectId); + expect(second?.measured_at).toBe(first?.measured_at); + expect(second?.last_alarm_at).toBe(first?.last_alarm_at); + } + ); + }); + + it('strips old terminal tool payloads in bounded cleanup batches and resumes by cursor', async () => { + const projectId = `storage-tool-cleanup-${crypto.randomUUID()}`; + await seedProjectGraph(projectId); + const stub = getStub(projectId); + await stub.ensureProjectId(projectId); + + const messageIds = await runInDurableObject(stub, async (instance) => { + const stoppedSession = await instance.createSession(null, 'Old terminal tool payloads'); + const activeSession = await instance.createSession(null, 'Active tool payload'); + const sleepingSession = await instance.createSession(null, 'Sleeping tool payload'); + + const stoppedOne = await instance.persistMessage( + stoppedSession, + 'tool', + 'visible stopped one', + makeToolMetadata('stopped-one'), + 'tool-stopped-one' + ); + const stoppedTwo = await instance.persistMessage( + stoppedSession, + 'tool', + 'visible stopped two', + makeToolMetadata('stopped-two'), + 'tool-stopped-two' + ); + const stoppedThree = await instance.persistMessage( + stoppedSession, + 'tool', + 'visible stopped three', + makeToolMetadata('stopped-three'), + 'tool-stopped-three' + ); + const active = await instance.persistMessage( + activeSession, + 'tool', + 'visible active', + makeToolMetadata('active'), + 'tool-active' + ); + const sleeping = await instance.persistMessage( + sleepingSession, + 'tool', + 'visible sleeping', + makeToolMetadata('sleeping'), + 'tool-sleeping' + ); + + await instance.stopSession(stoppedSession); + await instance.sleepSession(sleepingSession); + + return { stoppedOne, stoppedTwo, stoppedThree, active, sleeping }; + }); + + await withProjectDataStorageEnv( + { + PROJECT_DATA_STORAGE_LIMIT_BYTES: '10000', + PROJECT_DATA_STORAGE_MEASURE_INTERVAL_MS: '86400000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO: '0.2', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO: '0.1', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS: '2', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS: '0', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS: '60000', + }, + async () => { + const first = await runInDurableObject(stub, async (instance, state) => { + const before = state.storage.sql.databaseSize; + await instance.alarm(); + const after = state.storage.sql.databaseSize; + const rows = state.storage.sql + .exec( + `SELECT id, content, tool_metadata + FROM chat_messages + WHERE id IN (?, ?, ?, ?, ?) + ORDER BY id ASC`, + messageIds.stoppedOne, + messageIds.stoppedTwo, + messageIds.stoppedThree, + messageIds.active, + messageIds.sleeping + ) + .toArray() as Array<{ id: string; content: string; tool_metadata: string }>; + const alarm = await state.storage.getAlarm(); + return { before, after, rows, alarm }; + }); + + expect(first.after).toBeLessThan(first.before); + expect(first.alarm).toBeTypeOf('number'); + + const firstById = new Map(first.rows.map((row) => [row.id, row])); + const stoppedOneMeta = JSON.parse( + firstById.get(messageIds.stoppedOne)?.tool_metadata ?? '{}' + ) as Record; + const stoppedTwoMeta = JSON.parse( + firstById.get(messageIds.stoppedTwo)?.tool_metadata ?? '{}' + ) as Record; + const stoppedThreeMeta = JSON.parse( + firstById.get(messageIds.stoppedThree)?.tool_metadata ?? '{}' + ) as Record; + const activeMeta = JSON.parse( + firstById.get(messageIds.active)?.tool_metadata ?? '{}' + ) as Record; + const sleepingMeta = JSON.parse( + firstById.get(messageIds.sleeping)?.tool_metadata ?? '{}' + ) as Record; + + expect(stoppedOneMeta.content).toBeUndefined(); + expect(stoppedOneMeta.contentSize).toBeGreaterThan(0); + expect(stoppedOneMeta.toolCallId).toBe('tool-stopped-one'); + expect(stoppedTwoMeta.content).toBeUndefined(); + expect(stoppedTwoMeta.contentSize).toBeGreaterThan(0); + expect(Array.isArray(stoppedThreeMeta.content)).toBe(true); + expect(Array.isArray(activeMeta.content)).toBe(true); + expect(Array.isArray(sleepingMeta.content)).toBe(true); + expect(firstById.get(messageIds.stoppedOne)?.content).toBe('visible stopped one'); + + await runInDurableObject(stub, async (instance) => instance.alarm()); + const early = await runInDurableObject(stub, async (_instance, state) => + state.storage.sql + .exec('SELECT tool_metadata FROM chat_messages WHERE id = ?', messageIds.stoppedThree) + .toArray()[0] + ) as { tool_metadata: string }; + expect(Array.isArray((JSON.parse(early.tool_metadata) as Record).content)) + .toBe(true); + + await runInDurableObject(stub, async (_instance, state) => { + state.storage.sql.exec( + `UPDATE do_meta + SET value = ? + WHERE key = 'storageSafetyToolCleanupRecheckAt'`, + String(Date.now() - 1) + ); + }); + + await runInDurableObject(stub, async (instance) => instance.alarm()); + + const second = await runInDurableObject(stub, async (_instance, state) => { + const rows = state.storage.sql + .exec( + `SELECT id, tool_metadata + FROM chat_messages + WHERE id IN (?, ?, ?, ?, ?) + ORDER BY id ASC`, + messageIds.stoppedOne, + messageIds.stoppedTwo, + messageIds.stoppedThree, + messageIds.active, + messageIds.sleeping + ) + .toArray() as Array<{ id: string; tool_metadata: string }>; + const alarm = await state.storage.getAlarm(); + return { rows, alarm }; + }); + const secondById = new Map(second.rows.map((row) => [row.id, row])); + const stoppedThreeAfter = JSON.parse( + secondById.get(messageIds.stoppedThree)?.tool_metadata ?? '{}' + ) as Record; + const activeAfter = JSON.parse( + secondById.get(messageIds.active)?.tool_metadata ?? '{}' + ) as Record; + const sleepingAfter = JSON.parse( + secondById.get(messageIds.sleeping)?.tool_metadata ?? '{}' + ) as Record; + const telemetry = await readTelemetry(projectId); + + expect(stoppedThreeAfter.content).toBeUndefined(); + expect(stoppedThreeAfter.contentSize).toBeGreaterThan(0); + expect(Array.isArray(activeAfter.content)).toBe(true); + expect(Array.isArray(sleepingAfter.content)).toBe(true); + expect(telemetry?.last_purge_rows).toBe(1); + expect(second.alarm).toBeTypeOf('number'); + } + ); + }); + + it('bounds cumulative legacy tool metadata bytes even when row limit is high', async () => { + const projectId = `storage-tool-cleanup-byte-budget-${crypto.randomUUID()}`; + await seedProjectGraph(projectId); + const stub = getStub(projectId); + await stub.ensureProjectId(projectId); + + const messageIds = await runInDurableObject(stub, async (instance) => { + const sessionId = await instance.createSession(null, 'Byte-bounded terminal payloads'); + const ids: string[] = []; + for (let index = 0; index < 4; index++) { + ids.push( + await instance.persistMessage( + sessionId, + 'tool', + `visible byte bounded ${index}`, + makeToolMetadata(`byte-bounded-${index}`), + `tool-byte-bounded-${index}` + ) + ); + } + await instance.stopSession(sessionId); + return ids; + }); + + await withProjectDataStorageEnv( + { + PROJECT_DATA_STORAGE_LIMIT_BYTES: '10000', + PROJECT_DATA_STORAGE_MEASURE_INTERVAL_MS: '86400000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO: '0.2', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO: '0.1', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS: '500', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES: '150000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS: '0', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS: '60000', + }, + async () => { + await runInDurableObject(stub, async (instance) => instance.alarm()); + + const rows = await runInDurableObject(stub, async (_instance, state) => + state.storage.sql + .exec( + `SELECT id, tool_metadata + FROM chat_messages + WHERE id IN (?, ?, ?, ?) + ORDER BY created_at ASC, COALESCE(sequence, 0) ASC, id ASC`, + messageIds[0], + messageIds[1], + messageIds[2], + messageIds[3] + ) + .toArray() + ) as Array<{ id: string; tool_metadata: string }>; + + const metadata = rows.map((row) => JSON.parse(row.tool_metadata) as Record); + expect(metadata[0]?.content).toBeUndefined(); + expect(metadata[1]?.content).toBeUndefined(); + expect(Array.isArray(metadata[2]?.content)).toBe(true); + expect(Array.isArray(metadata[3]?.content)).toBe(true); + + const alarm = await runInDurableObject(stub, async (_instance, state) => + state.storage.getAlarm() + ); + const telemetry = await readTelemetry(projectId); + expect(telemetry?.last_purge_rows).toBe(2); + expect(alarm).toBeTypeOf('number'); + expect(alarm as number).toBeGreaterThan(Date.now()); + } + ); + }); + + it('quarantines an oversized single legacy metadata row and resumes after it', async () => { + const projectId = `storage-tool-cleanup-oversized-${crypto.randomUUID()}`; + await seedProjectGraph(projectId); + const stub = getStub(projectId); + await stub.ensureProjectId(projectId); + + const messageIds = await runInDurableObject(stub, async (instance, state) => { + const sessionId = await instance.createSession(null, 'Oversized terminal payload'); + const oversized = await instance.persistMessage( + sessionId, + 'tool', + 'visible oversized', + makeToolMetadata('oversized-placeholder'), + 'tool-oversized' + ); + const next = await instance.persistMessage( + sessionId, + 'tool', + 'visible after oversized', + makeToolMetadata('after-oversized'), + 'tool-after-oversized' + ); + await instance.stopSession(sessionId); + state.storage.sql.exec( + 'UPDATE chat_messages SET tool_metadata = ? WHERE id = ?', + makeLegacyToolMetadata('oversized-legacy', 220 * 1024), + oversized + ); + return { oversized, next }; + }); + + await withProjectDataStorageEnv( + { + PROJECT_DATA_STORAGE_LIMIT_BYTES: '10000', + PROJECT_DATA_STORAGE_MEASURE_INTERVAL_MS: '86400000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO: '0.2', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO: '0.1', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS: '500', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES: '100000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS: '0', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS: '60000', + }, + async () => { + await runInDurableObject(stub, async (instance) => instance.alarm()); + + const firstPass = await runInDurableObject(stub, async (_instance, state) => + state.storage.sql + .exec( + `SELECT id, tool_metadata + FROM chat_messages + WHERE id IN (?, ?) + ORDER BY created_at ASC, COALESCE(sequence, 0) ASC, id ASC`, + messageIds.oversized, + messageIds.next + ) + .toArray() + ) as Array<{ id: string; tool_metadata: string }>; + const firstMetadata = firstPass.map( + (row) => JSON.parse(row.tool_metadata) as Record + ); + expect(firstMetadata[0]).toMatchObject({ + storageSafetyTruncated: true, + contentTruncated: true, + storageSafetyCleanupReason: 'oversized_legacy_payload', + }); + expect(Array.isArray(firstMetadata[1]?.content)).toBe(true); + + await runInDurableObject(stub, async (_instance, state) => { + state.storage.sql.exec( + `UPDATE do_meta + SET value = ? + WHERE key = 'storageSafetyToolCleanupRecheckAt'`, + String(Date.now() - 1) + ); + }); + + await runInDurableObject(stub, async (instance) => instance.alarm()); + const secondPass = await runInDurableObject(stub, async (_instance, state) => + state.storage.sql + .exec('SELECT tool_metadata FROM chat_messages WHERE id = ?', messageIds.next) + .toArray()[0] + ) as { tool_metadata: string }; + const nextMetadata = JSON.parse(secondPass.tool_metadata) as Record; + expect(nextMetadata.content).toBeUndefined(); + expect(nextMetadata.contentSize).toBeGreaterThan(0); + } + ); + }); + + it('fail-closes poison candidates and clears stale due rechecks without alarm thrash', async () => { + const projectId = `storage-tool-cleanup-poison-${crypto.randomUUID()}`; + await seedProjectGraph(projectId); + const stub = getStub(projectId); + await stub.ensureProjectId(projectId); + + const messageIds = await runInDurableObject(stub, async (instance, state) => { + const sessionId = await instance.createSession(null, 'Poison terminal payload'); + const poison = await instance.persistMessage( + sessionId, + 'tool', + 'visible poison', + makeToolMetadata('poison-placeholder'), + 'tool-poison' + ); + const valid = await instance.persistMessage( + sessionId, + 'tool', + 'visible valid after poison', + makeToolMetadata('valid-after-poison'), + 'tool-valid-after-poison' + ); + await instance.stopSession(sessionId); + state.storage.sql.exec( + 'UPDATE chat_messages SET tool_metadata = ? WHERE id = ?', + makePoisonToolMetadata('legacy-poison'), + poison + ); + state.storage.sql.exec( + `INSERT INTO do_meta (key, value) + VALUES ('storageSafetyLastMeasuredAt', ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + String(Date.now()) + ); + state.storage.sql.exec( + `INSERT INTO do_meta (key, value) + VALUES ('storageSafetyToolCleanupRecheckAt', ?) + ON CONFLICT(key) DO UPDATE SET value = excluded.value`, + String(Date.now() - 10_000) + ); + return { poison, valid }; + }); + + await withProjectDataStorageEnv( + { + PROJECT_DATA_STORAGE_LIMIT_BYTES: '10000', + PROJECT_DATA_STORAGE_MEASURE_INTERVAL_MS: '86400000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO: '0.2', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO: '0.1', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS: '500', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES: '200000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS: '0', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS: '60000', + }, + async () => { + await runInDurableObject(stub, async (instance) => instance.alarm()); + + const stateAfter = await runInDurableObject(stub, async (_instance, state) => { + const rows = state.storage.sql + .exec( + `SELECT id, tool_metadata + FROM chat_messages + WHERE id IN (?, ?) + ORDER BY created_at ASC, COALESCE(sequence, 0) ASC, id ASC`, + messageIds.poison, + messageIds.valid + ) + .toArray() as Array<{ id: string; tool_metadata: string }>; + const metaRows = state.storage.sql + .exec( + `SELECT key, value + FROM do_meta + WHERE key IN ('storageSafetyToolCleanupRecheckAt', 'storageSafetyLastError') + ORDER BY key ASC` + ) + .toArray() as Array<{ key: string; value: string }>; + const alarm = await state.storage.getAlarm(); + return { rows, metaRows, alarm }; + }); + const metadata = stateAfter.rows.map( + (row) => JSON.parse(row.tool_metadata) as Record + ); + const metaByKey = new Map(stateAfter.metaRows.map((row) => [row.key, row.value])); + + expect(metadata[0]).toMatchObject({ + storageSafetyTruncated: true, + contentTruncated: true, + storageSafetyCleanupReason: 'poison_legacy_payload', + }); + expect(metadata[1]?.content).toBeUndefined(); + expect(metadata[1]?.contentSize).toBeGreaterThan(0); + expect(metaByKey.has('storageSafetyToolCleanupRecheckAt')).toBe(false); + expect(metaByKey.get('storageSafetyLastError')).toMatch(/failed closed 1 candidate/); + expect(stateAfter.alarm).toBeTypeOf('number'); + expect(stateAfter.alarm as number).toBeGreaterThan(Date.now()); + + const telemetry = await readTelemetry(projectId); + expect(telemetry?.last_error).toMatch(/failed closed 1 candidate/); + } + ); + }); + + it('preserves recent terminal tool payloads until the configured age floor passes', async () => { + const projectId = `storage-tool-cleanup-age-${crypto.randomUUID()}`; + await seedProjectGraph(projectId); + const stub = getStub(projectId); + await stub.ensureProjectId(projectId); + + const messageId = await runInDurableObject(stub, async (instance) => { + const sessionId = await instance.createSession(null, 'Recent terminal payload'); + const id = await instance.persistMessage( + sessionId, + 'tool', + 'visible recent terminal', + makeToolMetadata('recent-terminal'), + 'tool-recent-terminal' + ); + await instance.stopSession(sessionId); + return id; + }); + + await withProjectDataStorageEnv( + { + PROJECT_DATA_STORAGE_LIMIT_BYTES: '10000', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO: '0.2', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO: '0.1', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS: '10', + PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS: '1', + }, + async () => { + await runInDurableObject(stub, async (instance) => instance.alarm()); + const row = await runInDurableObject(stub, async (_instance, state) => + state.storage.sql + .exec('SELECT tool_metadata FROM chat_messages WHERE id = ?', messageId) + .toArray()[0] + ) as { tool_metadata: string }; + const meta = JSON.parse(row.tool_metadata) as Record; + expect(Array.isArray(meta.content)).toBe(true); + } + ); + }); + it('service measurement writes ProjectData storage telemetry directly', async () => { const projectId = `storage-service-measure-${crypto.randomUUID()}`; await seedProjectGraph(projectId); diff --git a/apps/api/wrangler.toml b/apps/api/wrangler.toml index ceb20d38f..c8be7bfb8 100644 --- a/apps/api/wrangler.toml +++ b/apps/api/wrangler.toml @@ -72,6 +72,14 @@ PROJECT_DATA_STORAGE_DEGRADED_RATIO = "0.95" PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO = "0.9" PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS = "500" PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES = "4" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED = "true" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO = "0.8" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO = "0.75" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS = "500" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES = "1048576" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS = "7" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS = "60000" +PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM = "25" MESSAGE_SIZE_THRESHOLD = "102400" ACTIVITY_RETENTION_DAYS = "90" SESSION_IDLE_TIMEOUT_MINUTES = "60" diff --git a/apps/www/src/content/docs/docs/reference/configuration.md b/apps/www/src/content/docs/docs/reference/configuration.md index dbce56963..6ea23e1b0 100644 --- a/apps/www/src/content/docs/docs/reference/configuration.md +++ b/apps/www/src/content/docs/docs/reference/configuration.md @@ -808,6 +808,14 @@ ProjectData stores a single prompt-delivery queue and checkpoint episodes keyed | `PROJECT_DATA_STORAGE_EMERGENCY_TARGET_RATIO` | `0.9` | Target usage ratio for explicit superadmin ProjectData emergency purge calls | | `PROJECT_DATA_STORAGE_EMERGENCY_BATCH_ROWS` | `500` | Oldest `activity_events` and `acp_session_events` rows deleted per table per emergency purge batch | | `PROJECT_DATA_STORAGE_EMERGENCY_MAX_BATCHES` | `4` | Maximum emergency purge batches per explicit call | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_ENABLED` | `true` | Enables automatic ProjectData cleanup that strips expandable `tool_metadata.content` payloads from old terminal-session tool messages under storage pressure | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TRIGGER_RATIO` | `0.8` | ProjectData storage usage ratio that starts automatic terminal-session tool payload cleanup | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_TARGET_RATIO` | `0.75` | ProjectData storage usage ratio below which automatic tool payload cleanup stops | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_ROWS` | `500` | Maximum tool-message rows inspected by one automatic cleanup alarm batch | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_BATCH_BYTES` | `1048576` | Maximum legacy `tool_metadata` bytes read into JS by one automatic cleanup alarm batch | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MIN_SESSION_AGE_DAYS` | `7` | Minimum terminal-session age before automatic cleanup may strip stored tool payload content | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_RECHECK_MS` | `60000` | Delay before the next automatic cleanup alarm batch when more candidates remain | +| `PROJECT_DATA_TOOL_PAYLOAD_CLEANUP_MAX_SESSIONS_PER_ALARM` | `25` | Maximum terminal sessions scanned by one automatic cleanup alarm batch | | `MESSAGE_SIZE_THRESHOLD` | `102400` | Max message size in bytes | | `ACTIVITY_RETENTION_DAYS` | `90` | Days to retain activity events | | `SESSION_IDLE_TIMEOUT_MINUTES` | `60` | Idle session timeout | diff --git a/tasks/archive/2026-08-24-projectdata-storage-protection.md b/tasks/archive/2026-08-24-projectdata-storage-protection.md new file mode 100644 index 000000000..b753d47a8 --- /dev/null +++ b/tasks/archive/2026-08-24-projectdata-storage-protection.md @@ -0,0 +1,152 @@ +# ProjectData storage-protection firebreak + +## Problem + +SAM's dogfooding ProjectData Durable Object hit Cloudflare's hard SQLite-backed +10 GB per-object ceiling on 2026-08-18. A manual purge of tool-call JSON payloads +reclaimed about 3.14 GB, but production telemetry on 2026-08-24 shows the same +object is growing back toward the wall. PR #1875 measures and classifies storage +pressure, but it does not automatically brake growth. PR #1873 is broader +sharding infrastructure and must not be landed wholesale. + +Build one focused, independently shippable protection patch that automatically +reclaims low-value storage in bounded Durable Object alarm batches while +preserving high-value chat history. + +## Research findings + +- Current production D1 telemetry (read-only query on 2026-08-24) shows project + `01KHRJGANBBWGDY1NZ0KVF0D4J` at `8,326,971,392` bytes / `10,000,000,000` + bytes, status `warning`, with `last_alert_at` and `last_purge_at` still NULL. +- Observability D1 contains 102 `Exceeded the maximum database size.` errors in + the last 7 days, all from August 18. The literal production message exists in + addition to the `SQLITE_FULL` phrasing. +- The downloaded investigation report + `.library/projectdata-do-storage-ceiling-status-2026-08-24.md/...` confirms + the −3.14 GB drop was Raphaël's manual purge of tool-call JSON data, making + `chat_messages.tool_metadata` / large tool payload JSON the dominant consumer. +- Current main already has `storage-safety.ts`, per-object `sql.databaseSize` + telemetry, a superadmin emergency purge for `activity_events` / + `acp_session_events`, and a 128 KiB write-path cap for new `tool_metadata`; + it does not auto-reclaim legacy dominant tool payload rows under storage + pressure. +- `measureAndPersistProjectDataStorage()` currently runs on every ProjectData + alarm tick, even when the storage measurement interval is not due. Because + ProjectData alarms are multiplexed with faster control loops, this causes more + D1 telemetry writes than the configured interval implies. +- PR #1873's sharding draft uses the wrong PRAGMA sizing formula and can keep + re-arming after DELETEs. It also has large-session `.toArray()` migration OOM + risk and poison-candidate control-loop risk. +- Cloudflare DO storage quota must be measured with + `ctx.storage.sql.databaseSize`; `DELETE` reclaims quota immediately because + freelist pages are subtracted. Do not use `page_count * page_size`, `VACUUM`, + or `auto_vacuum`. +- For Durable Object cleanup, use raw row access with bounded LIMIT/keyset + batches across separate alarm calls. Fully consume cursors before any await; + never materialize large sweeps with `.toArray()`. +- Independent review of PR #1901 found the first cleanup implementation was + row-bounded but not byte-bounded: it selected full `tool_metadata` values and + materialized up to 500 legacy payloads. Legacy individual payloads may be + much larger than the write-path cap, so cleanup must bound bytes read into JS, + not just candidate rows. +- The same review found poison-candidate risk: one cleanup candidate that throws + during strip/update can prevent cursor advancement and leave a stale + `storageSafetyToolCleanupRecheckAt`, making ProjectData alarms re-arm + immediately. +- Safe cleanup target: terminal-session `tool_metadata.content` payloads. + Stripping the heavy structured payload preserves the chat row, role, text + content, tool identity/status, and `contentSize`; recent/active/sleeping + history remains protected by a configurable age floor and terminal-status + predicate. + +## Implementation checklist + +- [x] Add configurable automated storage-cleanup settings to the ProjectData Env + surface and public/internal env references. +- [x] Add storage-safety helpers that decide when cleanup is due, persist cleanup + cursors/stats in `do_meta`, and schedule cleanup rechecks separately from + hourly measurement. +- [x] Implement one bounded keyset/LIMIT cleanup batch per alarm that strips + `tool_metadata.content` from old terminal-session rows using raw cursor access. +- [x] Add a configurable cleanup byte budget so candidate selection reads only + row identity plus `length(CAST(tool_metadata AS BLOB))`, then reads full + legacy metadata only when it fits the remaining per-alarm byte budget. +- [x] Add fail-closed oversized/poison candidate handling that writes a small + sentinel, records failure metadata, advances the cursor, and prevents stale + due recheck alarms from hot-looping. +- [x] Fix alarm measurement gating so storage measurement runs only when due + unless explicitly forced by an admin call. +- [x] Ensure the ProjectData alarm isolates storage measurement/cleanup failures + from unrelated control-loop steps and recalculates the next alarm candidate. +- [x] Add scenario-driven Worker-runtime tests for due measurement gating, + terminal-tool cleanup, active/sleeping preservation, bounded batches, cursor + continuation, and telemetry/purge metadata. +- [x] Run focused local checks and specialist validation. +- [x] Create a focused draft/do-not-merge PR and wait for CI evidence; do not + deploy to staging and do not merge. + +## Acceptance criteria + +- Automated cleanup begins when `sql.databaseSize / configured limit` reaches a + configurable trigger ratio and stops when it reaches a configurable target + ratio or candidate rows are exhausted. +- Cleanup processes at most one bounded batch per alarm call and continues via + persisted keyset cursor/recheck alarm, not a long synchronous sweep. +- Cleanup is byte/memory-bounded for legacy metadata: it does not select or + materialize full `tool_metadata` batches, and full metadata reads are capped + by a configurable per-alarm byte budget. +- A single oversized or poison legacy row cannot stall the sweep; it is handled + fail-closed with observability, cursor progress, and a sane future alarm. +- Cleanup uses raw row access and fully consumes the cursor before updates or + awaits. +- Cleanup preserves active/sleeping sessions and recent terminal sessions by + default, and it does not delete chat messages or user/assistant transcript + text. +- Existing explicit emergency purge remains available. +- `measureIntervalMs` is honored even when unrelated ProjectData alarms fire + more frequently. +- Tests prove the cleanup shrinks `tool_metadata`, preserves high-value rows, + respects row and byte bounds, handles oversized/poison candidates, clears stale + due rechecks, avoids alarm thrash, and records telemetry/purge metadata. + +## Validation + +- Draft/do-not-merge PR: +- `needs-human-review` label added because spawned local review agents timed out + before returning results. Manual local specialist checklists passed; do not + merge until human review clears the label. +- `pnpm --filter @simple-agent-manager/shared build && pnpm --filter @simple-agent-manager/providers build && pnpm --filter @simple-agent-manager/cloud-init build` +- `pnpm --filter @simple-agent-manager/api test -- tests/unit/services/durable-object-retry.test.ts` +- `pnpm --filter @simple-agent-manager/api typecheck` +- `pnpm --filter @simple-agent-manager/api build` +- `(cd apps/api && pnpm vitest run --config vitest.workers.config.ts tests/workers/project-data-storage-safety.test.ts --reporter verbose)` — 11 tests passed after adding byte-budget, oversized-row, poison-row, stale-recheck, and non-thrashing alarm coverage. +- `pnpm --filter @simple-agent-manager/api test -- tests/unit/durable-objects/project-data-messages.test.ts` +- `pnpm --filter @simple-agent-manager/api lint` +- `pnpm quality:file-sizes` — passed after extracting automated tool-payload + cleanup candidate processing into `tool-payload-cleanup-candidates.ts`; no + files exceed 800 lines. +- `pnpm quality:ast-checks` — passed after splitting the cleanup candidate + query into static SQL branches; existing warnings remain unrelated. +- `pnpm lint` — passed with pre-existing warnings unrelated to this patch. +- `pnpm format:check` — Prettier format ratchet passed. +- `git diff --check HEAD` + +## Specialist review notes + +- Cloudflare/DO review: cleanup uses `ctx.storage.sql.databaseSize`, keeps + alarm work isolated, uses bounded LIMIT/keyset batches over row identity plus + metadata byte length, and persists a recheck cursor instead of running a long + sweep. +- Data-loss review: automated cleanup strips only expandable + `tool_metadata.content` from old terminal sessions; it preserves chat rows, + message text, active/sleeping sessions, and recent terminal sessions by + default. +- Constitution/env review: operational thresholds, limits, batch sizes, age + floors, byte budgets, and recheck cadence are configurable via env vars and + documented in Worker env types, examples, `wrangler.toml`, public docs, and + `env-reference`. +- Test review: worker-runtime scenarios cover quota measurement, delete reclaim + semantics, interval gating, bounded cleanup, cursor continuation, + active/sleeping preservation, age-floor preservation, oversized and poison + legacy metadata, stale recheck clearing, non-thrashing alarm scheduling, and + telemetry metadata.