AgentHub/server/routes/workflows.js

/**
 * @file Express router for workflow intelligence endpoints, providing aggregated insights into workflow orchestration, tool usage patterns, subagent effectiveness, error propagation, concurrency, and session complexity. It queries the database for various metrics and patterns related to agents, sessions, and events, and returns a comprehensive JSON response for frontend visualization on the dashboard. The GET / bundle is computed by `server/lib/workflow-stats.js` in the read worker (`server/lib/read-pool.js`), off the event loop.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const { Router } = require("express");
const { db, stmts } = require("../db");
const { runRead } = require("../lib/read-pool");
const { WORKFLOW_SECTIONS, sessionIdFilter } = require("../lib/workflow-stats");

const router = Router();

// ── GET / — Aggregate workflow intelligence ──
// Runs in the read worker: the bundle scans every event and takes seconds on a
// large database, which inline would stall every other request meanwhile.
// `?sections=a,b` computes only those parts of it (System Health needs seven).
router.get("/", async (req, res) => {
  // Optional status filter: "active", "completed", or omit for all
  const statusFilter = req.query.status || null;
  let sections = null;
  if (typeof req.query.sections === "string") {
    sections = req.query.sections.split(",").filter((k) => WORKFLOW_SECTIONS.includes(k));
    if (sections.length === 0) {
      return res.status(400).json({
        error: {
          code: "INVALID_SECTIONS",
          message: `sections must name some of: ${WORKFLOW_SECTIONS.join(", ")}`,
        },
      });
    }
  }
  try {
    res.json(await runRead("workflows", { status: statusFilter, sections }));
  } catch (err) {
    if (err.code === "EREADTIMEOUT") {
      return res.status(503).json({ error: { code: err.code, message: err.message } });
    }
    res.status(500).json({ error: { message: err.message } });
  }
});

/**
 * How much each skill and each plugin is actually used.
 *
 * Skill calls are read off `Skill` tool events, one row per invocation:
 * `PreToolUse` only, because every call also emits a `PostToolUse` and counting
 * both would double every number. The skill's own name carries its plugin as a
 * `plugin:skill` prefix; a bare name is a built-in or personal skill and is
 * grouped under `null`.
 *
 * A plugin's total also includes calls to the MCP tools it ships
 * (`mcp__plugin_<plugin>__<tool>`) — a plugin used entirely through its MCP
 * server would otherwise report zero usage.
 */
router.get("/skill-usage", (req, res) => {
  try {
    const sf = sessionIdFilter(req.query.status || null);

    const skills = db
      .prepare(
        `SELECT json_extract(data, '$.tool_input.skill') AS name,
                COUNT(*) AS count,
                COUNT(DISTINCT session_id) AS sessions
         FROM events
         WHERE tool_name = 'Skill' AND event_type = 'PreToolUse'
           AND json_extract(data, '$.tool_input.skill') IS NOT NULL${sf.clause}
         GROUP BY name ORDER BY count DESC LIMIT 30`
      )
      .all(...sf.params)
      .map((r) => {
        const i = r.name.indexOf(":");
        return { ...r, plugin: i > 0 ? r.name.slice(0, i) : null };
      });

    const mcpCalls = db
      .prepare(
        `SELECT tool_name, COUNT(*) AS count FROM events
         WHERE tool_name LIKE 'mcp__plugin\\_%' ESCAPE '\\'
           AND event_type = 'PreToolUse'${sf.clause}
         GROUP BY tool_name`
      )
      .all(...sf.params);

    const byPlugin = new Map();
    const bump = (name, field, n) => {
      const e = byPlugin.get(name) || { name, skills: 0, skillCalls: 0, mcpCalls: 0 };
      e[field] += n;
      byPlugin.set(name, e);
    };
    for (const s of skills) {
      if (!s.plugin) continue;
      bump(s.plugin, "skills", 1);
      bump(s.plugin, "skillCalls", s.count);
    }
    for (const m of mcpCalls) {
      // mcp__plugin_<plugin>_<server>__<tool>. Both halves may contain
      // underscores, so the split is only unambiguous when the plugin names its
      // server after itself — the overwhelmingly common case (`agenthub_agenthub`,
      // `vercel_vercel`), which the dedupe below collapses back to one name.
      // ponytail: a plugin whose server has its own name reports under
      // "<plugin>_<server>"; read the real pairing off the MCP config if that
      // ever matters.
      const seg = m.tool_name.slice("mcp__plugin_".length).split("__")[0];
      const doubled = seg.match(/^(.+)_\1$/);
      const plugin = doubled ? doubled[1] : seg;
      if (plugin) bump(plugin, "mcpCalls", m.count);
    }
    const plugins = [...byPlugin.values()].sort(
      (a, b) => b.skillCalls + b.mcpCalls - (a.skillCalls + a.mcpCalls)
    );

    res.json({ skills, plugins });
  } catch (err) {
    res.status(500).json({ error: { message: err.message } });
  }
});

// ── GET /session/:id — Single session drill-in ──
router.get("/session/:id", (req, res) => {
  try {
    const sessionId = req.params.id;
    const session = stmts.getSession.get(sessionId);
    if (!session) return res.status(404).json({ error: { message: "Session not found" } });

    const agents = stmts.listAgentsBySession.all(sessionId);
    const events = db
      .prepare("SELECT * FROM events WHERE session_id = ? ORDER BY created_at ASC, id ASC")
      .all(sessionId);

    // Build agent tree
    const tree = buildAgentTree(agents);

    // Build tool timeline
    const toolTimeline = events
      .filter((e) => e.tool_name)
      .map((e) => ({
        id: e.id,
        tool_name: e.tool_name,
        event_type: e.event_type,
        agent_id: e.agent_id,
        created_at: e.created_at,
        summary: e.summary,
      }));

    // Agent swim lanes
    const swimLanes = agents.map((a) => ({
      id: a.id,
      name: a.name,
      type: a.type,
      subagent_type: a.subagent_type,
      status: a.status,
      started_at: a.started_at,
      ended_at: a.ended_at,
      parent_agent_id: a.parent_agent_id,
    }));

    res.json({ session, tree, toolTimeline, swimLanes, events: events.slice(0, 500) });
  } catch (err) {
    res.status(500).json({ error: { message: err.message } });
  }
});

// ── Build agent tree from flat list ──
function buildAgentTree(agents) {
  const map = {};
  const roots = [];
  for (const a of agents) {
    map[a.id] = {
      id: a.id,
      name: a.name,
      type: a.type,
      subagent_type: a.subagent_type,
      status: a.status,
      task: a.task,
      started_at: a.started_at,
      ended_at: a.ended_at,
      children: [],
    };
  }
  for (const a of agents) {
    if (a.parent_agent_id && map[a.parent_agent_id]) {
      map[a.parent_agent_id].children.push(map[a.id]);
    } else {
      roots.push(map[a.id]);
    }
  }
  return roots;
}

// ── Workflow-tool runs (issue #167) ───────────────────────────────────────
// Distinct from the analytics above: these are fleets spawned by the Claude
// Code "Workflow" tool, ingested from on-disk run journals into the workflows
// table (see server/lib/workflow-ingest.js). Routed under /runs so they never
// collide with the analytics root or /session/:id.

// Parse the JSON-blob columns into arrays for the client.
function hydrateWorkflow(row) {
  if (!row) return row;
  let phases = [];
  let progress = [];
  try {
    phases = row.phases ? JSON.parse(row.phases) : [];
  } catch {
    phases = [];
  }
  try {
    progress = row.progress ? JSON.parse(row.progress) : [];
  } catch {
    progress = [];
  }
  return { ...row, phases, progress };
}

// GET /runs — list workflow runs (filter by status or session_id), paginated.
router.get("/runs", (req, res) => {
  try {
    const limit = Math.min(Math.max(parseInt(req.query.limit, 10) || 50, 1), 1000);
    const offset = Math.max(parseInt(req.query.offset, 10) || 0, 0);
    const status = req.query.status && req.query.status !== "all" ? req.query.status : null;
    const sessionId = req.query.session_id || null;

    let rows;
    if (sessionId) {
      rows = stmts.listWorkflowsBySessionFilter.all(sessionId, limit, offset);
    } else if (status) {
      rows = stmts.listWorkflowsByStatus.all(status, limit, offset);
    } else {
      rows = stmts.listWorkflows.all(limit, offset);
    }

    const total = status
      ? stmts.countWorkflowsByStatus.get(status).n
      : stmts.countWorkflows.get().n;
    const counts = {};
    for (const r of stmts.workflowStatusCounts.all()) counts[r.status] = r.n;

    res.json({ runs: rows.map(hydrateWorkflow), total, counts, limit, offset });
  } catch (err) {
    res.status(500).json({ error: { code: "WORKFLOW_LIST_FAILED", message: err.message } });
  }
});

// GET /runs/:runId — one run with its linked inner agents + their events.
router.get("/runs/:runId", (req, res) => {
  try {
    const wf = stmts.getWorkflow.get(req.params.runId);
    if (!wf) {
      return res
        .status(404)
        .json({ error: { code: "WORKFLOW_NOT_FOUND", message: "Workflow run not found" } });
    }
    const agents = stmts.listAgentsByWorkflow.all(req.params.runId);
    // Events attributed to this run's inner agents (chronological).
    let events = [];
    if (agents.length > 0) {
      const ids = agents.map((a) => a.id);
      const placeholders = ids.map(() => "?").join(",");
      events = db
        .prepare(
          `SELECT * FROM events WHERE agent_id IN (${placeholders}) ORDER BY created_at ASC, id ASC LIMIT 5000`
        )
        .all(...ids);
    }
    res.json({ workflow: hydrateWorkflow(wf), agents, events });
  } catch (err) {
    res.status(500).json({ error: { code: "WORKFLOW_DETAIL_FAILED", message: err.message } });
  }
});

module.exports = router;