diff --git a/src/services/api-handlers.ts b/src/services/api-handlers.ts index 7e462ff0..37800fdb 100644 --- a/src/services/api-handlers.ts +++ b/src/services/api-handlers.ts @@ -1067,6 +1067,13 @@ export async function handleRunDreaming( const result = await dreamingService.run(directory); return { success: true, data: result }; } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (message.includes("already running")) { + return { success: false, error: "already_running" }; + } + if (message === "dreaming_provider_unavailable") { + return { success: false, error: "dreaming_provider_unavailable" }; + } log("handleRunDreaming: error", { error: String(error) }); return { success: false, error: String(error) }; } diff --git a/src/services/cleanup-service.ts b/src/services/cleanup-service.ts index d2b79cc9..882cd358 100644 --- a/src/services/cleanup-service.ts +++ b/src/services/cleanup-service.ts @@ -1,6 +1,7 @@ import { tursoShardManager } from "./turso/shard-manager.js"; import { tursoVectorSearch } from "./turso/vector-search.js"; import { tursoConnectionManager } from "./turso/connection-manager.js"; +import { isVacuumUnsupportedError } from "./turso/error-classifiers.js"; import { ensureTursoReady } from "./turso/ready.js"; import { CONFIG } from "../config.js"; import { log } from "./logger.js"; @@ -16,6 +17,9 @@ interface CleanupResult { aiSessionsDeleted: number; } +/** Process-lifetime: Turso rejected VACUUM; do not retry every cleanup. */ +let vacuumUnsupported = false; + export class CleanupService { private lastCleanupTime: number = 0; private isRunning: boolean = false; @@ -46,18 +50,36 @@ export class CleanupService { await ensureTursoReady(); const cutoffTime = Date.now() - CONFIG.autoCleanupRetentionDays * 24 * 60 * 60 * 1000; - const userShards = await tursoShardManager.getAllShards("user", ""); - const projectShards = await tursoShardManager.getAllShards("project", ""); - const allShards = [...userShards, ...projectShards]; + let allShards: Awaited> = []; + try { + const userShards = await tursoShardManager.getAllShards("user", ""); + const projectShards = await tursoShardManager.getAllShards("project", ""); + allShards = [...userShards, ...projectShards]; + } catch (error) { + log("Cleanup: step failed", { step: "list-shards", error: String(error) }); + } const pinnedMemoryIds = new Set(); for (const shard of allShards) { - const db = await tursoConnectionManager.getConnection(shard.dbPath); - const pinned = await db.all(`SELECT id FROM memories WHERE is_pinned = 1`); - pinned.forEach((row) => pinnedMemoryIds.add(String(row.id))); + try { + const db = await tursoConnectionManager.getConnection(shard.dbPath); + const pinned = await db.all(`SELECT id FROM memories WHERE is_pinned = 1`); + pinned.forEach((row) => pinnedMemoryIds.add(String(row.id))); + } catch (error) { + log("Cleanup: step failed", { + step: "pins", + shardId: shard.id, + error: String(error), + }); + } } - const promptCleanupResult = await userPromptManager.deleteOldPrompts(cutoffTime); + let promptCleanupResult = { deleted: 0, linkedMemoryIds: [] as string[] }; + try { + promptCleanupResult = await userPromptManager.deleteOldPrompts(cutoffTime); + } catch (error) { + log("Cleanup: step failed", { step: "deleteOldPrompts", error: String(error) }); + } const linkedMemoryIds = new Set(promptCleanupResult.linkedMemoryIds); const protectedMemoryIds = new Set([...pinnedMemoryIds, ...linkedMemoryIds]); @@ -68,16 +90,36 @@ export class CleanupService { let pinnedSkipped = 0; for (const shard of allShards) { - const db = await tursoConnectionManager.getConnection(shard.dbPath); - // Ensure deleted_at (v3) exists before filtering on it. - await tursoShardManager.ensureShardSchema(db); - const oldMemories = await db.all( - ` + let db; + try { + db = await tursoConnectionManager.getConnection(shard.dbPath); + await tursoShardManager.ensureShardSchema(db); + } catch (error) { + log("Cleanup: step failed", { + step: "shard-schema", + shardId: shard.id, + error: String(error), + }); + continue; + } + + let oldMemories: Array> = []; + try { + oldMemories = await db.all( + ` SELECT id, container_tag, is_pinned FROM memories WHERE updated_at < ? AND deleted_at IS NULL `, - [cutoffTime] - ); + [cutoffTime] + ); + } catch (error) { + log("Cleanup: step failed", { + step: "age-select", + shardId: shard.id, + error: String(error), + }); + continue; + } for (const memory of oldMemories) { try { @@ -90,7 +132,11 @@ export class CleanupService { continue; } - await tursoVectorSearch.deleteVector(db, String(memory.id)); + const hardDeleted = await tursoVectorSearch.deleteVector(db, String(memory.id)); + if (!hardDeleted) { + // IdxDelete quarantine (soft-delete) — do not count as hard-deleted. + continue; + } await tursoShardManager.decrementVectorCount(shard.id); totalDeleted++; @@ -122,11 +168,20 @@ export class CleanupService { const promptsDeleted = promptCleanupResult.deleted - linkedMemoryIds.size; - try { - await userPromptManager.vacuum(); - log("Cleanup: VACUUM done", { db: "user-prompts.db" }); - } catch (err) { - log("Cleanup: VACUUM skipped (DB busy)", { error: String(err) }); + if (!vacuumUnsupported) { + try { + await userPromptManager.vacuum(); + log("Cleanup: VACUUM done", { db: "user-prompts.db" }); + } catch (err) { + if (isVacuumUnsupportedError(err)) { + vacuumUnsupported = true; + log("Cleanup: VACUUM skipped (unsupported by Turso backend)", { + error: String(err), + }); + } else { + log("Cleanup: VACUUM skipped (DB busy)", { error: String(err) }); + } + } } let aiSessionsDeleted = 0; @@ -163,3 +218,8 @@ export class CleanupService { } export const cleanupService = new CleanupService(); + +/** Test helper: reset process-lifetime VACUUM unsupported flag. */ +export function resetVacuumUnsupportedForTests(): void { + vacuumUnsupported = false; +} diff --git a/src/services/client.ts b/src/services/client.ts index 33460f3b..c308d811 100644 --- a/src/services/client.ts +++ b/src/services/client.ts @@ -115,7 +115,10 @@ export class LocalMemoryClient { const closed = await releasePluginTurso(); if (closed) { // Process-wide like the pool: only the last instance may drop the model. - await embeddingService.dispose("client-close"); + // Optional: test stubs for embeddingService may omit dispose. + if (typeof embeddingService.dispose === "function") { + await embeddingService.dispose("client-close"); + } this.reset(); } } diff --git a/src/services/deduplication-service.ts b/src/services/deduplication-service.ts index 56792508..bcfef8ac 100644 --- a/src/services/deduplication-service.ts +++ b/src/services/deduplication-service.ts @@ -70,7 +70,8 @@ export class DeduplicationService { for (const dup of toDelete) { try { - await tursoVectorSearch.deleteVector(db, String(dup.id)); + const hardDeleted = await tursoVectorSearch.deleteVector(db, String(dup.id)); + if (!hardDeleted) continue; await tursoShardManager.decrementVectorCount(shard.id); exactDeleted++; } catch (error) { diff --git a/src/services/dreaming-service.ts b/src/services/dreaming-service.ts index ab36d320..aedce012 100644 --- a/src/services/dreaming-service.ts +++ b/src/services/dreaming-service.ts @@ -110,6 +110,7 @@ export interface DreamingRunResult { groupsConsidered: number; groupsMerged: number; groupsSkipped: number; + groupsFailed: number; groupsFlagged: number; memoriesSoftDeleted: number; } @@ -288,6 +289,10 @@ export class DreamingService { log("dreaming: skipped due to lock contention", { error: message }); return null; } + if (message === "dreaming_provider_unavailable") { + log("dreaming: skipped (provider unavailable)", { error: message }); + return null; + } throw error; } } @@ -306,6 +311,7 @@ export class DreamingService { groupsConsidered: 0, groupsMerged: 0, groupsSkipped: 0, + groupsFailed: 0, groupsFlagged: 0, memoriesSoftDeleted: 0, }; @@ -324,10 +330,11 @@ export class DreamingService { const v2Client = getV2Client(); if (!v2Client || !CONFIG.opencodeProvider || !CONFIG.opencodeModel) { log("dreaming: skipped (no opencode structured-output client)"); - return result; + throw new Error("dreaming_provider_unavailable"); } const { generateStructuredOutput } = await import("./ai/opencode-provider.js"); + let abortedForProvider = false; for (const group of batch) { try { @@ -402,13 +409,39 @@ export class DreamingService { result.groupsSkipped++; } } catch (error) { - result.groupsSkipped++; - log("dreaming: group failed", { error: String(error) }); + result.groupsFailed++; + const errMsg = error instanceof Error ? error.message : String(error); + log("dreaming: group failed", { error: errMsg }); + // Auth / provider outages will fail every remaining group the same way. + if (/ProviderRequestError|API_KEY is not set|401|Unauthorized/i.test(errMsg)) { + const processed = + result.groupsMerged + + result.groupsSkipped + + result.groupsFlagged + + result.groupsFailed; + const remaining = Math.max(0, result.groupsConsidered - processed); + result.groupsFailed += remaining; + abortedForProvider = true; + log("dreaming: aborting remaining groups after provider failure", { remaining }); + break; + } } } - writeState({ lastRunAt: Date.now() }); - log("dreaming: run complete", result); + const onlyFailures = + result.groupsConsidered > 0 && + result.groupsMerged === 0 && + result.groupsFlagged === 0 && + result.groupsSkipped === 0 && + result.groupsFailed > 0 && + (abortedForProvider || result.groupsFailed === result.groupsConsidered); + if (onlyFailures) { + // Do not advance the schedule when nothing could be processed. + log("dreaming: run complete with only failures; not advancing lastRunAt", result); + } else { + writeState({ lastRunAt: Date.now() }); + log("dreaming: run complete", result); + } return result; } finally { this.isRunning = false; diff --git a/src/services/markdown/reconcile.ts b/src/services/markdown/reconcile.ts index b055127f..91da677e 100644 --- a/src/services/markdown/reconcile.ts +++ b/src/services/markdown/reconcile.ts @@ -68,7 +68,8 @@ async function deleteMemoryById(id: string): Promise { const found = await findMemoryAcrossShards(id); if (!found) return false; const db = await tursoConnectionManager.getConnection(found.dbPath); - await tursoVectorSearch.deleteVector(db, id); + const hardDeleted = await tursoVectorSearch.deleteVector(db, id); + if (!hardDeleted) return false; await tursoShardManager.decrementVectorCount(found.shardId); return true; } diff --git a/src/services/turso/connection-manager.ts b/src/services/turso/connection-manager.ts index 7b70bc45..64346f67 100644 --- a/src/services/turso/connection-manager.ts +++ b/src/services/turso/connection-manager.ts @@ -16,7 +16,7 @@ import { export type ConnectFactory = (path: string, opts?: DatabaseOpts) => Promise; /** Always-on experimental flags for every Turso open. */ -export const TURSO_BASE_EXPERIMENTAL_FEATURES = ["encryption"] as const; +export const TURSO_BASE_EXPERIMENTAL_FEATURES = ["encryption", "vacuum"] as const; /** * Multiprocess WAL is Unix-only. On Windows/WASM/32-bit the flag is accepted but diff --git a/src/services/turso/error-classifiers.ts b/src/services/turso/error-classifiers.ts new file mode 100644 index 00000000..c31680ed --- /dev/null +++ b/src/services/turso/error-classifiers.ts @@ -0,0 +1,23 @@ +/** Turso rejects VACUUM unless the experimental `vacuum` feature is enabled. */ +const VACUUM_UNSUPPORTED_PATTERNS = [/VACUUM is an experimental feature/i, /experimental-vacuum/i]; + +/** DiskANN / index delete when the index entry is already gone (row may still exist). */ +const IDX_DELETE_MISMATCH_PATTERNS = [ + /IdxDelete:\s*no matching index entry/i, + /no matching index entry found for key/i, +]; + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +export function isVacuumUnsupportedError(error: unknown): boolean { + const message = errorMessage(error); + return VACUUM_UNSUPPORTED_PATTERNS.some((pattern) => pattern.test(message)); +} + +/** Only IdxDelete mismatch — not every `Corrupt database:*` error. */ +export function isIdxDeleteMismatchError(error: unknown): boolean { + const message = errorMessage(error); + return IDX_DELETE_MISMATCH_PATTERNS.some((pattern) => pattern.test(message)); +} diff --git a/src/services/turso/vector-search.ts b/src/services/turso/vector-search.ts index 76fdf033..921cbe98 100644 --- a/src/services/turso/vector-search.ts +++ b/src/services/turso/vector-search.ts @@ -1,5 +1,6 @@ import { tursoConnectionManager } from "./connection-manager.js"; import { tursoShardManager } from "./shard-manager.js"; +import { isIdxDeleteMismatchError } from "./error-classifiers.js"; import { log } from "../logger.js"; import type { MemoryListFilters, MemoryRecord, SearchResult, ShardInfo } from "./types.js"; import { @@ -340,8 +341,47 @@ export class TursoVectorSearch { }; } - async deleteVector(db: TursoDb, memoryId: string): Promise { - await db.run(`DELETE FROM memories WHERE id = ?`, [memoryId]); + /** + * Hard-delete a memory row. Returns true when the row is gone (deleted now or + * already missing). Returns false when IdxDelete leaves the row in place — + * callers must not decrement vector counts. Active stuck rows are soft-deleted + * once so age cleanup stops retrying the same broken index key. + */ + async deleteVector(db: TursoDb, memoryId: string): Promise { + try { + await db.run(`DELETE FROM memories WHERE id = ?`, [memoryId]); + return true; + } catch (error) { + if (!isIdxDeleteMismatchError(error)) throw error; + // IdxDelete with the row still present means the DELETE rolled back. + const remaining = await db.get(`SELECT id, deleted_at FROM memories WHERE id = ?`, [ + memoryId, + ]); + if (!remaining) return true; + + // Quarantine active rows so age-based cleanup stops re-hitting the same key. + if (remaining.deleted_at == null) { + const now = Date.now(); + try { + await db.run( + `UPDATE memories SET deleted_at = ?, updated_at = ? WHERE id = ? AND deleted_at IS NULL`, + [now, now, memoryId] + ); + log("deleteVector: IdxDelete quarantined active row via soft-delete", { + memoryId, + error: String(error), + }); + } catch (quarantineError) { + log("deleteVector: IdxDelete quarantine failed", { + memoryId, + error: String(quarantineError), + }); + } + } + // Soft-deleted / quarantined row that cannot be hard-purged without REINDEX. + // Return false quietly so trash purge neither fake-counts nor log-spams. + return false; + } } async softDeleteVector( @@ -406,11 +446,18 @@ export class TursoVectorSearch { `SELECT id FROM memories WHERE deleted_at IS NOT NULL AND deleted_at < ?`, [cutoffMs] ); - const ids = rows.map((r) => String(r.id)); - for (const id of ids) { - await this.deleteVector(db, id); + const purged: string[] = []; + for (const row of rows) { + const id = String(row.id); + try { + if (await this.deleteVector(db, id)) { + purged.push(id); + } + } catch (error) { + log("purgeDeletedBefore: delete error", { memoryId: id, error: String(error) }); + } } - return ids; + return purged; } async updateVector( diff --git a/src/services/user-prompt/user-prompt-manager.ts b/src/services/user-prompt/user-prompt-manager.ts index 242c9fc0..dd1ce2e1 100644 --- a/src/services/user-prompt/user-prompt-manager.ts +++ b/src/services/user-prompt/user-prompt-manager.ts @@ -1,7 +1,9 @@ import { join } from "node:path"; import { tursoConnectionManager } from "../turso/connection-manager.js"; +import { isIdxDeleteMismatchError } from "../turso/error-classifiers.js"; import { CONFIG } from "../../config.js"; import type { SqlValue, TursoDb, TursoTx } from "../turso/turso-db.js"; +import { log } from "../logger.js"; import { stripPrivateContent, isFullyPrivate } from "../privacy.js"; import { truncateToMaxBytes } from "../../utils/context-limit.js"; @@ -438,14 +440,47 @@ export class UserPromptManager { .map((row) => row.linked_memory_id) .filter((id): id is string => Boolean(id)); - const deleted = await db.transaction("write", async (tx) => { - await tx.run( - `DELETE FROM pending_capture_evidence WHERE prompt_id IN (SELECT id FROM user_prompts WHERE created_at < ?)`, - [cutoffTime] - ); - return tx.run(`DELETE FROM user_prompts WHERE created_at < ?`, [cutoffTime]); - }); - return { deleted, linkedMemoryIds }; + try { + const deleted = await db.transaction("write", async (tx) => { + await tx.run( + `DELETE FROM pending_capture_evidence WHERE prompt_id IN (SELECT id FROM user_prompts WHERE created_at < ?)`, + [cutoffTime] + ); + return tx.run(`DELETE FROM user_prompts WHERE created_at < ?`, [cutoffTime]); + }); + return { deleted, linkedMemoryIds }; + } catch (error) { + if (!isIdxDeleteMismatchError(error)) throw error; + // Batch DELETE hit a desynced index entry; fall back to per-row deletes. + log("deleteOldPrompts: IdxDelete on batch, falling back to per-row deletes", { + error: String(error), + }); + const rows = await db.all(`SELECT id FROM user_prompts WHERE created_at < ?`, [cutoffTime]); + let deleted = 0; + for (const row of rows) { + const promptId = String(row.id); + try { + await db.transaction("write", async (tx) => { + await tx.run(`DELETE FROM pending_capture_evidence WHERE prompt_id = ?`, [promptId]); + await tx.run(`DELETE FROM user_prompts WHERE id = ?`, [promptId]); + }); + deleted++; + } catch (rowError) { + if (isIdxDeleteMismatchError(rowError)) { + const remaining = await db.get(`SELECT id FROM user_prompts WHERE id = ?`, [promptId]); + if (!remaining) { + deleted++; + continue; + } + } + log("deleteOldPrompts: per-row delete failed", { + promptId, + error: String(rowError), + }); + } + } + return { deleted, linkedMemoryIds }; + } } async vacuum(): Promise { diff --git a/src/services/web-server.ts b/src/services/web-server.ts index e2050591..9b6bfb66 100644 --- a/src/services/web-server.ts +++ b/src/services/web-server.ts @@ -656,6 +656,9 @@ export class WebServer { if (path === "/api/dreaming/run" && method === "POST") { const result = await handleRunDreaming(); + if (!result.success && result.error === "already_running") { + return this.jsonResponse(result, 409); + } return this.jsonResponse(result); } diff --git a/tests/cleanup-resilience.test.ts b/tests/cleanup-resilience.test.ts new file mode 100644 index 00000000..0bdb5266 --- /dev/null +++ b/tests/cleanup-resilience.test.ts @@ -0,0 +1,201 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const tempDirs: string[] = []; + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +const cleanupServiceUrl = new URL("../src/services/cleanup-service.js", import.meta.url).href; +const configUrl = new URL("../src/config.js", import.meta.url).href; +const loggerUrl = new URL("../src/services/logger.js", import.meta.url).href; +const readyUrl = new URL("../src/services/turso/ready.js", import.meta.url).href; +const shardManagerUrl = new URL("../src/services/turso/shard-manager.js", import.meta.url).href; +const vectorSearchUrl = new URL("../src/services/turso/vector-search.js", import.meta.url).href; +const connectionManagerUrl = new URL("../src/services/turso/connection-manager.js", import.meta.url) + .href; +const promptManagerUrl = new URL( + "../src/services/user-prompt/user-prompt-manager.js", + import.meta.url +).href; +const aiProviderFactoryUrl = new URL("../src/services/ai/ai-provider-factory.js", import.meta.url) + .href; +const errorClassifiersUrl = new URL("../src/services/turso/error-classifiers.js", import.meta.url) + .href; + +function runCleanupScenario(opts: { + promptThrowsIdxDelete?: boolean; + vacuumError?: "unsupported" | "busy" | null; + deleteVectorThrowsIdxDelete?: boolean; +}) { + const dir = mkdtempSync(join(tmpdir(), "opencode-mem-cleanup-resilience-")); + tempDirs.push(dir); + const scriptPath = join(dir, "scenario.mjs"); + const promptThrows = opts.promptThrowsIdxDelete ?? false; + const vacuumError = opts.vacuumError ?? null; + const deleteThrows = opts.deleteVectorThrowsIdxDelete ?? false; + + const script = ` +import { mock } from "bun:test"; + +const logs = []; +let vacuumCalls = 0; + +mock.module(${JSON.stringify(configUrl)}, () => ({ + CONFIG: { + autoCleanupEnabled: true, + autoCleanupRetentionDays: 30, + storagePath: ${JSON.stringify(dir)}, + dreaming: { trashRetentionDays: 30 }, + }, +})); + +mock.module(${JSON.stringify(loggerUrl)}, () => ({ + log: (msg, meta) => { logs.push({ msg, meta }); }, + safeLog: () => {}, + serializeError: (error) => ({ error: String(error) }), +})); + +mock.module(${JSON.stringify(readyUrl)}, () => ({ + ensureTursoReady: async () => {}, +})); + +mock.module(${JSON.stringify(errorClassifiersUrl)}, () => ({ + isVacuumUnsupportedError: (error) => String(error).includes("experimental"), + isIdxDeleteMismatchError: (error) => String(error).includes("IdxDelete"), +})); + +mock.module(${JSON.stringify(shardManagerUrl)}, () => ({ + tursoShardManager: { + getAllShards: async (kind) => + kind === "project" + ? [{ id: "proj-1", dbPath: ${JSON.stringify(join(dir, "proj.db"))} }] + : [], + ensureShardSchema: async () => {}, + decrementVectorCount: async () => {}, + }, +})); + +mock.module(${JSON.stringify(vectorSearchUrl)}, () => ({ + tursoVectorSearch: { + deleteVector: async () => { + if (${deleteThrows}) { + throw new Error("Corrupt database: IdxDelete: no matching index entry found for key"); + } + }, + purgeDeletedBefore: async () => [], + }, +})); + +mock.module(${JSON.stringify(connectionManagerUrl)}, () => ({ + tursoConnectionManager: { + getConnection: async () => ({ + all: async (sql) => { + if (String(sql).includes("is_pinned")) return []; + if (String(sql).includes("updated_at")) { + return [{ id: "mem-1", container_tag: "x_project_y", is_pinned: 0 }]; + } + return []; + }, + }), + }, +})); + +mock.module(${JSON.stringify(promptManagerUrl)}, () => ({ + userPromptManager: { + deleteOldPrompts: async () => { + if (${promptThrows}) { + throw new Error("Corrupt database: IdxDelete: no matching index entry found for key"); + } + return { deleted: 0, linkedMemoryIds: [] }; + }, + vacuum: async () => { + vacuumCalls += 1; + if (${JSON.stringify(vacuumError)} === "unsupported") { + throw new Error("Parse error: VACUUM is an experimental feature"); + } + if (${JSON.stringify(vacuumError)} === "busy") { + throw new Error("database is locked"); + } + }, + }, +})); + +mock.module(${JSON.stringify(aiProviderFactoryUrl)}, () => ({ + AIProviderFactory: { + cleanupExpiredSessions: async () => 0, + }, +})); + +const { cleanupService, resetVacuumUnsupportedForTests } = await import(${JSON.stringify(cleanupServiceUrl)}); +resetVacuumUnsupportedForTests(); +const result1 = await cleanupService.runCleanup(); +const result2 = await cleanupService.runCleanup(); +console.log(JSON.stringify({ result1, result2, vacuumCalls, logs })); +`; + + writeFileSync(scriptPath, script); + + const result = Bun.spawnSync({ + cmd: [process.execPath, scriptPath], + stdout: "pipe", + stderr: "pipe", + }); + + const stdout = Buffer.from(result.stdout).toString("utf8").trim(); + const stderr = Buffer.from(result.stderr).toString("utf8").trim(); + + return { + exitCode: result.exitCode, + stdout, + stderr, + parsed: stdout ? JSON.parse(stdout) : null, + }; +} + +describe("cleanup resilience", () => { + it("continues when deleteOldPrompts throws IdxDelete", () => { + const result = runCleanupScenario({ promptThrowsIdxDelete: true }); + expect(result.stderr).toBe(""); + expect(result.exitCode).toBe(0); + expect(result.parsed.result1.deletedCount).toBeGreaterThanOrEqual(0); + const stepLog = result.parsed.logs.find( + (l: { msg: string }) => l.msg === "Cleanup: step failed" + ); + expect(stepLog?.meta?.step).toBe("deleteOldPrompts"); + }); + + it("treats IdxDelete from deleteVector as non-fatal via catch path", () => { + const result = runCleanupScenario({ deleteVectorThrowsIdxDelete: true }); + expect(result.exitCode).toBe(0); + expect(result.parsed.result1).toBeTruthy(); + }); + + it("logs VACUUM unsupported once and skips subsequent vacuum calls", () => { + const result = runCleanupScenario({ vacuumError: "unsupported" }); + expect(result.exitCode).toBe(0); + expect(result.parsed.vacuumCalls).toBe(1); + const unsupported = result.parsed.logs.filter( + (l: { msg: string }) => l.msg === "Cleanup: VACUUM skipped (unsupported by Turso backend)" + ); + expect(unsupported).toHaveLength(1); + const busy = result.parsed.logs.filter( + (l: { msg: string }) => l.msg === "Cleanup: VACUUM skipped (DB busy)" + ); + expect(busy).toHaveLength(0); + }); + + it("keeps DB busy label for lock errors", () => { + const result = runCleanupScenario({ vacuumError: "busy" }); + expect(result.exitCode).toBe(0); + const busy = result.parsed.logs.filter( + (l: { msg: string }) => l.msg === "Cleanup: VACUUM skipped (DB busy)" + ); + expect(busy.length).toBeGreaterThanOrEqual(1); + }); +}); diff --git a/tests/dreaming-run.test.ts b/tests/dreaming-run.test.ts new file mode 100644 index 00000000..fa89611f --- /dev/null +++ b/tests/dreaming-run.test.ts @@ -0,0 +1,311 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const tempDirs: string[] = []; + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +const configUrl = new URL("../src/config.js", import.meta.url).href; +const loggerUrl = new URL("../src/services/logger.js", import.meta.url).href; +const readyUrl = new URL("../src/services/turso/ready.js", import.meta.url).href; +const shardManagerUrl = new URL("../src/services/turso/shard-manager.js", import.meta.url).href; +const connectionManagerUrl = new URL("../src/services/turso/connection-manager.js", import.meta.url) + .href; +const vectorSearchUrl = new URL("../src/services/turso/vector-search.js", import.meta.url).href; +const vectorUtilsUrl = new URL("../src/services/turso/vector-utils.js", import.meta.url).href; +const opencodeProviderUrl = new URL("../src/services/ai/opencode-provider.js", import.meta.url) + .href; +const dreamingLockUrl = new URL("../src/services/dreaming-lock.js", import.meta.url).href; +const dreamingServiceUrl = new URL("../src/services/dreaming-service.js", import.meta.url).href; +const apiHandlersUrl = new URL("../src/services/api-handlers.js", import.meta.url).href; +const embeddingUrl = new URL("../src/services/embedding.js", import.meta.url).href; + +function runFixture(scriptBody: string): { + exitCode: number; + stdout: string; + stderr: string; + parsed: unknown; +} { + const dir = mkdtempSync(join(tmpdir(), "opencode-mem-dreaming-run-")); + tempDirs.push(dir); + const scriptPath = join(dir, "scenario.mjs"); + writeFileSync(scriptPath, scriptBody); + const result = Bun.spawnSync({ + cmd: [process.execPath, scriptPath], + stdout: "pipe", + stderr: "pipe", + }); + const stdout = Buffer.from(result.stdout).toString("utf8").trim(); + const stderr = Buffer.from(result.stderr).toString("utf8").trim(); + return { + exitCode: result.exitCode, + stdout, + stderr, + parsed: stdout ? JSON.parse(stdout) : null, + }; +} + +describe("dreaming run API mapping", () => { + function apiMappingFixture(runError: string): string { + const storageDir = mkdtempSync(join(tmpdir(), "dream-api-map-")); + tempDirs.push(storageDir); + const promptManagerUrl = new URL( + "../src/services/user-prompt/user-prompt-manager.js", + import.meta.url + ).href; + return ` +import { mock } from "bun:test"; +import { mkdirSync } from "node:fs"; + +const storagePath = ${JSON.stringify(storageDir)}; +mkdirSync(storagePath, { recursive: true }); + +mock.module(${JSON.stringify(configUrl)}, () => ({ + CONFIG: { storagePath, dreaming: { enabled: true } }, +})); +mock.module(${JSON.stringify(loggerUrl)}, () => ({ + log: () => {}, + safeLog: () => {}, + serializeError: (e) => ({ error: String(e) }), +})); +mock.module(${JSON.stringify(promptManagerUrl)}, () => ({ + userPromptManager: {}, +})); +mock.module(${JSON.stringify(dreamingServiceUrl)}, () => ({ + dreamingService: { + run: async () => { + throw new Error(${JSON.stringify(runError)}); + }, + getStatus: () => ({ isRunning: true }), + }, +})); + +const { handleRunDreaming } = await import(${JSON.stringify(apiHandlersUrl)}); +const out = await handleRunDreaming(); +console.log(JSON.stringify(out)); +`; + } + + it("maps concurrent runs to already_running", () => { + const result = runFixture(apiMappingFixture("Dreaming already running")); + expect(result.stderr).toBe(""); + expect(result.exitCode).toBe(0); + expect(result.parsed).toEqual({ success: false, error: "already_running" }); + }); + + it("maps missing provider to dreaming_provider_unavailable", () => { + const result = runFixture(apiMappingFixture("dreaming_provider_unavailable")); + expect(result.stderr).toBe(""); + expect(result.exitCode).toBe(0); + expect(result.parsed).toEqual({ success: false, error: "dreaming_provider_unavailable" }); + }); +}); + +describe("dreaming service groupsFailed", () => { + it("counts LLM failures as groupsFailed not groupsSkipped", () => { + const storageDir = mkdtempSync(join(tmpdir(), "dream-groups-failed-")); + tempDirs.push(storageDir); + const vecA = JSON.stringify([1, 0, 0, 0]); + const vecB = JSON.stringify([0.99, 0.01, 0, 0]); + const result = runFixture(` +import { mock } from "bun:test"; +import { mkdirSync } from "node:fs"; + +const storagePath = ${JSON.stringify(storageDir)}; +mkdirSync(storagePath, { recursive: true }); + +mock.module(${JSON.stringify(configUrl)}, () => ({ + CONFIG: { + storagePath, + opencodeProvider: "openai", + opencodeModel: "gpt-test", + dreaming: { + enabled: true, + maxGroupsPerRun: 5, + relatedThreshold: 0.5, + crossProject: false, + trashRetentionDays: 30, + intervalHours: 24, + }, + memory: { defaultScope: "project" }, + }, +})); + +mock.module(${JSON.stringify(loggerUrl)}, () => ({ + log: () => {}, + safeLog: () => {}, + serializeError: (e) => ({ error: String(e) }), +})); + +mock.module(${JSON.stringify(readyUrl)}, () => ({ + ensureTursoReady: async () => {}, +})); + +mock.module(${JSON.stringify(dreamingLockUrl)}, () => ({ + tryAcquireDreamingLock: () => () => {}, +})); + +mock.module(${JSON.stringify(embeddingUrl)}, () => ({ + embeddingService: {}, +})); + +mock.module(${JSON.stringify(shardManagerUrl)}, () => ({ + tursoShardManager: { + getAllShards: async (kind) => + kind === "project" ? [{ id: "p1", dbPath: "/tmp/p1.db" }] : [], + }, +})); + +mock.module(${JSON.stringify(connectionManagerUrl)}, () => ({ + tursoConnectionManager: { + getConnection: async () => ({}), + }, +})); + +mock.module(${JSON.stringify(vectorUtilsUrl)}, () => ({ + parseExtractedVector: (json) => { + const arr = JSON.parse(String(json)); + return new Float32Array(arr); + }, + formatTagsForEmbedding: (t) => t, +})); + +mock.module(${JSON.stringify(vectorSearchUrl)}, () => ({ + tursoVectorSearch: { + getAllMemoriesWithExtractedVectors: async () => [ + { + id: "a", + content: "alpha fact", + container_tag: "proj_x", + created_at: 1, + is_pinned: 0, + vector_json: ${JSON.stringify(vecA)}, + }, + { + id: "b", + content: "alpha fact again", + container_tag: "proj_x", + created_at: 2, + is_pinned: 0, + vector_json: ${JSON.stringify(vecB)}, + }, + ], + }, +})); + +mock.module(${JSON.stringify(opencodeProviderUrl)}, () => ({ + getV2Client: () => ({ session: {} }), + generateStructuredOutput: async () => { + throw new Error("DEEPSEEK_API_KEY is not set"); + }, +})); + +const { DreamingService } = await import(${JSON.stringify(dreamingServiceUrl)}); +const { existsSync, readFileSync } = await import("node:fs"); +const { join } = await import("node:path"); +const service = new DreamingService(); +const out = await service.run(); +const statePath = join(storagePath, "dreaming-state.json"); +const state = existsSync(statePath) ? JSON.parse(readFileSync(statePath, "utf8")) : null; +console.log(JSON.stringify({ out, state })); +`); + expect(result.stderr).toBe(""); + expect(result.exitCode).toBe(0); + const parsed = result.parsed as { + out: { + groupsConsidered: number; + groupsFailed: number; + groupsSkipped: number; + groupsMerged: number; + }; + state: { lastRunAt: number } | null; + }; + expect(parsed.out.groupsConsidered).toBeGreaterThanOrEqual(1); + expect(parsed.out.groupsFailed).toBe(parsed.out.groupsConsidered); + expect(parsed.out.groupsSkipped).toBe(0); + expect(parsed.out.groupsMerged).toBe(0); + // Total provider failure must not advance the idle schedule. + expect(parsed.state).toBeNull(); + }); + + it("throws when already running", () => { + const storageDir = mkdtempSync(join(tmpdir(), "dream-busy-")); + tempDirs.push(storageDir); + const result = runFixture(` +import { mock } from "bun:test"; +import { mkdirSync } from "node:fs"; + +const storagePath = ${JSON.stringify(storageDir)}; +mkdirSync(storagePath, { recursive: true }); + +let releaseReady; +const readyGate = new Promise((resolve) => { + releaseReady = resolve; +}); + +mock.module(${JSON.stringify(configUrl)}, () => ({ + CONFIG: { + storagePath, + opencodeProvider: "openai", + opencodeModel: "gpt-test", + dreaming: { + enabled: true, + maxGroupsPerRun: 5, + relatedThreshold: 0.9, + crossProject: false, + trashRetentionDays: 30, + intervalHours: 24, + }, + memory: { defaultScope: "project" }, + }, +})); +mock.module(${JSON.stringify(loggerUrl)}, () => ({ log: () => {}, safeLog: () => {}, serializeError: (e) => ({ error: String(e) }) })); +mock.module(${JSON.stringify(readyUrl)}, () => ({ + ensureTursoReady: () => readyGate, +})); +mock.module(${JSON.stringify(dreamingLockUrl)}, () => ({ + tryAcquireDreamingLock: () => () => {}, +})); +mock.module(${JSON.stringify(embeddingUrl)}, () => ({ embeddingService: {} })); +mock.module(${JSON.stringify(shardManagerUrl)}, () => ({ + tursoShardManager: { getAllShards: async () => [] }, +})); +mock.module(${JSON.stringify(connectionManagerUrl)}, () => ({ + tursoConnectionManager: { getConnection: async () => ({}) }, +})); +mock.module(${JSON.stringify(vectorSearchUrl)}, () => ({ + tursoVectorSearch: { getAllMemoriesWithExtractedVectors: async () => [] }, +})); +mock.module(${JSON.stringify(vectorUtilsUrl)}, () => ({ + parseExtractedVector: () => null, + formatTagsForEmbedding: (t) => t, +})); +mock.module(${JSON.stringify(opencodeProviderUrl)}, () => ({ + getV2Client: () => null, + generateStructuredOutput: async () => ({}), +})); + +const { DreamingService } = await import(${JSON.stringify(dreamingServiceUrl)}); +const service = new DreamingService(); +const first = service.run(); +await new Promise((r) => setTimeout(r, 20)); +let secondError = null; +try { + await service.run(); +} catch (e) { + secondError = e instanceof Error ? e.message : String(e); +} +releaseReady(); +await first; +console.log(JSON.stringify({ secondError })); +`); + expect(result.exitCode).toBe(0); + expect((result.parsed as { secondError: string }).secondError).toBe("Dreaming already running"); + }); +}); diff --git a/tests/plugin-turso-refcount.test.ts b/tests/plugin-turso-refcount.test.ts index 529220ab..f8511c03 100644 --- a/tests/plugin-turso-refcount.test.ts +++ b/tests/plugin-turso-refcount.test.ts @@ -79,11 +79,16 @@ describe("plugin Turso refcount (#443)", () => { const id = await userPromptManager.savePrompt("sess-b", "msg-b", "/tmp/p", "world"); // Fake a warmed local model so the close path has something to release. + // Provide dispose explicitly: earlier suite mocks may replace embeddingService + // with a stub that lacks the real EmbeddingService methods. const { embeddingService } = await import("../src/services/embedding.js"); const released: string[] = []; Object.assign(embeddingService, { pipe: { dispose: async () => released.push("dispose") }, isWarmedUp: true, + async dispose() { + await (this as { pipe?: { dispose?: () => Promise } }).pipe?.dispose?.(); + }, }); await memoryClient.close(); diff --git a/tests/turso-connection-manager.test.ts b/tests/turso-connection-manager.test.ts index 7ebabe65..9a2e3a51 100644 --- a/tests/turso-connection-manager.test.ts +++ b/tests/turso-connection-manager.test.ts @@ -170,7 +170,9 @@ describe("turso connection manager", () => { expect(opens[0]?.path).toBe(dbPath); expect(opens[0]?.opts).toMatchObject({ experimental: - process.platform === "win32" ? ["encryption"] : ["encryption", "multiprocess_wal"], + process.platform === "win32" + ? ["encryption", "vacuum"] + : ["encryption", "vacuum", "multiprocess_wal"], }); expect(opens[0]?.opts).not.toHaveProperty("encryption"); } finally { diff --git a/tests/turso-error-classifiers.test.ts b/tests/turso-error-classifiers.test.ts new file mode 100644 index 00000000..3f6ad43c --- /dev/null +++ b/tests/turso-error-classifiers.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from "bun:test"; +import { + isIdxDeleteMismatchError, + isVacuumUnsupportedError, +} from "../src/services/turso/error-classifiers.js"; + +describe("isVacuumUnsupportedError", () => { + it("matches Turso experimental VACUUM parse errors", () => { + expect( + isVacuumUnsupportedError( + new Error( + "SqliteError: prepare failed: Parse error: VACUUM is an experimental feature. Enable with --experimental-vacuum flag" + ) + ) + ).toBe(true); + expect(isVacuumUnsupportedError("experimental-vacuum flag required")).toBe(true); + }); + + it("ignores busy / lock errors", () => { + expect(isVacuumUnsupportedError(new Error("database is locked"))).toBe(false); + expect(isVacuumUnsupportedError("SQLITE_BUSY")).toBe(false); + }); +}); + +describe("isIdxDeleteMismatchError", () => { + it("matches IdxDelete missing index entry", () => { + expect( + isIdxDeleteMismatchError( + new Error( + "Corrupt database: IdxDelete: no matching index entry found for key [Value(Numeric(Integer(1))), Value(Numeric(Integer(2135)))] while seeking" + ) + ) + ).toBe(true); + expect(isIdxDeleteMismatchError("no matching index entry found for key [1, 2]")).toBe(true); + }); + + it("does not treat every Corrupt database error as mismatch", () => { + expect(isIdxDeleteMismatchError(new Error("Corrupt database: page checksum mismatch"))).toBe( + false + ); + }); +}); diff --git a/tests/turso-multiprocess.test.ts b/tests/turso-multiprocess.test.ts index b75c472c..4f7f6173 100644 --- a/tests/turso-multiprocess.test.ts +++ b/tests/turso-multiprocess.test.ts @@ -15,9 +15,17 @@ describe("turso multiprocess_wal", () => { it("buildConnectOptions enables encryption and multiprocess_wal on Unix only", async () => { const { buildConnectOptions, tursoExperimentalFeatures, supportsTursoMultiprocessWal } = await import("../src/services/turso/connection-manager.js"); - expect(tursoExperimentalFeatures("linux")).toEqual(["encryption", "multiprocess_wal"]); - expect(tursoExperimentalFeatures("darwin")).toEqual(["encryption", "multiprocess_wal"]); - expect(tursoExperimentalFeatures("win32")).toEqual(["encryption"]); + expect(tursoExperimentalFeatures("linux")).toEqual([ + "encryption", + "vacuum", + "multiprocess_wal", + ]); + expect(tursoExperimentalFeatures("darwin")).toEqual([ + "encryption", + "vacuum", + "multiprocess_wal", + ]); + expect(tursoExperimentalFeatures("win32")).toEqual(["encryption", "vacuum"]); expect(supportsTursoMultiprocessWal("win32")).toBe(false); expect(buildConnectOptions()).toEqual({ experimental: tursoExperimentalFeatures(), diff --git a/tests/turso-wal-index-recovery.test.ts b/tests/turso-wal-index-recovery.test.ts index 2d24eba6..b4d57c79 100644 --- a/tests/turso-wal-index-recovery.test.ts +++ b/tests/turso-wal-index-recovery.test.ts @@ -509,14 +509,22 @@ describe("turso WAL index recovery", () => { await import("../src/services/turso/connection-manager.js"); try { CONFIG.tursoMultiprocessWal = false; - expect(tursoExperimentalFeatures("linux")).toEqual(["encryption"]); - expect(buildConnectOptions()).toEqual({ experimental: ["encryption"] }); + expect(tursoExperimentalFeatures("linux")).toEqual(["encryption", "vacuum"]); + expect(buildConnectOptions()).toEqual({ experimental: ["encryption", "vacuum"] }); } finally { CONFIG.tursoMultiprocessWal = true; } - expect(tursoExperimentalFeatures("linux")).toEqual(["encryption", "multiprocess_wal"]); - expect(tursoExperimentalFeatures("linux", false)).toEqual(["encryption"]); - expect(tursoExperimentalFeatures("linux", true)).toEqual(["encryption", "multiprocess_wal"]); + expect(tursoExperimentalFeatures("linux")).toEqual([ + "encryption", + "vacuum", + "multiprocess_wal", + ]); + expect(tursoExperimentalFeatures("linux", false)).toEqual(["encryption", "vacuum"]); + expect(tursoExperimentalFeatures("linux", true)).toEqual([ + "encryption", + "vacuum", + "multiprocess_wal", + ]); }); }); diff --git a/web/src/App.svelte b/web/src/App.svelte index 2ea8a071..03716fa8 100644 --- a/web/src/App.svelte +++ b/web/src/App.svelte @@ -347,7 +347,12 @@ -