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. In server mode a member reads only their own sessions (`member` in the filter, `server/lib/scope.js`).
 * @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 { parseTool, BAD_TOOL, parseRange, BAD_RANGE } = require("../lib/source-filter");
const { memberId, bySessionWhere } = require("../lib/scope");

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 filters: status ("active", "completed", or omit for all), the AI
  // tool (`?tool=`, `sessions.tool`) and `?from=&to=` (sessions started in
  // those local days). A server-mode member reads only their own sessions.
  const statusFilter = req.query.status || null;
  const tool = parseTool(req);
  if (tool === false) return res.status(400).json(BAD_TOOL);
  const range = parseRange(req);
  if (range === false) return res.status(400).json(BAD_RANGE);
  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,
          tool,
          range,
          sections,
          member: memberId(req),
        },
        // Every open Workflows page shares one run (~1.5 s on a large DB).
        { maxAgeMs: 30_000 }
      )
    );
  } 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 tool = parseTool(req);
    if (tool === false) return res.status(400).json(BAD_TOOL);
    const range = parseRange(req);
    if (range === false) return res.status(400).json(BAD_RANGE);
    const sf = sessionIdFilter({
      status: req.query.status || null,
      tool,
      range,
      member: memberId(req),
    });

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

    // A range on tool_name ("`" sorts right after "_") instead of LIKE, and
    // `+event_type` to keep the planner off idx_events_type: both make the
    // query use idx_events_tool (784 ms → a few ms on 253K events).
    const mcpCalls = db
      .prepare(
        `SELECT tool_name, COUNT(*) AS count FROM events
         WHERE tool_name >= 'mcp__plugin_' AND tool_name < 'mcp__plugin\`'
           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.
// A server-mode member lists and counts only runs of their own sessions.
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;

    if (memberId(req) != null)
      return res.json(memberRuns(req, { limit, offset, status, sessionId }));

    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 for a member: the same answer over their own sessions' runs only. */
function memberRuns(req, { limit, offset, status, sessionId }) {
  const w = bySessionWhere(req);
  const order = "ORDER BY COALESCE(started_at, created_at) DESC LIMIT ? OFFSET ?";
  const rows = sessionId
    ? db
        .prepare(`SELECT * FROM workflows WHERE session_id = ? AND ${w.where} ${order}`)
        .all(sessionId, ...w.params, limit, offset)
    : status
      ? db
          .prepare(`SELECT * FROM workflows WHERE status = ? AND ${w.where} ${order}`)
          .all(status, ...w.params, limit, offset)
      : db
          .prepare(`SELECT * FROM workflows WHERE ${w.where} ${order}`)
          .all(...w.params, limit, offset);
  const total = db
    .prepare(
      `SELECT COUNT(*) AS n FROM workflows WHERE ${status ? "status = ? AND " : ""}${w.where}`
    )
    .get(...(status ? [status] : []), ...w.params).n;
  const counts = {};
  for (const r of db
    .prepare(`SELECT status, COUNT(*) AS n FROM workflows WHERE ${w.where} GROUP BY status`)
    .all(...w.params))
    counts[r.status] = r.n;
  return { runs: rows.map(hydrateWorkflow), total, counts, limit, offset };
}

// 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;