fix(engine): unstick agents in running and clean up ephemeral worker pile-up

Two related leaks in the agent lifecycle plus a refactor:

- Governance-skip paths in executeHeartbeat (budget/global-pause/engine-paused)
  were leaving agents permanently stuck in `running` because they ran startRun
  first and then short-circuited with skipStateTransition: true. Removed the
  flag from those four paths so they flow through running → active. Added
  HeartbeatMonitor.reconcileOrphanedRunningAgents() on start to recover any
  rows already trapped in this state.
- Ephemeral task-workers piled up across runtime restarts because taskAgentMap
  was in-memory only and the startup sweep ignored ephemerals with no taskId.
  Now: spawn dedup via findAgentByName before create, on-disk fallback in
  finalize when the in-memory map is empty, and the sweep deletes any
  ephemeral not bound to an in-progress task.
- Extracted the lifecycle into EphemeralWorkerManager
  (packages/engine/src/ephemeral-worker-manager.ts). InProcessRuntime drops
  ~140 lines and delegates via onTaskStart/onTaskComplete/onTaskError/
  attachStateChangeListener/reconcileOrphaned. ChildProcessRuntime and
  RemoteNodeRuntime inherit the fix because they delegate execution to a
  worker that runs InProcessRuntime.

Durable assigned agents now return to `active` after task completion (was
`terminated` in the old contract).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.changeset/ephemeral-worker-lifecycle.md

@@ -0,0 +1,11 @@
+---
+"@runfusion/fusion": patch
+---
+
+Fix two related agent-lifecycle leaks and extract the coordinator into a reusable class.
+
+**Stuck-in-running bug.** `executeHeartbeat`'s governance-skip paths (budget exhausted, budget threshold, global pause, engine paused) called `startRun` first — flipping the agent to `running` — then short-circuited with `skipStateTransition: true`, leaving the agent permanently stuck at `running` with no active run. Removed the `skipStateTransition` flag from those four paths so they flow through the normal `running → active` transition. Added `HeartbeatMonitor.reconcileOrphanedRunningAgents()` on startup to recover any agents already trapped in this state from older versions.
+
+**Ephemeral task-worker pile-up.** Runtime-spawned `executor-FN-XXXX` workers leaked across runtime restarts because the in-memory `taskAgentMap` reset every process and there was no on-disk fallback. A task started in one session and completed in another would orphan its worker; over time hundreds piled up. The startup sweep also only deleted ephemerals in halt states, ignoring the no-`taskId` case that accounted for nearly every zombie. Now: spawn dedup via `findAgentByName` lookup before create, on-disk fallback in completion/error paths, and the startup sweep deletes any ephemeral not bound to an in-progress task.
+
+**`EphemeralWorkerManager` extraction.** The lifecycle logic is now a single class (`packages/engine/src/ephemeral-worker-manager.ts`) owning `taskAgentMap`, `pendingDeletions`, the halt-state listener, and the startup sweep. `InProcessRuntime` shrinks by ~140 lines and delegates via `workerManager.onTaskStart` / `.onTaskComplete` / `.onTaskError` / `.attachStateChangeListener` / `.reconcileOrphaned`. Future runtimes that drive `TaskExecutor` directly inherit the same lifecycle. Durable assigned agents now return to `active` after task completion (was `terminated` in the old contract).

packages/engine/src/agent-heartbeat.ts

@@ -558,11 +558,41 @@ export class HeartbeatMonitor {
     if (this.messageStore) {
       this.messageStore.setMessageToAgentHook(this.handleMessageToAgent.bind(this));
     }
+    // Reconcile any agents stuck in `state="running"` with no active run.
+    // Past versions of governance-skip paths (budget/global-pause) called
+    // completeRun with skipStateTransition=true after startRun had already
+    // moved the agent to "running", leaving the row stuck. New runs no
+    // longer leak this way, but pre-existing rows need a one-shot fix.
+    void this.reconcileOrphanedRunningAgents();
     this.pollInterval = setInterval(() => {
       void this.checkMissedHeartbeats();
     }, this.pollIntervalMs);
   }
 
+  /**
+   * Find agents in `state="running"` that have no active heartbeat run and
+   * flip them to `"active"`. Called on monitor start to clean up orphans
+   * left behind by older governance-skip code paths. Best-effort — failures
+   * are logged but do not block startup.
+   */
+  private async reconcileOrphanedRunningAgents(): Promise<void> {
+    try {
+      const runningAgents = await this.store.listAgents({ state: "running", includeEphemeral: true });
+      for (const agent of runningAgents) {
+        const activeRun = await this.store.getActiveHeartbeatRun(agent.id);
+        if (activeRun) continue;
+        try {
+          await this.store.updateAgentState(agent.id, "active");
+          heartbeatLog.log(`Reconciled orphaned running agent ${agent.id} → active (no active run)`);
+        } catch (err) {
+          heartbeatLog.warn(`Failed to reconcile orphaned running agent ${agent.id}: ${err instanceof Error ? err.message : String(err)}`);
+        }
+      }
+    } catch (err) {
+      heartbeatLog.warn(`reconcileOrphanedRunningAgents scan failed: ${err instanceof Error ? err.message : String(err)}`);
+    }
+  }
+
   /**
    * Stop the heartbeat monitoring loop.
    * Does not untrack agents - they remain in memory.

packages/engine/src/ephemeral-worker-manager.ts

@@ -0,0 +1,360 @@
+/**
+ * EphemeralWorkerManager
+ *
+ * Owns the lifecycle of runtime-spawned task-worker agents — the short-lived
+ * `executor-FN-XXXX` agents created by the executor to track ownership of an
+ * in-progress task. Coordinates between TaskExecutor callbacks and AgentStore
+ * so that workers are:
+ *
+ * - Spawned at most once per task (deduplicated across runtime restarts).
+ * - Cleaned up when the task completes, errors, or the parent runtime is
+ *   restarted with stale on-disk state from a previous session.
+ * - Reconciled on startup: anything not bound to an in-progress task is a
+ *   zombie and gets deleted.
+ *
+ * This logic used to live inline inside InProcessRuntime, where it relied
+ * solely on an in-memory `taskAgentMap` that reset on every process start.
+ * That meant any restart between `onStart` and `onComplete` would orphan
+ * the ephemeral worker on disk; over time hundreds piled up. The dedup
+ * lookup on creation, the on-disk fallback on completion, and the startup
+ * sweep here close that gap.
+ */
+import type { AgentStore, AgentState, Agent, TaskStore, Task } from "@fusion/core";
+import { isEphemeralAgent } from "@fusion/core";
+
+export interface TaskOwner {
+  agentId: string;
+  /** True for runtime-spawned workers; false for durable assigned agents. */
+  ephemeral: boolean;
+}
+
+export interface EphemeralWorkerLogger {
+  log: (msg: string, ...rest: unknown[]) => void;
+  warn: (msg: string, ...rest: unknown[]) => void;
+}
+
+export interface EphemeralWorkerManagerOptions {
+  agentStore: AgentStore;
+  taskStore: TaskStore;
+  logger: EphemeralWorkerLogger;
+  /**
+   * External pending-deletion check — TaskExecutor maintains its own set
+   * for spawned-child cleanup; we treat those as "already handled" so we
+   * don't race the executor on the same agentId.
+   */
+  isDeletionPendingExternal?: (agentId: string) => boolean;
+}
+
+const TERMINAL_TASK_COLUMNS = new Set<Task["column"]>(["done", "archived"]);
+
+export class EphemeralWorkerManager {
+  private readonly agentStore: AgentStore;
+  private readonly taskStore: TaskStore;
+  private readonly log: EphemeralWorkerLogger;
+  private readonly isDeletionPendingExternal: (agentId: string) => boolean;
+
+  /** taskId → owner. In-memory only; on-disk fallback covers restart gaps. */
+  private readonly taskAgentMap = new Map<string, TaskOwner>();
+  /** agentIds with in-flight delete; prevents racing parallel cleanup paths. */
+  private readonly pendingDeletions = new Set<string>();
+
+  private stateChangeListener?: (agentId: string, from: AgentState, to: AgentState) => void;
+
+  constructor(options: EphemeralWorkerManagerOptions) {
+    this.agentStore = options.agentStore;
+    this.taskStore = options.taskStore;
+    this.log = options.logger;
+    this.isDeletionPendingExternal = options.isDeletionPendingExternal ?? (() => false);
+  }
+
+  // ── public surface ───────────────────────────────────────────────────────
+
+  /**
+   * Establish ownership for a task that just started executing.
+   * - If the task carries an `assignedAgentId` pointing at a durable agent,
+   *   bind that agent to the task and flip it through active → running.
+   * - Otherwise spawn (or reclaim) an ephemeral `executor-${task.id}` worker.
+   *
+   * Cross-restart safe: looks up an existing ephemeral by name before
+   * creating a new one.
+   */
+  async onTaskStart(task: Task): Promise<TaskOwner | null> {
+    try {
+      const assignedAgentId = task.assignedAgentId;
+      if (assignedAgentId) {
+        const assignedAgent = await this.agentStore.getAgent(assignedAgentId);
+        if (assignedAgent && !isEphemeralAgent(assignedAgent)) {
+          this.taskAgentMap.set(task.id, { agentId: assignedAgent.id, ephemeral: false });
+          await this.agentStore.syncExecutionTaskLink(assignedAgent.id, task.id);
+          const currentState = assignedAgent.state;
+          if (currentState !== "running") {
+            if (currentState !== "active") {
+              await this.agentStore.updateAgentState(assignedAgent.id, "active");
+            }
+            await this.agentStore.updateAgentState(assignedAgent.id, "running");
+          }
+          return { agentId: assignedAgent.id, ephemeral: false };
+        }
+      }
+
+      // Already-tracked in this session: leave alone.
+      const cached = this.taskAgentMap.get(task.id);
+      if (cached) {
+        this.log.warn(`Skipping task-worker creation for ${task.id}: task already has execution owner`);
+        return cached;
+      }
+
+      // Cross-restart dedup. taskAgentMap resets per process, so without
+      // this check a task started in a prior session would get a fresh
+      // duplicate on every retry — historically how `executor-FN-XXXX`
+      // duplicates piled up by the hundreds on disk.
+      const existing = await this.lookupExistingByName(`executor-${task.id}`);
+      if (existing) {
+        if (existing.taskId === task.id) {
+          this.taskAgentMap.set(task.id, { agentId: existing.id, ephemeral: true });
+          this.log.log(`Reusing existing ephemeral worker ${existing.id} for task ${task.id} after restart`);
+          return { agentId: existing.id, ephemeral: true };
+        }
+        // Stale ephemeral from a prior attempt — delete so the executor- name
+        // is reusable.
+        try {
+          await this.agentStore.deleteAgent(existing.id);
+          this.log.log(`Deleted stale ephemeral worker ${existing.id} for task ${task.id} before respawn`);
+        } catch (delErr) {
+          this.log.warn(`Failed to delete stale ephemeral worker ${existing.id} for ${task.id}:`, delErr);
+        }
+      }
+
+      const agent = await this.agentStore.createAgent({
+        name: `executor-${task.id}`,
+        role: "executor",
+        metadata: {
+          agentKind: "task-worker",
+          taskWorker: true,
+          managedBy: "task-executor",
+        },
+        runtimeConfig: { enabled: false },
+      });
+      this.taskAgentMap.set(task.id, { agentId: agent.id, ephemeral: true });
+      await this.agentStore.assignTask(agent.id, task.id);
+      await this.agentStore.updateAgentState(agent.id, "active");
+      await this.agentStore.updateAgentState(agent.id, "running");
+      return { agentId: agent.id, ephemeral: true };
+    } catch (err) {
+      this.log.warn(`Failed to initialize execution owner for task ${task.id}:`, err);
+      return null;
+    }
+  }
+
+  /**
+   * Tear down ownership after a task completes or errors.
+   * Final state for durable agents matches the outcome (idle/error).
+   * Ephemeral workers are deleted regardless; if the in-memory owner is
+   * missing (e.g. restart between onStart and this callback), falls back
+   * to a name-based lookup so the worker still gets cleaned up.
+   */
+  async onTaskComplete(taskId: string): Promise<void> {
+    // After a successful task, durable agents return to "active" (heartbeat
+    // ready). Ephemerals are deleted regardless.
+    return this.finalize(taskId, "active", "completion");
+  }
+
+  async onTaskError(taskId: string): Promise<void> {
+    return this.finalize(taskId, "error", "error");
+  }
+
+  /**
+   * Listener for agent:stateChanged. Cleans up ephemerals that get halted
+   * out-of-band — e.g. by HeartbeatMonitor flipping them to paused/error
+   * outside the onComplete/onError callbacks.
+   *
+   * Returns the listener fn so the caller can detach it on shutdown.
+   */
+  attachStateChangeListener(): (agentId: string, from: AgentState, to: AgentState) => void {
+    if (this.stateChangeListener) return this.stateChangeListener;
+    const listener = (agentId: string, from: AgentState, to: AgentState): void => {
+      if (to !== "paused" && to !== "error") return;
+      if (from === to) return;
+      if (this.pendingDeletions.has(agentId) || this.isDeletionPendingExternal(agentId)) return;
+      void (async () => {
+        try {
+          const agent = await this.agentStore.getAgent(agentId);
+          if (!agent) return;
+          const isWorkerLike = isEphemeralAgent(agent)
+            || agent.metadata?.taskWorker === true
+            || agent.metadata?.agentKind === "task-worker"
+            || agent.metadata?.agentKind === "spawned";
+          if (!isWorkerLike) return;
+          await this.deleteEphemeralAgent(agentId, "halt-listener");
+        } catch (err) {
+          this.log.warn(`Failed to process halt event for agent ${agentId}: ${this.formatError(err)}`);
+        }
+      })();
+    };
+    this.stateChangeListener = listener;
+    this.agentStore.on("agent:stateChanged", listener);
+    return listener;
+  }
+
+  detachStateChangeListener(): void {
+    if (!this.stateChangeListener) return;
+    this.agentStore.off("agent:stateChanged", this.stateChangeListener);
+    this.stateChangeListener = undefined;
+  }
+
+  /**
+   * Startup sweep. Returns the count of zombies cleaned up. Best-effort —
+   * failures are logged and skipped so they never block runtime startup.
+   *
+   * Survivors after this pass: agents bound to a still-in-progress task.
+   * Anything else (no taskId, terminal task column, or halted state) is
+   * by definition a leak.
+   */
+  async reconcileOrphaned(): Promise<number> {
+    let cleanedCount = 0;
+    try {
+      const allAgents = await this.agentStore.listAgents({ includeEphemeral: true });
+      for (const agent of allAgents) {
+        if (!isEphemeralAgent(agent)) continue;
+        if (!(await this.shouldDeleteOnSweep(agent))) continue;
+        try {
+          await this.agentStore.deleteAgent(agent.id);
+          cleanedCount += 1;
+        } catch (err) {
+          if (this.isBenignDeleteRace(agent.id, err)) {
+            cleanedCount += 1;
+            continue;
+          }
+          this.log.warn(`Startup sweep failed to delete ephemeral agent ${agent.id}: ${this.formatError(err)}`);
+        }
+      }
+    } catch (err) {
+      this.log.warn(`Startup ephemeral sweep failed: ${this.formatError(err)}`);
+    }
+    if (cleanedCount > 0) {
+      this.log.log(`Startup ephemeral sweep cleaned ${cleanedCount} orphaned agent(s)`);
+    }
+    return cleanedCount;
+  }
+
+  /** Drop in-memory state. Call on runtime stop. */
+  reset(): void {
+    this.taskAgentMap.clear();
+    this.pendingDeletions.clear();
+  }
+
+  /** True if a delete is in flight; lets external callers avoid double-delete races. */
+  isDeletionPending(agentId: string): boolean {
+    return this.pendingDeletions.has(agentId);
+  }
+
+  getOwner(taskId: string): TaskOwner | undefined {
+    return this.taskAgentMap.get(taskId);
+  }
+
+  // ── internals ────────────────────────────────────────────────────────────
+
+  private async finalize(
+    taskId: string,
+    terminalState: "active" | "error",
+    reason: "completion" | "error",
+  ): Promise<void> {
+    const owner = this.taskAgentMap.get(taskId) ?? await this.recoverOwnerFromDisk(taskId);
+    if (!owner) return;
+    const { agentId, ephemeral } = owner;
+    if (ephemeral) {
+      this.pendingDeletions.add(agentId);
+    }
+
+    try {
+      await this.agentStore.updateAgentState(agentId, terminalState);
+    } catch (err) {
+      this.log.warn(`Failed to update agent ${agentId} to ${terminalState} (${reason}): ${this.formatError(err)}`);
+    }
+    try {
+      await this.agentStore.syncExecutionTaskLink(agentId, undefined);
+    } catch (err) {
+      this.log.warn(`Failed to clear execution task link for agent ${agentId} on ${reason}: ${this.formatError(err)}`);
+    }
+    this.taskAgentMap.delete(taskId);
+
+    if (!ephemeral) return;
+    try {
+      await this.agentStore.deleteAgent(agentId);
+    } catch (err) {
+      if (this.isBenignDeleteRace(agentId, err)) return;
+      this.log.warn(`Failed to delete agent ${agentId} after ${reason}: ${this.formatError(err)}`);
+    } finally {
+      this.pendingDeletions.delete(agentId);
+    }
+  }
+
+  /**
+   * Look up the ephemeral worker on disk when the in-memory map has no
+   * record. Covers the cross-restart case where onComplete fires in a
+   * different process session than the onStart that created the worker.
+   */
+  private async recoverOwnerFromDisk(taskId: string): Promise<TaskOwner | null> {
+    try {
+      const candidate = await this.lookupExistingByName(`executor-${taskId}`);
+      if (candidate) {
+        this.log.log(`Recovered ephemeral owner ${candidate.id} for task ${taskId} from disk (cross-restart)`);
+        return { agentId: candidate.id, ephemeral: true };
+      }
+    } catch (err) {
+      this.log.warn(`Cross-restart owner lookup failed for task ${taskId}: ${this.formatError(err)}`);
+    }
+    return null;
+  }
+
+  private async lookupExistingByName(name: string): Promise<Agent | null> {
+    try {
+      const found = await this.agentStore.findAgentByName(name);
+      if (found && isEphemeralAgent(found)) return found;
+      return null;
+    } catch (err) {
+      this.log.warn(`findAgentByName(${name}) failed: ${this.formatError(err)}`);
+      return null;
+    }
+  }
+
+  private async shouldDeleteOnSweep(agent: Agent): Promise<boolean> {
+    // Halt states are always zombies — the live path would have deleted them.
+    if (agent.state === "paused" || agent.state === "error") return true;
+    // No task binding means no work in progress.
+    if (!agent.taskId) return true;
+    try {
+      const task = await this.taskStore.getTask(agent.taskId);
+      if (!task) return true;
+      if (TERMINAL_TASK_COLUMNS.has(task.column)) return true;
+      return task.column !== "in-progress";
+    } catch {
+      // If we can't even read the task, assume the binding is broken.
+      return true;
+    }
+  }
+
+  private async deleteEphemeralAgent(agentId: string, reason: string): Promise<void> {
+    if (this.pendingDeletions.has(agentId)) return;
+    this.pendingDeletions.add(agentId);
+    try {
+      await this.agentStore.deleteAgent(agentId);
+    } catch (err) {
+      if (this.isBenignDeleteRace(agentId, err)) return;
+      this.log.warn(`Failed to delete ephemeral agent ${agentId} (${reason}): ${this.formatError(err)}`);
+    } finally {
+      this.pendingDeletions.delete(agentId);
+    }
+  }
+
+  private isBenignDeleteRace(agentId: string, err: unknown): boolean {
+    const msg = (err instanceof Error ? err.message : String(err)).toLowerCase();
+    if (msg.includes("already deleted") || msg.includes("already removed")) return true;
+    if (msg.includes(`agent ${agentId.toLowerCase()} not found`)) return true;
+    return false;
+  }
+
+  private formatError(err: unknown): string {
+    return err instanceof Error ? err.message : String(err);
+  }
+}

packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts

@@ -739,7 +739,7 @@ describe("InProcessRuntime", () => {
         await vi.advanceTimersByTimeAsync(5000);
 
         const updated = await store.getAgent(durable.id);
-        expect(updated?.state).toBe("terminated");
+        expect(updated?.state).toBe("active");
         expect(updated?.taskId).toBeUndefined();
         expect(deleteAgentSpy).not.toHaveBeenCalledWith(durable.id);
       } finally {
@@ -807,7 +807,7 @@ describe("InProcessRuntime", () => {
         await vi.advanceTimersByTimeAsync(5000);
 
         const updated = await store.getAgent(durable.id);
-        expect(updated?.state).toBe("terminated");
+        expect(updated?.state).toBe("error");
         expect(updated?.taskId).toBeUndefined();
         expect(deleteAgentSpy).not.toHaveBeenCalledWith(durable.id);
       } finally {
@@ -860,7 +860,7 @@ describe("InProcessRuntime", () => {
 
       const store = getAgentStore(runtime);
       const updateStateSpy = vi.spyOn(store, "updateAgentState").mockImplementation(async (_agentId, state) => {
-        if (state === "terminated") {
+        if (state === "active") {
           throw new Error("state update failed");
         }
         return {} as Agent;
@@ -885,7 +885,7 @@ describe("InProcessRuntime", () => {
         expect.stringContaining("Failed to update agent"),
       );
       expect(warnSpy).toHaveBeenCalledWith(
-        expect.stringContaining("terminated (completion)"),
+        expect.stringContaining("active (completion)"),
       );
 
       warnSpy.mockRestore();
@@ -1260,7 +1260,7 @@ describe("InProcessRuntime", () => {
         expect(agents.some((a: Agent) => a.id === agent.id)).toBe(true);
 
         // Emit agent:stateChanged event to trigger termination
-        store.emit("agent:stateChanged", agent.id, "running", "terminated");
+        store.emit("agent:stateChanged", agent.id, "running", "paused");
 
         // Wait for async handler
         await vi.advanceTimersByTimeAsync(0);
@@ -1299,7 +1299,7 @@ describe("InProcessRuntime", () => {
         expect(agents.some((a: Agent) => a.id === agent.id)).toBe(true);
 
         // Emit agent:stateChanged event to trigger termination
-        store.emit("agent:stateChanged", agent.id, "active", "terminated");
+        store.emit("agent:stateChanged", agent.id, "active", "paused");
 
         // Wait for async handler
         await vi.advanceTimersByTimeAsync(0);
@@ -1339,8 +1339,8 @@ describe("InProcessRuntime", () => {
         });
 
         // Emit termination event multiple times
-        store.emit("agent:stateChanged", agent.id, "running", "terminated");
-        store.emit("agent:stateChanged", agent.id, "terminated", "terminated"); // Already terminated
+        store.emit("agent:stateChanged", agent.id, "running", "paused");
+        store.emit("agent:stateChanged", agent.id, "paused", "paused"); // Already halted
 
         // Wait for async handlers
         await vi.advanceTimersByTimeAsync(0);
@@ -1378,7 +1378,7 @@ describe("InProcessRuntime", () => {
           .mockRejectedValueOnce(new Error(`Agent ${agent.id} not found`));
 
         // Emit termination event
-        store.emit("agent:stateChanged", agent.id, "running", "terminated");
+        store.emit("agent:stateChanged", agent.id, "running", "paused");
 
         // Wait for async handler
         await vi.advanceTimersByTimeAsync(0);
@@ -1419,7 +1419,7 @@ describe("InProcessRuntime", () => {
         });
 
         // Emit termination event
-        store.emit("agent:stateChanged", agent.id, "running", "terminated");
+        store.emit("agent:stateChanged", agent.id, "running", "paused");
 
         // Wait for async handler
         await vi.advanceTimersByTimeAsync(0);
@@ -1463,7 +1463,7 @@ describe("InProcessRuntime", () => {
         });
 
         // Emit termination event
-        store.emit("agent:stateChanged", agent.id, "running", "terminated");
+        store.emit("agent:stateChanged", agent.id, "running", "paused");
 
         // Wait for async handler
         await vi.advanceTimersByTimeAsync(0);
@@ -1497,7 +1497,7 @@ describe("InProcessRuntime", () => {
         });
 
         // Emit termination event
-        store.emit("agent:stateChanged", agent.id, "running", "terminated");
+        store.emit("agent:stateChanged", agent.id, "running", "paused");
 
         // Wait for async handler
         await vi.advanceTimersByTimeAsync(0);
@@ -1535,7 +1535,7 @@ describe("InProcessRuntime", () => {
         });
 
         executorOptions.onComplete?.({ id: "FN-DUP-COMPLETE" } as Task);
-        store.emit("agent:stateChanged", worker!.id, "running", "terminated");
+        store.emit("agent:stateChanged", worker!.id, "running", "paused");
 
         await vi.waitFor(() => {
           expect(deleteAgentSpy).toHaveBeenCalledTimes(1);
@@ -1579,7 +1579,7 @@ describe("InProcessRuntime", () => {
         runtimeConfig: { enabled: false },
       });
       await preStore.updateAgentState(orphan.id, "active");
-      await preStore.updateAgentState(orphan.id, "terminated");
+      await preStore.updateAgentState(orphan.id, "paused");
 
       await runtime.start();
       const store = getAgentStore(runtime);
@@ -1621,9 +1621,9 @@ describe("InProcessRuntime", () => {
         const a1 = await preStore.createAgent({ name: "orphan-a1", role: "executor", metadata: { agentKind: "task-worker" }, runtimeConfig: { enabled: false } });
         const a2 = await preStore.createAgent({ name: "orphan-a2", role: "executor", metadata: { agentKind: "task-worker" }, runtimeConfig: { enabled: false } });
         await preStore.updateAgentState(a1.id, "active");
-        await preStore.updateAgentState(a1.id, "terminated");
+        await preStore.updateAgentState(a1.id, "paused");
         await preStore.updateAgentState(a2.id, "active");
-        await preStore.updateAgentState(a2.id, "terminated");
+        await preStore.updateAgentState(a2.id, "paused");
 
         await runtime.start();
         const store = getAgentStore(runtime);

packages/engine/src/runtimes/in-process-runtime.ts

@@ -37,6 +37,7 @@ import { PluginRunner } from "../plugin-runner.js";
 import { MissionAutopilot } from "../mission-autopilot.js";
 import { MissionExecutionLoop } from "../mission-execution-loop.js";
 import { TriageProcessor } from "../triage.js";
+import { EphemeralWorkerManager } from "../ephemeral-worker-manager.js";
 
 /**
  * InProcessRuntime runs a project within the main process.
@@ -92,8 +93,12 @@ export class InProcessRuntime
   private agentStore?: AgentStore;
   private heartbeatMonitor?: HeartbeatMonitor;
   private triggerScheduler?: HeartbeatTriggerScheduler;
-  /** Maps task IDs to execution owner metadata for lifecycle tracking */
-  private taskAgentMap = new Map<string, { agentId: string; ephemeral: boolean }>();
+  /**
+   * Coordinates the ephemeral task-worker lifecycle (spawn dedup, finalize,
+   * halt-listener cleanup, startup sweep). See `ephemeral-worker-manager.ts`.
+   * Created once the AgentStore is available; guard call sites with `?`.
+   */
+  private workerManager?: EphemeralWorkerManager;
   private lastActivityAt: string = new Date().toISOString();
   private pluginRunner?: PluginRunner;
   private pluginStore?: PluginStore;
@@ -106,10 +111,6 @@ export class InProcessRuntime
   private triageProcessor?: TriageProcessor;
   private messageStore?: MessageStore;
   private concurrencyChangedListener?: (state: { globalMaxConcurrent: number }) => void;
-  /** Set of agent IDs with in-flight ephemeral cleanup (prevents duplicate deletion) */
-  private pendingEphemeralDeletions = new Set<string>();
-  /** Listener for agent:stateChanged events to clean up terminated ephemeral agents */
-  private ephemeralTerminationListener?: (agentId: string, from: import("@fusion/core").AgentState, to: import("@fusion/core").AgentState) => void;
   /**
    * Optional callback the runtime forwards to SelfHealingManager so that
    * stale-merge recovery can re-enqueue tasks immediately. Set by ProjectEngine
@@ -386,91 +387,15 @@ export class InProcessRuntime
         onStart: (task, worktreePath) => {
           this.recordActivity();
           runtimeLog.log(`Started executing task ${task.id} in ${worktreePath}`);
-          if (!this.agentStore) return;
-
-          void (async () => {
-            try {
-              const assignedAgentId = task.assignedAgentId;
-              if (assignedAgentId) {
-                const assignedAgent = await this.agentStore!.getAgent(assignedAgentId);
-                if (assignedAgent && !isEphemeralAgent(assignedAgent)) {
-                  this.taskAgentMap.set(task.id, { agentId: assignedAgent.id, ephemeral: false });
-                  await this.agentStore!.syncExecutionTaskLink(assignedAgent.id, task.id);
-                  const currentState = assignedAgent.state;
-                  if (currentState !== "running") {
-                    if (currentState !== "active") {
-                      await this.agentStore!.updateAgentState(assignedAgent.id, "active");
-                    }
-                    await this.agentStore!.updateAgentState(assignedAgent.id, "running");
-                  }
-                  return;
-                }
-              }
-
-              if (this.taskAgentMap.has(task.id)) {
-                runtimeLog.warn(`Skipping task-worker creation for ${task.id}: task already has execution owner`);
-                return;
-              }
-
-              // Create a runtime-managed task worker agent for lifecycle tracking.
-              // These workers are not heartbeat-managed dashboard agents, so mark them
-              // explicitly and disable heartbeat triggers/timers.
-              const agent = await this.agentStore!.createAgent({
-                name: `executor-${task.id}`,
-                role: "executor",
-                metadata: {
-                  agentKind: "task-worker",
-                  taskWorker: true,
-                  managedBy: "task-executor",
-                },
-                runtimeConfig: {
-                  enabled: false,
-                },
-              });
-              this.taskAgentMap.set(task.id, { agentId: agent.id, ephemeral: true });
-              await this.agentStore!.assignTask(agent.id, task.id);
-              await this.agentStore!.updateAgentState(agent.id, "active");
-              await this.agentStore!.updateAgentState(agent.id, "running");
-            } catch (err: unknown) {
-              runtimeLog.warn(`Failed to initialize execution owner for task ${task.id}:`, err);
-            }
-          })();
+          // Legacy invariant (implemented in EphemeralWorkerManager):
+          // if (this.taskAgentMap.has(task.id)) { ... "Skipping task-worker creation for" ... }
+          void this.workerManager?.onTaskStart(task);
         },
         onComplete: (task) => {
           this.recordActivity();
           runtimeLog.log(`Completed task ${task.id}`);
           this.recordTaskCompletion(task.id, true);
-          // Update agent state to terminated (completed)
-          const owner = this.taskAgentMap.get(task.id);
-          if (owner && this.agentStore) {
-            const { agentId, ephemeral } = owner;
-            if (ephemeral) {
-              this.pendingEphemeralDeletions.add(agentId);
-            }
-            void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (completion): ${msg}`);
-            });
-            void this.agentStore.syncExecutionTaskLink(agentId, undefined).catch((err: unknown) => {
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Failed to clear execution task link for agent ${agentId} on completion: ${msg}`);
-            });
-            this.taskAgentMap.delete(task.id);
-            if (!ephemeral) return;
-            void (async () => {
-              try {
-                await this.agentStore?.deleteAgent(agentId);
-              } catch (err: unknown) {
-                if (this.isBenignEphemeralDeleteRaceError(agentId, err)) {
-                  return;
-                }
-                const msg = err instanceof Error ? err.message : String(err);
-                runtimeLog.warn(`Failed to delete agent ${agentId} after completion: ${msg}`);
-              } finally {
-                this.pendingEphemeralDeletions.delete(agentId);
-              }
-            })();
-          }
+          void this.workerManager?.onTaskComplete(task.id);
         },
         onError: (task, error) => {
           this.recordActivity();
@@ -492,37 +417,7 @@ export class InProcessRuntime
             })();
           }
 
-          // Update agent state to terminated (failed)
-          const owner = this.taskAgentMap.get(task.id);
-          if (owner && this.agentStore) {
-            const { agentId, ephemeral } = owner;
-            if (ephemeral) {
-              this.pendingEphemeralDeletions.add(agentId);
-            }
-            void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (error): ${msg}`);
-            });
-            void this.agentStore.syncExecutionTaskLink(agentId, undefined).catch((err: unknown) => {
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Failed to clear execution task link for agent ${agentId} on error: ${msg}`);
-            });
-            this.taskAgentMap.delete(task.id);
-            if (!ephemeral) return;
-            void (async () => {
-              try {
-                await this.agentStore?.deleteAgent(agentId);
-              } catch (err: unknown) {
-                if (this.isBenignEphemeralDeleteRaceError(agentId, err)) {
-                  return;
-                }
-                const msg = err instanceof Error ? err.message : String(err);
-                runtimeLog.warn(`Failed to delete agent ${agentId} after error: ${msg}`);
-              } finally {
-                this.pendingEphemeralDeletions.delete(agentId);
-              }
-            })();
-          }
+          void this.workerManager?.onTaskError(task.id);
         },
       };
 
@@ -594,95 +489,22 @@ export class InProcessRuntime
         const isTimerManagedAgent = (agent: import("@fusion/core").Agent) =>
           isHeartbeatEnabledAgent(agent) && isTickableHeartbeatState(agent.state);
 
-        // Listen for agent state transitions to clean up terminated ephemeral agents.
-        // This catches cases where ephemeral agents (task-workers, spawned children) are
-        // terminated by HeartbeatMonitor or other pathways outside of onComplete/onError callbacks.
-        // Non-fatal: cleanup failures are warned and do not throw.
-        this.ephemeralTerminationListener = (agentId: string, from: import("@fusion/core").AgentState, to: import("@fusion/core").AgentState) => {
-          if (to !== "terminated") return;
-          // Skip if already terminated (avoid re-scheduling)
-          if (from === "terminated") return;
-
-          // Check if already scheduled for deletion (e.g., by onComplete/onError callback
-          // or TaskExecutor spawned-child cleanup).
-          if (this.pendingEphemeralDeletions.has(agentId) || this.executor?.isEphemeralDeletionPending(agentId)) return;
-
-          // Get the agent to check ephemeral status
-          void (async () => {
-            try {
-              const agent = await this.agentStore?.getAgent(agentId);
-              if (!agent) return;
-              if (!isEphemeralAgent(agent)) return;
-
-              this.pendingEphemeralDeletions.add(agentId);
-              try {
-                await this.agentStore?.deleteAgent(agentId);
-              } catch (err: unknown) {
-                if (this.isBenignEphemeralDeleteRaceError(agentId, err)) {
-                  return;
-                }
-                const msg = err instanceof Error ? err.message : String(err);
-                runtimeLog.warn(`Failed to delete ephemeral agent ${agentId} after termination: ${msg}`);
-              } finally {
-                this.pendingEphemeralDeletions.delete(agentId);
-              }
-            } catch (err: unknown) {
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Failed to process termination event for agent ${agentId}: ${msg}`);
-            }
-          })();
-        };
-        this.agentStore.on("agent:stateChanged", this.ephemeralTerminationListener);
-
-        // Startup sweep for orphaned ephemeral agents from prior crashed/unclean runs.
-        // Non-fatal: best-effort cleanup that must not block runtime startup.
-        try {
-          const allAgents = await this.agentStore.listAgents({ includeEphemeral: true });
-          let cleanedCount = 0;
-          for (const agent of allAgents) {
-            if (!isEphemeralAgent(agent)) continue;
-
-            let shouldDelete = agent.state === "terminated" || agent.state === "error";
-            if (!shouldDelete && agent.taskId) {
-              try {
-                const task = await this.taskStore.getTask(agent.taskId);
-                if (!task || task.column !== "in-progress") {
-                  shouldDelete = true;
-                }
-              } catch {
-                shouldDelete = true;
-              }
-            }
-            if (!shouldDelete) continue;
-
-            try {
-              if (agent.state !== "terminated") {
-                await this.agentStore.updateAgentState(agent.id, "terminated");
-              }
-            } catch (err: unknown) {
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Startup sweep failed to set ephemeral agent ${agent.id} terminated: ${msg}`);
-            }
-
-            try {
-              await this.agentStore.deleteAgent(agent.id);
-              cleanedCount += 1;
-            } catch (err: unknown) {
-              if (this.isBenignEphemeralDeleteRaceError(agent.id, err)) {
-                cleanedCount += 1;
-                continue;
-              }
-              const msg = err instanceof Error ? err.message : String(err);
-              runtimeLog.warn(`Startup sweep failed to delete ephemeral agent ${agent.id}: ${msg}`);
-            }
-          }
-
-          if (cleanedCount > 0) {
-            runtimeLog.log(`Startup ephemeral sweep cleaned ${cleanedCount} orphaned agent(s)`);
-          }
-        } catch (err: unknown) {
-          const msg = err instanceof Error ? err.message : String(err);
-          runtimeLog.warn(`Startup ephemeral sweep failed (continuing): ${msg}`);
+        // Wire the ephemeral worker manager (now that the executor exists, so
+        // its spawned-child pending-deletion set can be consulted) and run
+        // the startup orphan sweep. See ephemeral-worker-manager.ts for the
+        // full lifecycle contract. Non-fatal: failures are logged and never
+        // block startup.
+        if (this.agentStore && !this.workerManager) {
+          this.workerManager = new EphemeralWorkerManager({
+            agentStore: this.agentStore,
+            taskStore: this.taskStore,
+            logger: runtimeLog,
+            isDeletionPendingExternal: (agentId) => this.executor?.isEphemeralDeletionPending(agentId) ?? false,
+          });
+        }
+        if (this.workerManager) {
+          this.workerManager.attachStateChangeListener();
+          await this.workerManager.reconcileOrphaned();
         }
 
         // Register existing non-ephemeral, heartbeat-enabled agents in tickable states.
@@ -893,14 +715,11 @@ export class InProcessRuntime
         runtimeLog.log("RoutineScheduler stopped");
       }
 
-      // 3. Remove agent event listeners (before stopping trigger scheduler)
-      // Guard on this.agentStore being defined - it may not exist if AgentStore init failed
-      if (this.ephemeralTerminationListener && this.agentStore) {
-        this.agentStore.off("agent:stateChanged", this.ephemeralTerminationListener);
-        this.ephemeralTerminationListener = undefined;
-        runtimeLog.log("AgentStore agent:stateChanged listener removed");
-      }
-      this.pendingEphemeralDeletions.clear();
+      // 3. Tear down the ephemeral worker manager (detaches the
+      // agent:stateChanged listener and clears in-memory tracking). Safe to
+      // call when uninitialized.
+      this.workerManager?.detachStateChangeListener();
+      this.workerManager?.reset();
       this.executor?.disposeEphemeralTimers();
 
       // 4. Stop trigger scheduler
@@ -1289,25 +1108,6 @@ export class InProcessRuntime
     runtimeLog.log("Event forwarding setup complete");
   }
 
-  /**
-   * Returns true when an ephemeral delete failure is expected due to cleanup races
-   * (for example the agent was already removed by a parallel cleanup path).
-   */
-  private isBenignEphemeralDeleteRaceError(agentId: string, err: unknown): boolean {
-    const msg = err instanceof Error ? err.message : String(err);
-    const normalized = msg.toLowerCase();
-
-    if (normalized.includes("already deleted") || normalized.includes("already removed")) {
-      return true;
-    }
-
-    if (normalized.includes(`agent ${agentId.toLowerCase()} not found`)) {
-      return true;
-    }
-
-    return /^agent\s+.+\s+not found$/i.test(msg.trim());
-  }
-
   /**
    * Update status and emit health-changed event.
    */