fusion/packages/core/src/workflow-analytics.ts

import { isReviewColumnRole, isWipColumnRole, type ColumnRoleTraitFlags } from "./column-roles.js";
import { resolveProjectColumnsForRoles, type ProjectLaneVocabularyStore } from "./project-lane-vocabulary.js";
import { sql } from "drizzle-orm";
import type { Database } from "./db.js";
import type { AsyncDataLayer } from "./postgres/data-layer.js";
import { BUILTIN_WORKFLOWS, getBuiltinWorkflow, isBuiltinWorkflowId } from "./builtin-workflows.js";
import { costFor, type CostResult, type ModelPricingOverrides } from "./model-pricing.js";
import type { TokenTotals } from "./token-analytics.js";

export interface WorkflowAnalyticsQuery {
  /** ISO-8601 lower bound (inclusive). */
  from?: string;
  /** ISO-8601 upper bound (inclusive). */
  to?: string;
  /** Epoch ms "now" used only for pricing-staleness. */
  now?: number;
  /** User-managed pricing overrides that take precedence over the built-in baseline. */
  pricingOverrides?: ModelPricingOverrides;
  /** Workflow id used for tasks without an explicit task_workflow_selection row. */
  defaultWorkflowId?: string;
  /*
  FNXC:WorkflowLifecycleColumns 2026-07-31-03:20 (batch-core feed):
  Resolved trait flags per column NAME, so the wip/review tallies below come from the board's own
  lanes. Omitted, core's role helpers fall back to the legacy ids — that degraded mode lives in
  `column-roles.ts` and is tested there, so this file carries no hand-written fallback.

  The failure this fixes is a SILENT ZERO. On a renamed board neither literal matched any row, so the
  analytics surface reported `tasksInProgress: 0` and `tasksInReview: 0` beside token and cost totals
  that were entirely correct. Plausible-looking numbers are worse than missing ones: a zero next to a
  populated cost column reads as "nobody is working", not as "this metric is broken".
  */
  columnFlagsByName?: ReadonlyMap<string, ColumnRoleTraitFlags>;
}

export interface WorkflowMetricTotals {
  tokens: TokenTotals;
  cost: CostResult;
  filesChanged: number;
  tasksCompleted: number;
  tasksInProgress: number;
  tasksInReview: number;
}

export interface WorkflowSummary extends WorkflowMetricTotals {
  workflowId: string;
  workflowName: string;
  workflowIcon?: string;
  isBuiltin: boolean;
}

export interface WorkflowAnalytics {
  from: string | null;
  to: string | null;
  totals: WorkflowMetricTotals;
  workflows: WorkflowSummary[];
}

interface WorkflowNameRow {
  id: string;
  name: string | null;
  icon: string | null;
}

interface TaskTokenRow {
  workflowId: string;
  inputTokens: number | null;
  outputTokens: number | null;
  cachedTokens: number | null;
  cacheWriteTokens: number | null;
  totalTokens: number | null;
  modelProvider: string | null;
  modelId: string | null;
  tokenUsageModelProvider: string | null;
  tokenUsageModelId: string | null;
}

interface CountByWorkflowRow {
  workflowId: string;
  count: number;
}

interface ModifiedFilesRow {
  workflowId: string;
  modifiedFiles: string | null;
}

function emptyTokenTotals(): TokenTotals {
  return {
    inputTokens: 0,
    outputTokens: 0,
    cachedTokens: 0,
    cacheWriteTokens: 0,
    totalTokens: 0,
    nTasks: 0,
  };
}

interface CostAccumulator {
  usd: number;
  anyPriced: boolean;
  anyUnavailable: boolean;
  anyStale: boolean;
}

function emptyCostAccumulator(): CostAccumulator {
  return { usd: 0, anyPriced: false, anyUnavailable: false, anyStale: false };
}

function finalizeCost(acc: CostAccumulator): CostResult {
  return {
    usd: acc.anyPriced ? acc.usd : null,
    unavailable: acc.anyUnavailable,
    stale: acc.anyStale,
  };
}

function addTokenRow(totals: TokenTotals, row: TaskTokenRow): void {
  totals.inputTokens += row.inputTokens ?? 0;
  totals.outputTokens += row.outputTokens ?? 0;
  totals.cachedTokens += row.cachedTokens ?? 0;
  totals.cacheWriteTokens += row.cacheWriteTokens ?? 0;
  totals.totalTokens +=
    row.totalTokens ??
    (row.inputTokens ?? 0) +
      (row.outputTokens ?? 0) +
      (row.cachedTokens ?? 0) +
      (row.cacheWriteTokens ?? 0);
  totals.nTasks += 1;
}

function addRowCost(
  acc: CostAccumulator,
  row: TaskTokenRow,
  now?: number,
  pricingOverrides?: ModelPricingOverrides,
): void {
  const result = costFor(
    {
      inputTokens: row.inputTokens ?? 0,
      outputTokens: row.outputTokens ?? 0,
      cachedTokens: row.cachedTokens ?? 0,
      cacheWriteTokens: row.cacheWriteTokens ?? 0,
    },
    {
      /*
       * FNXC:CommandCenter 2026-07-10-08:25:
       * Workflow cost analytics must mirror token analytics by pricing the actually-used token-usage model snapshot before legacy task model columns. FN-7757's static catalog rows could not help rows whose legacy model columns are NULL, so this fixes the durable resolution path without guessing prices for unknown models.
       */
      provider: row.tokenUsageModelProvider ?? row.modelProvider,
      model: row.tokenUsageModelId ?? row.modelId,
    },
    now,
    pricingOverrides,
  );
  if (result.stale) acc.anyStale = true;
  if (result.unavailable || result.usd === null) {
    acc.anyUnavailable = true;
  } else {
    acc.usd += result.usd;
    acc.anyPriced = true;
  }
}

function emptyMetricTotals(): WorkflowMetricTotals {
  return {
    tokens: emptyTokenTotals(),
    cost: { usd: null, unavailable: false, stale: false },
    filesChanged: 0,
    tasksCompleted: 0,
    tasksInProgress: 0,
    tasksInReview: 0,
  };
}

function countModifiedFiles(value: string | null): number {
  if (!value) return 0;
  let files: unknown;
  try {
    files = JSON.parse(value);
  } catch {
    return 0;
  }
  if (!Array.isArray(files)) return 0;
  let count = 0;
  for (const file of files) {
    if (typeof file === "string" && file.length > 0) count += 1;
  }
  return count;
}

function addRangeClauses(column: string, clauses: string[], params: string[], query: WorkflowAnalyticsQuery): void {
  if (query.from !== undefined) {
    clauses.push(`${column} >= ?`);
    params.push(query.from);
  }
  if (query.to !== undefined) {
    clauses.push(`${column} <= ?`);
    params.push(query.to);
  }
}

/** Resolve a workflow's display name + builtin flag from a name lookup. */
type WorkflowNameResolver = (workflowId: string) => { workflowName: string; workflowIcon?: string; isBuiltin: boolean };

function resolveWorkflowNameSync(db: Database, workflowId: string): { workflowName: string; workflowIcon?: string; isBuiltin: boolean } {
  const builtin = getBuiltinWorkflow(workflowId) ?? BUILTIN_WORKFLOWS.find((workflow) => workflow.id === workflowId);
  if (builtin) return { workflowName: builtin.name, isBuiltin: true };
  const row = db.prepare("SELECT id, name, icon FROM workflows WHERE id = ?").get(workflowId) as WorkflowNameRow | undefined;
  return {
    workflowName: row?.name && row.name.length > 0 ? row.name : workflowId,
    ...(row?.icon && row.icon.length > 0 ? { workflowIcon: row.icon } : {}),
    isBuiltin: isBuiltinWorkflowId(workflowId),
  };
}

/**
 * FNXC:PostgresCommandCenterAnalytics 2026-06-28-09:30:
 * Async (PostgreSQL) name resolver. Workflow names are prefetched once from
 * project.workflows into a Map (the async connection cannot issue per-id sync
 * prepared reads), so per-workflow name resolution is an in-memory lookup that
 * mirrors resolveWorkflowNameSync's builtin-first / NULLIF-empty fallback.
 */
function resolveWorkflowNameFromMap(
  names: Map<string, string>,
  workflowId: string,
): { workflowName: string; isBuiltin: boolean } {
  const builtin = getBuiltinWorkflow(workflowId) ?? BUILTIN_WORKFLOWS.find((workflow) => workflow.id === workflowId);
  if (builtin) return { workflowName: builtin.name, isBuiltin: true };
  const name = names.get(workflowId);
  return {
    workflowName: name && name.length > 0 ? name : workflowId,
    isBuiltin: isBuiltinWorkflowId(workflowId),
  };
}

function makeSummary(resolveName: WorkflowNameResolver, workflowId: string): WorkflowSummary {
  return {
    workflowId,
    ...resolveName(workflowId),
    ...emptyMetricTotals(),
  };
}

/** Pre-fetched per-workflow row sets shared by the sync + async aggregation. */
interface WorkflowAnalyticsRows {
  tokenRows: TaskTokenRow[];
  completedRows: CountByWorkflowRow[];
  currentRows: Array<CountByWorkflowRow & { columnName: string }>;
  fileRows: ModifiedFilesRow[];
}

/**
 * FNXC:PostgresCommandCenterAnalytics 2026-06-28-09:30:
 * Pure per-workflow aggregation shared by the sync (SQLite) and async
 * (PostgreSQL) fetch paths. No I/O — takes already-fetched row sets and a name
 * resolver, so both backends produce byte-identical totals/sorting/cost
 * semantics.
 */
function buildWorkflowAnalytics(
  rows: WorkflowAnalyticsRows,
  query: WorkflowAnalyticsQuery,
  resolveName: WorkflowNameResolver,
): WorkflowAnalytics {
  const summaries = new Map<string, WorkflowSummary>();
  const costAccumulators = new Map<string, CostAccumulator>();
  const totalTokens = emptyTokenTotals();
  const totalCost = emptyCostAccumulator();
  const pricingOverrides = query.pricingOverrides;

  const ensureSummary = (workflowId: string): WorkflowSummary => {
    const existing = summaries.get(workflowId);
    if (existing) return existing;
    const created = makeSummary(resolveName, workflowId);
    summaries.set(workflowId, created);
    costAccumulators.set(workflowId, emptyCostAccumulator());
    return created;
  };

  for (const row of rows.tokenRows) {
    const summary = ensureSummary(row.workflowId);
    const workflowCost = costAccumulators.get(row.workflowId) ?? emptyCostAccumulator();
    costAccumulators.set(row.workflowId, workflowCost);
    addTokenRow(summary.tokens, row);
    addTokenRow(totalTokens, row);
    addRowCost(workflowCost, row, query.now, pricingOverrides);
    addRowCost(totalCost, row, query.now, pricingOverrides);
  }

  for (const row of rows.completedRows) {
    ensureSummary(row.workflowId).tasksCompleted = row.count;
  }

  for (const row of rows.currentRows) {
    const summary = ensureSummary(row.workflowId);
    const columnFlags = query.columnFlagsByName?.get(row.columnName);
    if (isWipColumnRole(columnFlags, row.columnName)) summary.tasksInProgress = row.count;
    if (isReviewColumnRole(columnFlags, row.columnName)) summary.tasksInReview = row.count;
  }

  for (const row of rows.fileRows) {
    ensureSummary(row.workflowId).filesChanged += countModifiedFiles(row.modifiedFiles);
  }

  for (const [workflowId, summary] of summaries) {
    summary.cost = finalizeCost(costAccumulators.get(workflowId) ?? emptyCostAccumulator());
  }

  let filesChanged = 0;
  let tasksCompleted = 0;
  let tasksInProgress = 0;
  let tasksInReview = 0;
  for (const summary of summaries.values()) {
    filesChanged += summary.filesChanged;
    tasksCompleted += summary.tasksCompleted;
    tasksInProgress += summary.tasksInProgress;
    tasksInReview += summary.tasksInReview;
  }

  const sortedWorkflows = [...summaries.values()].sort((a, b) => {
    const tokenCmp = b.tokens.totalTokens - a.tokens.totalTokens;
    if (tokenCmp !== 0) return tokenCmp;
    return a.workflowId.localeCompare(b.workflowId);
  });

  return {
    from: query.from ?? null,
    to: query.to ?? null,
    totals: {
      tokens: totalTokens,
      cost: finalizeCost(totalCost),
      filesChanged,
      tasksCompleted,
      tasksInProgress,
      tasksInReview,
    },
    workflows: sortedWorkflows,
  };
}

/**
 * Aggregate store-derived per-workflow Command Center metrics over a date range.
 *
 * FNXC:CommandCenter 2026-06-27-12:00:
 * Per-workflow analytics derive from tasks ⨝ task_workflow_selection, with the project default workflow backfilling unselected tasks. The HTTP layer passes an already project-scoped Database handle, so this pure read-only aggregator adds observability for custom workflows without introducing schema or cross-project reads.
 *
 * FNXC:PostgresCommandCenterAnalytics 2026-06-28-09:30:
 * Now accepts a `Database | AsyncDataLayer` and is async. In backend (PostgreSQL)
 * mode it branches on `"ping" in dbOrLayer` and runs schema-qualified `project.*`
 * snake_case queries via the async layer; the sync SQLite branch is unchanged.
 */
export async function aggregateWorkflowAnalytics(
  dbOrLayer: Database | AsyncDataLayer,
  query: WorkflowAnalyticsQuery = {},
  /*
  FNXC:WorkflowResolvedColumns 2026-07-30-18:30:
  The store, used ONLY to resolve which columns carry the complete / wip / human-review traits.

  These queries filtered on `t."column" = 'done'` and `IN ('in-progress','in-review')`. Those ids sit
  inside SQL strings, which the lifecycle census cannot see because it parses TypeScript comparisons —
  so the sweep that converted this file's TS guards left the queries alone and the file scored as
  converted. On a renamed board every per-workflow completed count and throughput figure reads ZERO
  while the board is busy, with no error.

  Resolved per PROJECT: this aggregates a whole project, so the union of a role's columns across its
  workflows is the right set and a bound IN list is enough. (Per-task lanes need the
  superset-then-decide-in-JS shape instead — see cleanupStaleMergeQueueRowsInTransaction.)

  Omitted, the legacy ids answer, so an unconverted caller is byte-identical.
  */
  laneStore?: ProjectLaneVocabularyStore,
): Promise<WorkflowAnalytics> {
  const defaultWorkflowId = query.defaultWorkflowId ?? "builtin:coding";
  if ("ping" in dbOrLayer) {
    return aggregateWorkflowAnalyticsAsync(dbOrLayer, query, defaultWorkflowId, laneStore);
  }
  const db = dbOrLayer as Database;

  const workflowExpr = "COALESCE(NULLIF(s.workflowId, ''), ?)";

  const tokenClauses = ["t.tokenUsageLastUsedAt IS NOT NULL"];
  const tokenParams: string[] = [defaultWorkflowId];
  addRangeClauses("t.tokenUsageLastUsedAt", tokenClauses, tokenParams, query);
  const tokenRows = db
    .prepare(
      `SELECT
         ${workflowExpr} AS workflowId,
         t.tokenUsageInputTokens AS inputTokens,
         t.tokenUsageOutputTokens AS outputTokens,
         t.tokenUsageCachedTokens AS cachedTokens,
         t.tokenUsageCacheWriteTokens AS cacheWriteTokens,
         t.tokenUsageTotalTokens AS totalTokens,
         t.modelProvider,
         t.modelId,
         t.tokenUsageModelProvider,
         t.tokenUsageModelId
       FROM tasks t
       LEFT JOIN task_workflow_selection s ON s.taskId = t.id
       WHERE ${tokenClauses.join(" AND ")}`,
    )
    .all(...tokenParams) as TaskTokenRow[];

  const completedClauses = [`t."column" = 'done'`, "t.columnMovedAt IS NOT NULL"];
  const completedParams: string[] = [defaultWorkflowId];
  addRangeClauses("t.columnMovedAt", completedClauses, completedParams, query);
  const completedRows = db
    .prepare(
      `SELECT ${workflowExpr} AS workflowId, COUNT(*) AS count
       FROM tasks t
       LEFT JOIN task_workflow_selection s ON s.taskId = t.id
       WHERE ${completedClauses.join(" AND ")}
       GROUP BY workflowId`,
    )
    .all(...completedParams) as CountByWorkflowRow[];

  const currentClauses = [`t."column" IN ('in-progress', 'in-review')`];
  const currentParams: string[] = [defaultWorkflowId];
  /*
   * FNXC:CommandCenter 2026-06-27-17:45:
   * The Workflows tab describes all task counts as range-scoped analytics. Count active workflow tasks only when their current column transition (or updatedAt fallback for legacy rows) falls inside the selected range so stale in-progress/review tasks do not suppress the empty state for unrelated windows.
   */
  addRangeClauses("COALESCE(t.columnMovedAt, t.updatedAt)", currentClauses, currentParams, query);
  const currentRows = db
    .prepare(
      `SELECT ${workflowExpr} AS workflowId, t."column" AS columnName, COUNT(*) AS count
       FROM tasks t
       LEFT JOIN task_workflow_selection s ON s.taskId = t.id
       WHERE ${currentClauses.join(" AND ")}
       GROUP BY workflowId, t."column"`,
    )
    .all(...currentParams) as Array<CountByWorkflowRow & { columnName: string }>;

  const filesClauses = ["t.modifiedFiles IS NOT NULL", "t.modifiedFiles NOT IN ('', '[]')"];
  const filesParams: string[] = [defaultWorkflowId];
  addRangeClauses("t.updatedAt", filesClauses, filesParams, query);
  const fileRows = db
    .prepare(
      `SELECT ${workflowExpr} AS workflowId, t.modifiedFiles
       FROM tasks t
       LEFT JOIN task_workflow_selection s ON s.taskId = t.id
       WHERE ${filesClauses.join(" AND ")}`,
    )
    .all(...filesParams) as ModifiedFilesRow[];

  return buildWorkflowAnalytics(
    { tokenRows, completedRows, currentRows, fileRows },
    query,
    (workflowId) => resolveWorkflowNameSync(db, workflowId),
  );
}

/**
 * FNXC:PostgresCommandCenterAnalytics 2026-06-28-09:30:
 * PostgreSQL fetch path for {@link aggregateWorkflowAnalytics}. Every table is
 * schema-qualified (`project.*`) with snake_case columns because the async
 * connection has no `project` on the search_path. The `COALESCE(NULLIF(...))`
 * default-workflow backfill, range columns, GROUP BY shape, and integer
 * coercion mirror the sync branch exactly. `modified_files` is jsonb (postgres-js
 * returns it parsed) so it is re-stringified to feed the shared
 * countModifiedFiles helper unchanged.
 */
/* The legacy active ids are always retained: the classifier's fallback recognises them, so they can
   never be the "selected but unclassifiable" case this filter exists to prevent. */
const LEGACY_ACTIVE_LANES: readonly string[] = ["in-progress", "in-review"];

async function aggregateWorkflowAnalyticsAsync(
  layer: AsyncDataLayer,
  query: WorkflowAnalyticsQuery,
  defaultWorkflowId: string,
  laneStore?: ProjectLaneVocabularyStore,
): Promise<WorkflowAnalytics> {
  /*
  FNXC:PostgresCommandCenterAnalytics 2026-07-30-21:20 (#2866 review — greptile P1 x2, and the second
  one is a self-contradiction rather than an imprecision):

  THE SQL FILTER AND THE ROW CLASSIFIER MUST DROP THE SAME COLUMNS.

  `resolveProjectColumnsForRoles` unions every column any workflow gives the role, so a column id two
  workflows reuse with DIFFERENT traits stays in the filter. `columnFlagsByName` — built by the
  Command Center route and consumed ~200 lines below — deliberately DROPS such an id as conflicting,
  on the argument that a merged entry double-counts (see its own note, #2803 review).

  The two together produce rows that are SELECTED and then classified by nothing:
  `columnFlagsByName.get("checking")` is undefined, so `isWipColumnRole` and `isReviewColumnRole` both
  fall back to the legacy ids, `checking` matches neither, and the count silently evaporates. Worse
  than either half alone — the filter says the lane counts, the classifier says it cannot say, and the
  operator sees a workflow sitting at zero.

  Aligning on the CLASSIFIER's answer is the conservative direction: a column it refuses to judge is
  removed from the filter too, so the rows are never selected rather than selected and discarded. The
  totals are unchanged (they were already lost); what changes is that the two halves now agree, and a
  reader is not left hunting for where the rows went.

  CONSEQUENTLY THIS IS UNCOVERED, AND THAT IS INHERENT, NOT AN OMISSION. Mutating the filter away
  leaves the renamed-lane suite green, because the counts were already zero on both sides — the only
  observable difference is whether a workflow with no other rows appears with zeros or is absent.
  A test pinning THAT would be asserting an artifact of , not the invariant. The
  invariant worth covering is the one the first finding names, and it needs the per-workflow map.

  The FIRST finding — a project-wide union admitting one workflow's rows into another's completed
  count — is real and NOT fixed here: unlike `team-analytics`, these rows carry a workflow id, so a
  per-workflow lane map is feasible and is the right fix. It needs the lane resolution keyed by
  workflow rather than by project, which changes this function's contract with its caller. Recorded on
  the PR rather than folded into a review-response commit.
  */
  const droppedAsConflicting = (lane: string): boolean =>
    query.columnFlagsByName !== undefined && !query.columnFlagsByName.has(lane);
  const completeLanes = laneStore
    ? [...await resolveProjectColumnsForRoles(laneStore, ["complete"])]
    : ["done"];
  const activeLanes = (laneStore
    ? [...await resolveProjectColumnsForRoles(laneStore, ["countsTowardWip", "humanReview"])]
    : ["in-progress", "in-review"]
  ).filter((lane) => LEGACY_ACTIVE_LANES.includes(lane) || !droppedAsConflicting(lane));
  /* An IN list of bound parameters, not `= ANY(${array})`: drizzle expands a JS array in a template
     into a tuple, which PostgreSQL rejects for ANY. Each id stays a parameter. */
  const inList = (lanes: readonly string[]) => sql.join(lanes.map((lane) => sql`${lane}`), sql`, `);
  const wfExpr = sql`COALESCE(NULLIF(s.workflow_id, ''), ${defaultWorkflowId})`;

  const tokFrom = query.from !== undefined ? sql`AND t.token_usage_last_used_at >= ${query.from}` : sql``;
  const tokTo = query.to !== undefined ? sql`AND t.token_usage_last_used_at <= ${query.to}` : sql``;
  const tokenRowsRaw = (await layer.db.execute(
    sql`SELECT
          ${wfExpr}                            AS "workflowId",
          t.token_usage_input_tokens           AS "inputTokens",
          t.token_usage_output_tokens          AS "outputTokens",
          t.token_usage_cached_tokens          AS "cachedTokens",
          t.token_usage_cache_write_tokens     AS "cacheWriteTokens",
          t.token_usage_total_tokens           AS "totalTokens",
          t.model_provider                     AS "modelProvider",
          t.model_id                           AS "modelId",
          t.token_usage_model_provider         AS "tokenUsageModelProvider",
          t.token_usage_model_id               AS "tokenUsageModelId"
        FROM project.tasks t
        LEFT JOIN project.task_workflow_selection s ON s.task_id = t.id
        WHERE t.token_usage_last_used_at IS NOT NULL ${tokFrom} ${tokTo}`,
  )) as Array<Record<string, unknown>>;
  const tokenRows: TaskTokenRow[] = tokenRowsRaw.map((r) => ({
    workflowId: String(r.workflowId),
    inputTokens: r.inputTokens == null ? null : Number(r.inputTokens),
    outputTokens: r.outputTokens == null ? null : Number(r.outputTokens),
    cachedTokens: r.cachedTokens == null ? null : Number(r.cachedTokens),
    cacheWriteTokens: r.cacheWriteTokens == null ? null : Number(r.cacheWriteTokens),
    totalTokens: r.totalTokens == null ? null : Number(r.totalTokens),
    modelProvider: (r.modelProvider as string | null) ?? null,
    modelId: (r.modelId as string | null) ?? null,
    tokenUsageModelProvider: (r.tokenUsageModelProvider as string | null) ?? null,
    tokenUsageModelId: (r.tokenUsageModelId as string | null) ?? null,
  }));

  const compFrom = query.from !== undefined ? sql`AND t.column_moved_at >= ${query.from}` : sql``;
  const compTo = query.to !== undefined ? sql`AND t.column_moved_at <= ${query.to}` : sql``;
  const completedRowsRaw = (await layer.db.execute(
    sql`SELECT ${wfExpr} AS "workflowId", count(*)::int AS count
        FROM project.tasks t
        LEFT JOIN project.task_workflow_selection s ON s.task_id = t.id
        WHERE t."column" IN (${inList(completeLanes)}) AND t.column_moved_at IS NOT NULL ${compFrom} ${compTo}
        GROUP BY 1`,
  )) as Array<{ workflowId: string; count: number }>;
  const completedRows: CountByWorkflowRow[] = completedRowsRaw.map((r) => ({
    workflowId: String(r.workflowId),
    count: Number(r.count),
  }));

  const curFrom = query.from !== undefined ? sql`AND COALESCE(t.column_moved_at, t.updated_at) >= ${query.from}` : sql``;
  const curTo = query.to !== undefined ? sql`AND COALESCE(t.column_moved_at, t.updated_at) <= ${query.to}` : sql``;
  const currentRowsRaw = (await layer.db.execute(
    sql`SELECT ${wfExpr} AS "workflowId", t."column" AS "columnName", count(*)::int AS count
        FROM project.tasks t
        LEFT JOIN project.task_workflow_selection s ON s.task_id = t.id
        WHERE t."column" IN (${inList(activeLanes)}) ${curFrom} ${curTo}
        GROUP BY 1, t."column"`,
  )) as Array<{ workflowId: string; columnName: string; count: number }>;
  const currentRows: Array<CountByWorkflowRow & { columnName: string }> = currentRowsRaw.map((r) => ({
    workflowId: String(r.workflowId),
    columnName: String(r.columnName),
    count: Number(r.count),
  }));

  const filesFrom = query.from !== undefined ? sql`AND t.updated_at >= ${query.from}` : sql``;
  const filesTo = query.to !== undefined ? sql`AND t.updated_at <= ${query.to}` : sql``;
  const fileRowsRaw = (await layer.db.execute(
    sql`SELECT ${wfExpr} AS "workflowId", t.modified_files AS "modifiedFiles"
        FROM project.tasks t
        LEFT JOIN project.task_workflow_selection s ON s.task_id = t.id
        WHERE t.modified_files IS NOT NULL
          AND jsonb_typeof(t.modified_files) = 'array'
          AND jsonb_array_length(t.modified_files) > 0
          ${filesFrom} ${filesTo}`,
  )) as Array<{ workflowId: string; modifiedFiles: unknown }>;
  const fileRows: ModifiedFilesRow[] = fileRowsRaw.map((r) => ({
    workflowId: String(r.workflowId),
    modifiedFiles: r.modifiedFiles == null ? null : JSON.stringify(r.modifiedFiles),
  }));

  // Prefetch all custom workflow names once; builtins resolve in-memory.
  const workflowNameRows = (await layer.db.execute(
    sql`SELECT id, name FROM project.workflows`,
  )) as Array<{ id: string; name: string | null }>;
  const names = new Map<string, string>();
  for (const row of workflowNameRows) {
    if (row.name) names.set(String(row.id), row.name);
  }

  return buildWorkflowAnalytics(
    { tokenRows, completedRows, currentRows, fileRows },
    query,
    (workflowId) => resolveWorkflowNameFromMap(names, workflowId),
  );
}