From 02884ed8d2a9d6ff1a25e49e414fa5a97af82046 Mon Sep 17 00:00:00 2001 From: Paco Cartones <253313177+pacocartones@users.noreply.github.com> Date: Fri, 11 Sep 2026 22:41:53 +0200 Subject: [PATCH] fix(db): auto-clean conversation_turn_nodes and orphaned conversations (#12548) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two tables with no retention path at all and 775 MB of a 1.1 GB database is a real operational failure. Tying them to the existing `retention.callLogs` window rather than inventing a knob is right, and the reasoning is what makes it safe: once `cleanupCallLogs` purges the row `last_correlation_id` points at, the node can never render again. --- Validated in one consolidated worktree cut from `release/v3.8.51`, boarded with the rest of this batch — zero conflicts between the 19 PRs. - `typecheck:core` clean; `check:dashboard-typecheck` OK (206 pre-existing, all within the frozen baseline); `check:changelog-integrity` OK - complexity 2802 / baseline 3218 and cognitive-complexity 1267 / baseline 1437 — both under baseline - 226 of 228 focused assertions green across the batch's 23 test files. The 2 remaining belong to #12551, which is held separately. Two batch-owned defects were found and fixed in flight, both pure base drift: `173_xp_action_counts.sql` collided with `173_call_logs_video_content_removed.sql` (renumbered to 176 on #12651 — it aborted every DB open, which is what 53 of the first run's failures were), and the feature-flag catalog was missing the `SERVER_OWNED_TOOL_LOOP_ENABLED` row the base gained after #12552 was written. ⚠️ base-red inherited: #12732 — `Docs Gates`, `Merge integrity`, `No new ESLint warnings`, `Unit Tests fast-path` and `Fast Quality Gates` reproduce on the pure tip (provider count 356 vs the 358 the modules define, SKILL.md drift, and `open-sse/utils/stream.ts` at 3115 > frozen 3098, which this batch does not touch). Thanks @pacocartones — the `file:line` citations and the explicit out-of-scope notes on every one of these made a 19-PR batch reviewable in one pass. --- ...-db-cleanup-orphaned-conversation-nodes.md | 1 + .../api/settings/purge-usage-history/route.ts | 2 + src/lib/db/cleanup.ts | 128 +++++++++++- src/lib/db/cleanup/usagePurge.ts | 61 ++++-- ...b-cleanup-conversation-nodes-12453.test.ts | 190 ++++++++++++++++++ tests/unit/usage-history-reset.test.ts | 59 ++++++ 6 files changed, 421 insertions(+), 20 deletions(-) create mode 100644 changelog.d/fixes/12548-db-cleanup-orphaned-conversation-nodes.md create mode 100644 tests/unit/db-cleanup-conversation-nodes-12453.test.ts diff --git a/changelog.d/fixes/12548-db-cleanup-orphaned-conversation-nodes.md b/changelog.d/fixes/12548-db-cleanup-orphaned-conversation-nodes.md new file mode 100644 index 0000000000..1a5a015518 --- /dev/null +++ b/changelog.d/fixes/12548-db-cleanup-orphaned-conversation-nodes.md @@ -0,0 +1 @@ +- **fix(db):** Add `conversation_turn_nodes` and orphaned `agentic_conversations` to the auto-cleanup cycle under the existing `retention.callLogs` window, so identity nodes whose call-log content has already been purged no longer accumulate without bound in `storage.sqlite` (#12548 — thanks @pacocartones) diff --git a/src/app/api/settings/purge-usage-history/route.ts b/src/app/api/settings/purge-usage-history/route.ts index 12d1413567..7cedcf5e6c 100644 --- a/src/app/api/settings/purge-usage-history/route.ts +++ b/src/app/api/settings/purge-usage-history/route.ts @@ -54,6 +54,8 @@ export async function POST(request: Request) { deletedRoutingDecisions: result.deletedRoutingDecisions, deletedQuotaConsumption: result.deletedQuotaConsumption, deletedTokenLedger: result.deletedTokenLedger, + deletedConversationTurnNodes: result.deletedConversationTurnNodes, + deletedAgenticConversations: result.deletedAgenticConversations, errors: result.errors, }, { status: result.errors > 0 ? 500 : 200 } diff --git a/src/lib/db/cleanup.ts b/src/lib/db/cleanup.ts index a837909df7..01e84a7964 100644 --- a/src/lib/db/cleanup.ts +++ b/src/lib/db/cleanup.ts @@ -13,6 +13,7 @@ import { deleteAllFromTable, deleteCallLogArtifacts, deleteFromTableBefore, + deleteFromTableBeforeInBatches, tableExists, type DeleteByPeriodTarget, } from "./cleanup/usagePurge"; @@ -430,6 +431,103 @@ export async function cleanupCcrBlocks(): Promise { return result; } +/** + * Clean up conversation_turn_nodes older than the call-log retention window (#12453). + * + * The nodes are identity-only: the transcript view resolves each turn's display + * content from the call_logs row `last_correlation_id` points at. Once + * cleanupCallLogs purges that row the node can never render again, so the two + * tables share the dashboard database setting `retention.callLogs` instead of + * a knob of their own; `CALL_LOG_RETENTION_DAYS` configures the separate + * compliance cleanup path and does not override this window. Deleting an old + * node only affects reconnect anchors: a conversation resumed after the window + * mints a new id, which is already the documented anchor-miss behavior of + * resolveConversationId. `last_seen_at` has no index (migration 156), so + * each DELETE is a table scan. Bounded batches yield between writes so an + * existing large table cannot park the event loop for the whole cleanup pass. + */ +export async function cleanupConversationTurnNodes(): Promise { + const retention = getRetentionSettings(); + + const retentionDays = retention.callLogs; + const cutoffDate = new Date(); + cutoffDate.setDate(cutoffDate.getDate() - retentionDays); + const cutoffISO = cutoffDate.toISOString(); + + const result: CleanupResult = { deleted: 0, errors: 0 }; + + try { + result.deleted = await deleteFromTableBeforeInBatches( + { table: "conversation_turn_nodes", column: "last_seen_at", cutoff: "iso" }, + cutoffISO + ); + + console.log( + `[Cleanup] Deleted ${result.deleted} conversation_turn_nodes older than ${retentionDays} days` + ); + } catch (err: unknown) { + console.error("[Cleanup] Error cleaning conversation_turn_nodes:", err); + result.errors++; + } + + return result; +} + +/** + * Sweep agentic_conversations left without any conversation_turn_nodes (#12453). + * + * Runs after cleanupConversationTurnNodes so a root whose whole chain just + * expired goes in the same pass. The indexed `last_seen_at` predicate bounds + * the NOT EXISTS probe to roots that are already past the retention window. + * Deletion is batched for the same event-loop fairness guarantee as the + * preceding node cleanup. + */ +export async function cleanupAgenticConversations(): Promise { + const db = getDbInstance(); + const retention = getRetentionSettings(); + + const retentionDays = retention.callLogs; + const cutoffDate = new Date(); + cutoffDate.setDate(cutoffDate.getDate() - retentionDays); + const cutoffISO = cutoffDate.toISOString(); + + const result: CleanupResult = { deleted: 0, errors: 0 }; + + try { + if (!tableExists("agentic_conversations") || !tableExists("conversation_turn_nodes")) { + return result; + } + + const stmt = db.prepare( + `DELETE FROM agentic_conversations + WHERE rowid IN ( + SELECT rowid FROM agentic_conversations + WHERE last_seen_at < ? + AND NOT EXISTS ( + SELECT 1 FROM conversation_turn_nodes n + WHERE n.conversation_id = agentic_conversations.id + ) + LIMIT 10000 + )` + ); + while (true) { + const batch = stmt.run(cutoffISO).changes; + result.deleted += batch; + if (batch < 10_000) break; + await new Promise((resolve) => setImmediate(resolve)); + } + + console.log( + `[Cleanup] Deleted ${result.deleted} orphaned agentic_conversations older than ${retentionDays} days` + ); + } catch (err: unknown) { + console.error("[Cleanup] Error cleaning agentic_conversations:", err); + result.errors++; + } + + return result; +} + /** * Run all cleanup functions if auto-cleanup is enabled. */ @@ -463,6 +561,8 @@ export async function runAutoCleanup(): Promise<{ compressionRunTelemetry: await cleanupCompressionRunTelemetry(), proxyLogs: await cleanupProxyLogs(), ccrBlocks: await cleanupCcrBlocks(), + conversationTurnNodes: await cleanupConversationTurnNodes(), + agenticConversations: await cleanupAgenticConversations(), }; const totalDeleted = Object.values(results).reduce((sum, r) => sum + r.deleted, 0); @@ -588,6 +688,8 @@ export interface ResetUsageHistoryResult extends CleanupResult { deletedRoutingDecisions: number; deletedQuotaConsumption: number; deletedTokenLedger: number; + deletedConversationTurnNodes: number; + deletedAgenticConversations: number; } function isResetUsageHistoryPeriod(period: string): period is ResetUsageHistoryPeriod { @@ -604,10 +706,13 @@ function isResetUsageHistoryPeriod(period: string): period is ResetUsageHistoryP * first, since the whole point is to wipe the data the user selected. * * @param period - One of {@link RESET_USAGE_HISTORY_PERIODS}. `"all"` wipes - * every row in all three tables; any other value deletes rows strictly - * older than `now - period`. Throws on an invalid period. + * every reset target, including conversation identity metadata; any other + * value deletes only time-scoped usage/log rows older than `now - period`. + * Throws on an invalid period. */ -const RESET_TARGETS: Array = [ +const RESET_TARGETS: Array< + DeleteByPeriodTarget & { resultKey: keyof ResetUsageHistoryResult; allOnly?: boolean } +> = [ { table: "usage_history", column: "timestamp", cutoff: "iso", resultKey: "deletedUsageHistory" }, { table: "daily_usage_summary", @@ -660,6 +765,20 @@ const RESET_TARGETS: Array { @@ -684,6 +803,8 @@ export async function resetUsageHistory(period: string): Promise { - switch (target.cutoff) { - case "date": - return cutoffIso.slice(0, 10); - case "dateHour": - return `${cutoffIso.slice(0, 10)} ${cutoffIso.slice(11, 13)}:00:00`; - case "epochMs": - return new Date(cutoffIso).getTime(); - case "epochSeconds": - return Math.floor(new Date(cutoffIso).getTime() / 1000); - case "iso": - default: - return cutoffIso; - } - })(); - return getDbInstance() .prepare(`DELETE FROM ${target.table} WHERE ${target.column} < ?`) - .run(cutoff).changes; + .run(cutoffValue(target, cutoffIso)).changes; +} + +export async function deleteFromTableBeforeInBatches( + target: DeleteByPeriodTarget, + cutoffIso: string +): Promise { + if (!tableExists(target.table)) return 0; + + const statement = getDbInstance().prepare( + `DELETE FROM ${target.table} + WHERE rowid IN ( + SELECT rowid FROM ${target.table} + WHERE ${target.column} < ? + LIMIT ? + )` + ); + const cutoff = cutoffValue(target, cutoffIso); + let deleted = 0; + + while (true) { + const batch = statement.run(cutoff, DELETE_BATCH_SIZE).changes; + deleted += batch; + if (batch < DELETE_BATCH_SIZE) return deleted; + await new Promise((resolve) => setImmediate(resolve)); + } } export function collectCallLogArtifactsBefore(cutoffIso: string): string[] { diff --git a/tests/unit/db-cleanup-conversation-nodes-12453.test.ts b/tests/unit/db-cleanup-conversation-nodes-12453.test.ts new file mode 100644 index 0000000000..a285fbc1f7 --- /dev/null +++ b/tests/unit/db-cleanup-conversation-nodes-12453.test.ts @@ -0,0 +1,190 @@ +/** + * Issue #12453 — conversation_turn_nodes / agentic_conversations have no + * retention path, so storage.sqlite grows without bound (1.15M node rows, + * ~775 MB in four days on one busy coding-agent workload). + * + * The identity nodes only make sense while the call_logs row their + * last_correlation_id points at still exists, so both tables follow the + * existing `retention.callLogs` window instead of getting a knob of their own. + * + * These tests call the REAL cleanup functions against a real SQLite adapter + * seeded with test rows, exactly like telemetry-auto-cleanup-6848.test.ts. + * + * DATA_DIR isolation is self-contained (mkdtempSync below), not dependent on + * the test:unit harness's `--import ./tests/_setup/isolateDataDir.ts`: this + * file runs real DELETEs through getDbInstance(), which resolves to the + * developer's ~/.omniroute/storage.sqlite when DATA_DIR is unset. Do NOT + * remove the DATA_DIR override below. + */ + +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-12453-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const { cleanupConversationTurnNodes, cleanupAgenticConversations, runAutoCleanup } = + await import("../../src/lib/db/cleanup.ts"); +const { getDbInstance, resetDbInstance } = await import("../../src/lib/db/core.ts"); +const { getUserDatabaseSettings } = await import("../../src/lib/db/databaseSettings.ts"); + +test.after(() => { + resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 }); +}); + +const DAY_MS = 86_400_000; +const RETENTION_DAYS = getUserDatabaseSettings().retention.callLogs; +const OLD = new Date(Date.now() - (RETENTION_DAYS + 1) * DAY_MS).toISOString(); +const RECENT = new Date().toISOString(); + +function insertConversation(id: string, lastSeenAt: string): void { + getDbInstance()! + .prepare( + `INSERT INTO agentic_conversations + (id, api_key_id, fingerprint_hash, last_message_count, last_messages_hash, turn_count, first_seen_at, last_seen_at) + VALUES (?, 'key1', 'fp', 0, '', 1, ?, ?)` + ) + .run(id, lastSeenAt, lastSeenAt); +} + +function insertNode(id: string, conversationId: string, lastSeenAt: string): void { + getDbInstance()! + .prepare( + `INSERT INTO conversation_turn_nodes + (id, conversation_id, parent_id, role, content_hash, last_correlation_id, first_seen_at, last_seen_at) + VALUES (?, ?, NULL, 'user', 'hash', 'corr', ?, ?)` + ) + .run(id, conversationId, lastSeenAt, lastSeenAt); +} + +function count(table: string): number { + const row = getDbInstance()!.prepare(`SELECT COUNT(*) AS cnt FROM ${table}`).get() as { + cnt: number; + }; + return row.cnt; +} + +function ids(table: string): string[] { + const rows = getDbInstance()!.prepare(`SELECT id FROM ${table} ORDER BY id`).all() as Array<{ + id: string; + }>; + return rows.map((r) => r.id); +} + +test.beforeEach(() => { + const db = getDbInstance()!; + db.exec("DELETE FROM conversation_turn_nodes"); + db.exec("DELETE FROM agentic_conversations"); +}); + +test("#12453 cleanupConversationTurnNodes: deletes nodes older than the call-log retention window", async () => { + insertConversation("conv_a", RECENT); + insertNode("old-1", "conv_a", OLD); + insertNode("old-2", "conv_a", OLD); + insertNode("old-3", "conv_a", OLD); + insertNode("recent-1", "conv_a", RECENT); + insertNode("recent-2", "conv_a", RECENT); + + const result = await cleanupConversationTurnNodes(); + + assert.strictEqual(result.deleted, 3); + assert.strictEqual(result.errors, 0); + assert.deepStrictEqual(ids("conversation_turn_nodes"), ["recent-1", "recent-2"]); +}); + +test("#12453 cleanupConversationTurnNodes: yields between bounded delete batches", async () => { + insertConversation("conv_bulk", OLD); + const db = getDbInstance()!; + const insert = db.prepare( + `INSERT INTO conversation_turn_nodes + (id, conversation_id, parent_id, role, content_hash, last_correlation_id, first_seen_at, last_seen_at) + VALUES (?, 'conv_bulk', NULL, 'user', 'hash', 'corr', ?, ?)` + ); + db.transaction(() => { + for (let i = 0; i < 10_001; i++) insert.run(`bulk-${i}`, OLD, OLD); + })(); + + let eventLoopTurnObserved = false; + setImmediate(() => { + eventLoopTurnObserved = true; + }); + + const result = await cleanupConversationTurnNodes(); + + assert.strictEqual(result.deleted, 10_001); + assert.strictEqual(result.errors, 0); + assert.strictEqual(count("conversation_turn_nodes"), 0); + assert.strictEqual(eventLoopTurnObserved, true, "cleanup should yield after a full batch"); +}); + +test("#12453 cleanupAgenticConversations: sweeps stale conversations that have no nodes left", async () => { + // Stale and orphaned: every node already expired -> must go. + insertConversation("conv_orphan_old", OLD); + // Stale but still anchored by a live node -> must stay. + insertConversation("conv_anchored", OLD); + insertNode("live-1", "conv_anchored", RECENT); + // Fresh root whose nodes are not written yet (createConversation runs before + // the node insert in the same request) -> must stay. + insertConversation("conv_fresh_no_nodes", RECENT); + + const result = await cleanupAgenticConversations(); + + assert.strictEqual(result.deleted, 1); + assert.strictEqual(result.errors, 0); + assert.deepStrictEqual(ids("agentic_conversations"), ["conv_anchored", "conv_fresh_no_nodes"]); + assert.strictEqual(count("conversation_turn_nodes"), 1); +}); + +test("#12453 nodes expire first, then the conversation they anchored is swept in the same pass", async () => { + insertConversation("conv_dead", OLD); + insertNode("dead-1", "conv_dead", OLD); + insertNode("dead-2", "conv_dead", OLD); + + // Conversation-only sweep must not touch a root that still has (old) nodes. + const first = await cleanupAgenticConversations(); + assert.strictEqual(first.deleted, 0); + assert.strictEqual(count("agentic_conversations"), 1); + + const nodes = await cleanupConversationTurnNodes(); + assert.strictEqual(nodes.deleted, 2); + + const second = await cleanupAgenticConversations(); + assert.strictEqual(second.deleted, 1); + assert.strictEqual(count("agentic_conversations"), 0); +}); + +test("#12453 runAutoCleanup: registers both tables and reports them in results", async () => { + insertConversation("conv_x", OLD); + insertNode("x-1", "conv_x", OLD); + insertConversation("conv_y", RECENT); + insertNode("y-1", "conv_y", RECENT); + + const summary = await runAutoCleanup(); + + assert.ok(summary.results.conversationTurnNodes, "conversationTurnNodes missing from results"); + assert.ok(summary.results.agenticConversations, "agenticConversations missing from results"); + assert.strictEqual(summary.results.conversationTurnNodes.deleted, 1); + assert.strictEqual(summary.results.agenticConversations.deleted, 1); + assert.strictEqual(summary.results.conversationTurnNodes.errors, 0); + assert.strictEqual(summary.results.agenticConversations.errors, 0); + assert.deepStrictEqual(ids("conversation_turn_nodes"), ["y-1"]); + assert.deepStrictEqual(ids("agentic_conversations"), ["conv_y"]); +}); + +test("#12453 cleanupAgenticConversations: missing node table is a safe no-op", async () => { + insertConversation("conv_without_table", OLD); + const db = getDbInstance()!; + db.exec("ALTER TABLE conversation_turn_nodes RENAME TO conversation_turn_nodes_unavailable"); + + try { + const result = await cleanupAgenticConversations(); + assert.deepStrictEqual(result, { deleted: 0, errors: 0 }); + assert.strictEqual(count("agentic_conversations"), 1); + } finally { + db.exec("ALTER TABLE conversation_turn_nodes_unavailable RENAME TO conversation_turn_nodes"); + } +}); diff --git a/tests/unit/usage-history-reset.test.ts b/tests/unit/usage-history-reset.test.ts index 01f0d86d4e..86d254193c 100644 --- a/tests/unit/usage-history-reset.test.ts +++ b/tests/unit/usage-history-reset.test.ts @@ -58,6 +58,24 @@ test.after(() => { } }); +test("purge usage API exposes every conversation reset counter", () => { + const routeSource = fs.readFileSync( + path.join(process.cwd(), "src/app/api/settings/purge-usage-history/route.ts"), + "utf8" + ); + + assert.match( + routeSource, + /deletedConversationTurnNodes:\s*result\.deletedConversationTurnNodes/, + "the API response should expose deleted conversation nodes" + ); + assert.match( + routeSource, + /deletedAgenticConversations:\s*result\.deletedAgenticConversations/, + "the API response should expose deleted conversation roots" + ); +}); + test("resetUsageHistory: 'all' wipes usage_history, daily_usage_summary, and hourly_usage_summary; a period only deletes rows older than the cutoff; an invalid period throws", async () => { setup(); try { @@ -103,6 +121,17 @@ test("resetUsageHistory: 'all' wipes usage_history, daily_usage_summary, and hou "INSERT INTO combos (id, name, data, created_at, updated_at) VALUES (?, ?, ?, ?, ?)" ).run("combo-test", "Test Combo", "{}", recentIso, recentIso); + db.prepare( + `INSERT INTO agentic_conversations + (id, api_key_id, fingerprint_hash, last_message_count, last_messages_hash, turn_count, first_seen_at, last_seen_at) + VALUES ('conversation-test', 'key-test', 'fp', 0, '', 1, ?, ?)` + ).run(recentIso, recentIso); + db.prepare( + `INSERT INTO conversation_turn_nodes + (id, conversation_id, parent_id, role, content_hash, last_correlation_id, first_seen_at, last_seen_at) + VALUES ('turn-test', 'conversation-test', NULL, 'user', 'hash', 'recent-call', ?, ?)` + ).run(recentIso, recentIso); + db.prepare("INSERT INTO usage_history (provider, model, timestamp) VALUES (?, ?, ?)").run( "openai", "gpt-test", @@ -240,6 +269,16 @@ test("resetUsageHistory: 'all' wipes usage_history, daily_usage_summary, and hou assert.equal(countRows(db, "provider_nodes"), 1, "provider config should survive reset"); assert.equal(countRows(db, "api_keys"), 1, "API keys should survive reset"); assert.equal(countRows(db, "combos"), 1, "combos should survive reset"); + assert.equal( + countRows(db, "conversation_turn_nodes"), + 1, + "a timed reset should preserve conversation identity nodes" + ); + assert.equal( + countRows(db, "agentic_conversations"), + 1, + "a timed reset should preserve conversation roots" + ); assert.equal(countRows(db, "usage_history"), 1, "recent usage_history row should survive"); assert.equal(countRows(db, "call_logs"), 1, "recent call_logs row should survive"); @@ -310,6 +349,16 @@ test("resetUsageHistory: 'all' wipes usage_history, daily_usage_summary, and hou 1, "'all' should delete remaining call artifact" ); + assert.equal( + allResult.deletedConversationTurnNodes, + 1, + "'all' should delete conversation identity nodes" + ); + assert.equal( + allResult.deletedAgenticConversations, + 1, + "'all' should delete conversation roots" + ); assert.equal( fs.existsSync(recentArtifactPath), false, @@ -331,6 +380,16 @@ test("resetUsageHistory: 'all' wipes usage_history, daily_usage_summary, and hou 0, "'all' should empty hourly_usage_summary" ); + assert.equal( + countRows(db, "conversation_turn_nodes"), + 0, + "'all' should empty conversation_turn_nodes" + ); + assert.equal( + countRows(db, "agentic_conversations"), + 0, + "'all' should empty agentic_conversations" + ); assert.equal(countRows(db, "provider_nodes"), 1, "provider config should still survive 'all'"); assert.equal(countRows(db, "api_keys"), 1, "API keys should still survive 'all'"); assert.equal(countRows(db, "combos"), 1, "combos should still survive 'all'");