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

fix(FN-2326): harden TaskStore observability logging paths

Browse Source 
- Add a task-store logger and funnel activity listeners through a shared helper that logs source-event failures
- Replace silent catches with structured warn/error logs for workflow default resolution, title summarization, fs.watch, and polling paths
- Preserve best-effort behavior for activity recording and async summarization while attaching actionable error context
- Expand TaskStore tests to validate logging behavior for activity insert failures, listener rejections, workflow fallback, and watch/poll error handling
This commit is contained in:
Fusion
2026-04-23 12:36:52 -07:00
committed by gsxdsm
parent 445bdb2eea
commit c6842c9bab
2 changed files with 299 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6769,6 +6769,86 @@ Task with acceptance criteria
expect(logs[0].timestamp).toBeDefined();
});
it("recordActivity logs failures and stays best-effort", async () => {
const storeAny = store as any;
const originalPrepare = storeAny.db.prepare.bind(storeAny.db);
const prepareSpy = vi.spyOn(storeAny.db, "prepare").mockImplementation((sql: string) => {
if (sql.includes("INSERT INTO activityLog")) {
return {
run: () => {
throw new Error("activity insert failed");
},
};
}
return originalPrepare(sql);
});
const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
try {
await expect(
store.recordActivity({
type: "task:created",
taskId: "FN-404",
taskTitle: "Resilient record",
details: "Create event",
metadata: { source: "test" },
}),
).resolves.toMatchObject({
type: "task:created",
taskId: "FN-404",
taskTitle: "Resilient record",
details: "Create event",
});
const failureCall = errorSpy.mock.calls.find(
(call) => typeof call[0] === "string" && call[0].includes("[task-store] Failed to record activity"),
);
expect(failureCall).toBeDefined();
const [, context] = failureCall as [string, Record<string, unknown>];
expect(context).toMatchObject({
type: "task:created",
taskId: "FN-404",
taskTitle: "Resilient record",
detailsLength: "Create event".length,
hasMetadata: true,
error: "activity insert failed",
});
} finally {
prepareSpy.mockRestore();
errorSpy.mockRestore();
}
});
it("logs listener-level activity recording failures without throwing", async () => {
const task = await store.createTask({ description: "Listener test" });
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
const recordSpy = vi.spyOn(store, "recordActivity").mockRejectedValue(new Error("listener rejected"));
try {
expect(() => {
store.emit("task:created", task);
}).not.toThrow();
await Promise.resolve();
const warningCall = warnSpy.mock.calls.find(
(call) => typeof call[0] === "string" && call[0].includes("[task-store] Activity logging listener failed"),
);
expect(warningCall).toBeDefined();
const [, context] = warningCall as [string, Record<string, unknown>];
expect(context).toMatchObject({
sourceEvent: "task:created",
type: "task:created",
taskId: task.id,
error: "listener rejected",
});
} finally {
recordSpy.mockRestore();
warnSpy.mockRestore();
}
});
it("getActivityLog returns entries newest first", async () => {
await store.recordActivity({ type: "task:created", taskId: "FN-001", details: "First" });
await new Promise((r) => setTimeout(r, 10));
@@ -7543,6 +7623,31 @@ Task with acceptance criteria
expect(task.enabledWorkflowSteps).toEqual(["WS-001", "WS-002"]);
});
it("logs default-on resolution failures and still creates the task", async () => {
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
const listStepsSpy = vi.spyOn(store, "listWorkflowSteps").mockRejectedValue(new Error("workflow catalog unavailable"));
try {
const task = await store.createTask({ description: "Best effort defaults" });
expect(task.id).toMatch(/^FN-\d+$/);
expect(task.enabledWorkflowSteps).toBeUndefined();
const warningCall = warnSpy.mock.calls.find(
(call) => typeof call[0] === "string" && call[0].includes("[task-store] Failed to auto-apply default workflow steps during task creation"),
);
expect(warningCall).toBeDefined();
const [, context] = warningCall as [string, Record<string, unknown>];
expect(context).toMatchObject({
descriptionLength: "Best effort defaults".length,
error: "workflow catalog unavailable",
});
} finally {
listStepsSpy.mockRestore();
warnSpy.mockRestore();
}
});
it("should update task workflow steps and materialize built-in templates", async () => {
const task = await store.createTask({ description: "Editable task" });
@@ -7774,23 +7879,30 @@ Task with acceptance criteria
const consoleSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
const mockOnSummarize = vi.fn().mockRejectedValue(new Error("AI service failed"));
const task = await store.createTask(
{ description: "a".repeat(201) },
{ onSummarize: mockOnSummarize, settings: { autoSummarizeTitles: true } }
);
try {
const task = await store.createTask(
{ description: "a".repeat(201) },
{ onSummarize: mockOnSummarize, settings: { autoSummarizeTitles: true } }
);
expect(task.title).toBeUndefined();
expect(task.id).toMatch(/^FN-\d+$/); // Task still created
expect(task.title).toBeUndefined();
expect(task.id).toMatch(/^FN-\d+$/); // Task still created
// Wait for async error to be logged
await new Promise((resolve) => setTimeout(resolve, 10));
// Wait for async error to be logged
await new Promise((resolve) => setTimeout(resolve, 10));
expect(consoleSpy.mock.calls[0][0]).toMatch(/Title summarization failed for task/);
expect(consoleSpy.mock.calls[0][0]).toMatch(/AI service failed/);
expect(consoleSpy.mock.calls[0][0]).toMatch(/desc length: 201/);
expect(consoleSpy.mock.calls[0][0]).toMatch(/auto-summarize: true/);
consoleSpy.mockRestore();
expect(consoleSpy).toHaveBeenCalled();
const [message, context] = consoleSpy.mock.calls[0] as [string, Record<string, unknown>];
expect(message).toContain("[task-store] Title summarization failed for task");
expect(context).toMatchObject({
taskId: task.id,
descriptionLength: 201,
autoSummarizeEnabled: true,
error: "AI service failed",
});
} finally {
consoleSpy.mockRestore();
}
});
it("should trigger summarization at exactly 201 characters", async () => {
@@ -8096,35 +8208,65 @@ Task with acceptance criteria
expect(storeAny.pollingInProgress).toBe(false);
});
it("emits timing warning when polling is slow (>100ms)", async () => {
it("logs poll failures with context and keeps checkForChanges non-fatal", async () => {
// Start watching to enable polling
await store.watch();
const storeAny = store as any;
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
const originalGetLastModified = storeAny.db.getLastModified.bind(storeAny.db);
storeAny.db.getLastModified = vi.fn(() => {
throw new Error("poll db unavailable");
});
// Create a task to ensure there's something to poll
await store.createTask({ description: "slow poll test" });
try {
await expect(storeAny.checkForChanges()).resolves.toBeUndefined();
expect(storeAny.pollingInProgress).toBe(false);
// Wait for poll interval to trigger naturally
await new Promise((resolve) => setTimeout(resolve, 1100));
const pollFailureCall = warnSpy.mock.calls.find(
(call) =>
typeof call[0] === "string"
&& call[0].includes("[task-store] checkForChanges poll cycle failed"),
);
expect(pollFailureCall).toBeDefined();
const [, context] = pollFailureCall as [string, Record<string, unknown>];
expect(context).toMatchObject({
lastPollTime: storeAny.lastPollTime,
error: "poll db unavailable",
});
} finally {
storeAny.db.getLastModified = originalGetLastModified;
warnSpy.mockRestore();
}
});
// Reset the guard so we can call checkForChanges directly
storeAny.pollingInProgress = false;
it("logs watcher failures and keeps polling operational", async () => {
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
// Test the timing warning logic by directly manipulating the condition
// We verify the timing warning code path exists by checking the source
const storeSource = storeAny.checkForChanges.toString();
expect(storeSource).toContain("Date.now()");
expect(storeSource).toContain("elapsed > 100");
expect(storeSource).toContain("console.warn");
try {
await store.watch();
const storeAny = store as any;
// Verify the guard prevents overlapping calls
const firstCall = storeAny.checkForChanges();
const secondCall = storeAny.checkForChanges();
expect(firstCall).toBeInstanceOf(Promise);
expect(secondCall).toBeInstanceOf(Promise);
await Promise.all([firstCall, secondCall]);
expect(storeAny.pollingInProgress).toBe(false);
if (storeAny.watcher) {
storeAny.watcher.emit("error", new Error("watcher degraded"));
const watcherErrorCall = warnSpy.mock.calls.find(
(call) => typeof call[0] === "string" && call[0].includes("[task-store] fs.watch emitted an error; polling will continue"),
);
expect(watcherErrorCall).toBeDefined();
const [, context] = watcherErrorCall as [string, Record<string, unknown>];
expect(context).toMatchObject({ error: "watcher degraded" });
} else {
const fallbackCall = warnSpy.mock.calls.find(
(call) => typeof call[0] === "string" && call[0].includes("[task-store] fs.watch unavailable; falling back to polling-only updates"),
);
expect(fallbackCall).toBeDefined();
}
await expect(storeAny.checkForChanges()).resolves.toBeUndefined();
} finally {
warnSpy.mockRestore();
}
});
it("does not emit timing warning when polling is fast (<100ms)", async () => {

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

@@ -18,6 +18,7 @@ import { CentralCore } from "./central-core.js";
import { getTaskMergeBlocker } from "./task-merge.js";
import { ensureMemoryFileWithBackend } from "./project-memory.js";
import { runCommandAsync } from "./run-command.js";
import { createLogger } from "./logger.js";
/**
* Legacy backup directory default value from the old .kb storage structure.
@@ -30,6 +31,7 @@ const TASK_ACTIVITY_LOG_ENTRY_LIMIT = 1_000;
const TASK_ACTIVITY_LOG_OUTCOME_LIMIT = 4_000;
const ARCHIVE_AGENT_LOG_SNAPSHOT_LIMIT = 25;
const ARCHIVE_AGENT_LOG_SNIPPET_LIMIT = 160;
const storeLog = createLogger("task-store");
/**
* Reject branch names that would be unsafe to interpolate into a shell command.
@@ -839,55 +841,59 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
private setupActivityLogListeners(): void {
// Task created
this.on("task:created", (task) => {
this.recordActivity({
type: "task:created",
taskId: task.id,
taskTitle: task.title,
details: `Task ${task.id} created${task.title ? `: ${task.title}` : ""}`,
}).catch(() => {
// Best-effort: ignore recording errors
});
this.recordActivityFromListener(
{
type: "task:created",
taskId: task.id,
taskTitle: task.title,
details: `Task ${task.id} created${task.title ? `: ${task.title}` : ""}`,
},
"task:created",
);
});
// Task moved
this.on("task:moved", (data) => {
this.recordActivity({
type: "task:moved",
taskId: data.task.id,
taskTitle: data.task.title,
details: `Task ${data.task.id} moved: ${data.from} → ${data.to}`,
metadata: { from: data.from, to: data.to },
}).catch(() => {
// Best-effort: ignore recording errors
});
this.recordActivityFromListener(
{
type: "task:moved",
taskId: data.task.id,
taskTitle: data.task.title,
details: `Task ${data.task.id} moved: ${data.from} → ${data.to}`,
metadata: { from: data.from, to: data.to },
},
"task:moved",
);
});
// Task merged
this.on("task:merged", (result) => {
const status = result.merged ? "successfully merged" : "merge attempted";
this.recordActivity({
type: "task:merged",
taskId: result.task.id,
taskTitle: result.task.title,
details: `Task ${result.task.id} ${status} to main`,
metadata: { merged: result.merged, branch: result.branch },
}).catch(() => {
// Best-effort: ignore recording errors
});
this.recordActivityFromListener(
{
type: "task:merged",
taskId: result.task.id,
taskTitle: result.task.title,
details: `Task ${result.task.id} ${status} to main`,
metadata: { merged: result.merged, branch: result.branch },
},
"task:merged",
);
});
// Task updated (check for failures)
this.on("task:updated", (task) => {
if (task.status === "failed") {
this.recordActivity({
type: "task:failed",
taskId: task.id,
taskTitle: task.title,
details: `Task ${task.id} failed${task.error ? `: ${task.error}` : ""}`,
metadata: task.error ? { error: task.error } : undefined,
}).catch(() => {
// Best-effort: ignore recording errors
});
this.recordActivityFromListener(
{
type: "task:failed",
taskId: task.id,
taskTitle: task.title,
details: `Task ${task.id} failed${task.error ? `: ${task.error}` : ""}`,
metadata: task.error ? { error: task.error } : undefined,
},
"task:updated",
);
}
});
@@ -908,25 +914,41 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
if (importantChanges.length > 0) {
this.recordActivity({
type: "settings:updated",
details: `Settings updated: ${importantChanges.join(", ")}`,
metadata: { changes: importantChanges },
}).catch(() => {
// Best-effort: ignore recording errors
});
this.recordActivityFromListener(
{
type: "settings:updated",
details: `Settings updated: ${importantChanges.join(", ")}`,
metadata: { changes: importantChanges },
},
"settings:updated",
);
}
});
// Task deleted
this.on("task:deleted", (task) => {
this.recordActivity({
type: "task:deleted",
taskId: task.id,
taskTitle: task.title,
details: `Task ${task.id} deleted${task.title ? `: ${task.title}` : ""}`,
}).catch(() => {
// Best-effort: ignore recording errors
this.recordActivityFromListener(
{
type: "task:deleted",
taskId: task.id,
taskTitle: task.title,
details: `Task ${task.id} deleted${task.title ? `: ${task.title}` : ""}`,
},
"task:deleted",
);
});
}
private recordActivityFromListener(
entry: Omit<ActivityLogEntry, "id" | "timestamp">,
sourceEvent: string,
): void {
this.recordActivity(entry).catch((err) => {
storeLog.warn("Activity logging listener failed", {
sourceEvent,
type: entry.type,
taskId: entry.taskId,
error: err instanceof Error ? err.message : String(err),
});
});
}
@@ -1616,8 +1638,12 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
if (defaultOnSteps.length > 0) {
resolvedWorkflowSteps = defaultOnSteps;
}
} catch {
} catch (err) {
// Non-fatal: default-on resolution is best-effort
storeLog.warn("Failed to auto-apply default workflow steps during task creation", {
error: err instanceof Error ? err.message : String(err),
descriptionLength: input.description.length,
});
}
} else if (input.enabledWorkflowSteps.length === 0) {
// Explicitly empty array — user intentionally selected no steps
@@ -1642,15 +1668,27 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
}
} catch (err) {
// Log warning but don't crash
const errorMsg = err instanceof Error ? err.message : String(err);
const autoEnabled = options?.settings?.autoSummarizeTitles === true;
console.warn(
`[TaskStore] Title summarization failed for task ${id}: ${errorMsg}` +
` (desc length: ${input.description.length}, auto-summarize: ${autoEnabled})`
const errorMessage = err instanceof Error ? err.message : String(err);
storeLog.warn(
`Title summarization failed for task ${id}: ${errorMessage} (desc length: ${input.description.length}, auto-summarize: ${autoEnabled})`,
{
taskId: id,
descriptionLength: input.description.length,
autoSummarizeEnabled: autoEnabled,
error: errorMessage,
},
);
}
}).catch(() => {}); // Prevent unhandled rejection
}).catch((err) => {
const autoEnabled = options?.settings?.autoSummarizeTitles === true;
storeLog.error("Unexpected title summarization promise-chain failure", {
taskId: id,
descriptionLength: input.description.length,
autoSummarizeEnabled: autoEnabled,
error: err instanceof Error ? err.message : String(err),
});
});
}
return task;
@@ -3485,11 +3523,18 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
this.watcher = watch(this.tasksDir, { recursive: true }, (_event, _filename) => {
// No-op - we use polling now, but keep watcher for API compat
});
this.watcher.on("error", () => {
// Ignore errors
this.watcher.on("error", (err) => {
storeLog.warn("fs.watch emitted an error; polling will continue", {
error: err instanceof Error ? err.message : String(err),
tasksDir: this.tasksDir,
});
});
} catch {
} catch (err) {
// fs.watch may not be available - that's fine
storeLog.warn("fs.watch unavailable; falling back to polling-only updates", {
error: err instanceof Error ? err.message : String(err),
tasksDir: this.tasksDir,
});
}
// Poll for changes every second
@@ -3564,10 +3609,17 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
const elapsed = Date.now() - startTime;
if (elapsed > 100) {
console.warn(`[TaskStore] checkForChanges took ${elapsed}ms — event loop may have been blocked`);
storeLog.warn("checkForChanges took longer than expected", {
elapsedMs: elapsed,
thresholdMs: 100,
});
}
} catch {
// Ignore polling errors
} catch (err) {
storeLog.warn("checkForChanges poll cycle failed", {
lastKnownModified: this.lastKnownModified,
lastPollTime: this.lastPollTime,
error: err instanceof Error ? err.message : String(err),
});
} finally {
this.pollingInProgress = false;
}
@@ -5191,7 +5243,15 @@ ${notificationsSection}`;
this.db.bumpLastModified();
} catch (err) {
// Best-effort: log errors but don't break operations
console.error("Failed to record activity:", err);
storeLog.error("Failed to record activity", {
id: fullEntry.id,
type: fullEntry.type,
taskId: fullEntry.taskId,
taskTitle: fullEntry.taskTitle,
detailsLength: fullEntry.details.length,
hasMetadata: fullEntry.metadata !== undefined,
error: err instanceof Error ? err.message : String(err),
});
}
return fullEntry;