root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(FN-1952): bound recovery and task log growth

Browse Source
This commit is contained in:
gsxdsm
2026-04-16 19:32:35 -07:00
parent 80e49b8e40
commit 4a576c028c
16 changed files with 332 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/core/src/__tests__/task-documents.test.ts
View File

@@ -51,7 +51,7 @@ describe("TaskStore task documents", () => {
expect(tableNames.has("task_documents")).toBe(true);
expect(tableNames.has("task_document_revisions")).toBe(true);
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
const index = db
.prepare(

25
packages/core/src/db.test.ts
View File

@@ -119,7 +119,7 @@ describe("Database", () => {
});
it("seeds schema version", () => {
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
});
it("seeds lastModified", () => {
@@ -142,7 +142,7 @@ describe("Database", () => {
it("is idempotent - calling init() twice does not fail", () => {
expect(() => db.init()).not.toThrow();
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
});
it("does not overwrite existing config on re-init", () => {
@@ -749,7 +749,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 29 (includes v1→v2 through v26→v29)
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -774,11 +774,11 @@ describe("schema migrations", () => {
const db = new Database(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
// Re-init should not fail
db.init();
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
db.close();
});
@@ -794,7 +794,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "agentRatings" }]);
@@ -818,7 +818,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>;
expect(tables).toEqual([{ name: "mission_events" }]);
@@ -922,7 +922,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 29
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -1136,13 +1136,16 @@ describe("FTS5 full-text search", () => {
it("creates FTS5 triggers after init", () => {
const triggers = db.prepare(
"SELECT name FROM sqlite_master WHERE type='trigger'"
).all() as { name: string }[];
"SELECT name, sql FROM sqlite_master WHERE type='trigger'"
).all() as { name: string; sql: string }[];
const triggerNames = triggers.map((t) => t.name);
expect(triggerNames).toContain("tasks_fts_ai");
expect(triggerNames).toContain("tasks_fts_au");
expect(triggerNames).toContain("tasks_fts_ad");
const updateTrigger = triggers.find((t) => t.name === "tasks_fts_au");
expect(updateTrigger?.sql).toContain("AFTER UPDATE OF id, title, description, comments ON tasks");
});
it("populates FTS index from existing tasks on migration", () => {
@@ -1288,7 +1291,7 @@ describe("createDatabase factory", () => {
const db = createDatabase(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
expect(db.getLastModified()).toBeGreaterThan(0);
db.close();

47
packages/core/src/db.ts
View File

@@ -59,7 +59,7 @@ export function fromJson<T>(json: string | null | undefined): T | undefined {
// ── Schema Definition ────────────────────────────────────────────────
const SCHEMA_VERSION = 34;
const SCHEMA_VERSION = 35;
function normalizeTaskComments(
steeringComments: SteeringComment[] | undefined,
@@ -936,13 +936,22 @@ export class Database {
END
`);
// AFTER UPDATE trigger - reindex updated tasks (delete old + insert new)
const hasTaskTitle = this.hasColumn("tasks", "title");
const updateColumns = hasTaskTitle
? "id, title, description, comments"
: "id, description, comments";
const oldTitle = hasTaskTitle ? "COALESCE(old.title, '')" : "''";
const newTitle = hasTaskTitle ? "COALESCE(new.title, '')" : "''";
// AFTER UPDATE trigger - reindex updated tasks (delete old + insert new).
// Restrict this to searchable columns so log/status churn does not bloat
// the FTS index during long-running executor activity.
this.db.exec(`
CREATE TRIGGER IF NOT EXISTS tasks_fts_au AFTER UPDATE ON tasks BEGIN
CREATE TRIGGER IF NOT EXISTS tasks_fts_au AFTER UPDATE OF ${updateColumns} ON tasks BEGIN
INSERT INTO tasks_fts(tasks_fts, rowid, id, title, description, comments)
VALUES('delete', old.rowid, old.id, COALESCE(old.title, ''), old.description, COALESCE(old.comments, '[]'));
VALUES('delete', old.rowid, old.id, ${oldTitle}, old.description, COALESCE(old.comments, '[]'));
INSERT INTO tasks_fts(rowid, id, title, description, comments)
VALUES (new.rowid, new.id, COALESCE(new.title, ''), new.description, COALESCE(new.comments, '[]'));
VALUES (new.rowid, new.id, ${newTitle}, new.description, COALESCE(new.comments, '[]'));
END
`);
@@ -1412,6 +1421,34 @@ export class Database {
this.db.exec(`CREATE INDEX IF NOT EXISTS idxRoutinesScope ON routines(scope)`);
});
}
// Restrict task full-text-search maintenance to searchable fields only.
// Agent/activity logs live in tasks.log and are intentionally not searchable;
// log-only executor updates should not churn or bloat the FTS index.
if (version < 35) {
this.applyMigration(35, () => {
const hasTaskTitle = this.hasColumn("tasks", "title");
const updateColumns = hasTaskTitle
? "id, title, description, comments"
: "id, description, comments";
const oldTitle = hasTaskTitle ? "COALESCE(old.title, '')" : "''";
const newTitle = hasTaskTitle ? "COALESCE(new.title, '')" : "''";
this.db.exec(`
DROP TRIGGER IF EXISTS tasks_fts_au;
CREATE TRIGGER tasks_fts_au AFTER UPDATE OF ${updateColumns} ON tasks BEGIN
INSERT INTO tasks_fts(tasks_fts, rowid, id, title, description, comments)
VALUES('delete', old.rowid, old.id, ${oldTitle}, old.description, COALESCE(old.comments, '[]'));
INSERT INTO tasks_fts(rowid, id, title, description, comments)
VALUES (new.rowid, new.id, ${newTitle}, new.description, COALESCE(new.comments, '[]'));
END;
`);
if (hasTaskTitle) {
this.db.exec("INSERT INTO tasks_fts(tasks_fts) VALUES('rebuild')");
}
});
}
}
/**

8
packages/core/src/insight-store.test.ts
View File

@@ -776,7 +776,7 @@ describe("Migration: pre-33 DB upgrade", () => {
// Step 1: Create a fresh database at v33 (runs all migrations up to 33)
const db1 = createDatabase(legacyDir);
db1.init();
expect(db1.getSchemaVersion()).toBe(34);
expect(db1.getSchemaVersion()).toBe(35);
db1.close();
// Step 2: Manually downgrade to version 32 and drop insight tables
@@ -811,7 +811,7 @@ describe("Migration: pre-33 DB upgrade", () => {
expect(tableNamesBefore).not.toContain("project_insight_runs");
// Now run init — this triggers the v32→v33 migration
db3.init();
expect(db3.getSchemaVersion()).toBe(34);
expect(db3.getSchemaVersion()).toBe(35);
// Step 4: Verify insight tables exist after migration
const tablesAfter = db3.prepare(
@@ -842,12 +842,12 @@ describe("Migration: pre-33 DB upgrade", () => {
try {
const db1 = createDatabase(testDir);
db1.init();
expect(db1.getSchemaVersion()).toBe(34);
expect(db1.getSchemaVersion()).toBe(35);
db1.close();
const db2 = createDatabase(testDir);
expect(() => db2.init()).not.toThrow();
expect(db2.getSchemaVersion()).toBe(34);
expect(db2.getSchemaVersion()).toBe(35);
db2.close();
} finally {
rmSync(testDir, { recursive: true, force: true });

2
packages/core/src/mission-store.test.ts
View File

@@ -2544,7 +2544,7 @@ describe("MissionStore", () => {
describe("Loop State & Validator Run Schema (v31)", () => {
it("schema version is 32 after migration", () => {
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
});
it("mission_features table has loop state columns", () => {

2
packages/core/src/roadmap-store.test.ts
View File

@@ -739,7 +739,7 @@ describe("RoadmapStore", () => {
describe("schema version", () => {
it("schema version is 32 after init", () => {
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
});
});

2
packages/core/src/run-audit.test.ts
View File

@@ -465,7 +465,7 @@ describe("Run Audit", () => {
});
it("schema version is bumped to 28", () => {
expect(db.getSchemaVersion()).toBe(34);
expect(db.getSchemaVersion()).toBe(35);
});
});
});

29
packages/core/src/store.test.ts
View File

@@ -8481,6 +8481,35 @@ describe("RunMutationContext", () => {
}
});
it("logEntry() bounds retained activity entries and truncates large outcomes", async () => {
const localRoot = makeTmpDir();
const localGlobal = makeTmpDir();
try {
const localStore = new TaskStore(localRoot, localGlobal);
await localStore.init();
const task = await localStore.createTask({ description: "Test task" });
const longOutcome = "x".repeat(5_000);
for (let index = 0; index < 1_005; index += 1) {
await localStore.logEntry(task.id, `Action ${index}`, index === 1_004 ? longOutcome : undefined);
}
const updatedTask = await localStore.getTask(task.id);
expect(updatedTask.log).toHaveLength(1_000);
expect(updatedTask.log[0].action).toBe("Action 5");
const lastEntry = updatedTask.log[updatedTask.log.length - 1];
expect(lastEntry.action).toBe("Action 1004");
expect(lastEntry.outcome?.length).toBeLessThan(longOutcome.length);
expect(lastEntry.outcome).toContain("outcome truncated");
localStore.close();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
it("addComment() with runContext includes runContext in log entry", async () => {
const localRoot = makeTmpDir();
const localGlobal = makeTmpDir();

72
packages/core/src/store.ts
View File

@@ -25,6 +25,23 @@ import { runCommandAsync } from "./run-command.js";
* so that all backup operations use a consistent directory.
*/
const LEGACY_BACKUP_DIR = ".kb/backups";
const TASK_ACTIVITY_LOG_ENTRY_LIMIT = 1_000;
const TASK_ACTIVITY_LOG_OUTCOME_LIMIT = 4_000;
function truncateTaskLogOutcome(outcome: string | undefined): string | undefined {
if (!outcome || outcome.length <= TASK_ACTIVITY_LOG_OUTCOME_LIMIT) {
return outcome;
}
return `${outcome.slice(0, TASK_ACTIVITY_LOG_OUTCOME_LIMIT)}\n... outcome truncated to ${TASK_ACTIVITY_LOG_OUTCOME_LIMIT} characters ...`;
}
function compactTaskActivityLog(entries: TaskLogEntry[]): TaskLogEntry[] {
const recentEntries = entries.slice(-TASK_ACTIVITY_LOG_ENTRY_LIMIT);
return recentEntries.map((entry) => ({
...entry,
outcome: truncateTaskLogOutcome(entry.outcome),
}));
}
/**
* Canonicalizes a settings object by resolving legacy defaults.
@@ -335,7 +352,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
"modelPresetId", "modelProvider", "modelId",
"validatorModelProvider", "validatorModelId",
"planningModelProvider", "planningModelId",
"mergeRetries", "stuckKillCount", "recoveryRetryCount", "nextRecoveryAt",
"mergeRetries", "workflowStepRetries", "stuckKillCount", "recoveryRetryCount", "nextRecoveryAt",
"error", "summary", "thinkingLevel",
"createdAt", "updatedAt", "columnMovedAt",
"dependencies", "steps", "comments", "workflowStepResults", "steeringComments",
@@ -346,6 +363,45 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
].map((column) => `${prefix}${column}`).join(", ");
}
private getTaskSelectClauseWithActivityLogLimit(limit: number): string {
const columns = [
"id", "title", "description", "\"column\"", "status", "size", "reviewLevel", "currentStep",
"worktree", "blockedBy", "paused", "baseBranch", "branch", "baseCommitSha",
"modelPresetId", "modelProvider", "modelId",
"validatorModelProvider", "validatorModelId",
"planningModelProvider", "planningModelId",
"mergeRetries", "workflowStepRetries", "stuckKillCount", "recoveryRetryCount", "nextRecoveryAt",
"error", "summary", "thinkingLevel",
"createdAt", "updatedAt", "columnMovedAt",
"dependencies", "steps", "attachments", "steeringComments",
"comments", "workflowStepResults", "prInfo", "issueInfo", "mergeDetails",
"breakIntoSubtasks", "enabledWorkflowSteps", "modifiedFiles",
"missionId", "sliceId", "assignedAgentId", "assigneeUserId",
"checkedOutBy", "checkedOutAt",
];
const limitedLog = `
CASE
WHEN json_valid(log) AND json_array_length(log) > ${limit} THEN (
SELECT json_group_array(json(value))
FROM (
SELECT value
FROM (
SELECT key, value
FROM json_each(tasks.log)
ORDER BY key DESC
LIMIT ${limit}
)
ORDER BY key ASC
)
)
ELSE log
END AS log
`;
return [...columns, limitedLog].join(", ");
}
/**
* Upsert a task to the database. Used by create and update operations.
*/
@@ -423,8 +479,11 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
/**
* Read a task from SQLite by ID.
*/
private readTaskFromDb(id: string): Task | undefined {
const row = this.db.prepare('SELECT * FROM tasks WHERE id = ?').get(id);
private readTaskFromDb(id: string, options?: { activityLogLimit?: number }): Task | undefined {
const selectClause = options?.activityLogLimit
? this.getTaskSelectClauseWithActivityLogLimit(options.activityLogLimit)
: "*";
const row = this.db.prepare(`SELECT ${selectClause} FROM tasks WHERE id = ?`).get(id);
if (!row) return undefined;
return this.rowToTask(row);
}
@@ -1416,8 +1475,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
/**
* Read a task and its prompt content.
*/
async getTask(id: string): Promise<TaskDetail> {
const task = this.readTaskFromDb(id);
async getTask(id: string, options?: { activityLogLimit?: number }): Promise<TaskDetail> {
const task = this.readTaskFromDb(id, options);
if (!task) {
throw new Error(`Task ${id} not found`);
}
@@ -2110,12 +2169,13 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
const entry: TaskLogEntry = {
timestamp: new Date().toISOString(),
action,
outcome,
outcome: truncateTaskLogOutcome(outcome),
};
if (runContext) {
entry.runContext = runContext;
}
task.log.push(entry);
task.log = compactTaskActivityLog(task.log);
task.updatedAt = new Date().toISOString();
// When runContext is provided, record audit event atomically with task mutation