/**
 * Crow Orchestrator — MCP Server
 *
 * Exposes multi-agent orchestration as Crow tools via the MCP protocol.
 * Uses open-multi-agent as the engine, with Crow's MCP tools bridged
 * into a shared ToolRegistry.
 *
 * Tools:
 *   crow_orchestrate        — Start a multi-agent team on a goal (async, returns job ID)
 *   crow_orchestrate_status — Check/retrieve result of a running job
 *   crow_list_presets       — List available team presets
 *   crow_run_pipeline       — Execute a named pipeline immediately
 *   crow_schedule_pipeline  — Schedule a pipeline on a cron schedule
 *   crow_list_pipelines     — List available pipelines
 *   crow_list_remote_tools  — List tools available on remote Crow instances
 *
 * Jobs are stored in-memory and pruned after 1 hour.
 */

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { spawn } from "node:child_process";
import { appendFileSync, mkdirSync } from "fs";
import { resolve, dirname } from "path";
import { fileURLToPath } from "url";

import { OpenMultiAgent, ToolRegistry, registerBuiltInTools } from "open-multi-agent";
import { registerCrowTools, registerRemoteTools, registerAddonTools } from "./mcp-bridge.js";
import { presets } from "./presets.js";
import { resolvePreset } from "./preset-resolver.js";
import { pipelines } from "./pipelines.js";
import { startPipelineRunner } from "./pipeline-runner.js";
import { substituteGoalPlaceholders } from "./pipeline-vars.js";
import { createDbClient } from "../db.js";
import { ensureModelWarm, releaseModel, getLifecycleSnapshot, resetAllRefcounts } from "./lifecycle.js";
import { attachEventLogger, logEvent } from "./events.js";
import { loadProviders } from "./providers.js";

const __dirname = dirname(fileURLToPath(import.meta.url));

// ---------------------------------------------------------------------------
// Models config loader
// ---------------------------------------------------------------------------

function loadModelsConfig() {
  return loadProviders();
}

/**
 * Resolve provider config from models.json.
 * Returns { baseURL, apiKey } for the given provider name.
 */
function resolveProvider(modelsConfig, providerName) {
  const provider = modelsConfig.providers?.[providerName];
  if (!provider) {
    throw new Error(
      `Provider "${providerName}" not found in provider registry. ` +
      `Available: ${Object.keys(modelsConfig.providers || {}).join(", ")}`
    );
  }
  return {
    baseURL: provider.baseUrl,
    apiKey: provider.apiKey || undefined,
  };
}

/**
 * Resolve the default orchestrator provider and model.
 * Priority: env vars > first provider in models.json.
 */
function resolveDefaultOrchestratorConfig(modelsConfig) {
  const providers = modelsConfig.providers || {};
  const providerKeys = Object.keys(providers);

  // Resolve provider
  const envProvider = process.env.CROW_ORCHESTRATOR_PROVIDER;
  const providerName = (envProvider && providers[envProvider]) ? envProvider : providerKeys[0];
  if (!providerName) {
    throw new Error("No LLM providers configured in provider registry");
  }

  const provider = providers[providerName];

  // Resolve model
  const envModel = process.env.CROW_ORCHESTRATOR_MODEL;
  const modelId = envModel || provider.models?.[0]?.id;

  return {
    provider: providerName,
    model: modelId,
    baseURL: provider.baseUrl,
    apiKey: provider.apiKey || undefined,
  };
}

// ---------------------------------------------------------------------------
// Job storage
// ---------------------------------------------------------------------------

/** @type {Map<string, { status: string, result?: string, error?: string, startedAt: number }>} */
const jobs = new Map();

const JOB_TTL_MS = 60 * 60 * 1000; // 1 hour

function pruneJobs() {
  const now = Date.now();
  for (const [id, job] of jobs) {
    if (job.status !== "running" && now - job.startedAt > JOB_TTL_MS) {
      jobs.delete(id);
    }
  }
}

function generateJobId() {
  return `orch-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
}

// Prune completed jobs every 10 minutes
setInterval(pruneJobs, 10 * 60 * 1000).unref();

// ---------------------------------------------------------------------------
// Health check
// ---------------------------------------------------------------------------

/**
 * Check if a local/self-hosted LLM server is reachable.
 * Only checks URLs that look like local endpoints (localhost, private IPs).
 * Cloud APIs are assumed reachable (they authenticate, not health-check).
 */
async function checkLlmHealth(baseURL) {
  if (!baseURL) return true;

  // Only health-check local/private endpoints
  try {
    const url = new URL(baseURL);
    const host = url.hostname;
    const isLocal = host === "localhost" || host === "127.0.0.1" || host === "::1"
      || host.startsWith("10.") || host.startsWith("192.168.") || host.startsWith("100.");
    if (!isLocal) return true; // Cloud API — skip health check
  } catch {
    return true; // Can't parse URL, assume reachable
  }

  try {
    const healthUrl = baseURL.replace(/\/v1\/?$/, "/health");
    const res = await fetch(healthUrl, {
      signal: AbortSignal.timeout(5000),
    });
    return res.ok;
  } catch {
    return false;
  }
}

const PIPELINE_PREFIX = "pipeline:";

// ---------------------------------------------------------------------------
// Per-preset ToolRegistry (keyed by sorted category list)
// ---------------------------------------------------------------------------

/** @type {Map<string, ToolRegistry>} */
const registryCache = new Map();

/**
 * Get or create a ToolRegistry for the given MCP categories.
 * Caches by category set (except when "remote" is included — remote tools are dynamic).
 */
async function getRegistryForCategories(categories, connectedServers) {
  const hasRemote = categories.includes("remote");
  const hasAddons = categories.includes("addons");
  const localCategories = categories.filter((c) => c !== "remote" && c !== "addons");
  // Addons MUST be in the cache key so an "with-addons" registry is not
  // returned to a "without-addons" caller (and vice-versa). Without the
  // suffix, two callers with the same localCategories but different
  // hasAddons values would collide.
  const key = [...localCategories, ...(hasAddons ? ["+addons"] : [])].sort().join(",");
  // Addons CAN be cached: the addon set is loaded from a static
  // mcp-addons.json and the StdioClientTransport children stay alive
  // as long as the parent gateway lives. Pre-fix (commit 45563ec,
  // 2026-04-21), addons were treated like remote (un-cacheable) and
  // every pipeline run re-spawned every addon child without closing
  // the previous ones — a process leak that grew ~8 MCP children/min
  // in the bot pipeline era. Only "remote" connections are genuinely
  // dynamic and stay un-cacheable.
  const cacheable = !hasRemote;

  let registry;
  if (cacheable && registryCache.has(key)) {
    registry = registryCache.get(key);
  } else {
    registry = new ToolRegistry();
    registerBuiltInTools(registry);

    if (localCategories.length > 0) {
      const { toolCount } = await registerCrowTools(registry, { categories: localCategories });
      console.log(`[orchestrator] Registry [${key}]: ${toolCount} Crow tools + built-ins`);
    }

    if (hasAddons) {
      const { toolCount: addonCount } = await registerAddonTools(registry);
      if (addonCount > 0) {
        console.log(`[orchestrator] Registry [${key}+addons]: +${addonCount} addon tools`);
      }
    }

    if (cacheable) {
      registryCache.set(key, registry);
    }
  }

  // Register remote instance tools (not cached — connections are dynamic)
  if (hasRemote && connectedServers) {
    const { toolCount: remoteCount } = await registerRemoteTools(registry, connectedServers);
    if (remoteCount > 0) {
      console.log(`[orchestrator] Registry [${key}+remote]: +${remoteCount} remote tools`);
    }
  }

  return registry;
}

// ---------------------------------------------------------------------------
// Server factory
// ---------------------------------------------------------------------------

export function createOrchestratorServer(dbPath, options = {}) {
  const server = new McpServer(
    { name: "crow-orchestrator", version: "0.1.0" },
    options.instructions ? { instructions: options.instructions } : undefined
  );

  const modelsConfig = loadModelsConfig();
  const connectedServers = options.connectedServers || null;

  // Phase 5-full: attach the event logger so lifecycle + dispatch events
  // persist to orchestrator_events. No-op if no DB is available.
  try {
    const db = createDbClient(dbPath);
    attachEventLogger(db);
  } catch {}

  // Resolve default provider/model once at startup
  let defaults;
  try {
    defaults = resolveDefaultOrchestratorConfig(modelsConfig);
    console.log(`[orchestrator] Default provider: ${defaults.provider}, model: ${defaults.model}`);
  } catch (err) {
    console.warn(`[orchestrator] No default provider: ${err.message}`);
    defaults = { provider: null, model: null, baseURL: null, apiKey: null };
  }

  // -----------------------------------------------------------------------
  // Helper: resolve provider config for a preset (with per-agent overrides)
  // -----------------------------------------------------------------------

  function resolveAgentConfig(agent, preset) {
    const agentProvider = agent.provider || preset.provider || defaults.provider;
    const agentModel = agent.model || preset.model || defaults.model;
    if (!agentProvider) {
      throw new Error("No LLM provider configured. Set CROW_ORCHESTRATOR_PROVIDER or add a provider to the provider registry.");
    }
    const pc = resolveProvider(modelsConfig, agentProvider);
    return {
      name: agent.name,
      model: agentModel,
      provider: "openai", // All providers use OpenAI-compatible API
      apiKey: pc.apiKey,
      baseURL: pc.baseURL,
      systemPrompt: agent.systemPrompt,
      tools: agent.tools,
      maxTurns: agent.maxTurns || 6,
      maxTokens: agent.maxTokens || preset.maxTokens || 8192,
    };
  }

  // -----------------------------------------------------------------------
  // Helper: health check for a preset's default provider
  // -----------------------------------------------------------------------

  async function checkPresetHealth(preset) {
    const providerName = preset.provider || defaults.provider;
    if (!providerName) return { healthy: false, error: "No LLM provider configured" };

    let pc;
    try {
      pc = resolveProvider(modelsConfig, providerName);
    } catch (err) {
      return { healthy: false, error: err.message };
    }

    // Only health-check providers with a baseURL (local/self-hosted)
    if (pc.baseURL) {
      const healthy = await checkLlmHealth(pc.baseURL);
      if (!healthy) {
        return { healthy: false, error: `LLM server not reachable at ${pc.baseURL}` };
      }
    }

    return { healthy: true, providerConfig: pc };
  }

  // -----------------------------------------------------------------------
  // Background orchestration runner (inside closure for connectedServers access)
  // -----------------------------------------------------------------------

  async function runOrchestration(jobId, goal, preset, providerConfig) {
    const job = jobs.get(jobId);
    if (!job) return;

    // Phase 5-full: ensure all providers the preset agents need are warm.
    // Collect unique provider IDs from preset + per-agent overrides.
    const requiredProviders = new Set();
    if (preset.provider) requiredProviders.add(preset.provider);
    for (const a of preset.agents || []) {
      if (a.provider) requiredProviders.add(a.provider);
    }
    const warmed = [];
    for (const providerId of requiredProviders) {
      const r = await ensureModelWarm(providerId);
      await logEvent({
        run_id: jobId,
        event_type: r.ok ? "dispatch.provider_ready" : "dispatch.provider_failed",
        provider_id: providerId,
        preset: preset.name || null,
        data: r.ok ? null : { reason: r.reason },
      });
      if (r.ok) warmed.push(providerId);
      else {
        // A required provider failed to warm — abort this run cleanly
        for (const w of warmed) await releaseModel(w);
        const err = `required provider "${providerId}" unavailable: ${r.reason}`;
        job.status = "failed"; job.error = err;
        await logEvent({ run_id: jobId, event_type: "dispatch.aborted", preset: preset.name || null, data: { error: err } });
        return;
      }
    }

    await logEvent({ run_id: jobId, event_type: "dispatch.run_start", preset: preset.name || null, data: { goal: String(goal).slice(0, 200) } });

    try {
      const categories = preset.categories || ["memory", "projects"];
      const registry = await getRegistryForCategories(categories, connectedServers);

      // Expand wildcard tool references (e.g., "colibri:*")
      const expandedAgents = preset.agents.map((a) => {
        if (!a.tools || a.tools.length === 0) return a;
        const allTools = registry.list ? [...registry.list()] : [];
        const expanded = a.tools.flatMap((t) => {
          if (typeof t === "string" && t.endsWith(":*")) {
            const prefix = t.slice(0, -1); // "colibri:"
            const matches = allTools.filter((name) => name.startsWith(prefix));
            if (matches.length === 0) {
              console.warn(`[orchestrator] Wildcard "${t}" matched 0 tools (instance may be offline)`);
            }
            return matches;
          }
          return [t];
        });
        return { ...a, tools: expanded };
      });

      // Build agent configs with per-agent provider resolution
      const agentConfigs = expandedAgents.map((a) => resolveAgentConfig(a, preset));

      const presetProvider = preset.provider || defaults.provider;
      const presetModel = preset.model || defaults.model;
      const presetPC = resolveProvider(modelsConfig, presetProvider);

      const orchestrator = new OpenMultiAgent({
        maxConcurrency: preset.maxConcurrency || 1,
        defaultModel: presetModel,
        defaultProvider: "openai",
        defaultApiKey: presetPC.apiKey,
        defaultBaseURL: presetPC.baseURL,
        toolRegistry: registry,
        onProgress: (event) => {
          let extra = "";
          if (event.type === "task_complete" && event.data?.output) {
            extra = ` output=${event.data.output.length}chars`;
          }
          if (event.type === "agent_complete" && event.data?.toolCalls?.length > 0) {
            extra = ` toolCalls=${event.data.toolCalls.length}`;
          }
          if (event.type === "error") {
            const d = event.data;
            const msg = d?.message || d?.output || String(d);
            extra = ` error="${msg.slice(0, 300)}"`;
          }
          console.log(`[orchestrator] [${jobId}] ${event.type}${event.agent ? ` agent=${event.agent}` : ""}${event.task ? ` task=${event.task}` : ""}${extra}`);
        },
      });

      const team = orchestrator.createTeam("team", {
        name: "team",
        agents: agentConfigs,
        sharedMemory: true,
        maxConcurrency: preset.maxConcurrency || 1,
      });

      // Run with 5 minute timeout
      const result = await Promise.race([
        orchestrator.runTeam(team, goal),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error("Orchestration timed out after 5 minutes")), 300000)
        ),
      ]);

      // Extract coordinator's final synthesis
      const coordinatorResult = result.agentResults.get("coordinator");
      const output = coordinatorResult?.output || "(no coordinator output)";

      const fullResult = [
        output,
        "",
        "---",
        `Orchestration completed: ${result.success ? "success" : "partial failure"}`,
        `Total tokens: ${result.totalTokenUsage.input_tokens} in / ${result.totalTokenUsage.output_tokens} out`,
      ].join("\n");

      job.status = "completed";
      job.result = fullResult;

      await orchestrator.shutdown();
      await logEvent({
        run_id: jobId,
        event_type: "dispatch.run_complete",
        preset: preset.name || null,
        data: { tokens_in: result.totalTokenUsage.input_tokens, tokens_out: result.totalTokenUsage.output_tokens },
      });
    } catch (err) {
      job.status = "failed";
      job.error = err.message || String(err);
      console.error(`[orchestrator] Job ${jobId} failed:`, err.message);
      await logEvent({
        run_id: jobId,
        event_type: "dispatch.run_error",
        preset: preset.name || null,
        data: { error: String(err.message || err).slice(0, 500) },
      });
    } finally {
      // Release every provider we warmed up for this run.
      for (const providerId of requiredProviders) {
        await releaseModel(providerId).catch(() => {});
      }
    }
  }

  // -----------------------------------------------------------------------
  // MCP Tools
  // -----------------------------------------------------------------------

  // --- crow_list_presets ---

  server.tool(
    "crow_list_presets",
    "List available multi-agent team presets with their descriptions, provider, model, and agent names.",
    {},
    async () => {
      const lines = [];
      for (const [name, preset] of Object.entries(presets)) {
        const agentNames = preset.agents.map((a) => a.name).join(", ");
        const provider = preset.provider || defaults.provider || "(none)";
        const model = preset.model || defaults.model || "(none)";
        lines.push(
          `**${name}** — ${preset.description}\n` +
          `  Provider: ${provider}, Model: ${model}\n` +
          `  Agents: ${agentNames}`
        );
      }

      return {
        content: [{ type: "text", text: lines.join("\n\n") }],
      };
    }
  );

  // --- crow_orchestrate ---

  server.tool(
    "crow_orchestrate",
    "Start a multi-agent team on a goal. Returns a job ID immediately — the orchestration runs in the background. Use crow_orchestrate_status to check progress and retrieve results.",
    {
      goal: z.string().min(1).describe("The high-level goal for the agent team to accomplish"),
      preset: z.string().optional().describe('Team preset name (default: "research"). Use crow_list_presets to see options.'),
    },
    async ({ goal, preset: presetName }) => {
      const name = presetName || "research";
      const presetDb = createDbClient();
      let preset;
      try { preset = await resolvePreset(presetDb, name); }
      finally { presetDb.close(); }
      if (!preset) {
        return {
          content: [{
            type: "text",
            text: `Unknown preset: "${name}". Available: ${Object.keys(presets).join(", ")}`,
          }],
          isError: true,
        };
      }

      const { healthy, error } = await checkPresetHealth(preset);
      if (!healthy) {
        return { content: [{ type: "text", text: error }], isError: true };
      }

      const jobId = generateJobId();
      jobs.set(jobId, { status: "running", startedAt: Date.now() });

      runOrchestration(jobId, goal, preset, null).catch((err) => {
        console.error(`[orchestrator] Job ${jobId} unexpected error:`, err);
        const job = jobs.get(jobId);
        if (job && job.status === "running") {
          job.status = "failed";
          job.error = err.message;
        }
      });

      return {
        content: [{
          type: "text",
          text: JSON.stringify({ jobId, status: "running", preset: presetName || "research", goal }),
        }],
      };
    }
  );

  // --- crow_orchestrate_status ---

  server.tool(
    "crow_orchestrate_status",
    "Check the status of a multi-agent orchestration job. Returns the result when completed.",
    {
      jobId: z.string().describe("Job ID returned by crow_orchestrate"),
    },
    async ({ jobId }) => {
      const job = jobs.get(jobId);
      if (!job) {
        return {
          content: [{ type: "text", text: `Job not found: ${jobId}. It may have expired (jobs are kept for 1 hour).` }],
          isError: true,
        };
      }

      const response = { status: job.status };
      if (job.result) response.result = job.result;
      if (job.error) response.error = job.error;

      return {
        content: [{ type: "text", text: JSON.stringify(response, null, 2) }],
      };
    }
  );

  // --- crow_run_pipeline ---

  server.tool(
    "crow_run_pipeline",
    "Execute a named pipeline immediately. Pipelines are predefined multi-agent workflows (e.g. memory-consolidation, daily-summary). Returns a job ID — poll with crow_orchestrate_status.",
    {
      pipeline: z.string().describe("Pipeline name (use crow_list_pipelines to see options)"),
    },
    async ({ pipeline: pipelineName }) => {
      const pipeline = pipelines[pipelineName];
      if (!pipeline) {
        return {
          content: [{
            type: "text",
            text: `Unknown pipeline: "${pipelineName}". Available: ${Object.keys(pipelines).join(", ")}`,
          }],
          isError: true,
        };
      }

      const presetDb = createDbClient();
      let preset;
      try { preset = await resolvePreset(presetDb, pipeline.preset); }
      finally { presetDb.close(); }
      if (!preset) {
        return {
          content: [{ type: "text", text: `Pipeline "${pipelineName}" references unknown preset: "${pipeline.preset}"` }],
          isError: true,
        };
      }

      const { healthy, error } = await checkPresetHealth(preset);
      if (!healthy) {
        return { content: [{ type: "text", text: error }], isError: true };
      }

      const jobId = generateJobId();
      jobs.set(jobId, { status: "running", startedAt: Date.now(), pipeline: pipelineName });

      const expandedGoal = substituteGoalPlaceholders(pipeline.goal);
      runOrchestration(jobId, expandedGoal, preset, null).catch((err) => {
        console.error(`[orchestrator] Pipeline job ${jobId} error:`, err);
        const job = jobs.get(jobId);
        if (job && job.status === "running") {
          job.status = "failed";
          job.error = err.message;
        }
      });

      return {
        content: [{
          type: "text",
          text: JSON.stringify({ jobId, status: "running", pipeline: pipelineName, goal: expandedGoal }),
        }],
      };
    }
  );

  // --- crow_schedule_pipeline ---

  server.tool(
    "crow_schedule_pipeline",
    "Schedule a pipeline to run on a cron schedule. Creates an entry in Crow's schedules table with a pipeline: prefix.",
    {
      pipeline: z.string().describe("Pipeline name (e.g. memory-consolidation, daily-summary, research-digest)"),
      cron_expression: z.string().max(50).optional().describe("Cron expression (e.g. '0 3 * * *'). If omitted, uses the pipeline's default schedule."),
      description: z.string().max(500).optional().describe("Optional description override"),
    },
    async ({ pipeline: pipelineName, cron_expression, description }) => {
      const pipeline = pipelines[pipelineName];
      if (!pipeline) {
        return {
          content: [{
            type: "text",
            text: `Unknown pipeline: "${pipelineName}". Available: ${Object.keys(pipelines).join(", ")}`,
          }],
          isError: true,
        };
      }

      const cron = cron_expression || pipeline.defaultCron;

      let nextRun = null;
      try {
        const { CronExpressionParser } = await import("cron-parser");
        const interval = CronExpressionParser.parse(cron);
        nextRun = interval.next().toISOString();
      } catch {
        return {
          content: [{ type: "text", text: `Invalid cron expression: "${cron}". Use standard 5-field format.` }],
          isError: true,
        };
      }

      const db = createDbClient();
      const task = `${PIPELINE_PREFIX}${pipelineName}`;
      const desc = description || pipeline.description;

      const { rows: existing } = await db.execute({
        sql: "SELECT id FROM schedules WHERE task = ?",
        args: [task],
      });

      if (existing.length > 0) {
        await db.execute({
          sql: "UPDATE schedules SET cron_expression = ?, description = ?, next_run = ?, enabled = 1, updated_at = datetime('now') WHERE task = ?",
          args: [cron, desc, nextRun, task],
        });
        return {
          content: [{
            type: "text",
            text: `Pipeline schedule updated (id: #${existing[0].id}, pipeline: ${pipelineName}, cron: ${cron}, next run: ${nextRun})`,
          }],
        };
      }

      const result = await db.execute({
        sql: "INSERT INTO schedules (task, cron_expression, description, next_run) VALUES (?, ?, ?, ?)",
        args: [task, cron, desc, nextRun],
      });

      return {
        content: [{
          type: "text",
          text: `Pipeline scheduled (id: #${Number(result.lastInsertRowid)}, pipeline: ${pipelineName}, cron: ${cron}, next run: ${nextRun})`,
        }],
      };
    }
  );

  // --- crow_list_pipelines ---

  server.tool(
    "crow_list_pipelines",
    "List available pipelines with their descriptions and default schedules.",
    {},
    async () => {
      const lines = [];
      for (const [name, pipeline] of Object.entries(pipelines)) {
        lines.push(
          `**${name}** — ${pipeline.description}\n` +
          `  Preset: ${pipeline.preset}, Default cron: ${pipeline.defaultCron}\n` +
          `  Stores result: ${pipeline.storeResult ? `yes (category: ${pipeline.resultCategory})` : "no"}`
        );
      }
      return {
        content: [{ type: "text", text: lines.join("\n\n") }],
      };
    }
  );

  // --- crow_list_remote_tools ---

  server.tool(
    "crow_list_remote_tools",
    "List tools available on remote Crow instances. Shows connected instances and their exposed tools.",
    {},
    async () => {
      if (!connectedServers) {
        return {
          content: [{ type: "text", text: "No remote instances available (orchestrator running in stdio mode)." }],
        };
      }

      const lines = [];
      for (const [key, entry] of connectedServers) {
        if (!entry.isRemote) continue;
        const status = entry.status === "connected" ? "connected" : `offline (${entry.error || "unknown"})`;
        const toolNames = (entry.tools || []).map((t) => t.name);
        lines.push(
          `**${entry.instanceName || key}** — ${status}\n` +
          `  Gateway: ${entry.gatewayUrl || "unknown"}\n` +
          `  Tools (${toolNames.length}): ${toolNames.join(", ") || "(none)"}`
        );
      }

      if (lines.length === 0) {
        return { content: [{ type: "text", text: "No remote instances registered." }] };
      }

      return {
        content: [{ type: "text", text: lines.join("\n\n") }],
      };
    }
  );

  // --- crow_lifecycle_snapshot (Phase 5-full) ---
  server.tool(
    "crow_lifecycle_snapshot",
    "Return the current orchestrator lifecycle snapshot (refcounts per provider, last released timestamps).",
    {},
    async () => {
      const snap = getLifecycleSnapshot();
      if (Object.keys(snap).length === 0) {
        return { content: [{ type: "text", text: "No providers currently tracked." }] };
      }
      const lines = Object.entries(snap).map(([k, v]) => {
        const age = v.lastReleasedAt ? ` (released ${Math.round((Date.now() - v.lastReleasedAt) / 1000)}s ago)` : "";
        return `  ${k}: refs=${v.refs}${age}`;
      });
      return { content: [{ type: "text", text: "Lifecycle snapshot:\n" + lines.join("\n") }] };
    }
  );

  // --- crow_reset_refcounts (operator kill switch) ---
  server.tool(
    "crow_reset_refcounts",
    "Operator kill switch: clear all lifecycle refcounts and reconcile against live provider health. Use when refcount state drifts from reality.",
    {},
    async () => {
      const r = await resetAllRefcounts();
      return { content: [{ type: "text", text: `Refcounts reset. Reconciled ${r.reconciled} providers.` }] };
    }
  );

  return server;
}

// ---------------------------------------------------------------------------
// Pipeline runner integration
// ---------------------------------------------------------------------------

/**
 * Run one orchestration team to completion, returning the result.
 *
 * Extracted from the closure previously embedded in startOrchestratorPipelines
 * so it can also be invoked from a clean child process (see
 * scripts/run-pipeline-subprocess.mjs). That subprocess pathway is the
 * workaround for MPA, whose long-lived gateway process has wedged libsql
 * state (see servers/sharing/peer-pull-sync.js:9-17) — every fresh libsql
 * client opened inside MPA's gateway fails with SQLITE_IOERR_SHORT_READ on
 * write, so the whole pipeline has to run in a fresh process.
 *
 * @param {string} goal - Orchestrator goal text
 * @param {string} presetName - Preset key (presets.js)
 * @param {object} [options]
 * @param {object} [options.modelsConfig] - Parsed models.json (loaded if omitted)
 * @param {Map}    [options.connectedServers] - Remote instance connections (null when run as subprocess)
 * @param {object} [options.pipelineDefaults] - Default provider/model config
 */

// ---------------------------------------------------------------------------
// Watermark rescue
// ---------------------------------------------------------------------------
// When a bot preset sends a Gmail threaded reply but the agent runs out of
// turns/tokens before calling bot_conversations_patch, the next tick re-reads
// the same user message and double-replies. This hook inspects the toolCalls
// captured during the run, detects "reply sent but no watermark patch", and
// patches bot_conversations.last_user_msg_at directly. Idempotent and only
// advances the watermark forward; safe to invoke after every run.

const WATERMARK_RESCUE_REPLY_TOOLS = new Set(["gmail_send_threaded_to_self"]);

async function rescueBotWatermark(toolCalls) {
  if (!Array.isArray(toolCalls) || toolCalls.length === 0) return;

  const sentReplies = toolCalls.filter((tc) => {
    if (!tc || !WATERMARK_RESCUE_REPLY_TOOLS.has(tc.toolName)) return false;
    const threadId = tc?.input?.thread_id;
    if (!threadId) return false;
    try {
      const out = typeof tc.output === "string" ? JSON.parse(tc.output) : tc.output;
      return out?.success === true && out?.data?.sent === true;
    } catch {
      return false;
    }
  });
  if (sentReplies.length === 0) return;

  const alreadyPatched = toolCalls.some(
    (tc) => tc?.toolName === "bot_conversations_patch" && tc?.input?.last_user_msg_at,
  );
  if (alreadyPatched) return;

  const db = createDbClient();
  try {
    for (const reply of sentReplies) {
      const threadId = reply.input.thread_id;

      let watermark = null;
      for (const tc of toolCalls) {
        if (tc?.toolName !== "gmail_get_thread") continue;
        if (tc?.input?.thread_id !== threadId) continue;
        try {
          const parsed = typeof tc.output === "string" ? JSON.parse(tc.output) : tc.output;
          const msgs = parsed?.data?.messages || [];
          let latestMs = 0;
          for (const m of msgs) {
            const ts = parseInt(m?.internal_date, 10);
            if (Number.isFinite(ts) && ts > latestMs) latestMs = ts;
          }
          if (latestMs > 0) watermark = new Date(latestMs).toISOString();
        } catch {}
      }
      if (!watermark) watermark = new Date().toISOString();

      try {
        const res = await db.execute({
          sql: `UPDATE bot_conversations
                SET last_user_msg_at = ?, updated_at = datetime('now')
                WHERE gmail_thread_id = ?
                  AND (last_user_msg_at IS NULL OR last_user_msg_at < ?)`,
          args: [watermark, threadId, watermark],
        });
        if (res.rowsAffected > 0) {
          console.log(
            `[watermark-rescue] thread=${threadId} watermark=${watermark} rows=${res.rowsAffected}`,
          );
        }
      } catch (err) {
        console.warn(`[watermark-rescue] UPDATE failed for thread=${threadId}: ${err.message}`);
      }
    }
  } finally {
    try { db.close(); } catch {}
  }
}

export async function runOrchestrationStandalone(goal, presetName, options = {}) {
  const modelsConfig = options.modelsConfig || loadModelsConfig();
  const connectedServers = options.connectedServers || null;

  let pipelineDefaults = options.pipelineDefaults;
  if (!pipelineDefaults) {
    try { pipelineDefaults = resolveDefaultOrchestratorConfig(modelsConfig); }
    catch { pipelineDefaults = { provider: null, model: null, baseURL: null, apiKey: null }; }
  }

  const presetDb = createDbClient();
  let preset;
  try { preset = await resolvePreset(presetDb, presetName); }
  finally { presetDb.close(); }
  if (!preset) {
    return { status: "failed", error: `Unknown preset: "${presetName}"` };
  }

  const providerName = preset.provider || pipelineDefaults.provider;
  if (!providerName) {
    return { status: "failed", error: "No LLM provider configured" };
  }

  let pc;
  try {
    pc = resolveProvider(modelsConfig, providerName);
  } catch (err) {
    return { status: "failed", error: err.message };
  }

  if (pc.baseURL) {
    const healthy = await checkLlmHealth(pc.baseURL);
    if (!healthy) {
      return { status: "failed", error: `LLM not reachable at ${pc.baseURL}` };
    }
  }

  const jobId = generateJobId();
  jobs.set(jobId, { status: "running", startedAt: Date.now() });

  const categories = preset.categories || ["memory", "projects"];
  const registry = await getRegistryForCategories(categories, connectedServers);

  const presetModel = preset.model || pipelineDefaults.model;
  const agentConfigs = preset.agents.map((a) => {
    const agentProvider = a.provider || preset.provider || pipelineDefaults.provider;
    const agentModel = a.model || preset.model || pipelineDefaults.model;
    const agentPC = resolveProvider(modelsConfig, agentProvider);
    return {
      name: a.name,
      model: agentModel,
      provider: "openai",
      apiKey: agentPC.apiKey,
      baseURL: agentPC.baseURL,
      systemPrompt: a.systemPrompt,
      tools: a.tools,
      maxTurns: a.maxTurns || 6,
      maxTokens: a.maxTokens || preset.maxTokens || 8192,
    };
  });

  const presetPC = resolveProvider(modelsConfig, providerName);

  // Captured for the post-run watermark rescue (see rescueBotWatermark above).
  const _collectedToolCalls = [];

  try {
    const orchestrator = new OpenMultiAgent({
      maxConcurrency: preset.maxConcurrency || 1,
      defaultModel: presetModel,
      defaultProvider: "openai",
      defaultApiKey: presetPC.apiKey,
      defaultBaseURL: presetPC.baseURL,
      toolRegistry: registry,
      onProgress: (event) => {
        // Always persist the full event to a daily JSONL trace file. This is
        // the only place we capture agent reasoning, tool calls, and tool
        // results — the journal log truncates at 200 chars per line so
        // long agent outputs are unreadable there. Trace files live next
        // to the MPA db and rotate daily.
        try {
          const traceDir = `${process.env.CROW_HOME || "/home/kh0pp/.crow-mpa"}/data/agent-traces`;
          mkdirSync(traceDir, { recursive: true });
          const day = new Date().toISOString().slice(0, 10);
          const line = JSON.stringify({
            at: new Date().toISOString(),
            run_id: jobId,
            preset: presetName,
            event_type: event.type,
            agent: event.agent || null,
            data: event.data,
          }) + "\n";
          appendFileSync(`${traceDir}/agent-trace-${day}.jsonl`, line);
        } catch (err) {
          console.warn(`[pipeline-runner] trace write failed: ${err.message}`);
        }

        if (event.type === "agent_complete" && Array.isArray(event.data?.toolCalls)) {
          for (const tc of event.data.toolCalls) _collectedToolCalls.push(tc);
        }

        if (event.type === "error") {
          const d = event.data;
          const msg = d?.message || d?.output || String(d);
          console.log(`[pipeline-runner] ${event.type} ${event.agent || ""} error="${msg.slice(0, 200)}"`);
        } else if (process.env.CROW_PIPELINE_TRACE === "1") {
          const d = event.data;
          const summary = typeof d === "string" ? d : d?.tool || d?.message || JSON.stringify(d).slice(0, 200);
          console.log(`[pipeline-runner] ${event.type} ${event.agent || ""} ${summary}`);
        }
      },
    });

    // Linear single-agent presets (one entry in agents[]) skip the coordinator
    // decomposition and run via runAgent — OpenMultiAgent's runTeam keeps the
    // agent's turn budget across multiple decomposed tasks but empirically
    // dispatches only the first task before synthesising the rest in prose
    // (never firing the follow-up tool calls the goal specifies). For
    // linear tool chains (retrieve → format → store → notify) the direct
    // conversation-loop path gets every tool called as instructed.
    let runResult;
    if (agentConfigs.length === 1) {
      const single = agentConfigs[0];
      const agentResult = await Promise.race([
        orchestrator.runAgent(single, goal),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error("Pipeline timed out after 30 minutes")), 1800000)
        ),
      ]);
      runResult = {
        success: !agentResult.error,
        output: agentResult.output || "(no agent output)",
        totalTokenUsage: agentResult.tokenUsage || { input_tokens: 0, output_tokens: 0 },
      };
    } else {
      const team = orchestrator.createTeam("team", {
        name: "team",
        agents: agentConfigs,
        sharedMemory: true,
        maxConcurrency: preset.maxConcurrency || 1,
      });

      const teamResult = await Promise.race([
        orchestrator.runTeam(team, goal),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error("Pipeline timed out after 30 minutes")), 1800000)
        ),
      ]);
      const coordinatorResult = teamResult.agentResults.get("coordinator");
      runResult = {
        success: teamResult.success,
        output: coordinatorResult?.output || "(no coordinator output)",
        totalTokenUsage: teamResult.totalTokenUsage,
      };
    }

    await orchestrator.shutdown();

    try {
      await rescueBotWatermark(_collectedToolCalls);
    } catch (err) {
      console.warn(`[watermark-rescue] failed: ${err.message}`);
    }

    return {
      status: "completed",
      result: [
        runResult.output,
        "",
        "---",
        `Pipeline completed: ${runResult.success ? "success" : "partial failure"}`,
        `Total tokens: ${runResult.totalTokenUsage.input_tokens} in / ${runResult.totalTokenUsage.output_tokens} out`,
      ].join("\n"),
    };
  } catch (err) {
    return { status: "failed", error: err.message || String(err) };
  }
}

// Path to the subprocess CLI. Set once via fileURLToPath to keep ESM pathing sane.
const SUBPROCESS_SCRIPT = resolve(__dirname, "../../scripts/run-pipeline-subprocess.mjs");
// Matches the 5-minute orchestrator internal timeout with a small buffer so
// the subprocess kill isn't racing orchestrator.shutdown().
const SUBPROCESS_TIMEOUT_MS = 6 * 60 * 1000;

/**
 * Dispatch one orchestration to a child process. The child runs
 * runOrchestrationStandalone inside its own Node instance, writes JSON to
 * stdout, and exits. Called from the gateway's runOrchestrationSync when
 * CROW_PIPELINE_SUBPROCESS=1 is set (MPA's systemd drop-in).
 *
 * Primary doesn't set the flag and keeps the in-process path unchanged.
 */
function runOrchestrationInSubprocess(goal, presetName) {
  return new Promise((resolveP) => {
    const payload = JSON.stringify({ goal, presetName });
    const child = spawn(
      process.execPath,
      [SUBPROCESS_SCRIPT, "--from-stdin"],
      { stdio: ["pipe", "pipe", "pipe"], env: process.env }
    );

    let stdout = "";
    let stderr = "";
    let settled = false;

    const timer = setTimeout(() => {
      if (settled) return;
      settled = true;
      try { child.kill("SIGKILL"); } catch {}
      resolveP({ status: "failed", error: `pipeline subprocess timed out after ${SUBPROCESS_TIMEOUT_MS}ms` });
    }, SUBPROCESS_TIMEOUT_MS);

    child.stdout.on("data", (d) => { stdout += d.toString(); });
    child.stderr.on("data", (d) => { stderr += d.toString(); });
    child.on("error", (err) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      resolveP({ status: "failed", error: `pipeline subprocess spawn failed: ${err.message}` });
    });
    child.on("close", (code) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      const lastLine = stdout.trim().split("\n").filter(Boolean).pop() || "";
      try {
        resolveP(JSON.parse(lastLine));
      } catch {
        resolveP({
          status: "failed",
          error: `pipeline subprocess exit ${code}, stdout did not contain a JSON result line. stderr=${stderr.slice(0, 400)}`,
        });
      }
    });

    child.stdin.write(payload);
    child.stdin.end();
  });
}

/**
 * Start the background pipeline runner. Call from gateway startup.
 *
 * @param {object} db - libsql database client
 * @param {object} [options]
 * @param {Map} [options.connectedServers] - Remote instance connections (from proxy.js)
 */
export function startOrchestratorPipelines(db, options = {}) {
  const modelsConfig = loadModelsConfig();
  const connectedServers = options.connectedServers || null;

  let pipelineDefaults;
  try {
    pipelineDefaults = resolveDefaultOrchestratorConfig(modelsConfig);
  } catch {
    pipelineDefaults = { provider: null, model: null, baseURL: null, apiKey: null };
  }

  async function runOrchestrationSync(goal, presetName) {
    if (process.env.CROW_PIPELINE_SUBPROCESS === "1") {
      return runOrchestrationInSubprocess(goal, presetName);
    }
    return runOrchestrationStandalone(goal, presetName, {
      modelsConfig,
      connectedServers,
      pipelineDefaults,
    });
  }

  async function storeResultAsMemory(title, content, category) {
    await db.execute({
      sql: `INSERT INTO memories (content, category, importance, tags)
            VALUES (?, ?, 6, 'pipeline,automated')`,
      args: [`${title}\n\n${content}`, category],
    });
  }

  startPipelineRunner(db, {
    runOrchestration: runOrchestrationSync,
    storeResult: storeResultAsMemory,
  });
}
