From 304d7af3a193aa62acc8c532d742c66203905ee1 Mon Sep 17 00:00:00 2001 From: EyJunge1 <149941075+EyJunge1@users.noreply.github.com> Date: Mon, 2 Nov 2026 11:07:00 +0100 Subject: [PATCH 1/7] fix: harden cleanup vacuum and dreaming run feedback Enable Turso experimental vacuum, treat IdxDelete as idempotent during cleanup, and surface dreaming concurrency/failures without raw English error toasts. --- src/services/api-handlers.ts | 7 + src/services/cleanup-service.ts | 94 ++++-- src/services/dreaming-service.ts | 6 +- src/services/turso/connection-manager.ts | 2 +- src/services/turso/error-classifiers.ts | 23 ++ src/services/turso/vector-search.ts | 25 +- .../user-prompt/user-prompt-manager.ts | 48 ++- src/services/web-server.ts | 3 + tests/cleanup-resilience.test.ts | 201 ++++++++++++ tests/dreaming-run.test.ts | 302 ++++++++++++++++++ tests/turso-error-classifiers.test.ts | 42 +++ tests/turso-multiprocess.test.ts | 14 +- tests/turso-wal-index-recovery.test.ts | 18 +- web/src/App.svelte | 7 +- web/src/lib/i18n/translations.ts | 12 + .../lib/stores/memories-explorer.svelte.ts | 44 ++- 16 files changed, 798 insertions(+), 50 deletions(-) create mode 100644 src/services/turso/error-classifiers.ts create mode 100644 tests/cleanup-resilience.test.ts create mode 100644 tests/dreaming-run.test.ts create mode 100644 tests/turso-error-classifiers.test.ts 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..85ff7f03 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 { @@ -122,11 +164,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 +214,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/dreaming-service.ts b/src/services/dreaming-service.ts index ab36d320..bd375d71 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; } @@ -306,6 +307,7 @@ export class DreamingService { groupsConsidered: 0, groupsMerged: 0, groupsSkipped: 0, + groupsFailed: 0, groupsFlagged: 0, memoriesSoftDeleted: 0, }; @@ -324,7 +326,7 @@ 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"); @@ -402,7 +404,7 @@ export class DreamingService { result.groupsSkipped++; } } catch (error) { - result.groupsSkipped++; + result.groupsFailed++; log("dreaming: group failed", { error: String(error) }); } } 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..c16995e7 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 { @@ -341,7 +342,15 @@ export class TursoVectorSearch { } async deleteVector(db: TursoDb, memoryId: string): Promise { - await db.run(`DELETE FROM memories WHERE id = ?`, [memoryId]); + try { + await db.run(`DELETE FROM memories WHERE id = ?`, [memoryId]); + } catch (error) { + if (isIdxDeleteMismatchError(error)) { + // Index entry already gone; treat as idempotent success. + return; + } + throw error; + } } async softDeleteVector( @@ -406,11 +415,17 @@ 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 { + 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..c9374e05 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,44 @@ 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)) { + 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..f2d70535 --- /dev/null +++ b/tests/dreaming-run.test.ts @@ -0,0 +1,302 @@ +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 service = new DreamingService(); +const out = await service.run(); +console.log(JSON.stringify(out)); +`); + expect(result.stderr).toBe(""); + expect(result.exitCode).toBe(0); + const out = result.parsed as { + groupsConsidered: number; + groupsFailed: number; + groupsSkipped: number; + groupsMerged: number; + }; + expect(out.groupsConsidered).toBeGreaterThanOrEqual(1); + expect(out.groupsFailed).toBeGreaterThanOrEqual(1); + expect(out.groupsSkipped).toBe(0); + expect(out.groupsMerged).toBe(0); + }); + + 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/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 @@ -