Setup and full context: https://lopes.id/log/log-health-monitoring

log-health.gs

// log-health.gs — Apps Script for Log-Source Health Monitoring
//
// Delivers Slack notifications for the log_health pipeline:
//   - checkAnomalies (every 10 min): posts new SPIKE/DIP/FULL_STOP/NEW_SOURCE
//     alerts from the anomalies table.
//   - dailyDigest (daily): source-by-source status, 24h alert tally,
//     unclassified NEW_SOURCE callout, and candidate-only watchlist.
//   - weeklyPipelineHealth (Mondays): reads pipeline_health_status; posts if
//     any log_health scheduled job didn't run in the last 25h.
//   - weeklyUnspecifiedGrowth (Mondays): reads unspecified_growth_status;
//     posts if UNSPECIFIED_LOG_TYPE volume grew >20% WoW.
//
// Setup and full context: https://lopes.id/log/log-health-monitoring
//
// Before using, replace:
//   - PROJECT_ID   → the GCP project that hosts the log_health dataset
//   - SLACK_WEBHOOK → your Slack incoming-webhook URL
//   - `chronicle-google` in embedded SQL → project holding ingestion_metrics
//   - `chronicle-self` in embedded SQL   → project holding the log_health dataset

const PROJECT_ID = "chronicle-self";
const SLACK_WEBHOOK = "https://hooks.slack.com/services/T00000000/B00000000/xxxxxxxxxxxxxxxxxxxxxxxx";

// Attachment colors — one per alert/status type.
const C_FULL_STOP = "#E01E5A"; // crimson
const C_SPIKE = "#E8A838"; // amber
const C_DIP = C_SPIKE; // shared with SPIKE — direction is carried by the emoji, not color
const C_NEW_SOURCE = "#36C5F0"; // sky blue
const C_DEAD = "#BF2600"; // dark red
const C_ALERTING = "#E01E5A";
const C_APPROACHING = "#E8A838";
const C_NEUTRAL = "#DDDDDD"; // light grey
const C_WARNING = "#E8A838";

// postSlack — posts a structured attachment message to the configured webhook.
//   fallback: plain-text summary shown in notification toasts and email digests.
//   attachments: array of { color, blocks } objects.
function postSlack(fallback, attachments) {
  UrlFetchApp.fetch(SLACK_WEBHOOK, {
    method: "post",
    contentType: "application/json",
    payload: JSON.stringify({ text: fallback, attachments: attachments }),
  });
}

// makeAttachment — color ribbon + optional bold title, body, and small footer.
//   isCode (default true): wraps body in triple-backticks for monospace tables.
//   Pass isCode=false for prose bodies that already contain mrkdwn formatting.
function makeAttachment(color, title, body, footer, isCode) {
  const blocks = [];
  if (title) {
    blocks.push({
      type: "section",
      text: { type: "mrkdwn", text: "*" + title + "*" },
    });
  }
  if (body) {
    const text = isCode !== false ? "```\n" + body + "\n```" : body;
    blocks.push({ type: "section", text: { type: "mrkdwn", text: text } });
  }
  if (footer) {
    blocks.push({
      type: "context",
      elements: [{ type: "mrkdwn", text: footer }],
    });
  }
  return { color: color, blocks: blocks };
}

// [§7 — checkAnomalies] Posts un-sent anomalies to Slack, then marks them sent.
//   Reads only normalized columns (actual_value/expected_value/delta_value/
//   severity_pct/value_unit + cluster summary) — all math is computed in BQ
//   by Step B/C/E so this function is a pure formatter. No source_config
//   join: FULL_STOP rows carry their own max_silence_hours via expected_value.
//   NEW_SOURCE rows re-fire every 24h until the source is classified — each
//   re-fire is a separate row with sent_to_slack=FALSE.
function checkAnomalies() {
  const sql = `SELECT detected_at, source_key, anomaly_type, hour_bucket,
                      actual_value, expected_value, delta_value, severity_pct, value_unit,
                      breach_first_at, breach_last_at, breach_hits,
                      breach_overflow_evts, breach_overflow_pct,
                      log_count
               FROM \`chronicle-self.log_health.anomalies\`
               WHERE sent_to_slack = FALSE
               ORDER BY detected_at`;
  const rows =
    BigQuery.Jobs.query({ query: sql, useLegacySql: false }, PROJECT_ID).rows ||
    [];
  if (!rows.length) return;

  const padR = (s, w) => String(s).padEnd(w);
  const padL = (s, w) => String(s).padStart(w);

  // fmtFull — number with thousands separators; '—' for null/NaN.
  // fmtShort — short-scale (k/M/G) for OVER and OVERFLOW where magnitude matters more than precision.
  const fmtFull = (n) => {
    const v = parseFloat(n);
    return Number.isFinite(v) ? Math.round(v).toLocaleString("en-US") : "—";
  };
  const fmtShort = (n) => {
    const v = parseFloat(n);
    if (!Number.isFinite(v)) return "—";
    const abs = Math.abs(v);
    if (abs >= 1e9) return (v / 1e9).toFixed(1) + "G";
    if (abs >= 1e6) return (v / 1e6).toFixed(1) + "M";
    if (abs >= 1e3) return Math.round(v / 1e3) + "k";
    return Math.round(v).toString();
  };
  // withUnit appends 'h' only when the row's value_unit signals hours
  //   (FULL_STOP). SPIKE/DIP carry 'evt' and render bare numbers.
  const withUnit = (s, u) => (s === "—" || u !== "h" ? s : s + "h");
  // BigQuery returns TIMESTAMP as seconds-since-epoch strings; JS Date needs ms.
  // fmtHour — HH:MM UTC. Used for SPIKE/DIP window endpoints.
  const fmtHour = (ts) =>
    ts
      ? Utilities.formatDate(new Date(parseFloat(ts) * 1000), "UTC", "HH:mm")
      : "—";
  // Compact window: "HH:MM→HH:MM" or single "HH:MM" when first=last (magnitude-only promotions).
  const fmtWindow = (first, last) => {
    if (!first || !last) return "—";
    const f = fmtHour(first),
      l = fmtHour(last);
    return f === l ? f : f + "→" + l;
  };
  // Appends the breach-hit count to the window, e.g. "14:00→22:00 (4×)".
  //   Omitted when hits<=1 — a single-hit (magnitude-only) promotion needs no count.
  const fmtWindowWithHits = (first, last, hits) => {
    const w = fmtWindow(first, last);
    const h = parseInt(hits, 10);
    return Number.isFinite(h) && h > 1 ? w + " (" + h + "×)" : w;
  };
  // fmtSignedMag — the one shared "±magnitude (percent%)" cell renderer, used by
  //   FULL_STOP's OVER, and SPIKE/DIP's PEAK and OVERFLOW. `sign` is always
  //   caller-supplied — magnitude/pct arrive pre-ABS'd from BigQuery, so the
  //   renderer can never infer direction on its own.
  //   `floor` (FULL_STOP only): the row is a lower bound (silence aged out of
  //   Step C's lookback window, value_unit='h+') — prefix "≥" and drop the ±
  //   sign, since a floored overage is always non-negative and "≥" carries it.
  const fmtSignedMag = (sign, magnitude, pct, unit, floor) => {
    const m = parseFloat(magnitude),
      p = parseFloat(pct);
    if (!Number.isFinite(m) || !Number.isFinite(p)) return "—";
    const pre = floor ? "≥" : "";
    const sgn = floor ? "" : sign;
    return (
      pre +
      sgn +
      fmtShort(m) +
      (unit === "h" ? "h" : "") +
      " (" +
      pre +
      Math.round(p) +
      "%)"
    );
  };
  // truncSrc — fit source key into exactly w chars; truncate with ellipsis if longer.
  const truncSrc = (s, w) =>
    s.length <= w ? s.padEnd(w) : s.slice(0, w - 1) + "…";

  // Layout A (FULL_STOP): SOURCE + SILENT + LIMIT + OVER = ~64 chars
  //   SILENT = hours since last event, LIMIT = max_silence_hours,
  //   OVER = excess hours as "+magnitude (pct%)" via fmtSignedMag.
  const FS_SRC_W = 32,
    FS_NUM_W = 8,
    FS_OVER_W = 13;
  // Layout B (SPIKE/DIP): TIME + SOURCE + PEAK + OVERFLOW
  //   TIME = HH:MM→HH:MM UTC breach window, with the hit count folded in as
  //          " (N×)" when N>1 (fmtWindowWithHits) — the window IS the span of
  //          those N hits, so the count belongs there rather than its own column.
  //   PEAK = worst-hour excess as "±magnitude (%)" (magnitude from delta_value,
  //          % from severity_pct — both scoped to the peak hour in the cluster).
  //   OVERFLOW = cluster-total excess as "±magnitude (weighted %)" summed across
  //          all breach hours — same fmtSignedMag renderer as PEAK, so both are
  //          always signed consistently.
  //   Order matches narrative reading: window → worst → total.
  const SD_TIME_W = 17,
    SD_SRC_W = 25,
    SD_PEAK_W = 14,
    SD_OVR_W = 14;

  const byType = { FULL_STOP: [], NEW_SOURCE: [], SPIKE: [], DIP: [] };
  rows.forEach((r) => {
    const v = r.f.map((c) => c.v);
    byType[v[2]].push({
      ts: v[0],
      src: v[1],
      type: v[2],
      hour_bucket: v[3],
      actual: v[4],
      expected: v[5],
      delta: v[6],
      severity: v[7],
      unit: v[8],
      breach_first: v[9],
      breach_last: v[10],
      breach_hits: v[11],
      breach_evts: v[12],
      breach_pct: v[13],
      legacy_cnt: v[14],
    });
  });

  // Layout A — FULL_STOP: SOURCE + SILENT + LIMIT + OVER
  const renderFullStopHeader =
    padR("SOURCE", FS_SRC_W) +
    " " +
    padL("SILENT", FS_NUM_W) +
    " " +
    padL("LIMIT", FS_NUM_W) +
    " " +
    padL("OVER", FS_OVER_W) +
    "\n";
  //   value_unit='h+' ⇒ SILENT/OVER are floors (feed silent past the 336h
  //   lookback; exact duration unmeasurable) — prefix SILENT with "≥".
  const renderFullStopRow = (a) => {
    const floor = a.unit === "h+";
    return (
      truncSrc(a.src, FS_SRC_W) +
      " " +
      padL((floor ? "≥" : "") + withUnit(fmtFull(a.actual), "h"), FS_NUM_W) +
      " " +
      padL(withUnit(fmtFull(a.expected), "h"), FS_NUM_W) +
      " " +
      padL(fmtSignedMag("+", a.delta, a.severity, "h", floor), FS_OVER_W) +
      "\n"
    );
  };

  // Layout B — SPIKE/DIP: TIME + SOURCE + PEAK + OVERFLOW
  const renderSpikeDipHeader =
    padR("TIME", SD_TIME_W) +
    " " +
    padR("SOURCE", SD_SRC_W) +
    " " +
    padL("PEAK", SD_PEAK_W) +
    " " +
    padL("OVERFLOW", SD_OVR_W) +
    "\n";
  const renderSpikeDipRow = (a) => {
    const sign = a.type === "DIP" ? "-" : "+";
    return (
      padR(
        fmtWindowWithHits(a.breach_first, a.breach_last, a.breach_hits),
        SD_TIME_W,
      ) +
      " " +
      truncSrc(a.src, SD_SRC_W) +
      " " +
      padL(fmtSignedMag(sign, a.delta, a.severity), SD_PEAK_W) +
      " " +
      padL(fmtSignedMag(sign, a.breach_evts, a.breach_pct), SD_OVR_W) +
      "\n"
    );
  };

  // buildTableBody — sorts by severity desc, returns raw table text (no backtick wrap).
  const buildTableBody = (bucket, header, rowFn) => {
    bucket.sort((a, b) => parseFloat(b.severity) - parseFloat(a.severity));
    let s = header;
    bucket.forEach((a) => {
      s += rowFn(a);
    });
    return s.trimEnd();
  };

  const nowUtc = Utilities.formatDate(new Date(), "UTC", "yyyy-MM-dd HH:mm");
  const attachments = [];

  // FULL_STOP first (most severe — feed went dark).
  if (byType.FULL_STOP.length) {
    attachments.push(
      makeAttachment(
        C_FULL_STOP,
        ":red_circle: FULL STOP (" + byType.FULL_STOP.length + ")",
        buildTableBody(
          byType.FULL_STOP,
          renderFullStopHeader,
          renderFullStopRow,
        ),
        "SILENT/LIMIT in hours; OVER = hours over limit (% of limit); ≥ = silent past the 336h lookback, floor only · UTC",
      ),
    );
  }

  if (byType.SPIKE.length) {
    attachments.push(
      makeAttachment(
        C_SPIKE,
        ":arrow_up: SPIKE (" + byType.SPIKE.length + ")",
        buildTableBody(byType.SPIKE, renderSpikeDipHeader, renderSpikeDipRow),
        ":clock3: All times UTC",
      ),
    );
  }

  if (byType.DIP.length) {
    attachments.push(
      makeAttachment(
        C_DIP,
        ":arrow_down: DIP (" + byType.DIP.length + ")",
        buildTableBody(byType.DIP, renderSpikeDipHeader, renderSpikeDipRow),
        ":clock3: All times UTC",
      ),
    );
  }

  // NEW_SOURCE stays distinct — no baseline exists, so the canonical
  //   SILENT/LIMIT/OVER/% columns don't apply. actual_value carries
  //   the 24h event count (populated by Step D); legacy_cnt is a fallback
  //   for pre-migration rows that still wrote to log_count only.
  if (byType.NEW_SOURCE.length) {
    byType.NEW_SOURCE.sort(
      (a, b) =>
        parseFloat(b.actual || b.legacy_cnt) -
        parseFloat(a.actual || a.legacy_cnt),
    );
    const NS_SRC_W = 38,
      NS_NUM_W = 12,
      NS_DATE_W = 16;
    const fmtDate = (ts) =>
      ts
        ? Utilities.formatDate(
            new Date(parseFloat(ts) * 1000),
            "UTC",
            "yyyy-MM-dd HH:mm",
          )
        : "—";
    let nsBody =
      padR("SOURCE", NS_SRC_W) +
      " " +
      padL("24H EVENTS", NS_NUM_W) +
      " " +
      padL("FIRST SEEN", NS_DATE_W) +
      "\n";
    byType.NEW_SOURCE.forEach((a) => {
      nsBody +=
        padR(a.src, NS_SRC_W) +
        " " +
        padL(fmtFull(a.actual != null ? a.actual : a.legacy_cnt), NS_NUM_W) +
        " " +
        padL(fmtDate(a.hour_bucket), NS_DATE_W) +
        "\n";
    });
    attachments.push(
      makeAttachment(
        C_NEW_SOURCE,
        ":large_blue_circle: NEW SOURCE (" + byType.NEW_SOURCE.length + ")",
        nsBody.trimEnd(),
        null,
      ),
    );
  }

  postSlack(
    ":rotating_light: `Log-Source Health Alerts — " + nowUtc + " UTC`",
    attachments,
  );

  BigQuery.Jobs.query(
    {
      query:
        "UPDATE `chronicle-self.log_health.anomalies` SET sent_to_slack = TRUE WHERE sent_to_slack = FALSE",
      useLegacySql: false,
    },
    PROJECT_ID,
  );
}

// [§7 — dailyDigest] Tabulated status of every source, grouped by status, plus 24h alert
//   tally, an Unclassified-NEW_SOURCE callout, and a candidate-only watchlist.
//   Code-block tables (Slack monospace), sorted by severity within each section.
function dailyDigest() {
  // Compute hours_over and pct_of_threshold in SQL so Apps Script is a pure
  //   formatter; same math source-of-truth as Step C's FULL_STOP severity.
  const inventorySql = `
    WITH latest_per_source AS (
      SELECT log_type AS source_key, MAX(start_time) AS latest_event
      FROM \`chronicle-google.datalake.ingestion_metrics\`
      WHERE start_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 336 HOUR)
        AND event_count > 0
      GROUP BY log_type
    ),
    computed AS (
      SELECT c.source_key,
             c.detection_method,
             c.max_silence_hours,
             CAST(TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), l.latest_event, HOUR) AS INT64) AS hours_silent
      FROM \`chronicle-self.log_health.source_config\` c
      LEFT JOIN latest_per_source l ON l.source_key = c.source_key
      WHERE c.enabled
    )
    SELECT source_key,
           detection_method,
           max_silence_hours,
           hours_silent,
           CAST(hours_silent - max_silence_hours AS INT64) AS hours_over,
           CAST(ROUND(SAFE_DIVIDE(hours_silent, max_silence_hours) * 100) AS INT64) AS pct_of_threshold,
           CASE
             WHEN hours_silent IS NULL THEN 'DEAD'
             WHEN hours_silent > max_silence_hours THEN 'ALERTING'
             WHEN hours_silent > max_silence_hours * 0.75 THEN 'APPROACHING'
             ELSE 'HEALTHY'
           END AS status
    FROM computed
    ORDER BY source_key
  `;

  const alertSql = `
    SELECT anomaly_type, COUNT(*) AS n
    FROM \`chronicle-self.log_health.anomalies\`
    WHERE detected_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
    GROUP BY anomaly_type
  `;

  // Candidate-only sources — SPIKE/DIP detections that did NOT escalate to alerts
  //   (failed Step E's persistence ≥4/24h AND magnitude ≥200% bars). Surface them as
  //   capacity-planning input: "these are misbehaving but not enough to page anyone".
  //   HAVING hits >= 2 drops single-hit transient bursts that aren't worth even reading.
  const candidateSql = `
    SELECT source_key, anomaly_type, COUNT(*) AS hits,
           ROUND(MAX(severity_pct), 0) AS worst_pct
    FROM \`chronicle-self.log_health.anomaly_candidates\`
    WHERE detected_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
      AND promoted = FALSE
    GROUP BY source_key, anomaly_type
    HAVING hits >= 2
    ORDER BY hits DESC, worst_pct DESC
    LIMIT 30
  `;

  // Unclassified sources — log_types that fired NEW_SOURCE in the last 25h and
  //   haven't been added to source_config or excluded_log_types yet.
  //   The 25h window mirrors Step D's 24h dedup + 1h buffer: if a source stopped
  //   sending events, Step D stops inserting NEW_SOURCE rows (HAVING events_24h > 0),
  //   so last_alerted_at falls outside the window and it drops off this list.
  //   Sources that were deprecated (like PAN_CORTEX_XDR_EVENTS) self-clear within
  //   25h of their last event — no manual cleanup needed. See §9.10 (a) for the runbook.
  const unclassifiedSql = `
    WITH new_source_history AS (
      SELECT source_key,
             MIN(detected_at) AS first_alerted_at,
             MAX(detected_at) AS last_alerted_at,
             COUNT(*) AS times_alerted
      FROM \`chronicle-self.log_health.anomalies\`
      WHERE anomaly_type = 'NEW_SOURCE'
      GROUP BY source_key
    )
    SELECT h.source_key,
           h.times_alerted,
           CAST(TIMESTAMP_DIFF(CURRENT_TIMESTAMP(), h.first_alerted_at, DAY) AS INT64) AS days_pending
    FROM new_source_history h
    WHERE h.source_key NOT IN (SELECT source_key FROM \`chronicle-self.log_health.source_config\`)
      AND h.source_key NOT IN (SELECT log_type   FROM \`chronicle-self.log_health.excluded_log_types\`)
      AND h.last_alerted_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 25 HOUR)
    ORDER BY days_pending DESC, h.source_key
  `;

  // Run all four queries.
  const invRows =
    BigQuery.Jobs.query(
      { query: inventorySql, useLegacySql: false },
      PROJECT_ID,
    ).rows || [];
  const alertRows =
    BigQuery.Jobs.query({ query: alertSql, useLegacySql: false }, PROJECT_ID)
      .rows || [];
  const candRows =
    BigQuery.Jobs.query(
      { query: candidateSql, useLegacySql: false },
      PROJECT_ID,
    ).rows || [];
  const unclRows =
    BigQuery.Jobs.query(
      { query: unclassifiedSql, useLegacySql: false },
      PROJECT_ID,
    ).rows || [];

  // Bucket sources by status. hours_over and pct_of_threshold come
  //   pre-computed from inventorySql so Apps Script does no math.
  const groups = { DEAD: [], ALERTING: [], APPROACHING: [], HEALTHY: [] };
  invRows.forEach((r) => {
    const [src, method, max_h, silent_h, over_h, pct_th, status] = r.f.map(
      (c) => c.v,
    );
    groups[status].push({
      src,
      method,
      max_h: parseInt(max_h),
      silent_h: parseInt(silent_h),
      over_h: parseInt(over_h),
      pct_th: parseInt(pct_th),
    });
  });

  // Roll up the 24h alert counts.
  const c24 = { SPIKE: 0, DIP: 0, FULL_STOP: 0, NEW_SOURCE: 0 };
  alertRows.forEach((r) => {
    const [type, n] = r.f.map((c) => c.v);
    c24[type] = parseInt(n);
  });

  // Formatting helpers — same approach as checkAnomalies.
  const padR = (s, w) => String(s).padEnd(w);
  const padL = (s, w) => String(s).padStart(w);
  const SRC_W = 38,
    NUM_W = 7,
    PCT_W = 5;

  const today = Utilities.formatDate(
    new Date(),
    "America/Sao_Paulo",
    "yyyy-MM-dd",
  );
  const attachments = [];

  // DEAD — no severity dimension; sort by source name.
  if (groups.DEAD.length) {
    groups.DEAD.sort((a, b) => a.src.localeCompare(b.src));
    let body = padR("SOURCE", SRC_W) + " " + padL("METHOD", 10) + "\n";
    groups.DEAD.forEach((s) => {
      body += padR(s.src, SRC_W) + " " + padL(s.method, 10) + "\n";
    });
    attachments.push(
      makeAttachment(
        C_DEAD,
        ":skull: DEAD (" + groups.DEAD.length + ") — no data in 14+ days",
        body.trimEnd(),
        null,
      ),
    );
  }

  // ALERTING — sort by overage descending.
  if (groups.ALERTING.length) {
    groups.ALERTING.sort((a, b) => b.over_h - a.over_h);
    let body =
      padR("SOURCE", SRC_W) +
      " " +
      padL("SILENT", NUM_W) +
      "  " +
      padL("LIMIT", NUM_W) +
      "  " +
      padL("OVER", NUM_W) +
      "\n";
    groups.ALERTING.forEach((s) => {
      body +=
        padR(s.src, SRC_W) +
        " " +
        padL(s.silent_h + "h", NUM_W) +
        "  " +
        padL(s.max_h + "h", NUM_W) +
        "  " +
        padL("+" + s.over_h + "h", NUM_W) +
        "\n";
    });
    attachments.push(
      makeAttachment(
        C_ALERTING,
        ":red_circle: ALERTING (" +
          groups.ALERTING.length +
          ") — silent past max_silence_hours",
        body.trimEnd(),
        null,
      ),
    );
  }

  // APPROACHING — sort by % of threshold descending.
  if (groups.APPROACHING.length) {
    groups.APPROACHING.sort((a, b) => b.pct_th - a.pct_th);
    let body =
      padR("SOURCE", SRC_W) +
      " " +
      padL("SILENT", NUM_W) +
      "  " +
      padL("LIMIT", NUM_W) +
      "  " +
      padL("%", PCT_W) +
      "\n";
    groups.APPROACHING.forEach((s) => {
      body +=
        padR(s.src, SRC_W) +
        " " +
        padL(s.silent_h + "h", NUM_W) +
        "  " +
        padL(s.max_h + "h", NUM_W) +
        "  " +
        padL(s.pct_th + "%", PCT_W) +
        "\n";
    });
    attachments.push(
      makeAttachment(
        C_APPROACHING,
        ":large_yellow_circle: APPROACHING (" +
          groups.APPROACHING.length +
          ") — silent past 75% of threshold",
        body.trimEnd(),
        null,
      ),
    );
  }

  // HEALTHY summary + 24h tally — prose, no code block.
  const summaryText =
    ":large_green_circle: *HEALTHY*: " +
    groups.HEALTHY.length +
    " sources\n" +
    ":bell: *Last 24h alerts*: " +
    c24.SPIKE +
    " SPIKE · " +
    c24.DIP +
    " DIP · " +
    c24.FULL_STOP +
    " FULL_STOP · " +
    c24.NEW_SOURCE +
    " NEW_SOURCE";
  attachments.push(makeAttachment(C_NEUTRAL, null, summaryText, null, false));

  // Unclassified NEW_SOURCEs — pending classification (§9.10 a).
  if (unclRows.length) {
    let body =
      padR("SOURCE", SRC_W) +
      " " +
      padL("DAYS PENDING", 13) +
      "  " +
      padL("PINGS", 6) +
      "\n";
    unclRows.forEach((r) => {
      const [src, times, days] = r.f.map((c) => c.v);
      body +=
        padR(src, SRC_W) +
        " " +
        padL(days + "d", 13) +
        "  " +
        padL(times, 6) +
        "\n";
    });
    attachments.push(
      makeAttachment(
        C_APPROACHING,
        ":question: Unclassified NEW_SOURCEs (" +
          unclRows.length +
          ") — add to source_config or excluded_log_types",
        body.trimEnd(),
        null,
      ),
    );
  }

  // Candidate-only — sources that produced SPIKE/DIP candidates without escalating.
  //   Read as: "model thinks these are anomalous but they don't meet persistence/magnitude
  //   bars for paging. Treat as capacity-planning input."
  if (candRows.length) {
    let body =
      padR("SOURCE", SRC_W) +
      " " +
      padL("TYPE", 6) +
      "  " +
      padL("HITS", 5) +
      "  " +
      padL("WORST %", 8) +
      "\n";
    candRows.forEach((r) => {
      const [src, type, hits, worst] = r.f.map((c) => c.v);
      body +=
        padR(src, SRC_W) +
        " " +
        padL(type, 6) +
        "  " +
        padL(hits, 5) +
        "  " +
        padL(Math.round(parseFloat(worst)) + "%", 8) +
        "\n";
    });
    attachments.push(
      makeAttachment(
        C_NEUTRAL,
        ":eye: Candidate-only last 24h (" +
          candRows.length +
          ") — fired but did not escalate",
        body.trimEnd(),
        null,
      ),
    );
  }

  postSlack(
    ":bar_chart: `Log-Source Health Daily Digest — " + today + "`",
    attachments,
  );
}

// [§6.3 / §7 — weeklyPipelineHealth] Polls pipeline_health_status (overwritten weekly
//   by §6.3's scheduled query). Empty table = healthy week. Any rows = a log_health
//   job didn't run in the last 25h or finished non-DONE; posts a Slack alert.
function weeklyPipelineHealth() {
  const sql = `SELECT statement_type, last_state, last_run
               FROM \`chronicle-self.log_health.pipeline_health_status\`
               ORDER BY statement_type`;
  const rows =
    BigQuery.Jobs.query({ query: sql, useLegacySql: false }, PROJECT_ID).rows ||
    [];
  if (!rows.length) return;

  const padR = (s, w) => String(s).padEnd(w);
  const padL = (s, w) => String(s).padStart(w);
  const fmtWhen = (ts) =>
    ts
      ? Utilities.formatDate(
          new Date(parseFloat(ts) * 1000),
          "UTC",
          "yyyy-MM-dd HH:mm",
        )
      : "—";

  const nowUtc = Utilities.formatDate(
    new Date(),
    "UTC",
    "yyyy-MM-dd HH:mm",
  );
  let body =
    padR("STATEMENT_TYPE", 22) +
    " " +
    padR("LAST STATE", 12) +
    " " +
    padL("LAST RUN (UTC)", 18) +
    "\n";
  rows.forEach((r) => {
    const [stmt, last_state, last_run] = r.f.map((c) => c.v);
    body +=
      padR(stmt || "—", 22) +
      " " +
      padR(last_state || "—", 12) +
      " " +
      padL(fmtWhen(last_run), 18) +
      "\n";
  });

  postSlack(":warning: `Log-Source Health — Pipeline issue(s) detected`", [
    makeAttachment(
      C_WARNING,
      ":warning: Pipeline Issues — last 7d (" + rows.length + ")",
      body.trimEnd(),
      nowUtc + " UTC",
    ),
  ]);
}

// [§6.4 / §7 — weeklyUnspecifiedGrowth] Polls unspecified_growth_status (overwritten
//   weekly by §6.4's scheduled query). Empty table = stable; any row = WoW growth in
//   UNSPECIFIED_LOG_TYPE volume exceeded 20% (signal of an integration dropping events
//   into the Chronicle catch-all bucket).
function weeklyUnspecifiedGrowth() {
  const sql = `SELECT current_week, prior_week, growth_pct
               FROM \`chronicle-self.log_health.unspecified_growth_status\``;
  const rows =
    BigQuery.Jobs.query({ query: sql, useLegacySql: false }, PROJECT_ID).rows ||
    [];
  if (!rows.length) return;

  const [current, prior, growth] = rows[0].f.map((c) => c.v);
  const fmt = (n) => Math.round(parseFloat(n)).toLocaleString("en-US");
  const nowBrt = Utilities.formatDate(
    new Date(),
    "America/Sao_Paulo",
    "yyyy-MM-dd HH:mm",
  );

  const bodyText =
    "Current week: *" +
    fmt(current) +
    "* events · Prior week: " +
    fmt(prior) +
    " events\n" +
    "Growth: *" +
    growth +
    "%* (threshold: 20%)\n" +
    "_Investigate which integrations are dropping events into the Chronicle catch-all bucket._";

  postSlack(":warning: `Log-Source Health — Unspecified-Log-Type Growth`", [
    makeAttachment(
      C_WARNING,
      ":chart_with_upwards_trend: UNSPECIFIED_LOG_TYPE — WoW Growth",
      bodyText,
      nowBrt + " BRT",
      false,
    ),
  ]);
}