feat(FN-1404): add run-audit instrumentation layer for agent run mutations

- Add shared RunAuditService in packages/engine/src/run-audit.ts for tracking agent run state transitions - Instrument HeartbeatMonitor to log run mutations (create, complete, error, skip) with timestamps and context - Instrument TaskExecutor to log run mutations during task execution lifecycle - Instrument Merger to log run mutations during merge workflow - Add run-audit pattern documentation to project memory - Fix: remove audit calls from non-run recovery methods to avoid false positives
2026-04-10 02:56:57 -07:00
parent 61adfc87de
commit 174ef490e7
5 changed files with 360 additions and 3 deletions
--- a/.fusion/memory.md
+++ b/.fusion/memory.md
@@ -34,6 +34,7 @@
 - Checkout leasing is explicit: use `checkoutTask`/`releaseTask` (or `/api/tasks/:id/checkout` + `/release`) for ownership, treat 409 conflicts as non-retryable contention, and let `HeartbeatMonitor.executeHeartbeat()` only validate `checkedOutBy` (never auto-acquire leases).
 - The null-as-delete pattern for settings: In `TaskStore.updateSettings()`, `null` values in the settings patch are treated as "delete this key from settings" (since `JSON.stringify` drops `undefined` keys). This allows the frontend to explicitly clear a setting by sending `null`. The key is deleted from both `config.settings` and `projectPatch` before merging, so cleared settings fall back to `DEFAULT_SETTINGS`.
 - `TaskStore.logEntry()`, `addComment()`, `addSteeringComment()`, `pauseTask()` accept an optional `RunMutationContext` parameter for audit trail correlation. Always pass it when the caller is an engine module (executor, heartbeat monitor) to maintain the audit trail. The executor constructs a synthetic `runContext` with `runId: "exec-{taskId}-{timestamp}-{random}"` since it doesn't use `AgentHeartbeatRun`.
+- **Run-Audit Instrumentation (FN-1404)**: The engine instruments mutation calls with audit events via `createRunAuditor()` from `run-audit.ts`. Each active run (heartbeat, executor, merger) creates an `EngineRunContext` with `runId`, `agentId`, `taskId`, and `phase`. The auditor no-ops cleanly when no run context exists (backward compatible with manual/non-run paths). Use `generateSyntheticRunId()` for executor/merger synthetic IDs. Audit events are emitted for git mutations (worktree/branch/create/remove/reset), database mutations (task:update/move/comment/assign/checkout), and filesystem mutations (file:capture-modified).

 ## Color Theme System

--- a/packages/engine/src/agent-heartbeat.ts
+++ b/packages/engine/src/agent-heartbeat.ts
@@ -23,6 +23,7 @@ import { Type, type Static } from "@mariozechner/pi-ai";
 import { createTaskCreateTool, createTaskLogTool, createTaskLogToolWithContext, taskCreateParams } from "./agent-tools.js";
 import { AgentLogger } from "./agent-logger.js";
 import { heartbeatLog } from "./logger.js";
+import { createRunAuditor, type EngineRunContext } from "./run-audit.js";

 // Lazy import for pi — avoids pulling the pi SDK into the module graph
 // when heartbeat execution isn't needed.
@@ -646,6 +647,18 @@ export class HeartbeatMonitor {
        source,
      };

+      // Build engine run context for audit instrumentation
+      const engineRunContext: EngineRunContext = {
+        runId: run.id,
+        agentId,
+        source,
+        phase: "heartbeat",
+      };
+
+      // Create run auditor for audit trail (FN-1404)
+      // Uses TaskStore.recordRunAuditEvent when available; no-ops otherwise
+      const audit = createRunAuditor(taskStore, engineRunContext);
+
      let agentLogger: AgentLogger | null = null;
      const flushAgentLogger = async (): Promise<void> => {
        if (!agentLogger) {
@@ -709,6 +722,8 @@ export class HeartbeatMonitor {
            // Persist assignment to AgentStore so subsequent runs retain linkage.
            if (agent.taskId !== taskId) {
              await this.store.assignTask(agentId, taskId, runContext);
+              // Audit trail: record assignment mutation (FN-1404)
+              await audit.database({ type: "task:assign", target: taskId });
            }

            // FN-1253 compatibility: if checkout API is available on TaskStore,
@@ -719,6 +734,8 @@ export class HeartbeatMonitor {
            if (typeof checkoutTask === "function") {
              try {
                await checkoutTask.call(taskStore, taskId, agentId, runContext);
+                // Audit trail: record checkout mutation (FN-1404)
+                await audit.database({ type: "task:checkout", target: taskId });
              } catch {
                heartbeatLog.log(`Task ${taskId} already checked out — skipping`);
                taskId = undefined;
@@ -737,6 +754,9 @@ export class HeartbeatMonitor {
            },
          };
          await this.store.saveRun(updatedRun);
+
+          // Update engine run context with resolved taskId for audit trail (FN-1404)
+          engineRunContext.taskId = taskId;
        }

        if (!taskId) {
@@ -824,6 +844,8 @@ export class HeartbeatMonitor {

          const blockedMessage = `Task is blocked by ${blockedBy}; waiting for dependency/context changes before retrying.`;
          await taskStore.addComment(taskId, blockedMessage, "agent", undefined, runContext);
+          // Audit trail: record comment mutation (FN-1404)
+          await audit.database({ type: "task:comment:add", target: taskId, metadata: { blockedBy } });
          await this.store.setLastBlockedState(agentId, currentBlockedState);

          heartbeatLog.log(`Task ${taskId} is blocked by ${blockedBy} — recorded blocked state`);
@@ -875,7 +897,7 @@ export class HeartbeatMonitor {
        const { createKbAgent, promptWithFallback } = await import("./pi.js");

        // Build tools with task creation tracking and run context for mutation correlation
-        const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId, runContext);
+        const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId, runContext, audit);
        heartbeatTools.push(heartbeatDoneTool);

        agentLogger = new AgentLogger({
@@ -1035,9 +1057,16 @@ export class HeartbeatMonitor {
   * @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
+   * @param audit - Optional run auditor for audit trail (FN-1404)
   * @returns Array of ToolDefinitions for the heartbeat session
   */
-  createHeartbeatTools(agentId: string, taskStore: TaskStore, taskId: string, runContext?: RunMutationContext): ToolDefinition[] {
+  createHeartbeatTools(
+    agentId: string,
+    taskStore: TaskStore,
+    taskId: string,
+    runContext?: RunMutationContext,
+    audit?: ReturnType<typeof createRunAuditor>,
+  ): ToolDefinition[] {
    const tools: ToolDefinition[] = [];

    // Wrap createTaskCreateTool with tracking and agent-link logging
@@ -1060,6 +1089,9 @@ export class HeartbeatMonitor {
          // Non-critical — task was created, just the log failed
        }

+        // Audit trail: record task creation (FN-1404)
+        await audit?.database({ type: "task:create", target: createdTaskId });
+
        // Accumulate for inclusion in run resultJson
        if (!this.runCreatedTasks.has(agentId)) {
          this.runCreatedTasks.set(agentId, []);
--- a/packages/engine/src/executor.ts
+++ b/packages/engine/src/executor.ts
@@ -25,6 +25,7 @@ import { isContextLimitError } from "./context-limit-detector.js";
 import { StepSessionExecutor, type StepSessionExecutorOptions, type StepResult } from "./step-session-executor.js";
 import { resolveAgentInstructions, buildSystemPromptWithInstructions } from "./agent-instructions.js";
 import type { AgentReflectionService } from "./agent-reflection.js";
+import { createRunAuditor, generateSyntheticRunId, type EngineRunContext } from "./run-audit.js";
 import {
  createReflectOnPerformanceTool,
  createTaskCreateTool as sharedCreateTaskCreateTool,
@@ -803,11 +804,23 @@ export class TaskExecutor {

    // Construct run context for mutation correlation
    // Use a synthetic correlation ID: task ID + timestamp + random suffix
+    const syntheticRunId = generateSyntheticRunId("exec", task.id);
    this.currentRunContext = {
-      runId: `exec-${task.id}-${Date.now()}-${Math.random().toString(36).slice(2, 6)}`,
+      runId: syntheticRunId,
      agentId: task.assignedAgentId ?? "executor",
    };

+    // Build engine run context for audit instrumentation (FN-1404)
+    const engineRunContext: EngineRunContext = {
+      runId: syntheticRunId,
+      agentId: task.assignedAgentId ?? "executor",
+      taskId: task.id,
+      phase: "execute",
+    };
+
+    // Create run auditor for TaskStore-backed audit emission (no-ops if store doesn't support it)
+    const audit = createRunAuditor(this.store, engineRunContext);
+
    // Hoist worktreePath so it's accessible in the catch block for dep-abort cleanup
    // Determine worktree name based on settings
    let worktreePath: string;
@@ -873,6 +886,8 @@ export class TaskExecutor {
              acquiredFromPool = true;
              executorLog.log(`Acquired worktree from pool: ${pooled}`);
              await this.store.updateTask(task.id, { worktree: worktreePath, branch: actualBranch });
+              // Audit trail: record worktree reuse (FN-1404)
+              await audit.git({ type: "worktree:reuse", target: worktreePath, metadata: { 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})`, undefined, this.currentRunContext);
@@ -899,6 +914,9 @@ export class TaskExecutor {
          const created = await this.createWorktree(branchName, worktreePath, task.id, baseBranch ?? undefined);
          worktreePath = created.path;
          await this.store.updateTask(task.id, { worktree: created.path, branch: created.branch });
+          // Audit trail: record worktree creation and branch creation (FN-1404)
+          await audit.git({ type: "worktree:create", target: created.path, metadata: { branch: created.branch } });
+          await audit.git({ type: "branch:create", target: 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})`, undefined, this.currentRunContext);
@@ -951,6 +969,9 @@ export class TaskExecutor {
        const created = await this.createWorktree(branchName, worktreePath, task.id);
        worktreePath = created.path;
        await this.store.updateTask(task.id, { worktree: created.path, branch: created.branch });
+        // Audit trail: record worktree creation and branch creation (FN-1404)
+        await audit.git({ type: "worktree:create", target: created.path, metadata: { branch: created.branch } });
+        await audit.git({ type: "branch:create", target: created.branch });
      }

      // Capture the base commit SHA for diff computation whenever a task
@@ -965,6 +986,8 @@ export class TaskExecutor {
          }).trim();
          await this.store.updateTask(task.id, { baseCommitSha });
          executorLog.log(`${task.id}: captured baseCommitSha ${baseCommitSha.slice(0, 7)}`);
+          // Audit trail: record base commit capture for later diff computation (FN-1404)
+          await audit.git({ type: "commit:create", target: baseCommitSha, metadata: { purpose: "base" } });
        } catch (err: any) {
          executorLog.log(`Failed to capture baseCommitSha for ${task.id}: ${err.message}`);
          // Non-fatal: task can continue without baseCommitSha
@@ -1055,18 +1078,24 @@ export class TaskExecutor {
            if (modifiedFiles.length > 0) {
              await this.store.updateTask(task.id, { modifiedFiles });
              executorLog.log(`${task.id}: captured ${modifiedFiles.length} modified files`);
+              // Audit trail: record filesystem mutation (FN-1404)
+              await audit.filesystem({ type: "file:capture-modified", target: task.id, metadata: { files: modifiedFiles } });
            }

            const workflowSuccess = await this.runWorkflowSteps(task, worktreePath, settings);
            if (!workflowSuccess) {
              await this.store.updateTask(task.id, { status: "failed", error: "Workflow step failed" });
              await this.store.moveTask(task.id, "in-review");
+              // Audit trail: record task move (FN-1404)
+              await audit.database({ type: "task:move", target: task.id, metadata: { to: "in-review" } });
              executorLog.log(`✗ ${task.id} workflow step failed → in-review`);
              this.options.onError?.(task, new Error("Workflow step failed"));
              return;
            }

            await this.store.moveTask(task.id, "in-review");
+            // Audit trail: record task move (FN-1404)
+            await audit.database({ type: "task:move", target: task.id, metadata: { to: "in-review" } });
            executorLog.log(`✓ ${task.id} completed (step-session) → in-review`);
            this.options.onComplete?.(task);
          } else {
@@ -1122,6 +1151,8 @@ export class TaskExecutor {
              if (worktreePath && existsSync(worktreePath)) {
                try {
                  execSync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir, stdio: "pipe" });
+                  // Audit trail: record worktree removal (FN-1404)
+                  await audit.git({ type: "worktree:remove", target: worktreePath });
                } catch {}
              }
              await this.store.updateTask(task.id, {
@@ -1643,6 +1674,8 @@ export class TaskExecutor {
            try {
              execSync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir, stdio: "pipe" });
              executorLog.log(`Removed old worktree for paused task: ${worktreePath}`);
+              // Audit trail: record worktree removal (FN-1404)
+              await audit.git({ type: "worktree:remove", target: worktreePath });
            } catch (cleanupErr: any) {
              executorLog.warn(`Failed to remove old worktree ${worktreePath}: ${cleanupErr.message}`);
            }
@@ -1725,6 +1758,8 @@ export class TaskExecutor {
              try {
                execSync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir, stdio: "pipe" });
                executorLog.log(`Removed old worktree for transient retry: ${worktreePath}`);
+                // Audit trail: record worktree removal (FN-1404)
+                await audit.git({ type: "worktree:remove", target: worktreePath });
              } catch (cleanupErr: any) {
                executorLog.warn(`Failed to remove old worktree ${worktreePath}: ${cleanupErr.message}`);
              }
@@ -1785,6 +1820,8 @@ export class TaskExecutor {
            try {
              execSync(`git worktree remove "${worktreePath}" --force`, { cwd: this.rootDir, stdio: "pipe" });
              executorLog.log(`Removed old worktree for stuck-killed retry: ${worktreePath}`);
+              // Audit trail: record worktree removal (FN-1404)
+              await audit.git({ type: "worktree:remove", target: worktreePath });
            } catch (cleanupErr: any) {
              executorLog.warn(`Failed to remove old worktree ${worktreePath}: ${cleanupErr.message}`);
            }
@@ -1795,6 +1832,8 @@ export class TaskExecutor {
          // when execute() started (e.g., resumed orphan), we skip the redundant move.
          if (task.column !== "todo") {
            await this.store.moveTask(task.id, "todo");
+            // Audit trail: record task move (FN-1404)
+            await audit.database({ type: "task:move", target: task.id, metadata: { to: "todo" } });
            executorLog.log(`${task.id} moved to todo for retry after stuck kill`);
          } else {
            executorLog.log(`${task.id} already in todo — skipping redundant move`);
--- a/packages/engine/src/merger.ts
+++ b/packages/engine/src/merger.ts
@@ -12,6 +12,7 @@ import { withRateLimitRetry } from "./rate-limit-retry.js";
 import { resolveAgentInstructions, buildSystemPromptWithInstructions } from "./agent-instructions.js";
 import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
 import { Type } from "@sinclair/typebox";
+import { createRunAuditor, generateSyntheticRunId, type EngineRunContext } from "./run-audit.js";

 /** Conflict type classification for merge conflict resolution */
 export type ConflictType =
@@ -709,6 +710,18 @@ export async function aiMergeTask(
    branchDeleted: false,
  };

+  // Build merge-run context for audit instrumentation (FN-1404)
+  const mergeRunId = generateSyntheticRunId("merge", taskId);
+  const engineRunContext: EngineRunContext = {
+    runId: mergeRunId,
+    agentId: "merger",
+    taskId,
+    phase: "merge",
+  };
+
+  // Create run auditor for TaskStore-backed audit emission (no-ops if store doesn't support it)
+  const audit = createRunAuditor(store, engineRunContext);
+
  if (!worktreePath) {
    mergerLog.warn(`${taskId}: no worktree path set — skipping worktree cleanup`);
  }
@@ -747,6 +760,8 @@ export async function aiMergeTask(
    } catch {
      // No commit SHA available — task will show summary fallback
    }
+    // Audit trail: record merge completion (FN-1404)
+    await audit.database({ type: "task:move", target: taskId, metadata: { to: "done", merged: false } });
    await completeTask(store, taskId, result);
    return result;
  }
@@ -771,11 +786,15 @@ export async function aiMergeTask(
        cwd: rootDir,
        stdio: "pipe",
      });
+      // Audit trail: record git checkout (FN-1404)
+      await audit.git({ type: "branch:checkout", target: mainBranch });
    }
  } catch {
    // Fallback: try checking out main directly
    try {
      execSync("git checkout main", { cwd: rootDir, stdio: "pipe" });
+      // Audit trail: record git checkout (FN-1404)
+      await audit.git({ type: "branch:checkout", target: "main" });
    } catch {
      mergerLog.warn(`${taskId}: unable to verify/checkout main branch — proceeding on current HEAD`);
    }
@@ -860,6 +879,8 @@ export async function aiMergeTask(
        mergerLog.log(`${taskId}: attempt ${attemptNum} failed, cleaning up for retry...`);
        try {
          execSync("git reset --merge", { cwd: rootDir, stdio: "pipe" });
+          // Audit trail: record git reset for merge cleanup (FN-1404)
+          await audit.git({ type: "reset:hard", target: branch, metadata: { purpose: "merge-cleanup", attempt: attemptNum } });
        } catch { /* ignore cleanup errors */ }
      }

@@ -875,6 +896,8 @@ export async function aiMergeTask(
          result._buildRetried = true;
          try {
            execSync("git reset --merge", { cwd: rootDir, stdio: "pipe" });
+            // Audit trail: record git reset for build retry (FN-1404)
+            await audit.git({ type: "reset:hard", target: branch, metadata: { purpose: "build-retry" } });
          } catch { /* ignore cleanup errors */ }
          return false; // Retry
        }
@@ -886,6 +909,8 @@ export async function aiMergeTask(
        mergerLog.log(`${taskId}: attempt ${attemptNum} error, cleaning up for retry...`);
        try {
          execSync("git reset --merge", { cwd: rootDir, stdio: "pipe" });
+          // Audit trail: record git reset for retry (FN-1404)
+          await audit.git({ type: "reset:hard", target: branch, metadata: { purpose: "merge-retry", attempt: attemptNum } });
        } catch { /* ignore cleanup errors */ }
        return false; // Allow retry
      }
@@ -972,10 +997,14 @@ export async function aiMergeTask(
  try {
    execSync(`git branch -d "${branch}"`, { cwd: rootDir, stdio: "pipe" });
    result.branchDeleted = true;
+    // Audit trail: record branch deletion (FN-1404)
+    await audit.git({ type: "branch:delete", target: branch });
  } catch {
    try {
      execSync(`git branch -D "${branch}"`, { cwd: rootDir, stdio: "pipe" });
      result.branchDeleted = true;
+      // Audit trail: record branch deletion (force) (FN-1404)
+      await audit.git({ type: "branch:delete", target: branch, metadata: { force: true } });
    } catch { /* non-fatal */ }
  }

@@ -994,6 +1023,8 @@ export async function aiMergeTask(
          cwd: rootDir,
          stdio: "pipe",
        });
+        // Audit trail: record worktree removal (FN-1404)
+        await audit.git({ type: "worktree:remove", target: worktreePath });
        result.worktreeRemoved = true;
      } catch { /* non-fatal */ }
    }
@@ -1008,6 +1039,18 @@ export async function aiMergeTask(
  }

  // 9. Move task to done
+  // Audit trail: record merge completion (FN-1404)
+  await audit.database({
+    type: "task:move",
+    target: taskId,
+    metadata: {
+      to: "done",
+      merged: true,
+      resolutionStrategy: result.resolutionStrategy,
+      resolutionMethod: result.resolutionMethod,
+      attemptsMade: result.attemptsMade,
+    },
+  });
  await completeTask(store, taskId, result);
  return result;
 }
--- a/packages/engine/src/run-audit.ts
+++ b/packages/engine/src/run-audit.ts
@@ -0,0 +1,242 @@
+/**
+ * Engine run-audit instrumentation helpers.
+ *
+ * Provides a shared layer for emitting run-audit events from heartbeat execution,
+ * task execution, and merge operations. Uses the core TaskStore APIs introduced
+ * by FN-1403 for event persistence.
+ *
+ * ## Run Context
+ *
+ * Every active run (heartbeat, executor, merger) has an associated run context
+ * that enables correlation of mutations back to the specific run that caused them:
+ *
+ * ```typescript
+ * interface EngineRunContext {
+ *   runId: string;           // Stable run identifier (heartbeat run ID, or synthetic for executor/merger)
+ *   agentId: string;          // Agent performing the mutation
+ *   taskId?: string;          // Task being operated on (if applicable)
+ *   phase?: string;           // Execution phase: "heartbeat", "execute", "merge-attempt-N"
+ *   source?: string;          // Invocation source: "timer", "on_demand", "assignment", etc.
+ * }
+ * ```
+ *
+ * ## Usage
+ *
+ * ```typescript
+ * // Create auditor with a run context (no-ops if context is null/undefined)
+ * const auditor = createRunAuditor(store, runContext);
+ *
+ * // Emit audit events for different mutation domains
+ * await auditor.git({ type: "branch:create", target: branchName });
+ * await auditor.database({ type: "task:update", target: taskId });
+ * await auditor.filesystem({ type: "file:write", target: filePath });
+ * ```
+ *
+ * ## Backward Compatibility
+ *
+ * All audit functions are no-ops when:
+ * - The auditor was created with a null/undefined context
+ * - The TaskStore doesn't have `recordRunAuditEvent` (not yet migrated)
+ *
+ * This ensures manual/non-run paths are unaffected by audit instrumentation.
+ */
+
+import type { TaskStore, RunAuditEventInput, RunAuditDomain } from "@fusion/core";
+
+/** Structured context for a run correlation ID. */
+export interface EngineRunContext {
+  /** Stable run identifier. For heartbeat runs, this is the AgentHeartbeatRun.id.
+   *  For executor/merger runs, this is a synthetic ID (e.g., "exec-{taskId}-{timestamp}" or "merge-{taskId}-{timestamp}"). */
+  runId: string;
+  /** Agent ID performing the mutation. */
+  agentId: string;
+  /** Task ID being operated on (if applicable). */
+  taskId?: string;
+  /** Execution phase for disambiguating sub-operations (e.g., "heartbeat", "execute", "merge-attempt-1"). */
+  phase?: string;
+  /** Invocation source for heartbeat runs (e.g., "timer", "on_demand", "assignment"). */
+  source?: string;
+}
+
+// ── Git mutation types ─────────────────────────────────────────────────────────
+
+export type GitMutationType =
+  | "worktree:create"
+  | "worktree:remove"
+  | "worktree:reuse"
+  | "branch:create"
+  | "branch:delete"
+  | "branch:checkout"
+  | "commit:create"
+  | "commit:amend"
+  | "reset:hard"
+  | "merge:start"
+  | "merge:resolve"
+  | "stash:push"
+  | "stash:pop";
+
+// ── Database mutation types ────────────────────────────────────────────────────
+
+export type DatabaseMutationType =
+  | "task:create"
+  | "task:update"
+  | "task:move"
+  | "task:log-entry"
+  | "task:comment:add"
+  | "task:steering-comment:add"
+  | "task:assign"
+  | "task:checkout"
+  | "task:release"
+  | "task:pause"
+  | "task:unpause"
+  | "task:dependency:add"
+  | "document:write"
+  | "workflow-step:result";
+
+// ── Filesystem mutation types ─────────────────────────────────────────────────
+
+export type FilesystemMutationType =
+  | "file:write"
+  | "file:delete"
+  | "file:capture-modified"
+  | "attachment:create"
+  | "attachment:delete"
+  | "prompt:write"
+  | "prompt:update"
+  | "session:write"
+  | "session:delete";
+
+/** Input for a git-domain audit event. */
+export interface GitAuditInput {
+  type: GitMutationType;
+  /** Target of the mutation (e.g., branch name, worktree path, commit SHA). */
+  target: string;
+  /** Optional structured metadata (e.g., { branch: "fusion/fn-001", from: "main" }). */
+  metadata?: Record<string, unknown>;
+}
+
+/** Input for a database-domain audit event. */
+export interface DatabaseAuditInput {
+  type: DatabaseMutationType;
+  /** Target of the mutation (e.g., task ID, document key). */
+  target: string;
+  /** Optional structured metadata. */
+  metadata?: Record<string, unknown>;
+}
+
+/** Input for a filesystem-domain audit event. */
+export interface FilesystemAuditInput {
+  type: FilesystemMutationType;
+  /** Target of the mutation (e.g., file path). */
+  target: string;
+  /** Optional structured metadata (e.g., { size: 1234, mimeType: "image/png" }). */
+  metadata?: Record<string, unknown>;
+}
+
+/** Interface for emitting run-audit events. */
+export interface RunAuditor {
+  /** Emit a git-domain audit event. No-op if no run context is available. */
+  git(input: GitAuditInput): Promise<void>;
+  /** Emit a database-domain audit event. No-op if no run context is available. */
+  database(input: DatabaseAuditInput): Promise<void>;
+  /** Emit a filesystem-domain audit event. No-op if no run context is available. */
+  filesystem(input: FilesystemAuditInput): Promise<void>;
+}
+
+/**
+ * Create a run auditor for a given run context.
+ *
+ * Returns an auditor that no-ops when:
+ * - `context` is null/undefined
+ * - The TaskStore doesn't expose `recordRunAuditEvent` (backward compatibility)
+ *
+ * @param store - TaskStore instance (must expose `recordRunAuditEvent`)
+ * @param context - Active run context, or null/undefined for non-run paths
+ */
+export function createRunAuditor(store: TaskStore, context: EngineRunContext | null | undefined): RunAuditor {
+  // No-op auditor for non-run paths
+  if (!context) {
+    return {
+      git: async () => { /* no-op */ },
+      database: async () => { /* no-op */ },
+      filesystem: async () => { /* no-op */ },
+    };
+  }
+
+  // Check if the store supports audit recording
+  const hasRecordAuditEvent = typeof store.recordRunAuditEvent === "function";
+
+  if (!hasRecordAuditEvent) {
+    // Store hasn't been migrated to FN-1403 yet — return no-op auditor
+    return {
+      git: async () => { /* no-op */ },
+      database: async () => { /* no-op */ },
+      filesystem: async () => { /* no-op */ },
+    };
+  }
+
+  return {
+    git: async (input: GitAuditInput) => {
+      const eventInput: RunAuditEventInput = {
+        taskId: context.taskId,
+        agentId: context.agentId,
+        runId: context.runId,
+        domain: "git",
+        mutationType: input.type,
+        target: input.target,
+        metadata: {
+          phase: context.phase,
+          ...(context.source ? { source: context.source } : {}),
+          ...input.metadata,
+        },
+      };
+      await store.recordRunAuditEvent(eventInput);
+    },
+
+    database: async (input: DatabaseAuditInput) => {
+      const eventInput: RunAuditEventInput = {
+        taskId: input.target.startsWith("FN-") || input.target.startsWith("KB-") ? input.target : context.taskId,
+        agentId: context.agentId,
+        runId: context.runId,
+        domain: "database",
+        mutationType: input.type,
+        target: input.target,
+        metadata: {
+          phase: context.phase,
+          ...(context.source ? { source: context.source } : {}),
+          ...input.metadata,
+        },
+      };
+      await store.recordRunAuditEvent(eventInput);
+    },
+
+    filesystem: async (input: FilesystemAuditInput) => {
+      const eventInput: RunAuditEventInput = {
+        taskId: context.taskId,
+        agentId: context.agentId,
+        runId: context.runId,
+        domain: "filesystem",
+        mutationType: input.type,
+        target: input.target,
+        metadata: {
+          phase: context.phase,
+          ...(context.source ? { source: context.source } : {}),
+          ...input.metadata,
+        },
+      };
+      await store.recordRunAuditEvent(eventInput);
+    },
+  };
+}
+
+/**
+ * Generate a synthetic run ID for executor/merger runs that don't use AgentHeartbeatRun.
+ *
+ * Format: "{prefix}-{taskId}-{timestamp}-{random4chars}"
+ * Example: "exec-FN-001-1712345678-a1b2"
+ */
+export function generateSyntheticRunId(prefix: string, taskId: string): string {
+  const timestamp = Date.now();
+  const random = Math.random().toString(36).slice(2, 6);
+  return `${prefix}-${taskId}-${timestamp}-${random}`;
+}