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

feat(FN-2156): migrate agent log storage to SQLite

Browse Source 
- Add an agentLogEntries table and schema migration updates for SQLite-backed agent log persistence
- Persist appended agent logs in SQLite and read task agent logs from the database instead of filesystem-only JSONL
- Import legacy agent log JSONL data into SQLite with type-safe handling for older log field shapes
- Preserve agent logs across task updates and archive flows, and update docs plus tests (including schema assertions) to cover the new behavior
- Add a changeset for @gsxdsm/fusion describing the agent log storage migration
This commit is contained in:
Fusion
2026-04-19 13:39:40 -07:00
committed by gsxdsm
parent 3f8161a90a
commit 5b6392d849
12 changed files with 373 additions and 221 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(39);
expect(db.getSchemaVersion()).toBe(40);
const index = db
.prepare(

2
packages/core/src/db-migrate.ts
View File

@@ -510,7 +510,7 @@ async function migrateAgents(kbDir: string, db: Database): Promise<void> {
/**
* Create backups of legacy files by renaming them with .bak suffix.
* Note: .fusion/tasks/ is NOT renamed because blob files (PROMPT.md, agent.log,
* Note: .fusion/tasks/ is NOT renamed because blob files (PROMPT.md,
* attachments) remain on the filesystem. Only task.json files inside each
* task directory are the "migrated" data now in SQLite. We rename individual
* task.json files to task.json.bak instead.

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

@@ -64,6 +64,7 @@ describe("Database", () => {
expect(tableNames).toContain("agents");
expect(tableNames).toContain("agentHeartbeats");
expect(tableNames).toContain("agentRuns");
expect(tableNames).toContain("agentLogEntries");
expect(tableNames).toContain("agentTaskSessions");
expect(tableNames).toContain("agentApiKeys");
expect(tableNames).toContain("agentConfigRevisions");
@@ -119,6 +120,8 @@ describe("Database", () => {
expect(indexNames).toContain("idxTaskDocumentRevisionsTaskKey");
expect(indexNames).toContain("idxAgentRunsAgentIdStartedAt");
expect(indexNames).toContain("idxAgentRunsStatus");
expect(indexNames).toContain("idxAgentLogEntriesTaskIdTimestamp");
expect(indexNames).toContain("idxAgentLogEntriesTaskIdType");
expect(indexNames).toContain("idxAgentApiKeysAgentId");
expect(indexNames).toContain("idxAgentConfigRevisionsAgentIdCreatedAt");
expect(indexNames).toContain("idxTasksCreatedAt");
@@ -128,7 +131,7 @@ describe("Database", () => {
});
it("seeds schema version", () => {
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
});
it("seeds lastModified", () => {
@@ -151,7 +154,7 @@ describe("Database", () => {
it("is idempotent - calling init() twice does not fail", () => {
expect(() => db.init()).not.toThrow();
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
});
it("does not overwrite existing config on re-init", () => {
@@ -758,7 +761,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 29 (includes v1→v2 through v26→v29)
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -783,11 +786,11 @@ describe("schema migrations", () => {
const db = new Database(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
// Re-init should not fail
db.init();
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
db.close();
});
@@ -803,7 +806,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
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" }]);
@@ -827,7 +830,7 @@ describe("schema migrations", () => {
db.init();
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
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" }]);
@@ -931,7 +934,7 @@ describe("schema migrations", () => {
db.init();
// Verify version bumped to 29
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
// Verify new columns exist and existing data is intact
const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>;
@@ -1300,7 +1303,7 @@ describe("createDatabase factory", () => {
const db = createDatabase(kbDir);
db.init();
expect(db.getSchemaVersion()).toBe(39);
expect(db.getSchemaVersion()).toBe(40);
expect(db.getLastModified()).toBeGreaterThan(0);
db.close();

34
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 = 39;
const SCHEMA_VERSION = 40;
function normalizeTaskComments(
steeringComments: SteeringComment[] | undefined,
@@ -283,6 +283,19 @@ CREATE TABLE IF NOT EXISTS agentRuns (
CREATE INDEX IF NOT EXISTS idxAgentRunsAgentIdStartedAt ON agentRuns(agentId, startedAt);
CREATE INDEX IF NOT EXISTS idxAgentRunsStatus ON agentRuns(status);
CREATE TABLE IF NOT EXISTS agentLogEntries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
taskId TEXT NOT NULL,
timestamp TEXT NOT NULL,
text TEXT NOT NULL,
type TEXT NOT NULL,
detail TEXT,
agent TEXT,
FOREIGN KEY (taskId) REFERENCES tasks(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idxAgentLogEntriesTaskIdTimestamp ON agentLogEntries(taskId, timestamp);
CREATE INDEX IF NOT EXISTS idxAgentLogEntriesTaskIdType ON agentLogEntries(taskId, type);
CREATE TABLE IF NOT EXISTS agentTaskSessions (
agentId TEXT NOT NULL,
taskId TEXT NOT NULL,
@@ -1591,6 +1604,25 @@ export class Database {
});
}
if (version < 40) {
this.applyMigration(40, () => {
this.db.exec(`
CREATE TABLE IF NOT EXISTS agentLogEntries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
taskId TEXT NOT NULL,
timestamp TEXT NOT NULL,
text TEXT NOT NULL,
type TEXT NOT NULL,
detail TEXT,
agent TEXT,
FOREIGN KEY (taskId) REFERENCES tasks(id) ON DELETE CASCADE
)
`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxAgentLogEntriesTaskIdTimestamp ON agentLogEntries(taskId, timestamp)`);
this.db.exec(`CREATE INDEX IF NOT EXISTS idxAgentLogEntriesTaskIdType ON agentLogEntries(taskId, type)`);
});
}
}
/**

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(39);
expect(db1.getSchemaVersion()).toBe(40);
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(39);
expect(db3.getSchemaVersion()).toBe(40);
// 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(39);
expect(db1.getSchemaVersion()).toBe(40);
db1.close();
const db2 = createDatabase(testDir);
expect(() => db2.init()).not.toThrow();
expect(db2.getSchemaVersion()).toBe(39);
expect(db2.getSchemaVersion()).toBe(40);
db2.close();
} finally {
rmSync(testDir, { recursive: true, force: true });

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

@@ -2627,8 +2627,8 @@ describe("MissionStore", () => {
// ── Loop State & Validator Run Schema Tests ───────────────────────────
describe("Loop State & Validator Run Schema (v31)", () => {
it("schema version is 39 after migration", () => {
expect(db.getSchemaVersion()).toBe(39);
it("schema version is 40 after migration", () => {
expect(db.getSchemaVersion()).toBe(40);
});
it("mission_features table has loop state columns", () => {

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

@@ -738,8 +738,8 @@ describe("RoadmapStore", () => {
});
describe("schema version", () => {
it("schema version is 39 after init", () => {
expect(db.getSchemaVersion()).toBe(39);
it("schema version is 40 after init", () => {
expect(db.getSchemaVersion()).toBe(40);
});
});

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

@@ -464,8 +464,8 @@ describe("Run Audit", () => {
expect(indexNames).toContain("idxRunAuditEventsTimestamp");
});
it("schema version is bumped to 39", () => {
expect(db.getSchemaVersion()).toBe(39);
it("schema version is bumped to 40", () => {
expect(db.getSchemaVersion()).toBe(40);
});
});
});

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

@@ -88,6 +88,21 @@ describe("TaskStore", () => {
return dir;
}
function insertLogEntryWithTimestamp(
targetStore: TaskStore,
taskId: string,
text: string,
type: string,
timestamp: string,
detail?: string,
agent?: string,
): void {
(targetStore as any).db.prepare(`
INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
VALUES (?, ?, ?, ?, ?, ?)
`).run(taskId, timestamp, text, type, detail ?? null, agent ?? null);
}
// ── Prompt generation (no duplicate description) ───────────────
describe("prompt generation", () => {
@@ -3031,20 +3046,6 @@ describe("TaskStore", () => {
expect(fetched.comments).toHaveLength(1);
});
it("appendAgentLog recreates missing task directory before writing agent.log", async () => {
const task = await createTestTask();
const dir = await deleteTaskDir(task.id);
await store.appendAgentLog(task.id, "Recovered log", "text");
expect(existsSync(dir)).toBe(true);
expect(existsSync(join(dir, "agent.log"))).toBe(true);
const logs = await store.getAgentLogs(task.id);
expect(logs).toHaveLength(1);
expect(logs[0].text).toBe("Recovered log");
});
it("addAttachment recreates missing task directory and attachment directory", async () => {
const task = await createTestTask();
const dir = await deleteTaskDir(task.id);
@@ -3569,12 +3570,22 @@ Task with acceptance criteria
});
describe("agent log persistence", () => {
it("appendAgentLog creates agent.log and getAgentLogs reads it back", async () => {
it("appendAgentLog inserts into agentLogEntries and getAgentLogs reads it back", async () => {
const task = await createTestTask();
await store.appendAgentLog(task.id, "Hello world", "text");
await store.appendAgentLog(task.id, "Read", "tool");
const rows = (store as any).db.prepare(`
SELECT taskId, text, type FROM agentLogEntries
WHERE taskId = ?
ORDER BY timestamp ASC
`).all(task.id) as Array<{ taskId: string; text: string; type: string }>;
expect(rows).toEqual([
{ taskId: task.id, text: "Hello world", type: "text" },
{ taskId: task.id, text: "Read", type: "tool" },
]);
const logs = await store.getAgentLogs(task.id);
expect(logs).toHaveLength(2);
expect(logs[0].text).toBe("Hello world");
@@ -3584,13 +3595,13 @@ Task with acceptance criteria
expect(logs[1].type).toBe("tool");
});
it("getAgentLogs returns empty array when no log file exists", async () => {
it("getAgentLogs returns empty array when no log entries exist", async () => {
const task = await createTestTask();
const logs = await store.getAgentLogs(task.id);
expect(logs).toEqual([]);
});
it("getAgentLogs returns empty array when the task directory is missing", async () => {
it("getAgentLogs returns empty array when task directory is missing", async () => {
const task = await createTestTask();
await deleteTaskDir(task.id);
@@ -3635,30 +3646,35 @@ Task with acceptance criteria
expect(logs[0]).not.toHaveProperty("detail");
});
it("handles multiple appends correctly (JSONL format)", async () => {
it("handles multiple appends correctly", async () => {
const task = await createTestTask();
for (let i = 0; i < 5; i++) {
await store.appendAgentLog(task.id, `chunk ${i}`, "text");
}
const logs = await store.getAgentLogs(task.id);
expect(logs).toHaveLength(5);
expect(logs[0].text).toBe("chunk 0");
expect(logs[4].text).toBe("chunk 4");
});
it("can return only the most recent agent log entries while skipping malformed lines", async () => {
it("getAgentLogCount returns the number of persisted log entries", async () => {
const task = await createTestTask();
expect(await store.getAgentLogCount(task.id)).toBe(0);
await store.appendAgentLog(task.id, "chunk 0", "text");
await store.appendAgentLog(task.id, "chunk 1", "tool");
expect(await store.getAgentLogCount(task.id)).toBe(2);
});
it("returns the most recent agent log entries from SQLite in chronological order", async () => {
const task = await createTestTask();
const dir = join(rootDir, ".fusion", "tasks", task.id);
const logPath = join(dir, "agent.log");
for (let i = 0; i < 5; i++) {
if (i === 2) {
await appendFile(logPath, "{not valid json}\n");
}
await store.appendAgentLog(task.id, `chunk ${i}`, "text");
}
const logs = await store.getAgentLogs(task.id, { limit: 2 });
expect(logs.map((entry) => entry.text)).toEqual(["chunk 3", "chunk 4"]);
});
@@ -3815,20 +3831,12 @@ Task with acceptance criteria
it("getAgentLogsByTimeRange filters entries by start and end timestamps (inclusive)", async () => {
const task = await createTestTask();
const dir = (store as any).taskDir(task.id);
const { mkdirSync, writeFileSync } = await import("node:fs");
const { join } = await import("node:path");
// Write entries at specific timestamps directly to the JSONL file
mkdirSync(dir, { recursive: true });
const entries = [
{ timestamp: "2024-01-01T00:00:00.000Z", taskId: task.id, text: "before start", type: "text" },
{ timestamp: "2024-01-01T01:00:00.000Z", taskId: task.id, text: "at start", type: "text" },
{ timestamp: "2024-01-01T02:00:00.000Z", taskId: task.id, text: "middle", type: "text" },
{ timestamp: "2024-01-01T03:00:00.000Z", taskId: task.id, text: "at end", type: "text" },
{ timestamp: "2024-01-01T04:00:00.000Z", taskId: task.id, text: "after end", type: "text" },
];
writeFileSync(join(dir, "agent.log"), entries.map((e) => JSON.stringify(e)).join("\n") + "\n");
insertLogEntryWithTimestamp(store, task.id, "before start", "text", "2024-01-01T00:00:00.000Z");
insertLogEntryWithTimestamp(store, task.id, "at start", "text", "2024-01-01T01:00:00.000Z");
insertLogEntryWithTimestamp(store, task.id, "middle", "text", "2024-01-01T02:00:00.000Z");
insertLogEntryWithTimestamp(store, task.id, "at end", "text", "2024-01-01T03:00:00.000Z");
insertLogEntryWithTimestamp(store, task.id, "after end", "text", "2024-01-01T04:00:00.000Z");
const logs = await store.getAgentLogsByTimeRange(
task.id,
@@ -3842,18 +3850,10 @@ Task with acceptance criteria
it("getAgentLogsByTimeRange uses current time when endIso is null", async () => {
const task = await createTestTask();
const dir = (store as any).taskDir(task.id);
const { mkdirSync, writeFileSync } = await import("node:fs");
const { join } = await import("node:path");
mkdirSync(dir, { recursive: true });
const entries = [
{ timestamp: "2024-01-01T00:00:00.000Z", taskId: task.id, text: "entry1", type: "text" },
{ timestamp: "2024-06-01T00:00:00.000Z", taskId: task.id, text: "entry2", type: "text" },
];
writeFileSync(join(dir, "agent.log"), entries.map((e) => JSON.stringify(e)).join("\n") + "\n");
insertLogEntryWithTimestamp(store, task.id, "entry1", "text", "2024-01-01T00:00:00.000Z");
insertLogEntryWithTimestamp(store, task.id, "entry2", "text", "2024-06-01T00:00:00.000Z");
// With null end, should include all entries after start
const logs = await store.getAgentLogsByTimeRange(
task.id,
"2024-01-01T00:00:00.000Z",
@@ -3865,15 +3865,7 @@ Task with acceptance criteria
it("getAgentLogsByTimeRange returns empty array when no entries match", async () => {
const task = await createTestTask();
const dir = (store as any).taskDir(task.id);
const { mkdirSync, writeFileSync } = await import("node:fs");
const { join } = await import("node:path");
mkdirSync(dir, { recursive: true });
const entries = [
{ timestamp: "2024-01-01T00:00:00.000Z", taskId: task.id, text: "entry1", type: "text" },
];
writeFileSync(join(dir, "agent.log"), entries.map((e) => JSON.stringify(e)).join("\n") + "\n");
insertLogEntryWithTimestamp(store, task.id, "entry1", "text", "2024-01-01T00:00:00.000Z");
const logs = await store.getAgentLogsByTimeRange(
task.id,
@@ -3884,7 +3876,7 @@ Task with acceptance criteria
expect(logs).toEqual([]);
});
it("getAgentLogsByTimeRange returns empty array when no log file exists", async () => {
it("getAgentLogsByTimeRange returns empty array when no entries exist", async () => {
const task = await createTestTask();
const logs = await store.getAgentLogsByTimeRange(
@@ -3895,6 +3887,87 @@ Task with acceptance criteria
expect(logs).toEqual([]);
});
it("deleting a task cascades agent log entry deletion", async () => {
const task = await createTestTask();
await store.appendAgentLog(task.id, "cascade me", "text");
const before = (store as any).db.prepare(
"SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
).get(task.id) as { count: number };
expect(before.count).toBe(1);
await store.deleteTask(task.id);
const after = (store as any).db.prepare(
"SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
).get(task.id) as { count: number };
expect(after.count).toBe(0);
});
it("importLegacyAgentLogs imports JSONL entries from existing agent.log files", async () => {
const task = await createTestTask();
const dir = join(rootDir, ".fusion", "tasks", task.id);
const legacyEntries = [
{
timestamp: "2024-01-01T00:00:00.000Z",
taskId: task.id,
text: "legacy line 1",
type: "text",
},
{
timestamp: "2024-01-01T01:00:00.000Z",
taskId: task.id,
text: "legacy line 2",
type: "tool",
detail: "legacy detail",
agent: "executor",
},
];
await writeFile(join(dir, "agent.log"), `${legacyEntries.map((entry) => JSON.stringify(entry)).join("\n")}\n`);
const imported = await store.importLegacyAgentLogs();
expect(imported).toBe(2);
const logs = await store.getAgentLogs(task.id);
expect(logs).toHaveLength(2);
expect(logs.map((log) => log.text)).toEqual(["legacy line 1", "legacy line 2"]);
expect(logs[1].detail).toBe("legacy detail");
expect(logs[1].agent).toBe("executor");
});
it("importLegacyAgentLogsOnce is idempotent via __meta guard", async () => {
const task = await createTestTask();
const dir = join(rootDir, ".fusion", "tasks", task.id);
const logPath = join(dir, "agent.log");
(store as any).db.prepare("DELETE FROM __meta WHERE key = ?").run("agentLogLegacyFileImportVersion");
await writeFile(logPath, `${JSON.stringify({
timestamp: "2024-01-01T00:00:00.000Z",
taskId: task.id,
text: "legacy line 1",
type: "text",
})}\n`);
await (store as any).importLegacyAgentLogsOnce();
expect(await store.getAgentLogCount(task.id)).toBe(1);
await appendFile(logPath, `${JSON.stringify({
timestamp: "2024-01-01T01:00:00.000Z",
taskId: task.id,
text: "legacy line 2",
type: "text",
})}\n`);
await (store as any).importLegacyAgentLogsOnce();
expect(await store.getAgentLogCount(task.id)).toBe(1);
const migrationRow = (store as any).db.prepare(
"SELECT value FROM __meta WHERE key = ?",
).get("agentLogLegacyFileImportVersion") as { value: string } | undefined;
expect(migrationRow?.value).toBe("1");
});
});
describe("task comments", () => {

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

@@ -1,6 +1,6 @@
import { EventEmitter } from "node:events";
import { randomUUID } from "node:crypto";
import { appendFile, mkdir, open, readFile, writeFile, rename, unlink } from "node:fs/promises";
import { mkdir, readdir, readFile, writeFile, rename, unlink } from "node:fs/promises";
import { join } from "node:path";
import { existsSync, watch, type FSWatcher } from "node:fs";
import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode } from "./types.js";
@@ -230,7 +230,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await migrateFromLegacy(this.kbDir, this._db);
}
await this.migrateActiveArchivedTasksToArchiveDb();
await this.importLegacyAgentLogsOnce();
// Write config.json for backward compatibility if it doesn't exist
if (!existsSync(this.configPath)) {
const config = await this.readConfig();
@@ -593,7 +594,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
*/
private upsertTask(task: Task): void {
this.db.prepare(`
INSERT OR REPLACE INTO tasks (
INSERT INTO tasks (
id, title, description, "column", status, size, reviewLevel, currentStep,
worktree, blockedBy, paused, baseBranch, branch, baseCommitSha, modelPresetId, modelProvider,
modelId, validatorModelProvider, validatorModelId, planningModelProvider, planningModelId, mergeRetries,
@@ -606,6 +607,58 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
)
ON CONFLICT(id) DO UPDATE SET
title = excluded.title,
description = excluded.description,
"column" = excluded."column",
status = excluded.status,
size = excluded.size,
reviewLevel = excluded.reviewLevel,
currentStep = excluded.currentStep,
worktree = excluded.worktree,
blockedBy = excluded.blockedBy,
paused = excluded.paused,
baseBranch = excluded.baseBranch,
branch = excluded.branch,
baseCommitSha = excluded.baseCommitSha,
modelPresetId = excluded.modelPresetId,
modelProvider = excluded.modelProvider,
modelId = excluded.modelId,
validatorModelProvider = excluded.validatorModelProvider,
validatorModelId = excluded.validatorModelId,
planningModelProvider = excluded.planningModelProvider,
planningModelId = excluded.planningModelId,
mergeRetries = excluded.mergeRetries,
workflowStepRetries = excluded.workflowStepRetries,
stuckKillCount = excluded.stuckKillCount,
postReviewFixCount = excluded.postReviewFixCount,
recoveryRetryCount = excluded.recoveryRetryCount,
nextRecoveryAt = excluded.nextRecoveryAt,
error = excluded.error,
summary = excluded.summary,
thinkingLevel = excluded.thinkingLevel,
createdAt = excluded.createdAt,
updatedAt = excluded.updatedAt,
columnMovedAt = excluded.columnMovedAt,
dependencies = excluded.dependencies,
steps = excluded.steps,
log = excluded.log,
attachments = excluded.attachments,
steeringComments = excluded.steeringComments,
comments = excluded.comments,
workflowStepResults = excluded.workflowStepResults,
prInfo = excluded.prInfo,
issueInfo = excluded.issueInfo,
mergeDetails = excluded.mergeDetails,
breakIntoSubtasks = excluded.breakIntoSubtasks,
enabledWorkflowSteps = excluded.enabledWorkflowSteps,
modifiedFiles = excluded.modifiedFiles,
missionId = excluded.missionId,
sliceId = excluded.sliceId,
assignedAgentId = excluded.assignedAgentId,
assigneeUserId = excluded.assigneeUserId,
checkedOutBy = excluded.checkedOutBy,
checkedOutAt = excluded.checkedOutAt
`).run(
task.id,
task.title ?? null,
@@ -3459,8 +3512,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
/**
* Append an agent log entry to the task's agent log file (JSONL format).
* Each entry is a single JSON line appended to `.fusion/tasks/{ID}/agent.log`.
* Insert an agent log entry into the agentLogEntries SQLite table.
* Also emits an `agent:log` event for live streaming.
*
* @param taskId - The task ID (e.g. "KB-001")
@@ -3476,88 +3528,34 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
detail?: string,
agent?: AgentLogEntry["agent"],
): Promise<void> {
const timestamp = new Date().toISOString();
const entry: AgentLogEntry = {
timestamp: new Date().toISOString(),
timestamp,
taskId,
text,
type,
...(detail !== undefined && { detail }),
...(agent !== undefined && { agent }),
};
const dir = this.taskDir(taskId);
const logPath = join(dir, "agent.log");
await mkdir(dir, { recursive: true });
await appendFile(logPath, JSON.stringify(entry) + "\n");
this.db.prepare(`
INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
VALUES (?, ?, ?, ?, ?, ?)
`).run(taskId, timestamp, text, type, detail ?? null, agent ?? null);
this.db.bumpLastModified();
this.emit("agent:log", entry);
}
private parseAgentLogLine(line: string): AgentLogEntry | null {
const trimmed = line.trim();
if (!trimmed) return null;
try {
return JSON.parse(trimmed) as AgentLogEntry;
} catch {
return null;
}
}
private parseAgentLogContent(content: string): AgentLogEntry[] {
const entries: AgentLogEntry[] = [];
for (const line of content.split("\n")) {
const entry = this.parseAgentLogLine(line);
if (entry) entries.push(entry);
}
return entries;
}
private async readAgentLogTail(logPath: string, limit: number): Promise<AgentLogEntry[]> {
const handle = await open(logPath, "r");
try {
const { size } = await handle.stat();
if (size === 0) return [];
const chunkSize = 64 * 1024;
let position = size;
let buffer = Buffer.alloc(0);
const entriesNewestFirst: AgentLogEntry[] = [];
while (position > 0 && entriesNewestFirst.length < limit) {
const readSize = Math.min(chunkSize, position);
position -= readSize;
const chunk = Buffer.allocUnsafe(readSize);
const { bytesRead } = await handle.read(chunk, 0, readSize, position);
if (bytesRead <= 0) break;
buffer = Buffer.concat([chunk.subarray(0, bytesRead), buffer]);
while (entriesNewestFirst.length < limit) {
const newlineIndex = buffer.lastIndexOf(10);
if (newlineIndex === -1) break;
const lineBuffer = buffer.subarray(newlineIndex + 1);
buffer = buffer.subarray(0, newlineIndex);
if (lineBuffer.length === 0) continue;
const entry = this.parseAgentLogLine(lineBuffer.toString("utf-8"));
if (entry) {
entriesNewestFirst.push(entry);
}
}
}
if (entriesNewestFirst.length < limit && buffer.length > 0) {
const entry = this.parseAgentLogLine(buffer.toString("utf-8"));
if (entry) {
entriesNewestFirst.push(entry);
}
}
return entriesNewestFirst.reverse();
} finally {
await handle.close();
}
private mapAgentLogRow(row: Record<string, unknown>): AgentLogEntry {
return {
timestamp: row.timestamp as string,
taskId: row.taskId as string,
text: row.text as string,
type: row.type as AgentLogEntry["type"],
...(row.detail != null && { detail: row.detail as string }),
...(row.agent != null && { agent: row.agent as AgentLogEntry["agent"] }),
};
}
async addTaskComment(id: string, text: string, author: string): Promise<Task> {
@@ -4108,27 +4106,23 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
/**
* Read all historical agent log entries for a task from its agent log file.
* Read historical agent log entries for a task from SQLite.
* Returns entries in chronological order (oldest first).
*
* Each entry's `text` and `detail` fields are returned in full — there is
* no per-entry truncation at the persistence layer. The 500-entry cap
* no per-entry truncation at the persistence layer. The 500-entry cap
* (`MAX_LOG_ENTRIES`) in the dashboard hooks is a whole-list limit only.
*
* @param taskId - The task ID (e.g. "KB-001")
* @param options - Optional pagination options
* @param options.limit - Maximum number of entries to return (most recent)
* @param options.offset - Number of most-recent entries to skip (for pagination)
* @returns Array of agent log entries, empty if no log file exists
* @returns Array of agent log entries
*/
async getAgentLogs(
taskId: string,
options?: { limit?: number; offset?: number },
): Promise<AgentLogEntry[]> {
const dir = this.taskDir(taskId);
const logPath = join(dir, "agent.log");
if (!existsSync(logPath)) return [];
const limit = options?.limit !== undefined
? (Number.isFinite(options.limit) ? Math.max(0, Math.floor(options.limit)) : 0)
: undefined;
@@ -4136,78 +4130,50 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
? (Number.isFinite(options.offset) ? Math.max(0, Math.floor(options.offset)) : 0)
: 0;
// If limit is specified, use readAgentLogTail for efficiency.
if (limit === 0) return [];
if (limit !== undefined) {
if (limit === 0) return [];
// Offset means "skip this many most-recent entries", so read enough
// tail entries to include the requested older page.
const readCount = offset > 0 ? limit + offset : limit;
const entries = await this.readAgentLogTail(logPath, readCount);
const rows = this.db.prepare(`
SELECT * FROM agentLogEntries
WHERE taskId = ?
ORDER BY timestamp DESC
LIMIT ?
`).all(taskId, readCount) as Array<Record<string, unknown>>;
const entries = rows.map((row) => this.mapAgentLogRow(row)).reverse();
if (offset > 0) {
return entries.slice(0, Math.max(0, entries.length - offset));
}
return entries;
}
// No limit specified - read entire file
const content = await readFile(logPath, "utf-8");
const entries = this.parseAgentLogContent(content);
const rows = this.db.prepare(`
SELECT * FROM agentLogEntries
WHERE taskId = ?
ORDER BY timestamp ASC
`).all(taskId) as Array<Record<string, unknown>>;
const entries = rows.map((row) => this.mapAgentLogRow(row));
if (offset > 0) {
return entries.slice(0, -offset);
return entries.slice(0, Math.max(0, entries.length - offset));
}
return entries;
}
/**
* Count total number of log entries in the agent log file.
* Uses efficient newline counting to avoid parsing entire file.
* Count total number of persisted agent log entries for a task in SQLite.
*
* @param taskId - The task ID (e.g. "KB-001")
* @returns Total number of log entries, or 0 if no log file exists
* @returns Total number of log entries
*/
async getAgentLogCount(taskId: string): Promise<number> {
const dir = this.taskDir(taskId);
const logPath = join(dir, "agent.log");
if (!existsSync(logPath)) return 0;
const handle = await open(logPath, "r");
try {
const { size } = await handle.stat();
if (size === 0) return 0;
const chunkSize = 64 * 1024;
const buffer = Buffer.allocUnsafe(chunkSize);
let position = 0;
let count = 0;
let hasNonWhitespace = false;
let lastByte = 0;
while (position < size) {
const readSize = Math.min(chunkSize, size - position);
const { bytesRead } = await handle.read(buffer, 0, readSize, position);
if (bytesRead <= 0) break;
for (let i = 0; i < bytesRead; i++) {
const byte = buffer[i];
if (byte === 10) count++;
if (byte > 32) hasNonWhitespace = true;
lastByte = byte;
}
position += bytesRead;
}
if (!hasNonWhitespace) return 0;
return lastByte === 10 ? count : count + 1;
} finally {
await handle.close();
}
const row = this.db.prepare(
"SELECT COUNT(*) as count FROM agentLogEntries WHERE taskId = ?",
).get(taskId) as { count: number } | undefined;
return row?.count ?? 0;
}
/**
* Get agent log entries for a task filtered by a time range.
*
* Returns all log entries whose `timestamp` falls within [startIso, endIso]
* (inclusive on both ends). If endIso is null (active run), the current
* time is used as the upper bound.
* Get persisted agent log entries for a task filtered by an inclusive time range.
*
* @param taskId - The task ID (e.g. "KB-001")
* @param startIso - ISO-8601 start timestamp (inclusive)
@@ -4219,11 +4185,82 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
startIso: string,
endIso: string | null,
): Promise<AgentLogEntry[]> {
const allEntries = await this.getAgentLogs(taskId);
const end = endIso ?? new Date().toISOString();
return allEntries.filter((entry) => {
return entry.timestamp >= startIso && entry.timestamp <= end;
});
const rows = this.db.prepare(`
SELECT * FROM agentLogEntries
WHERE taskId = ? AND timestamp >= ? AND timestamp <= ?
ORDER BY timestamp ASC
`).all(taskId, startIso, end) as Array<Record<string, unknown>>;
return rows.map((row) => this.mapAgentLogRow(row));
}
async importLegacyAgentLogs(): Promise<number> {
if (!existsSync(this.tasksDir)) return 0;
const entries = await readdir(this.tasksDir, { withFileTypes: true });
let imported = 0;
const insertStmt = this.db.prepare(`
INSERT INTO agentLogEntries (taskId, timestamp, text, type, detail, agent)
VALUES (?, ?, ?, ?, ?, ?)
`);
for (const entry of entries) {
if (!entry.isDirectory()) continue;
const logPath = join(this.tasksDir, entry.name, "agent.log");
if (!existsSync(logPath)) continue;
try {
const content = await readFile(logPath, "utf-8");
for (const line of content.split("\n")) {
const trimmed = line.trim();
if (!trimmed) continue;
try {
const parsed = JSON.parse(trimmed) as Record<string, unknown>;
const timestamp = typeof parsed.timestamp === "string" ? parsed.timestamp : null;
const parsedTaskId = typeof parsed.taskId === "string" ? parsed.taskId : null;
const type = typeof parsed.type === "string" ? parsed.type : null;
if (!timestamp || !parsedTaskId || !type) continue;
const text = typeof parsed.text === "string" ? parsed.text : "";
const detail = typeof parsed.detail === "string" ? parsed.detail : null;
const agent = typeof parsed.agent === "string" ? parsed.agent : null;
insertStmt.run(parsedTaskId, timestamp, text, type, detail, agent);
imported += 1;
} catch {
// Skip malformed JSONL lines.
}
}
} catch {
// Skip unreadable files.
}
}
if (imported > 0) {
this.db.bumpLastModified();
}
return imported;
}
private async importLegacyAgentLogsOnce(): Promise<void> {
const migrationKey = "agentLogLegacyFileImportVersion";
const migrationVersion = "1";
const row = this.db.prepare("SELECT value FROM __meta WHERE key = ?").get(migrationKey) as
| { value: string }
| undefined;
if (row?.value === migrationVersion) {
return;
}
await this.importLegacyAgentLogs();
this.db.prepare(`
INSERT INTO __meta (key, value) VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
`).run(migrationKey, migrationVersion);
this.db.bumpLastModified();
}
// ── Archive Cleanup Methods ─────────────────────────────────────────
@@ -4325,7 +4362,9 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* Restore a task from an archive entry.
* Recreates task directory with task.json and PROMPT.md.
* Clears transient execution state (worktree, status, blockedBy, etc.).
* Does NOT recreate agent.log (intentionally lost during archive).
* Agent log entries are stored in SQLite and are deleted by FK cascade when
* the task row is removed; archive snapshots (`agentLogFull`/`agentLogSnapshot`)
* preserve point-in-time log data inside the archived task record.
*/
private async restoreFromArchive(entry: import("./types.js").ArchivedTaskEntry): Promise<Task> {
const dir = this.taskDir(entry.id);