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

feat(FN-1257): add runContext audit trail for task mutations

Browse Source 
- Add RunMutationContext type to track which agent run caused a mutation
- Thread runContext through TaskStore.logEntry, addComment, addSteeringComment, and pauseTask
- Propagate runContext from HeartbeatMonitor.executeHeartbeat to task store operations
- Propagate runContext from TaskExecutor.execute to task store operations
- Add GET /api/agents/:id/runs/:runId/mutations endpoint to query mutations by runId
- Add createTaskLogToolWithContext for heartbeat tools with run context support
- Add comprehensive tests for RunMutationContext across store and heartbeat modules
- Update memory.md with RunMutationContext usage convention
This commit is contained in:
gsxdsm
2026-04-09 13:59:43 -07:00
parent e26321790c
commit 7d484dd972
12 changed files with 674 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
packages/core/src/agent-store.ts
View File

@@ -44,6 +44,7 @@ import type {
Task,
} from "./types.js";
import { AGENT_VALID_TRANSITIONS, agentToConfigSnapshot, diffConfigSnapshots, CheckoutConflictError } from "./types.js";
import type { RunMutationContext } from "./types.js";
import type { TaskStore } from "./store.js";
import { computeAccessState } from "./agent-permissions.js";
import { Database } from "./db.js";
@@ -795,7 +796,7 @@ export class AgentStore extends EventEmitter {
* @param taskId - The task ID to assign, or undefined to unassign
* @returns The updated agent
*/
async assignTask(agentId: string, taskId: string | undefined): Promise<Agent> {
async assignTask(agentId: string, taskId: string | undefined, runContext?: RunMutationContext): Promise<Agent> {
return this.withLock(agentId, async () => {
const agent = await this.getAgent(agentId);
if (!agent) {
@@ -816,6 +817,11 @@ export class AgentStore extends EventEmitter {
this.emit("agent:assigned", updated, taskId);
}
// Log the assignment to the task when a non-empty taskId is provided
if (taskId && this.taskStore) {
await this.taskStore.logEntry(taskId, `Task assigned to agent ${agentId}`, undefined, runContext);
}
return updated;
});
}
@@ -824,7 +830,7 @@ export class AgentStore extends EventEmitter {
* Acquire a checkout lease for a task.
* Throws CheckoutConflictError when another agent already holds the lease.
*/
async checkoutTask(agentId: string, taskId: string): Promise<Task> {
async checkoutTask(agentId: string, taskId: string, runContext?: RunMutationContext): Promise<Task> {
if (!this.taskStore) {
throw new Error("TaskStore not configured for checkout operations");
}
@@ -848,14 +854,14 @@ export class AgentStore extends EventEmitter {
}
const updated = await this.taskStore.updateTask(taskId, { checkedOutBy: agentId });
await this.taskStore.logEntry(taskId, `Checked out by agent ${agentId}`);
await this.taskStore.logEntry(taskId, `Checked out by agent ${agentId}`, undefined, runContext);
return updated;
}
/**
* Release a checkout lease for a task.
*/
async releaseTask(agentId: string, taskId: string): Promise<Task> {
async releaseTask(agentId: string, taskId: string, runContext?: RunMutationContext): Promise<Task> {
if (!this.taskStore) {
throw new Error("TaskStore not configured for checkout operations");
}
@@ -874,20 +880,20 @@ export class AgentStore extends EventEmitter {
}
const updated = await this.taskStore.updateTask(taskId, { checkedOutBy: null });
await this.taskStore.logEntry(taskId, `Released by agent ${agentId}`);
await this.taskStore.logEntry(taskId, `Released by agent ${agentId}`, undefined, runContext);
return updated;
}
/**
* Force release a task checkout lease regardless of holder.
*/
async forceReleaseTask(taskId: string): Promise<Task> {
async forceReleaseTask(taskId: string, runContext?: RunMutationContext): Promise<Task> {
if (!this.taskStore) {
throw new Error("TaskStore not configured for checkout operations");
}
const updated = await this.taskStore.updateTask(taskId, { checkedOutBy: null });
await this.taskStore.logEntry(taskId, "Checkout force-released");
await this.taskStore.logEntry(taskId, "Checkout force-released", undefined, runContext);
return updated;
}

2
packages/core/src/index.ts
View File

@@ -1,5 +1,5 @@
export { COLUMNS, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, agentToConfigSnapshot, diffConfigSnapshots, CheckoutConflictError } from "./types.js";
export type { Column, IssueInfo, IssueState, PrInfo, PrStatus, Task, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskCreateInput, TaskDetail, InboxTask, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, Settings, GlobalSettings, ProjectSettings, SettingsScope, TaskStep, StepStatus, TaskLogEntry, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, Mailbox, CheckoutLease } from "./types.js";
export type { Column, IssueInfo, IssueState, PrInfo, PrStatus, Task, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskCreateInput, TaskDetail, InboxTask, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, Settings, GlobalSettings, ProjectSettings, SettingsScope, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, Mailbox, CheckoutLease } from "./types.js";
export { AGENT_VALID_TRANSITIONS } from "./types.js";
export {
BUILTIN_AGENT_PROMPTS,

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

@@ -6118,142 +6118,322 @@ Task with acceptance criteria
});
});
describe("searchTasks", () => {
it("searches tasks by ID", async () => {
const task1 = await store.createTask({ description: "First task" });
const task2 = await store.createTask({ description: "Second task" });
const results = await store.searchTasks("FN-001");
expect(results).toHaveLength(1);
expect(results[0].id).toBe("FN-001");
expect(results.some((t) => t.id === "FN-002")).toBe(false);
});
describe("searchTasks", () => {
it("searches tasks by ID", async () => {
const task1 = await store.createTask({ description: "First task" });
const task2 = await store.createTask({ description: "Second task" });
it("searches tasks by title", async () => {
await store.createTask({ title: "Fix login bug", description: "Login issue" });
await store.createTask({ title: "Add dashboard feature", description: "New UI" });
const results = await store.searchTasks("FN-001");
const results = await store.searchTasks("dashboard");
expect(results).toHaveLength(1);
expect(results[0].id).toBe("FN-001");
expect(results.some((t) => t.id === "FN-002")).toBe(false);
});
expect(results).toHaveLength(1);
expect(results[0].title).toBe("Add dashboard feature");
});
it("searches tasks by title", async () => {
await store.createTask({ title: "Fix login bug", description: "Login issue" });
await store.createTask({ title: "Add dashboard feature", description: "New UI" });
it("searches tasks by description", async () => {
await store.createTask({ description: "Fix the login button on the homepage" });
await store.createTask({ description: "Update the settings page layout" });
const results = await store.searchTasks("dashboard");
const results = await store.searchTasks("homepage");
expect(results).toHaveLength(1);
expect(results[0].title).toBe("Add dashboard feature");
});
expect(results).toHaveLength(1);
expect(results[0].description).toContain("homepage");
});
it("searches tasks by description", async () => {
await store.createTask({ description: "Fix the login button on the homepage" });
await store.createTask({ description: "Update the settings page layout" });
it("searches tasks by comment text", async () => {
const task = await store.createTask({ description: "A task" });
// Add a comment containing a unique word
await store.addComment(task.id, "Need to prioritize the xylophone implementation", "tester");
const results = await store.searchTasks("homepage");
const results = await store.searchTasks("xylophone");
expect(results).toHaveLength(1);
expect(results[0].description).toContain("homepage");
});
expect(results).toHaveLength(1);
expect(results[0].id).toBe(task.id);
});
it("searches tasks by comment text", async () => {
const task = await store.createTask({ description: "A task" });
// Add a comment containing a unique word
await store.addComment(task.id, "Need to prioritize the xylophone implementation", "tester");
it("is case insensitive", async () => {
await store.createTask({ title: "UPPERCASE SEARCH TEST", description: "Testing case insensitivity" });
const results = await store.searchTasks("xylophone");
const results = await store.searchTasks("uppercase");
expect(results).toHaveLength(1);
expect(results[0].id).toBe(task.id);
});
expect(results).toHaveLength(1);
expect(results[0].title).toBe("UPPERCASE SEARCH TEST");
});
it("is case insensitive", async () => {
await store.createTask({ title: "UPPERCASE SEARCH TEST", description: "Testing case insensitivity" });
it("falls back to listTasks for empty query", async () => {
await store.createTask({ description: "Task 1" });
await store.createTask({ description: "Task 2" });
const results = await store.searchTasks("uppercase");
const results = await store.searchTasks("");
const allTasks = await store.listTasks();
expect(results).toHaveLength(1);
expect(results[0].title).toBe("UPPERCASE SEARCH TEST");
});
expect(results).toHaveLength(allTasks.length);
});
it("falls back to listTasks for empty query", async () => {
await store.createTask({ description: "Task 1" });
await store.createTask({ description: "Task 2" });
it("falls back to listTasks for whitespace-only query", async () => {
await store.createTask({ description: "Task 1" });
const results = await store.searchTasks("");
const allTasks = await store.listTasks();
const results = await store.searchTasks(" ");
expect(results).toHaveLength(allTasks.length);
});
expect(results).toHaveLength(1);
});
it("falls back to listTasks for whitespace-only query", async () => {
await store.createTask({ description: "Task 1" });
it("uses OR semantics for multi-word queries", async () => {
await store.createTask({ title: "Fix login", description: "Button issues" });
await store.createTask({ title: "Add dashboard", description: "New features" });
const results = await store.searchTasks(" ");
const results = await store.searchTasks("login dashboard");
expect(results).toHaveLength(1);
});
expect(results).toHaveLength(2);
});
it("uses OR semantics for multi-word queries", async () => {
await store.createTask({ title: "Fix login", description: "Button issues" });
await store.createTask({ title: "Add dashboard", description: "New features" });
it("returns empty array for non-existent query", async () => {
await store.createTask({ description: "Regular task description" });
const results = await store.searchTasks("login dashboard");
const results = await store.searchTasks("xyznonexistent12345");
expect(results).toHaveLength(2);
});
expect(results).toHaveLength(0);
});
it("returns empty array for non-existent query", async () => {
await store.createTask({ description: "Regular task description" });
it("respects limit option", async () => {
await store.createTask({ description: "Task 1" });
await store.createTask({ description: "Task 2" });
await store.createTask({ description: "Task 3" });
await store.createTask({ description: "Task 4" });
await store.createTask({ description: "Task 5" });
const results = await store.searchTasks("xyznonexistent12345");
const results = await store.searchTasks("", { limit: 2 });
expect(results).toHaveLength(0);
});
expect(results).toHaveLength(2);
});
it("respects limit option", async () => {
await store.createTask({ description: "Task 1" });
await store.createTask({ description: "Task 2" });
await store.createTask({ description: "Task 3" });
await store.createTask({ description: "Task 4" });
await store.createTask({ description: "Task 5" });
it("respects offset option", async () => {
await store.createTask({ description: "Task 1" });
await store.createTask({ description: "Task 2" });
await store.createTask({ description: "Task 3" });
const results = await store.searchTasks("", { limit: 2 });
const allResults = await store.searchTasks("");
const offsetResults = await store.searchTasks("", { offset: 1 });
expect(results).toHaveLength(2);
});
expect(allResults.length).toBe(3);
expect(offsetResults.length).toBe(2);
expect(offsetResults[0].id).toBe(allResults[1].id);
});
it("respects offset option", async () => {
await store.createTask({ description: "Task 1" });
await store.createTask({ description: "Task 2" });
await store.createTask({ description: "Task 3" });
it("immediately indexes new comments", async () => {
const task = await store.createTask({ description: "A task without comments" });
const uniqueWord = `unique_search_term_${Date.now()}`;
const allResults = await store.searchTasks("");
const offsetResults = await store.searchTasks("", { offset: 1 });
// Initially should not be found
const beforeResults = await store.searchTasks(uniqueWord);
expect(beforeResults).toHaveLength(0);
expect(allResults.length).toBe(3);
expect(offsetResults.length).toBe(2);
expect(offsetResults[0].id).toBe(allResults[1].id);
});
// Add comment with unique word
await store.addComment(task.id, `Important note about the ${uniqueWord} feature`, "tester");
it("immediately indexes new comments", async () => {
const task = await store.createTask({ description: "A task without comments" });
const uniqueWord = `unique_search_term_${Date.now()}`;
// Should now be found immediately (trigger fires synchronously)
const afterResults = await store.searchTasks(uniqueWord);
expect(afterResults).toHaveLength(1);
expect(afterResults[0].id).toBe(task.id);
});
// Initially should not be found
const beforeResults = await store.searchTasks(uniqueWord);
expect(beforeResults).toHaveLength(0);
it("sanitizes FTS5 special characters from query", async () => {
await store.createTask({ title: "Test with special chars", description: "Query parsing test" });
// Add comment with unique word
await store.addComment(task.id, `Important note about the ${uniqueWord} feature`, "tester");
// This should not throw and should work correctly
const results = await store.searchTasks("test + special (chars)");
// Should now be found immediately (trigger fires synchronously)
const afterResults = await store.searchTasks(uniqueWord);
expect(afterResults).toHaveLength(1);
expect(afterResults[0].id).toBe(task.id);
});
expect(results.length).toBeGreaterThanOrEqual(0); // Should not throw
});
it("sanitizes FTS5 special characters from query", async () => {
await store.createTask({ title: "Test with special chars", description: "Query parsing test" });
// This should not throw and should work correctly
const results = await store.searchTasks("test + special (chars)");
expect(results.length).toBeGreaterThanOrEqual(0); // Should not throw
});
});
describe("RunMutationContext", () => {
it("logEntry() with runContext includes runContext field", 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 runContext = { runId: "run-123", agentId: "agent-456" };
await localStore.logEntry(task.id, "Test action", "Test outcome", runContext);
const updatedTask = await localStore.getTask(task.id);
expect(updatedTask.log).toHaveLength(1);
expect(updatedTask.log[0].runContext).toEqual(runContext);
expect(updatedTask.log[0].action).toBe("Test action");
expect(updatedTask.log[0].outcome).toBe("Test outcome");
localStore.stopWatching();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
it("logEntry() without runContext has no runContext field (backward compat)", 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" });
await localStore.logEntry(task.id, "Test action", "Test outcome");
const updatedTask = await localStore.getTask(task.id);
expect(updatedTask.log).toHaveLength(1);
expect(updatedTask.log[0].runContext).toBeUndefined();
expect(updatedTask.log[0].action).toBe("Test action");
localStore.stopWatching();
} 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();
try {
const localStore = new TaskStore(localRoot, localGlobal);
await localStore.init();
const task = await localStore.createTask({ description: "Test task" });
const runContext = { runId: "run-789", agentId: "agent-101" };
await localStore.addComment(task.id, "Test comment", "user", undefined, runContext);
const updatedTask = await localStore.getTask(task.id);
expect(updatedTask.comments).toHaveLength(1);
expect(updatedTask.comments![0].text).toBe("Test comment");
expect(updatedTask.log).toHaveLength(1);
expect(updatedTask.log[0].runContext).toEqual(runContext);
localStore.stopWatching();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
it("addSteeringComment() forwards runContext to addComment", 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 runContext = { runId: "run-abc", agentId: "agent-def", source: "timer" };
await localStore.addSteeringComment(task.id, "Steering comment", "agent", runContext);
const updatedTask = await localStore.getTask(task.id);
expect(updatedTask.steeringComments).toHaveLength(1);
expect(updatedTask.steeringComments![0].text).toBe("Steering comment");
expect(updatedTask.log).toHaveLength(1);
expect(updatedTask.log[0].runContext).toEqual(runContext);
localStore.stopWatching();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
it("getMutationsForRun(runId) returns only entries matching the runId, sorted by timestamp", async () => {
const localRoot = makeTmpDir();
const localGlobal = makeTmpDir();
try {
const localStore = new TaskStore(localRoot, localGlobal);
await localStore.init();
const task1 = await localStore.createTask({ description: "Task 1" });
const task2 = await localStore.createTask({ description: "Task 2" });
// Add entries with different runIds
await localStore.logEntry(task1.id, "Action 1", undefined, { runId: "run-target", agentId: "agent-1" });
await new Promise(r => setTimeout(r, 10)); // Ensure different timestamps
await localStore.logEntry(task2.id, "Action 2", undefined, { runId: "run-target", agentId: "agent-1" });
await new Promise(r => setTimeout(r, 10));
await localStore.logEntry(task1.id, "Action 3", undefined, { runId: "run-other", agentId: "agent-2" });
const mutations = await localStore.getMutationsForRun("run-target");
expect(mutations).toHaveLength(2);
expect(mutations.map(m => m.action)).toEqual(["Action 1", "Action 2"]);
// Verify sorted by timestamp
expect(new Date(mutations[0].timestamp).getTime()).toBeLessThan(new Date(mutations[1].timestamp).getTime());
localStore.stopWatching();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
it("getMutationsForRun(unknownRunId) returns empty array", 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" });
await localStore.logEntry(task.id, "Some action", undefined, { runId: "run-existing", agentId: "agent-1" });
const mutations = await localStore.getMutationsForRun("run-does-not-exist");
expect(mutations).toEqual([]);
localStore.stopWatching();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
it("getMutationsForRun() collects entries across multiple tasks", async () => {
const localRoot = makeTmpDir();
const localGlobal = makeTmpDir();
try {
const localStore = new TaskStore(localRoot, localGlobal);
await localStore.init();
const task1 = await localStore.createTask({ description: "Task 1" });
const task2 = await localStore.createTask({ description: "Task 2" });
const task3 = await localStore.createTask({ description: "Task 3" });
await localStore.logEntry(task1.id, "Entry 1", undefined, { runId: "run-shared", agentId: "agent-x" });
await localStore.logEntry(task2.id, "Entry 2", undefined, { runId: "run-shared", agentId: "agent-x" });
await localStore.logEntry(task3.id, "Entry 3", undefined, { runId: "run-other", agentId: "agent-y" });
const mutations = await localStore.getMutationsForRun("run-shared");
expect(mutations).toHaveLength(2);
expect(mutations.map(m => m.action).sort()).toEqual(["Entry 1", "Entry 2"]);
localStore.stopWatching();
} finally {
await rm(localRoot, { recursive: true, force: true });
await rm(localGlobal, { recursive: true, force: true });
}
});
});
});

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

@@ -4,7 +4,7 @@ import { randomUUID } from "node:crypto";
import { appendFile, mkdir, 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, InboxTask } from "./types.js";
import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, InboxTask, TaskLogEntry, RunMutationContext } from "./types.js";
import { VALID_TRANSITIONS, DEFAULT_SETTINGS, GLOBAL_SETTINGS_KEYS, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js";
import { GlobalSettingsStore } from "./global-settings.js";
import { Database, toJson, toJsonNullable, fromJson } from "./db.js";
@@ -1394,6 +1394,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
async updateTask(
id: string,
updates: { title?: string; description?: string; prompt?: string; worktree?: string | null; status?: string | null; dependencies?: string[]; blockedBy?: string | null; assignedAgentId?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; paused?: boolean; baseBranch?: string | null; branch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; mergeRetries?: number; stuckKillCount?: number | null; recoveryRetryCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null },
runContext?: RunMutationContext,
): Promise<Task> {
return this.withTaskLock(id, async () => {
// Validate that task doesn't depend on itself
@@ -1427,10 +1428,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.column = "triage";
task.status = undefined;
task.columnMovedAt = new Date().toISOString();
task.log.push({
const depLogEntry: TaskLogEntry = {
timestamp: new Date().toISOString(),
action: "Moved to triage for re-specification — new dependency added",
});
};
if (runContext) {
depLogEntry.runContext = runContext;
}
task.log.push(depLogEntry);
movedToTriage = true;
}
}
@@ -1613,7 +1618,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* Pause or unpause a task. Paused tasks are excluded from all automated
* agent and scheduler interaction. Logs the action and emits `task:updated`.
*/
async pauseTask(id: string, paused: boolean): Promise<Task> {
async pauseTask(id: string, paused: boolean, runContext?: RunMutationContext): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.readTaskJson(dir);
@@ -1631,10 +1636,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
const now = new Date().toISOString();
task.updatedAt = now;
task.log.push({
const logEntry: TaskLogEntry = {
timestamp: now,
action: paused ? "Task paused" : "Task unpaused",
});
};
if (runContext) {
logEntry.runContext = runContext;
}
task.log.push(logEntry);
await this.atomicWriteTaskJson(dir, task);
if (this.isWatching) this.taskCache.set(id, { ...task });
@@ -1704,7 +1713,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
/**
* Add a log entry to a task.
*/
async logEntry(id: string, action: string, outcome?: string): Promise<Task> {
async logEntry(id: string, action: string, outcome?: string, runContext?: RunMutationContext): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.readTaskJson(dir);
@@ -1714,11 +1723,15 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
task.log = [];
}
task.log.push({
const entry: TaskLogEntry = {
timestamp: new Date().toISOString(),
action,
outcome,
});
};
if (runContext) {
entry.runContext = runContext;
}
task.log.push(entry);
task.updatedAt = new Date().toISOString();
await this.atomicWriteTaskJson(dir, task);
@@ -1729,6 +1742,25 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
});
}
/**
* Get all task log entries correlated with a specific run ID.
* Scans all tasks' logs for entries whose runContext.runId matches.
*/
async getMutationsForRun(runId: string): Promise<TaskLogEntry[]> {
const allTasks = await this.listTasks();
const mutations: TaskLogEntry[] = [];
for (const task of allTasks) {
if (!task.log) continue;
for (const entry of task.log) {
if (entry.runContext?.runId === runId) {
mutations.push(entry);
}
}
}
// Sort by timestamp ascending
return mutations.sort((a, b) => a.timestamp.localeCompare(b.timestamp));
}
/**
* Sync steps from PROMPT.md into task.json (called when steps are empty).
*/
@@ -2570,9 +2602,9 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
* `steeringComments` (for executor real-time injection).
* Unlike regular comments, steering comments never trigger auto-refinement.
*/
async addSteeringComment(id: string, text: string, author: "user" | "agent" = "user"): Promise<Task> {
async addSteeringComment(id: string, text: string, author: "user" | "agent" = "user", runContext?: RunMutationContext): Promise<Task> {
// Write to unified comments (skip refinement — steering is for agent injection, not follow-up tasks)
const task = await this.addComment(id, text, author, { skipRefinement: true });
const task = await this.addComment(id, text, author, { skipRefinement: true }, runContext);
// Also write to steeringComments so the executor's real-time injection listener can detect new entries
const updated = await this.withTaskLock(id, async () => {
@@ -2669,6 +2701,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
text: string,
author: string = "user",
options?: { skipRefinement?: boolean },
runContext?: RunMutationContext,
): Promise<Task> {
// Phase 1: Add comment under lock
const task = await this.withTaskLock(id, async () => {
@@ -2696,10 +2729,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
task.comments.push(comment);
task.updatedAt = new Date().toISOString();
task.log.push({
const logEntry: TaskLogEntry = {
timestamp: task.updatedAt,
action: `Comment added by ${author}`,
});
};
if (runContext) {
logEntry.runContext = runContext;
}
task.log.push(logEntry);
await this.atomicWriteTaskJson(dir, task);
if (this.isWatching) this.taskCache.set(id, { ...task });
@@ -2739,6 +2776,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
await this.logEntry(
id,
`User comment invalidated spec approval — task needs re-specification`,
undefined,
runContext,
);
} catch {
// Best-effort: don't fail the comment if the status update fails

12
packages/core/src/types.ts
View File

@@ -404,10 +404,22 @@ export interface TaskStep {
status: StepStatus;
}
/** Correlation metadata linking a task mutation to the agent run that caused it. */
export interface RunMutationContext {
/** The heartbeat run ID that initiated this mutation. */
runId: string;
/** The agent ID that performed the mutation. */
agentId: string;
/** Optional invocation source of the run (e.g., "on_demand", "timer", "assignment"). */
source?: string;
}
export interface TaskLogEntry {
timestamp: string;
action: string;
outcome?: string;
/** Correlation metadata linking this entry to the agent run that produced it. */
runContext?: RunMutationContext;
}
export type ActivityEventType = "task:created" | "task:moved" | "task:updated" | "task:deleted" | "task:merged" | "task:failed" | "settings:updated";

54
packages/dashboard/src/__tests__/routes-agent-runs.test.ts
View File

@@ -481,4 +481,58 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
expect((response.body as any).run).toBeDefined();
});
});
describe("GET /api/agents/:id/runs/:runId/mutations", () => {
beforeEach(() => {
// Reset scoped store mock
mockScopedStore = createMockScopedStore();
(createScopedStore as any).mockReturnValue(mockScopedStore);
});
it("returns mutation trail for a valid run", async () => {
const mockRun = createMockRun();
mockGetRunDetail.mockResolvedValue(mockRun);
// Mock getMutationsForRun on the scoped store
const mockMutations = [
{ timestamp: "2026-01-01T00:01:00.000Z", action: "Action 1", runContext: { runId: "run-123", agentId: "agent-001" } },
{ timestamp: "2026-01-01T00:02:00.000Z", action: "Action 2", runContext: { runId: "run-123", agentId: "agent-001" } },
];
mockScopedStore.getMutationsForRun = vi.fn().mockResolvedValue(mockMutations);
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-123/mutations");
expect(response.status).toBe(200);
expect(response.body).toEqual({
runId: "run-123",
mutations: mockMutations,
});
expect(mockScopedStore.getMutationsForRun).toHaveBeenCalledWith("run-123");
});
it("returns 404 for unknown run", async () => {
mockGetRunDetail.mockResolvedValue(null);
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-unknown/mutations");
expect(response.status).toBe(404);
expect(response.body).toHaveProperty("error");
});
it("returns empty mutations array for run with no correlated entries", async () => {
const mockRun = createMockRun();
mockGetRunDetail.mockResolvedValue(mockRun);
// Mock getMutationsForRun returning empty array
mockScopedStore.getMutationsForRun = vi.fn().mockResolvedValue([]);
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-empty/mutations");
expect(response.status).toBe(200);
expect(response.body).toEqual({
runId: "run-empty",
mutations: [],
});
});
});
});

49
packages/dashboard/src/routes.ts
View File

@@ -162,6 +162,22 @@ function slugifyPresetName(name: string): string {
return slug || "preset";
}
/**
* Extract RunMutationContext from the X-Run-Context header.
* Used to correlate dashboard mutations with agent runs for audit trails.
*/
function extractRunContext(req: { headers: { [key: string]: string | string[] | undefined } }): import("@fusion/core").RunMutationContext | undefined {
const header = req.headers['x-run-context'];
if (typeof header !== 'string') return undefined;
try {
const parsed = JSON.parse(header);
if (parsed && typeof parsed.runId === 'string' && typeof parsed.agentId === 'string') {
return parsed as import("@fusion/core").RunMutationContext;
}
} catch { /* invalid JSON, ignore */ }
return undefined;
}
function validateModelPresets(value: unknown): ModelPreset[] | undefined {
if (value === undefined) return undefined;
if (!Array.isArray(value)) {
@@ -9192,6 +9208,39 @@ Output ONLY the prompt text (no markdown, no explanations).`;
}
});
/**
* GET /api/agents/:id/runs/:runId/mutations
* Get the mutation trail for a specific agent run.
* Returns all TaskLogEntry objects correlated with the given runId via runContext.
*/
router.get("/agents/:id/runs/:runId/mutations", async (req, res) => {
try {
const scopedStore = await getScopedStore(req);
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
// Verify the run exists
const run = await agentStore.getRunDetail(req.params.id, req.params.runId);
if (!run) {
throw notFound("Run not found");
}
// Query mutation trail
const mutations = await scopedStore.getMutationsForRun(req.params.runId);
res.json({ runId: req.params.runId, mutations });
} catch (err: any) {
if (err instanceof ApiError) {
throw err;
}
if (err.message?.includes("not found")) {
throw notFound(err.message);
} else {
rethrowAsApiError(err);
}
}
});
/**
* GET /api/agents/:id/chain-of-command
* Fetch agent reporting chain from self to top-most manager.

117
packages/engine/src/agent-heartbeat.test.ts
View File

@@ -3031,4 +3031,121 @@ describe("HeartbeatTriggerScheduler", () => {
expect(callback).not.toHaveBeenCalled();
});
});
describe("Run context propagation", () => {
it("createHeartbeatTools passes runContext to taskStore.logEntry", async () => {
// Create a minimal mock TaskStore
const mockTaskStore = {
createTask: vi.fn().mockResolvedValue({ id: "FN-NEW", description: "New task" }),
logEntry: vi.fn().mockResolvedValue({}),
getTask: vi.fn().mockResolvedValue({
id: "FN-001",
description: "Test task",
column: "todo",
log: [],
}),
} as unknown as import("@fusion/core").TaskStore;
const monitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const runContext = { runId: "run-123", agentId: "agent-456", source: "timer" };
// Create tools with run context
const tools = monitor.createHeartbeatTools("agent-456", mockTaskStore, "FN-001", runContext);
// Find the task_log tool and execute it
const taskLogTool = tools.find(t => t.name === "task_log");
expect(taskLogTool).toBeDefined();
const result = await taskLogTool!.execute("call-1", { message: "Test log entry", outcome: undefined }, undefined as any, undefined as any, undefined as any);
// Verify logEntry was called with runContext
expect(mockTaskStore.logEntry).toHaveBeenCalledWith(
"FN-001",
"Test log entry",
undefined,
runContext,
);
});
it("createHeartbeatTools tracks task creations with runContext", async () => {
// Create a minimal mock TaskStore
const mockTaskStore = {
createTask: vi.fn().mockResolvedValue({ id: "FN-NEW", description: "New task created" }),
logEntry: vi.fn().mockResolvedValue({}),
getTask: vi.fn().mockResolvedValue({
id: "FN-001",
description: "Test task",
column: "todo",
log: [],
}),
} as unknown as import("@fusion/core").TaskStore;
const monitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const runContext = { runId: "run-789", agentId: "agent-abc", source: "on_demand" };
// Create tools with run context
const tools = monitor.createHeartbeatTools("agent-abc", mockTaskStore, "FN-001", runContext);
// Find the task_create tool and execute it
const taskCreateTool = tools.find(t => t.name === "task_create");
expect(taskCreateTool).toBeDefined();
const result = await taskCreateTool!.execute("call-1", { description: "New task created" }, undefined as any, undefined as any, undefined as any);
// Verify logEntry was called with runContext for the created task
expect(mockTaskStore.logEntry).toHaveBeenCalledWith(
"FN-NEW",
"Created by agent agent-abc during heartbeat run",
undefined,
runContext,
);
});
it("createHeartbeatTools works without runContext (backward compat)", async () => {
// Create a minimal mock TaskStore
const mockTaskStore = {
createTask: vi.fn().mockResolvedValue({ id: "FN-NEW", description: "New task" }),
logEntry: vi.fn().mockResolvedValue({}),
getTask: vi.fn().mockResolvedValue({
id: "FN-001",
description: "Test task",
column: "todo",
log: [],
}),
} as unknown as import("@fusion/core").TaskStore;
const monitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
// Create tools without run context
const tools = monitor.createHeartbeatTools("agent-456", mockTaskStore, "FN-001");
// Find the task_log tool and execute it
const taskLogTool = tools.find(t => t.name === "task_log");
expect(taskLogTool).toBeDefined();
const result = await taskLogTool!.execute("call-1", { message: "Test log entry", outcome: undefined }, undefined as any, undefined as any, undefined as any);
// Verify logEntry was called without runContext
expect(mockTaskStore.logEntry).toHaveBeenCalledWith(
"FN-001",
"Test log entry",
undefined,
undefined,
);
});
});
});

34
packages/engine/src/agent-heartbeat.ts
View File

@@ -17,10 +17,10 @@
* - onTerminated: Called when an unresponsive agent is terminated
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, BlockedStateSnapshot } from "@fusion/core";
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, BlockedStateSnapshot, RunMutationContext } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
import { createTaskCreateTool, createTaskLogTool, taskCreateParams } from "./agent-tools.js";
import { createTaskCreateTool, createTaskLogTool, createTaskLogToolWithContext, taskCreateParams } from "./agent-tools.js";
import { AgentLogger } from "./agent-logger.js";
import { heartbeatLog } from "./logger.js";
@@ -639,6 +639,13 @@ export class HeartbeatMonitor {
contextSnapshot: Object.keys(runContextSnapshot).length > 0 ? runContextSnapshot : undefined,
});
// Build run context for mutation correlation
const runContext: RunMutationContext = {
runId: run.id,
agentId,
source,
};
let agentLogger: AgentLogger | null = null;
const flushAgentLogger = async (): Promise<void> => {
if (!agentLogger) {
@@ -701,17 +708,17 @@ export class HeartbeatMonitor {
// Persist assignment to AgentStore so subsequent runs retain linkage.
if (agent.taskId !== taskId) {
await this.store.assignTask(agentId, taskId);
await this.store.assignTask(agentId, taskId, runContext);
}
// FN-1253 compatibility: if checkout API is available on TaskStore,
// try to claim the lease. On conflict, skip this task gracefully.
const checkoutTask = (taskStore as TaskStore & {
checkoutTask?: (taskId: string, agentId: string) => Promise<unknown>;
checkoutTask?: (taskId: string, agentId: string, runContext?: RunMutationContext) => Promise<unknown>;
}).checkoutTask;
if (typeof checkoutTask === "function") {
try {
await checkoutTask.call(taskStore, taskId, agentId);
await checkoutTask.call(taskStore, taskId, agentId, runContext);
} catch {
heartbeatLog.log(`Task ${taskId} already checked out — skipping`);
taskId = undefined;
@@ -816,7 +823,7 @@ export class HeartbeatMonitor {
}
const blockedMessage = `Task is blocked by ${blockedBy}; waiting for dependency/context changes before retrying.`;
await taskStore.addComment(taskId, blockedMessage, "agent");
await taskStore.addComment(taskId, blockedMessage, "agent", undefined, runContext);
await this.store.setLastBlockedState(agentId, currentBlockedState);
heartbeatLog.log(`Task ${taskId} is blocked by ${blockedBy} — recorded blocked state`);
@@ -867,8 +874,8 @@ export class HeartbeatMonitor {
// Lazy-load createKbAgent and promptWithFallback
const { createKbAgent, promptWithFallback } = await import("./pi.js");
// Build tools with task creation tracking
const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId);
// Build tools with task creation tracking and run context for mutation correlation
const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId, runContext);
heartbeatTools.push(heartbeatDoneTool);
agentLogger = new AgentLogger({
@@ -1027,9 +1034,10 @@ export class HeartbeatMonitor {
* @param agentId - The agent ID (used for tracking and logging)
* @param taskStore - TaskStore for task creation and logging
* @param taskId - The assigned task ID (for task_log context)
* @param runContext - Optional run context for mutation correlation
* @returns Array of ToolDefinitions for the heartbeat session
*/
createHeartbeatTools(agentId: string, taskStore: TaskStore, taskId: string): ToolDefinition[] {
createHeartbeatTools(agentId: string, taskStore: TaskStore, taskId: string, runContext?: RunMutationContext): ToolDefinition[] {
const tools: ToolDefinition[] = [];
// Wrap createTaskCreateTool with tracking and agent-link logging
@@ -1045,9 +1053,9 @@ export class HeartbeatMonitor {
const taskIdMatch = responseText.match(/Created (FN-\d+|KB-\d+|\w+-\d+):/);
const createdTaskId = taskIdMatch?.[1] ?? "unknown";
// Log agent link on the created task
// Log agent link on the created task with run context for correlation
try {
await taskStore.logEntry(createdTaskId, `Created by agent ${agentId} during heartbeat run`);
await taskStore.logEntry(createdTaskId, `Created by agent ${agentId} during heartbeat run`, undefined, runContext);
} catch {
// Non-critical — task was created, just the log failed
}
@@ -1066,8 +1074,8 @@ export class HeartbeatMonitor {
};
tools.push(trackedCreateTool);
// task_log tool (standard, no tracking needed)
tools.push(createTaskLogTool(taskStore, taskId));
// task_log tool (with run context for mutation correlation)
tools.push(createTaskLogToolWithContext(taskStore, taskId, runContext));
return tools;
}

28
packages/engine/src/agent-tools.ts
View File

@@ -7,7 +7,7 @@
* The parameter schemas are canonical here — executor.ts imports and reuses them.
*/
import type { TaskDocument, TaskDocumentCreateInput, TaskStore } from "@fusion/core";
import type { TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
import type { AgentReflectionService } from "./agent-reflection.js";
@@ -107,6 +107,32 @@ export function createTaskLogTool(store: TaskStore, taskId: string): ToolDefinit
};
}
/**
* Create a `task_log` tool with run context for mutation correlation.
*
* @param store - TaskStore for task persistence
* @param taskId - The task ID to log entries against
* @param runContext - Optional run context for mutation correlation
* @returns ToolDefinition for the `task_log` tool
*/
export function createTaskLogToolWithContext(store: TaskStore, taskId: string, runContext?: RunMutationContext): ToolDefinition {
return {
name: "task_log",
label: "Log Entry",
description:
"Log an important action, decision, or issue for this task. " +
"Use for significant events — not every small step.",
parameters: taskLogParams,
execute: async (_id: string, params: Static<typeof taskLogParams>) => {
await store.logEntry(taskId, params.message, params.outcome, runContext);
return {
content: [{ type: "text" as const, text: `Logged: ${params.message}` }],
details: {},
};
},
};
}
/**
* Create a `task_document_write` tool that stores a named task document.
*

83
packages/engine/src/executor.ts
View File

@@ -1,7 +1,7 @@
import { execSync } from "node:child_process";
import { join } from "node:path";
import { existsSync } from "node:fs";
import type { TaskStore, Task, TaskDetail, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability } from "@fusion/core";
import type { TaskStore, Task, TaskDetail, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext } from "@fusion/core";
import type { AgentStore } from "@fusion/core";
import { buildExecutionMemoryInstructions, resolveAgentPrompt } from "@fusion/core";
import { findWorktreeUser } from "./merger.js";
@@ -293,6 +293,8 @@ export class TaskExecutor {
/** Token cap detector for proactive context compaction. */
private tokenCapDetector = new TokenCapDetector();
private _modelRegistry?: InstanceType<typeof ModelRegistry>;
/** Current run context for mutation correlation. Set at execute() start, cleared in finally. */
private currentRunContext: RunMutationContext | undefined;
private get modelRegistry(): InstanceType<typeof ModelRegistry> {
if (!this._modelRegistry) {
@@ -396,7 +398,7 @@ export class TaskExecutor {
executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`);
try {
await this.clearResumeFailureState(task);
await this.store.logEntry(task.id, "Resuming execution after unpause");
await this.store.logEntry(task.id, "Resuming execution after unpause", undefined, this.currentRunContext);
} catch { /* non-critical */ }
this.execute(task).catch((err) =>
executorLog.error(`Failed to resume unpaused ${task.id}:`, err),
@@ -429,13 +431,13 @@ export class TaskExecutor {
if (model) {
await activeEntry.session.setModel(model);
executorLog.log(`${task.id}: executor model hot-swapped to ${newProvider}/${newModelId}`);
await this.store.logEntry(task.id, `Model changed to ${newProvider}/${newModelId}`);
await this.store.logEntry(task.id, `Model changed to ${newProvider}/${newModelId}`, undefined, this.currentRunContext);
} else {
executorLog.log(`${task.id}: model ${newProvider}/${newModelId} not found in registry for hot-swap`);
}
} catch (err: any) {
executorLog.error(`${task.id}: failed to hot-swap model: ${err.message}`);
await this.store.logEntry(task.id, `Model change failed: ${err.message}`);
await this.store.logEntry(task.id, `Model change failed: ${err.message}`, undefined, this.currentRunContext);
}
}
}
@@ -689,6 +691,13 @@ export class TaskExecutor {
// Fetch settings early — needed for worktree naming and later configuration
const settings = await this.store.getSettings();
// Construct run context for mutation correlation
// Use a synthetic correlation ID: task ID + timestamp + random suffix
this.currentRunContext = {
runId: `exec-${task.id}-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`,
agentId: task.assignedAgentId ?? "executor",
};
// Hoist worktreePath so it's accessible in the catch block for dep-abort cleanup
// Determine worktree name based on settings
let worktreePath: string;
@@ -755,9 +764,9 @@ export class TaskExecutor {
await this.store.updateTask(task.id, { worktree: worktreePath, branch: actualBranch });
if (actualBranch !== branchName) {
executorLog.log(`Branch conflict resolved: using ${actualBranch} instead of ${branchName}`);
await this.store.logEntry(task.id, `Acquired worktree from pool: ${worktreePath} (branch conflict: using ${actualBranch})`);
await this.store.logEntry(task.id, `Acquired worktree from pool: ${worktreePath} (branch conflict: using ${actualBranch})`, undefined, this.currentRunContext);
} else {
await this.store.logEntry(task.id, `Acquired worktree from pool: ${worktreePath}`);
await this.store.logEntry(task.id, `Acquired worktree from pool: ${worktreePath}`, undefined, this.currentRunContext);
}
} catch (poolErr: any) {
// Pool preparation failed — release the worktree back and fall through
@@ -767,6 +776,8 @@ export class TaskExecutor {
await this.store.logEntry(
task.id,
`Pool worktree preparation failed (${poolErr.message}), creating fresh worktree`,
undefined,
this.currentRunContext,
);
}
}
@@ -779,11 +790,11 @@ export class TaskExecutor {
await this.store.updateTask(task.id, { worktree: created.path, branch: created.branch });
if (created.branch !== branchName) {
executorLog.log(`Branch conflict resolved: using ${created.branch} instead of ${branchName}`);
await this.store.logEntry(task.id, `Worktree created at ${worktreePath} (branch conflict: using ${created.branch})`);
await this.store.logEntry(task.id, `Worktree created at ${worktreePath} (branch conflict: using ${created.branch})`, undefined, this.currentRunContext);
} else if (baseBranch) {
await this.store.logEntry(task.id, `Worktree created at ${worktreePath} (based on ${baseBranch})`);
await this.store.logEntry(task.id, `Worktree created at ${worktreePath} (based on ${baseBranch})`, undefined, this.currentRunContext);
} else {
await this.store.logEntry(task.id, `Worktree created at ${worktreePath}`);
await this.store.logEntry(task.id, `Worktree created at ${worktreePath}`, undefined, this.currentRunContext);
}
// Run worktree init command for fresh worktrees (skip for pooled — caches are warm)
@@ -794,10 +805,10 @@ export class TaskExecutor {
stdio: "pipe",
timeout: 120_000,
});
await this.store.logEntry(task.id, "Worktree init command completed", settings.worktreeInitCommand);
await this.store.logEntry(task.id, "Worktree init command completed", settings.worktreeInitCommand, this.currentRunContext);
} catch (err: any) {
const message = err.stderr?.toString() || err.message || "Unknown error";
await this.store.logEntry(task.id, `Worktree init command failed: ${message}`);
await this.store.logEntry(task.id, `Worktree init command failed: ${message}`, undefined, this.currentRunContext);
}
}
@@ -811,13 +822,13 @@ export class TaskExecutor {
stdio: "pipe",
timeout: 120_000,
});
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' completed`, scriptCommand);
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' completed`, scriptCommand, this.currentRunContext);
} catch (err: any) {
const message = err.stderr?.toString() || err.message || "Unknown error";
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' failed: ${message}`);
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' failed: ${message}`, undefined, this.currentRunContext);
}
} else {
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' not found in scripts map — skipping`);
await this.store.logEntry(task.id, `Setup script '${settings.setupScript}' not found in scripts map — skipping`, undefined, this.currentRunContext);
}
}
}
@@ -915,7 +926,7 @@ export class TaskExecutor {
}
if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
await this.store.logEntry(task.id, "Execution paused — step sessions terminated, moved to todo");
await this.store.logEntry(task.id, "Execution paused — step sessions terminated, moved to todo", undefined, this.currentRunContext);
await this.store.moveTask(task.id, "todo");
return;
}
@@ -960,7 +971,7 @@ export class TaskExecutor {
onRetry: (attempt, delayMs, error) => {
const delaySec = Math.round(delayMs / 1000);
executorLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`).catch(() => {});
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`, undefined, this.currentRunContext).catch(() => {});
},
});
@@ -976,7 +987,7 @@ export class TaskExecutor {
await this.handleDepAbortCleanup(task.id, worktreePath);
} else if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
await this.store.logEntry(task.id, "Execution paused during step-session");
await this.store.logEntry(task.id, "Execution paused during step-session", undefined, this.currentRunContext);
await this.store.moveTask(task.id, "todo");
} else if (this.stuckAborted.has(task.id)) {
stuckRequeue = this.stuckAborted.get(task.id) ?? true;
@@ -994,7 +1005,7 @@ export class TaskExecutor {
const delay = formatDelay(decision.delayMs);
if (!isSilentTransientError(err.message)) {
executorLog.warn(`⚡ ${task.id} transient error — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}: ${err.message}`);
await this.store.logEntry(task.id, `Transient error (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`);
await this.store.logEntry(task.id, `Transient error (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`, undefined, this.currentRunContext);
}
if (worktreePath && existsSync(worktreePath)) {
try {
@@ -1024,7 +1035,7 @@ export class TaskExecutor {
this.options.onError?.(task, err);
} else {
executorLog.error(`✗ ${task.id} step-session execution failed:`, err.message);
await this.store.logEntry(task.id, `Step-session execution failed: ${err.message}`);
await this.store.logEntry(task.id, `Step-session execution failed: ${err.message}`, undefined, this.currentRunContext);
await this.store.updateTask(task.id, { status: "failed", error: err.message });
await this.store.moveTask(task.id, "in-review");
executorLog.log(`✗ ${task.id} step-session execution failed → in-review`);
@@ -1162,10 +1173,10 @@ export class TaskExecutor {
if (isResuming) {
executorLog.log(`${task.id}: resumed session from ${task.sessionFile}`);
await this.store.logEntry(task.id, `Resumed agent session after unpause (model: ${describeModel(session)})`);
await this.store.logEntry(task.id, `Resumed agent session after unpause (model: ${describeModel(session)})`, undefined, this.currentRunContext);
} else {
executorLog.log(`${task.id}: using model ${describeModel(session)}`);
await this.store.logEntry(task.id, `Executor using model: ${describeModel(session)}`);
await this.store.logEntry(task.id, `Executor using model: ${describeModel(session)}`, undefined, this.currentRunContext);
// Persist session file path so pause/resume can reopen it
if (sessionFile) {
await this.store.updateTask(task.id, { sessionFile });
@@ -1230,6 +1241,8 @@ export class TaskExecutor {
await this.store.logEntry(
task.id,
`Context compacted at ${compactResult.tokensBefore} tokens (token cap: ${settings.tokenCap})`,
undefined,
this.currentRunContext,
);
}
return compactResult;
@@ -1250,7 +1263,7 @@ export class TaskExecutor {
if (loopState?.pending) {
loopState.pending = false;
executorLog.log(`${task.id} consuming loop recovery — resuming with fresh context`);
await this.store.logEntry(task.id, "Resuming execution after context compaction — taking a different approach");
await this.store.logEntry(task.id, "Resuming execution after context compaction — taking a different approach", undefined, this.currentRunContext);
// Reset activity tracking so the detector doesn't immediately re-trigger
stuckDetector?.recordProgress(task.id);
@@ -1312,7 +1325,7 @@ export class TaskExecutor {
implicitCheck.steps.every((s) => s.status === "done" || s.status === "skipped")) {
taskDone = true;
executorLog.log(`${task.id} all steps done — treating as implicit task_done`);
await this.store.logEntry(task.id, "All steps complete — implicit task_done (agent did not call tool explicitly)");
await this.store.logEntry(task.id, "All steps complete — implicit task_done (agent did not call tool explicitly)", undefined, this.currentRunContext);
}
}
@@ -1342,7 +1355,7 @@ export class TaskExecutor {
} else {
// Agent finished without calling task_done — retry once with a fresh session
executorLog.log(`⚠ ${task.id} finished without task_done — retrying with new session`);
await this.store.logEntry(task.id, "Agent finished without calling task_done — retrying with new session");
await this.store.logEntry(task.id, "Agent finished without calling task_done — retrying with new session", undefined, this.currentRunContext);
// Dispose old session and create a fresh one
this.activeSessions.delete(task.id);
@@ -1404,7 +1417,7 @@ export class TaskExecutor {
implicitCheck.steps.every((s) => s.status === "done" || s.status === "skipped")) {
taskDone = true;
executorLog.log(`${task.id} all steps done — treating as implicit task_done`);
await this.store.logEntry(task.id, "All steps complete — implicit task_done (agent did not call tool explicitly)");
await this.store.logEntry(task.id, "All steps complete — implicit task_done (agent did not call tool explicitly)", undefined, this.currentRunContext);
}
}
@@ -1431,7 +1444,7 @@ export class TaskExecutor {
} else {
const errorMessage = "Agent finished without calling task_done (after retry)";
await this.store.updateTask(task.id, { status: "failed", error: errorMessage });
await this.store.logEntry(task.id, `${errorMessage} — moved to in-review for inspection`);
await this.store.logEntry(task.id, `${errorMessage} — moved to in-review for inspection`, undefined, this.currentRunContext);
await this.store.moveTask(task.id, "in-review");
executorLog.log(`✗ ${task.id} failed after retry — no task_done → in-review`);
this.options.onError?.(task, new Error(errorMessage));
@@ -1458,7 +1471,7 @@ export class TaskExecutor {
onRetry: (attempt, delayMs, error) => {
const delaySec = Math.round(delayMs / 1000);
executorLog.warn(`⏳ ${task.id} rate limited — retry ${attempt} in ${delaySec}s: ${error.message}`);
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`).catch(() => {});
this.store.logEntry(task.id, `Rate limited — retry ${attempt} in ${delaySec}s`, undefined, this.currentRunContext).catch(() => {});
},
});
@@ -1480,7 +1493,7 @@ export class TaskExecutor {
const toColumn = transitionMatch?.[2] ?? "unknown";
const logMessage = `Task already moved from '${fromColumn}' — skipping transition to '${toColumn}'`;
executorLog.log(`${task.id} ${logMessage}`);
await this.store.logEntry(task.id, logMessage, err.message);
await this.store.logEntry(task.id, logMessage, err.message, this.currentRunContext);
// Task finished successfully (just already moved), so call onComplete
this.options.onComplete?.(task);
} else if (this.pausedAborted.has(task.id)) {
@@ -1495,8 +1508,8 @@ export class TaskExecutor {
executorLog.warn(`Failed to remove old worktree ${worktreePath}: ${cleanupErr.message}`);
}
}
await this.store.updateTask(task.id, { worktree: null, branch: null });
await this.store.logEntry(task.id, "Execution paused — agent terminated, moved to todo");
await this.store.updateTask(task.id, { worktree: undefined, branch: undefined });
await this.store.logEntry(task.id, "Execution paused — agent terminated, moved to todo", undefined, this.currentRunContext);
await this.store.moveTask(task.id, "todo");
} else if (this.stuckAborted.has(task.id)) {
// Task was killed by stuck task detector — defer requeue to finally block
@@ -1515,7 +1528,7 @@ export class TaskExecutor {
const activeEntry = this.activeSessions.get(task.id);
if (activeEntry) {
executorLog.log(`${task.id} context limit error — attempting compact-and-resume`);
await this.store.logEntry(task.id, `Context limit error — attempting compact-and-resume: ${err.message}`);
await this.store.logEntry(task.id, `Context limit error — attempting compact-and-resume: ${err.message}`, undefined, this.currentRunContext);
const compactResult = await compactSessionContext(activeEntry.session);
if (compactResult) {
@@ -1565,7 +1578,7 @@ export class TaskExecutor {
// Silent transient errors (e.g., "request was aborted") are noisy — skip logging
if (!isSilentTransientError(err.message)) {
executorLog.warn(`⚡ ${task.id} transient error — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}: ${err.message}`);
await this.store.logEntry(task.id, `Transient error (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`);
await this.store.logEntry(task.id, `Transient error (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`, undefined, this.currentRunContext);
}
// Clean up the old worktree so the retry gets a fresh one
if (worktreePath && existsSync(worktreePath)) {
@@ -1588,7 +1601,7 @@ export class TaskExecutor {
// Recovery budget exhausted — escalate to real failure
executorLog.error(`✗ ${task.id} transient error retries exhausted (${MAX_RECOVERY_RETRIES} attempts): ${err.message}`);
await this.store.logEntry(task.id, `Transient error retries exhausted after ${MAX_RECOVERY_RETRIES} attempts: ${err.message}`);
await this.store.logEntry(task.id, `Transient error retries exhausted after ${MAX_RECOVERY_RETRIES} attempts: ${err.message}`, undefined, this.currentRunContext);
await this.store.updateTask(task.id, {
status: "failed",
error: err.message,
@@ -1601,7 +1614,7 @@ export class TaskExecutor {
return;
}
executorLog.error(`✗ ${task.id} execution failed:`, err.message);
await this.store.logEntry(task.id, `Execution failed: ${err.message}`);
await this.store.logEntry(task.id, `Execution failed: ${err.message}`, undefined, this.currentRunContext);
await this.store.updateTask(task.id, { status: "failed", error: err.message });
await this.store.moveTask(task.id, "in-review");
executorLog.log(`✗ ${task.id} execution failed → in-review`);
@@ -1609,6 +1622,8 @@ export class TaskExecutor {
}
} finally {
this.executing.delete(task.id);
// Clear run context at end of execute() lifecycle
this.currentRunContext = undefined;
// Reset loop recovery state at end of execute() lifecycle.
// State is in-memory and per-run — should not persist across attempts.