/**
 * 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);
  }
}