From 2fd214b4f2962408459d3ede10de5ca051591853 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 00:44:04 -0700 Subject: [PATCH] FN-5975: extend archive FTS maintenance coverage Add archive-database FTS maintenance and document the expanded compaction policy. - add archive FTS maintenance helpers for optimize, rebuild, size measurement, and row counts - extend self-healing maintenance to compact and rebuild archived_tasks_fts on a slower archive-specific cadence - cover archive FTS maintenance with new core and engine tests and update architecture/storage docs Files changed: docs/architecture.md | 2 +- docs/storage.md | 12 +- .../__tests__/archive-db-fts-maintenance.test.ts | 221 +++++++++++++++++++++ packages/core/src/archive-db.ts | 61 +++++- packages/core/src/db.ts | 21 +- packages/core/src/store.ts | 20 ++ .../src/__tests__/fts-maintenance-archive.test.ts | 207 +++++++++++++++++++ packages/engine/src/self-healing.ts | 80 ++++++++ 8 files changed, 608 insertions(+), 16 deletions(-) Fusion-Task-Id: FN-5975 Fusion-Task-Lineage: 13645c8c-3126-4d1b-af23-7cab1a8bb276 --- docs/architecture.md | 2 +- docs/storage.md | 12 +- .../archive-db-fts-maintenance.test.ts | 221 ++++++++++++++++++ packages/core/src/archive-db.ts | 61 ++++- packages/core/src/db.ts | 21 +- packages/core/src/store.ts | 20 ++ .../__tests__/fts-maintenance-archive.test.ts | 207 ++++++++++++++++ packages/engine/src/self-healing.ts | 80 +++++++ 8 files changed, 608 insertions(+), 16 deletions(-) create mode 100644 packages/core/src/__tests__/archive-db-fts-maintenance.test.ts create mode 100644 packages/engine/src/__tests__/fts-maintenance-archive.test.ts diff --git a/docs/architecture.md b/docs/architecture.md index d2a6df9fae..1212f2c2f2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -669,7 +669,7 @@ Runtime action-gate flow (v1): - `GridlockDetector` (`gridlock-detector.ts`) — detects all-blocked todo pipelines and emits notification events (plus explicit clear signals when gridlock resolves) - `TransientErrorDetector` (`transient-error-detector.ts`) — retriable error classification - `SelfHealingManager` (`self-healing.ts`) — auto-unpause/maintenance recovery actions - - Batch 1 maintenance now includes `fts-maintenance` for the live task search index. When `fts5Available === true`, every maintenance tick runs an incremental `merge`, every 4th tick escalates to `optimize`, and any index larger than `32 MiB` or `1 MiB × live task count` is fully rebuilt. Each pass emits `task:fts-maintenance` run-audit telemetry with before/after byte counts. + - Batch 1 maintenance now includes one `fts-maintenance` step for both search indexes. The live `tasks_fts` branch still runs `merge` every tick, `optimize` every 4th tick, and `rebuild` above `32 MiB` or `1 MiB × live task count`. The archive `archived_tasks_fts` branch is lighter because archive writes are mostly append-only: `merge` every 8th tick, `optimize` every 24th tick, and `rebuild` above `64 MiB` or `512 KiB × archived row count`. Each branch is independently guarded by `fts5Available` and emits `task:fts-maintenance` run-audit telemetry with distinct `target` values (`tasks_fts` vs `archived_tasks_fts`). - `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`. - Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`. diff --git a/docs/storage.md b/docs/storage.md index c979bf25f6..bb89e74a89 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -143,14 +143,12 @@ Important execution nuance: - Full-row task persistence is still intentional for create/restore/replication-class paths: `insertTask` / `atomicCreateTaskJson` remain plain `INSERT`, and direct replication-style upserts (`upsertTaskWithFtsRecovery`, for example task-metadata snapshot application) still use the generated full-row `INSERT ... ON CONFLICT DO UPDATE` form. - After a partial SQLite update, Fusion rewrites compatibility `task.json` from a fresh DB read so the disk mirror stays byte-aligned with the authoritative row even on narrow SQL patches. - Checkout lease renewal has its own targeted path (`renewCheckoutLease`), updating only `checkoutRunId`, `checkoutLeaseRenewedAt`, and `updatedAt` instead of routing through the broad `updateTask(...)` mutator. -- `Database.getFtsIndexBytes()` measures index size via `SELECT SUM(LENGTH(block)) FROM tasks_fts_data`. Fusion intentionally does **not** rely on `dbstat`, because node:sqlite builds do not guarantee `SQLITE_ENABLE_DBSTAT_VTAB`. -- `SelfHealingManager` Batch 1 now runs `fts-maintenance` when `fts5Available === true`: - - every maintenance tick: incremental `merge` compaction - - every 4th maintenance tick: heavier `optimize` - - immediate full `rebuild` when `tasks_fts` exceeds either `32 MiB` absolute or `1 MiB × live task count` -- Each maintenance pass emits run-audit telemetry with `mutationType: "task:fts-maintenance"` and `metadata` including `mode`, `bytesBefore`, `bytesAfter`, `taskCount`, `rebuilt`, and the threshold values. +- Both `Database.getFtsIndexBytes()` and `ArchiveDatabase.getFtsIndexBytes()` measure index size via `SELECT SUM(LENGTH(block)) FROM _data`. Fusion intentionally does **not** rely on `dbstat`, because node:sqlite builds do not guarantee `SQLITE_ENABLE_DBSTAT_VTAB`. +- `SelfHealingManager` Batch 1 runs one `fts-maintenance` step with per-index `fts5Available` guards: + - `tasks_fts`: every maintenance tick runs incremental `merge`, every 4th tick escalates to `optimize`, and an immediate full `rebuild` fires when the index exceeds either `32 MiB` absolute or `1 MiB × live task count`. + - `archived_tasks_fts`: because the archive DB is mostly append-only, maintenance runs less often — incremental `merge` every 8th tick, `optimize` every 24th tick, and a full `rebuild` only when the index exceeds either `64 MiB` absolute or `512 KiB × archived row count`. +- Each maintenance pass emits run-audit telemetry with `mutationType: "task:fts-maintenance"`; the live index uses `target: "tasks_fts"`, the archive index uses `target: "archived_tasks_fts"`, and metadata includes the before/after byte counts plus row-count/threshold details for that target. - `rebuildFts5Index()` and migration 103 also set conservative FTS5 merge policy (`automerge=8`, `crisismerge=16`) so legitimate text edits merge segments sooner without forcing the heaviest optimize path on every write. -- `archived_tasks_fts` is intentionally **not** compacted by this task. The archive DB is effectively append-only for completed tasks, so it does not see the same constant-update churn as the live board; archive FTS compaction is deferred to a follow-up if archive bloat is observed. ### Attached live-FTS DB investigation (FN-5976) diff --git a/packages/core/src/__tests__/archive-db-fts-maintenance.test.ts b/packages/core/src/__tests__/archive-db-fts-maintenance.test.ts new file mode 100644 index 0000000000..cf9dd5d3b3 --- /dev/null +++ b/packages/core/src/__tests__/archive-db-fts-maintenance.test.ts @@ -0,0 +1,221 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { ArchiveDatabase } from "../archive-db.js"; +import type { ArchivedTaskEntry } from "../types.js"; + +type ArchiveEntryOverrides = Partial & { title?: string | null }; + +function makeTmpDir(prefix = "kb-archive-fts-"): string { + return mkdtempSync(join(tmpdir(), prefix)); +} + +function makeEntry(id: string, overrides: ArchiveEntryOverrides = {}): ArchivedTaskEntry { + const timestamp = overrides.archivedAt ?? "2026-06-03T00:00:00.000Z"; + return { + id, + lineageId: overrides.lineageId ?? id, + column: "archived", + title: overrides.title === null ? undefined : overrides.title ?? `title ${id}`, + description: overrides.description ?? `description ${id}`, + comments: overrides.comments ?? [], + dependencies: overrides.dependencies ?? [], + steps: overrides.steps ?? [], + currentStep: overrides.currentStep ?? 0, + log: overrides.log ?? [], + createdAt: overrides.createdAt ?? timestamp, + updatedAt: overrides.updatedAt ?? timestamp, + archivedAt: timestamp, + columnMovedAt: overrides.columnMovedAt ?? timestamp, + prompt: overrides.prompt, + }; +} + +describe("ArchiveDatabase FTS maintenance", () => { + let prevDisableFts5: string | undefined; + + beforeEach(() => { + prevDisableFts5 = process.env.FUSION_DISABLE_FTS5; + }); + + afterEach(() => { + if (prevDisableFts5 === undefined) { + delete process.env.FUSION_DISABLE_FTS5; + } else { + process.env.FUSION_DISABLE_FTS5 = prevDisableFts5; + } + }); + + it("rebuilds a churned disk-backed archive index down to a bounded size", async () => { + const dir = makeTmpDir(); + const archive = new ArchiveDatabase(dir); + + try { + archive.init(); + if (!archive.fts5Available) { + expect(archive.rebuildFts5Index()).toBe(false); + return; + } + + const payload = "alpha ".repeat(1200); + for (let i = 0; i < 180; i++) { + archive.upsert(makeEntry("FN-ARCHIVE-1", { + archivedAt: new Date(1717372800000 + i * 1000).toISOString(), + updatedAt: new Date(1717372800000 + i * 1000).toISOString(), + title: `release-note-${i}`, + description: `${payload}${i}`, + comments: [{ id: `c-${i}`, text: `${payload}comment-${i}`, author: "tester", createdAt: new Date(1717372800000 + i * 1000).toISOString() }], + })); + } + + const grownBytes = archive.getFtsIndexBytes(); + expect(grownBytes).not.toBeNull(); + expect(grownBytes!).toBeGreaterThan(0); + expect(archive.getArchivedRowCount()).toBe(1); + + expect(archive.rebuildFts5Index()).toBe(true); + const rebuiltBytes = archive.getFtsIndexBytes(); + expect(rebuiltBytes).not.toBeNull(); + expect(rebuiltBytes!).toBeLessThan(grownBytes!); + expect(rebuiltBytes!).toBeLessThan(1 * 1024 * 1024); + expect(archive.search("release-note-179", 10).map((entry) => entry.id)).toContain("FN-ARCHIVE-1"); + } finally { + archive.close(); + await rm(dir, { recursive: true, force: true }); + } + }); + + it("supports optimize and merge compaction on disk-backed archives", async () => { + const dir = makeTmpDir(); + const archive = new ArchiveDatabase(dir); + + try { + archive.init(); + if (!archive.fts5Available) { + expect(archive.optimizeFts5("merge")).toBe(false); + expect(archive.optimizeFts5("optimize")).toBe(false); + return; + } + + archive.upsert(makeEntry("FN-ARCHIVE-2", { + description: "optimize target alpha beta gamma", + comments: [{ id: "c-1", text: "merge optimize searchable", author: "tester", createdAt: "2026-06-03T00:00:00.000Z" }], + })); + + expect(archive.optimizeFts5("merge")).toBe(true); + expect(archive.optimizeFts5("optimize")).toBe(true); + expect(archive.search("searchable", 10).map((entry) => entry.id)).toContain("FN-ARCHIVE-2"); + } finally { + archive.close(); + await rm(dir, { recursive: true, force: true }); + } + }); + + it("keeps archive search results identical before and after compaction across null fields, hyphenated tokens, churn, and deletes", async () => { + const dir = makeTmpDir(); + const archive = new ArchiveDatabase(dir); + + try { + archive.init(); + const rawDb = (archive as any).db; + + archive.upsert(makeEntry("FN-ARCHIVE-3", { + title: "release-note-guard", + description: "archive special-char target", + comments: [{ id: "c-2", text: "comment-needle", author: "tester", createdAt: "2026-06-03T00:00:00.000Z" }], + })); + archive.upsert(makeEntry("FN-ARCHIVE-4", { + title: null, + description: "null title searchable phrase", + comments: [], + })); + rawDb.prepare("UPDATE archived_tasks SET comments = NULL WHERE id = ?").run("FN-ARCHIVE-4"); + archive.upsert(makeEntry("FN-ARCHIVE-5", { + title: "delete-me", + description: "deleted archive needle", + })); + archive.delete("FN-ARCHIVE-5"); + + for (let i = 0; i < 60; i++) { + archive.upsert(makeEntry("FN-ARCHIVE-3", { + archivedAt: new Date(1717372800000 + i * 1000).toISOString(), + updatedAt: new Date(1717372800000 + i * 1000).toISOString(), + title: `release-note-guard ${i}`, + description: `archive special-char target marker-${i}`, + comments: [{ id: `c-${i}`, text: `comment-needle marker-${i}`, author: "tester", createdAt: new Date(1717372800000 + i * 1000).toISOString() }], + })); + } + + const queryResultsBefore = { + hyphen: archive.search("release-note-guard", 10).map((entry) => entry.id).sort(), + nullTitle: archive.search("searchable phrase", 10).map((entry) => entry.id).sort(), + comment: archive.search("comment-needle", 10).map((entry) => entry.id).sort(), + special: archive.search("test + special (chars)", 10).map((entry) => entry.id).sort(), + deleted: archive.search("deleted archive needle", 10).map((entry) => entry.id).sort(), + }; + + expect(queryResultsBefore.hyphen).toContain("FN-ARCHIVE-3"); + expect(queryResultsBefore.nullTitle).toContain("FN-ARCHIVE-4"); + expect(queryResultsBefore.comment).toContain("FN-ARCHIVE-3"); + expect(queryResultsBefore.deleted).not.toContain("FN-ARCHIVE-5"); + + expect(archive.optimizeFts5("optimize")).toBe(archive.fts5Available); + expect(archive.rebuildFts5Index()).toBe(archive.fts5Available); + + const queryResultsAfter = { + hyphen: archive.search("release-note-guard", 10).map((entry) => entry.id).sort(), + nullTitle: archive.search("searchable phrase", 10).map((entry) => entry.id).sort(), + comment: archive.search("comment-needle", 10).map((entry) => entry.id).sort(), + special: archive.search("test + special (chars)", 10).map((entry) => entry.id).sort(), + deleted: archive.search("deleted archive needle", 10).map((entry) => entry.id).sort(), + }; + + expect(queryResultsAfter).toEqual(queryResultsBefore); + } finally { + archive.close(); + await rm(dir, { recursive: true, force: true }); + } + }); + + it("treats maintenance seams as safe no-ops when FTS5 is disabled or in-memory", async () => { + process.env.FUSION_DISABLE_FTS5 = "1"; + const disabledDir = makeTmpDir("kb-archive-fts-disabled-"); + const disabledArchive = new ArchiveDatabase(disabledDir); + + try { + disabledArchive.init(); + disabledArchive.upsert(makeEntry("FN-ARCHIVE-6", { title: null, description: "fallback-like alpha-beta" })); + expect(disabledArchive.fts5Available).toBe(false); + expect(disabledArchive.getFtsIndexBytes()).toBeNull(); + expect(disabledArchive.optimizeFts5("merge")).toBe(false); + expect(disabledArchive.optimizeFts5("optimize")).toBe(false); + expect(disabledArchive.rebuildFts5Index()).toBe(false); + expect(disabledArchive.search("alpha-beta", 10).map((entry) => entry.id)).toEqual(["FN-ARCHIVE-6"]); + } finally { + disabledArchive.close(); + await rm(disabledDir, { recursive: true, force: true }); + } + + delete process.env.FUSION_DISABLE_FTS5; + const memoryArchive = new ArchiveDatabase("/tmp/fusion-archive-memory-test", { inMemory: true }); + try { + memoryArchive.init(); + memoryArchive.upsert(makeEntry("FN-ARCHIVE-7", { description: "memory archive search" })); + expect(() => memoryArchive.getArchivedRowCount()).not.toThrow(); + expect(memoryArchive.search("memory archive", 10).map((entry) => entry.id)).toContain("FN-ARCHIVE-7"); + if (memoryArchive.fts5Available) { + expect(memoryArchive.optimizeFts5("merge")).toBe(true); + expect(memoryArchive.rebuildFts5Index()).toBe(true); + } else { + expect(memoryArchive.optimizeFts5("merge")).toBe(false); + expect(memoryArchive.rebuildFts5Index()).toBe(false); + } + } finally { + memoryArchive.close(); + rmSync("/tmp/fusion-archive-memory-test", { recursive: true, force: true }); + } + }); +}); diff --git a/packages/core/src/archive-db.ts b/packages/core/src/archive-db.ts index 799f5758ce..b729b93a7f 100644 --- a/packages/core/src/archive-db.ts +++ b/packages/core/src/archive-db.ts @@ -2,9 +2,11 @@ import { DatabaseSync } from "./sqlite-adapter.js"; import { existsSync, mkdirSync } from "node:fs"; import { join } from "node:path"; import type { ArchivedTaskEntry } from "./types.js"; -import { probeFts5 } from "./db.js"; +import { isFts5CorruptionError, probeFts5 } from "./db.js"; import { hasTitleIdDrift, normalizeTitleForTaskId } from "./task-title-id-drift.js"; +const ARCHIVED_TASKS_FTS_MERGE_PAGES = 16; + const BASE_SCHEMA_SQL = ` CREATE TABLE IF NOT EXISTS archived_tasks ( id TEXT PRIMARY KEY, @@ -157,6 +159,63 @@ export class ArchiveDatabase { this.db.prepare("DELETE FROM archived_tasks WHERE id = ?").run(id); } + rebuildFts5Index(): boolean { + if (!this._fts5Available) { + return false; + } + + try { + this.db.exec("INSERT INTO archived_tasks_fts(archived_tasks_fts) VALUES('rebuild')"); + return true; + } catch (error) { + console.warn("[fusion:archive-db] Failed to rebuild archive FTS5 index", error); + throw error; + } + } + + optimizeFts5(mode: "optimize" | "merge" = "optimize"): boolean { + if (!this._fts5Available) { + return false; + } + + try { + if (mode === "merge") { + this.db.exec( + `INSERT INTO archived_tasks_fts(archived_tasks_fts, rank) VALUES('merge', ${ARCHIVED_TASKS_FTS_MERGE_PAGES})`, + ); + } else { + this.db.exec("INSERT INTO archived_tasks_fts(archived_tasks_fts) VALUES('optimize')"); + } + return true; + } catch (error) { + if (isFts5CorruptionError(error)) { + return this.rebuildFts5Index(); + } + throw error; + } + } + + /** + * Estimate archive FTS index bytes using the shadow-table block payload. + * Prefer this over `dbstat` because node:sqlite builds do not guarantee + * `SQLITE_ENABLE_DBSTAT_VTAB`, while `archived_tasks_fts_data` exists anywhere FTS5 does. + */ + getFtsIndexBytes(): number | null { + if (!this._fts5Available) { + return null; + } + + const row = this.db.prepare("SELECT COALESCE(SUM(LENGTH(block)), 0) AS bytes FROM archived_tasks_fts_data").get() as + | { bytes?: number } + | undefined; + return typeof row?.bytes === "number" ? row.bytes : 0; + } + + getArchivedRowCount(): number { + const row = this.db.prepare("SELECT COUNT(*) AS count FROM archived_tasks").get() as { count?: number } | undefined; + return typeof row?.count === "number" ? row.count : 0; + } + /** * Full-text search over archived tasks. Accepts a raw user query and routes * through FTS5 when available, or a LIKE-based scan when not. diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index ae83d9179a..eb50c0ae20 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -147,6 +147,19 @@ export function probeFts5(db: DatabaseSync): boolean { } } +/** + * Check whether an error appears to be an FTS5 corruption/integrity failure. + */ +export function isFts5CorruptionError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error ?? ""); + const lower = message.toLowerCase(); + return ( + lower.includes("corruption found reading blob") || + lower.includes("database disk image is malformed") || + (lower.includes("fts5") && lower.includes("corrupt")) + ); +} + // ── Schema Definition ──────────────────────────────────────────────── const SCHEMA_VERSION = 114; @@ -4703,13 +4716,7 @@ export class Database { * Check whether an error appears to be an FTS5 corruption/integrity failure. */ isFts5CorruptionError(error: unknown): boolean { - const message = error instanceof Error ? error.message : String(error ?? ""); - const lower = message.toLowerCase(); - return ( - lower.includes("corruption found reading blob") || - lower.includes("database disk image is malformed") || - (lower.includes("fts5") && lower.includes("corrupt")) - ); + return isFts5CorruptionError(error); } /** diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 532dfe2bd9..e9e4ed5ab0 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -14660,18 +14660,38 @@ ${stepsSection}`; return this.db.fts5Available; } + get archiveFts5Available(): boolean { + return this.archiveDb.fts5Available; + } + optimizeFts5(mode?: "optimize" | "merge"): boolean { return this.db.optimizeFts5(mode); } + optimizeArchiveFts5(mode?: "optimize" | "merge"): boolean { + return this.archiveDb.optimizeFts5(mode); + } + getFtsIndexBytes(): number | null { return this.db.getFtsIndexBytes(); } + getArchiveFtsIndexBytes(): number | null { + return this.archiveDb.getFtsIndexBytes(); + } + getTaskRowCount(): number { return this.db.getTaskRowCount(); } + getArchivedRowCount(): number { + return this.archiveDb.getArchivedRowCount(); + } + + rebuildArchiveFts5Index(): boolean { + return this.archiveDb.rebuildFts5Index(); + } + /** * Run a WAL checkpoint and return checkpoint stats. * diff --git a/packages/engine/src/__tests__/fts-maintenance-archive.test.ts b/packages/engine/src/__tests__/fts-maintenance-archive.test.ts new file mode 100644 index 0000000000..9d11aa00b3 --- /dev/null +++ b/packages/engine/src/__tests__/fts-maintenance-archive.test.ts @@ -0,0 +1,207 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import { mkdtempSync, rmSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { TaskStore, type Settings, type TaskStore as TaskStoreType } from "@fusion/core"; + +import { SelfHealingManager } from "../self-healing.js"; + +function createMockStore(overrides: Record = {}): TaskStoreType & EventEmitter { + const emitter = new EventEmitter(); + return Object.assign(emitter, { + getSettings: vi.fn().mockResolvedValue({ + maintenanceIntervalMs: 0, + globalPause: false, + enginePaused: false, + } as unknown as Settings), + listTasks: vi.fn().mockResolvedValue([]), + recordRunAuditEvent: vi.fn().mockResolvedValue(undefined), + fts5Available: true, + archiveFts5Available: false, + getFtsIndexBytes: vi.fn().mockReturnValueOnce(4096).mockReturnValueOnce(2048), + getTaskRowCount: vi.fn().mockReturnValue(4), + optimizeFts5: vi.fn().mockReturnValue(true), + getDatabase: vi.fn().mockReturnValue({ rebuildFts5Index: vi.fn().mockReturnValue(true) }), + getArchiveFtsIndexBytes: vi.fn(), + getArchivedRowCount: vi.fn(), + optimizeArchiveFts5: vi.fn(), + rebuildArchiveFts5Index: vi.fn(), + ...overrides, + }) as unknown as TaskStoreType & EventEmitter; +} + +function makeTmpDir(prefix: string): string { + return mkdtempSync(join(tmpdir(), prefix)); +} + +const createdDirs = new Set(); + +function trackDir(path: string): string { + createdDirs.add(path); + return path; +} + +async function createStore(options?: { disableFts5?: boolean; inMemoryDb?: boolean }) { + const prevEnv = process.env.FUSION_DISABLE_FTS5; + if (options?.disableFts5) { + process.env.FUSION_DISABLE_FTS5 = "1"; + } else if (prevEnv === "1") { + delete process.env.FUSION_DISABLE_FTS5; + } + + const rootDir = trackDir(makeTmpDir("kb-engine-archive-fts-root-")); + const globalDir = trackDir(makeTmpDir("kb-engine-archive-fts-global-")); + const store = new TaskStore(rootDir, globalDir, { inMemoryDb: options?.inMemoryDb === true }); + await store.init(); + const manager = new SelfHealingManager(store, { rootDir }); + + return { + rootDir, + globalDir, + store, + manager, + restoreEnv() { + if (prevEnv === undefined) { + delete process.env.FUSION_DISABLE_FTS5; + } else { + process.env.FUSION_DISABLE_FTS5 = prevEnv; + } + }, + }; +} + +async function cleanupStore(context: Awaited> | undefined) { + if (!context) return; + context.manager.stop(); + context.store.close(); + context.restoreEnv(); + await rm(context.rootDir, { recursive: true, force: true }); + await rm(context.globalDir, { recursive: true, force: true }); + createdDirs.delete(context.rootDir); + createdDirs.delete(context.globalDir); +} + +afterEach(async () => { + vi.restoreAllMocks(); + for (const dir of Array.from(createdDirs)) { + try { + await rm(dir, { recursive: true, force: true }); + } catch { + rmSync(dir, { recursive: true, force: true }); + } finally { + createdDirs.delete(dir); + } + } +}); + +describe("SelfHealingManager archive FTS maintenance", () => { + it("skips the archive branch without disturbing live maintenance when archive FTS is unavailable", async () => { + const store = createMockStore(); + const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" }); + (manager as any).maintenanceTickCounter = 1; + + await (manager as any).maintainTaskFts(); + + expect(store.optimizeFts5).toHaveBeenCalledWith("merge"); + expect(store.getArchiveFtsIndexBytes).not.toHaveBeenCalled(); + expect(store.optimizeArchiveFts5).not.toHaveBeenCalled(); + expect(store.recordRunAuditEvent).toHaveBeenCalledTimes(1); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:fts-maintenance", + target: "tasks_fts", + })); + }); + + it("compacts a real disk-backed archive index and preserves archive search results", async () => { + let ctx: Awaited> | undefined; + try { + ctx = await createStore(); + const { store, manager } = ctx; + const archiveDb = (store as any).archiveDb; + if (!archiveDb.fts5Available) { + expect(store.archiveFts5Available).toBe(false); + return; + } + + const archivedTask = await store.createTask({ + title: "archive maintenance seed", + description: "archive-maintenance-needle", + column: "done", + }); + await store.archiveTask(archivedTask.id); + + const archivedEntry = await store.findInArchive(archivedTask.id); + expect(archivedEntry).toBeDefined(); + const seedEntry = archivedEntry!; + + const payload = "alpha ".repeat(1600); + for (let i = 0; i < 240; i++) { + archiveDb.upsert({ + ...seedEntry, + title: `archive-maintenance-seed-${i}`, + description: `${payload}archive-maintenance-needle marker-${i}`, + comments: [{ id: `c-${i}`, text: `${payload}comment-${i}`, author: "tester", createdAt: new Date(1717372800000 + i * 1000).toISOString() }], + archivedAt: new Date(1717372800000 + i * 1000).toISOString(), + updatedAt: new Date(1717372800000 + i * 1000).toISOString(), + }); + } + + const grownBytes = store.getArchiveFtsIndexBytes(); + expect(grownBytes).not.toBeNull(); + expect(grownBytes!).toBeGreaterThan(512 * 1024); + + const beforeResults = await store.searchTasks("archive-maintenance-needle"); + expect(beforeResults.map((task) => task.id)).toContain(archivedTask.id); + + (manager as any).maintenanceTickCounter = 24; + await (manager as any).maintainTaskFts(); + + const compactedBytes = store.getArchiveFtsIndexBytes(); + expect(compactedBytes).not.toBeNull(); + expect(compactedBytes!).toBeLessThan(grownBytes!); + expect(compactedBytes!).toBeLessThan(store.getArchivedRowCount() * 512 * 1024); + + const afterResults = await store.searchTasks("archive-maintenance-needle"); + expect(afterResults.map((task) => task.id)).toContain(archivedTask.id); + + const auditEvents = store.getRunAuditEvents({ mutationType: "task:fts-maintenance", limit: 20 }) + .filter((event) => event.target === "archived_tasks_fts"); + expect(auditEvents.length).toBeGreaterThan(0); + expect(auditEvents.at(-1)).toEqual(expect.objectContaining({ + mutationType: "task:fts-maintenance", + target: "archived_tasks_fts", + metadata: expect.objectContaining({ + rowCount: 1, + }), + })); + } finally { + await cleanupStore(ctx); + } + }); + + it("keeps archive fallback search working when FTS5 is disabled", async () => { + let ctx: Awaited> | undefined; + try { + ctx = await createStore({ disableFts5: true }); + const { store, manager } = ctx; + + const archivedTask = await store.createTask({ + title: "archive fallback target", + description: "archive-fallback-needle", + column: "done", + }); + await store.archiveTask(archivedTask.id); + + await expect((manager as any).maintainTaskFts()).resolves.toBeUndefined(); + expect(store.archiveFts5Available).toBe(false); + + const results = await store.searchTasks("archive-fallback-needle"); + expect(results.map((task) => task.id)).toContain(archivedTask.id); + } finally { + await cleanupStore(ctx); + } + }); +}); diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 709f631074..5d4483e6be 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -78,6 +78,13 @@ const FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS = 4; // bounded so sustained text churn heals before segment growth becomes material. const FTS_REBUILD_THRESHOLD_BYTES = 32 * 1024 * 1024; const FTS_REBUILD_BYTES_PER_TASK = 1 * 1024 * 1024; +// The archive index is mostly append-only, so maintenance can run much less +// often than the live task index. We still cap total growth because archive +// rows retain full title/description/comments payloads for the project's life. +const ARCHIVE_FTS_MAINTENANCE_MERGE_CADENCE_TICKS = 8; +const ARCHIVE_FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS = 24; +const ARCHIVE_FTS_REBUILD_THRESHOLD_BYTES = 64 * 1024 * 1024; +const ARCHIVE_FTS_REBUILD_BYTES_PER_TASK = 512 * 1024; export const STALE_ACTIVE_BRANCH_EXECUTION_GRACE_MS = 10 * 60_000; export const COMPLETION_HANDOFF_LIMBO_GRACE_MS = 5 * 60_000; export const MAX_COMPLETION_HANDOFF_LIMBO_RECOVERIES = 3; @@ -8820,6 +8827,17 @@ export class SelfHealingManager { } private async maintainTaskFts(): Promise { + await this.maintainLiveTaskFts(); + + try { + await this.maintainArchiveTaskFts(); + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.error(`Archive FTS maintenance failed: ${errorMessage}`); + } + } + + private async maintainLiveTaskFts(): Promise { if (!this.store.fts5Available) { log.log('Maintenance batch 1 step "fts-maintenance" skipped — FTS5 unavailable'); return; @@ -8881,6 +8899,68 @@ export class SelfHealingManager { } } + private async maintainArchiveTaskFts(): Promise { + if (!this.store.archiveFts5Available) { + log.log('Maintenance batch 1 step "fts-maintenance" archive skipped — FTS5 unavailable'); + return; + } + + const bytesBefore = this.store.getArchiveFtsIndexBytes(); + if (bytesBefore === null) { + log.log('Maintenance batch 1 step "fts-maintenance" archive skipped — FTS shadow tables unavailable'); + return; + } + + const rowCount = this.store.getArchivedRowCount(); + const relativeThresholdBytes = rowCount > 0 ? rowCount * ARCHIVE_FTS_REBUILD_BYTES_PER_TASK : null; + const shouldRebuild = bytesBefore >= ARCHIVE_FTS_REBUILD_THRESHOLD_BYTES + || (relativeThresholdBytes !== null && bytesBefore > relativeThresholdBytes); + const shouldOptimize = !shouldRebuild + && ARCHIVE_FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS > 0 + && this.maintenanceTickCounter % ARCHIVE_FTS_MAINTENANCE_OPTIMIZE_CADENCE_TICKS === 0; + const mode = shouldRebuild ? "rebuild" : shouldOptimize ? "optimize" : "merge"; + + if (mode === "merge" + && ARCHIVE_FTS_MAINTENANCE_MERGE_CADENCE_TICKS > 1 + && this.maintenanceTickCounter % ARCHIVE_FTS_MAINTENANCE_MERGE_CADENCE_TICKS !== 0) { + log.log('Maintenance batch 1 step "fts-maintenance" archive skipped — merge cadence not due'); + return; + } + + let rebuilt = false; + if (mode === "rebuild") { + rebuilt = this.store.rebuildArchiveFts5Index(); + } else { + this.store.optimizeArchiveFts5(mode); + } + + const bytesAfter = this.store.getArchiveFtsIndexBytes(); + log.log(`Maintenance batch 1 step "fts-maintenance" archive ${mode}: ${bytesBefore} → ${bytesAfter ?? "unknown"} bytes (archived=${rowCount})`); + + try { + await createRunAuditor(this.store, { + runId: generateSyntheticRunId("self-heal-fts-maintenance", "archived_tasks_fts"), + agentId: "self-healing", + phase: "maintenance-fts", + }).database({ + type: "task:fts-maintenance" as DatabaseMutationType, + target: "archived_tasks_fts", + metadata: { + mode, + bytesBefore, + bytesAfter, + rowCount, + rebuilt, + absoluteThresholdBytes: ARCHIVE_FTS_REBUILD_THRESHOLD_BYTES, + relativeThresholdBytes, + }, + }); + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + log.warn(`Failed to write archived task:fts-maintenance run-audit event: ${errorMessage}`); + } + } + /** Run a best-effort passive WAL checkpoint without forcing live writers to truncate. */ private checkpointWal(): void { try {