diff --git a/bugzooka/analysis/log_analyzer.py b/bugzooka/analysis/log_analyzer.py
index 2d187d6..ed8f599 100644
--- a/bugzooka/analysis/log_analyzer.py
+++ b/bugzooka/analysis/log_analyzer.py
@@ -1,5 +1,6 @@
 import logging
 import asyncio
+import tempfile
 from pydantic import BaseModel, Field
 
 from langchain_core.tools import StructuredTool
@@ -15,7 +16,12 @@
 from bugzooka.analysis.log_summarizer import (
     download_prow_logs,
     generate_prompt,
+    find_pod_latency_file,
+    download_pod_latency_file,
+    parse_slowest_pod,
+    download_node_journal,
 )
+from bugzooka.analysis.node_log_analyzer import analyze_node_journal, format_result_markdown
 from bugzooka.integrations.inference_client import (
     get_inference_client,
     analyze_with_agentic,
@@ -26,7 +32,8 @@
 from bugzooka.integrations.mcp_client import initialize_global_resources_async
 from bugzooka.core.config import get_prompt_config
 from bugzooka.analysis.prow_analyzer import analyze_prow_artifacts, ProwAnalysisResult
-from bugzooka.core.utils import extract_job_details
+from bugzooka.core.utils import extract_job_details, extract_gcs_path
+from bugzooka.analysis.log_summarizer import get_prow_inner_artifact_files
 
 logger = logging.getLogger(__name__)
 
@@ -228,3 +235,71 @@ def _run():
     result = _run()
     retry_count = _run.statistics.get("attempt_number", 1) - 1
     return result, retry_count
+
+
+def run_node_rca_analysis(job_url: str, step_hint: str = "") -> str:
+    """
+    Full pipeline: prow URL → slowest pod → node journal → RCA markdown.
+
+    Steps:
+      1. Extract GCS path from prow view URL
+      2. Discover log_folder via get_prow_inner_artifact_files
+      3. Find podLatencyMeasurement JSON in openshift-qe-* step dir
+      4. Download + parse JSON to find slowest pod and its node
+      5. Download + decompress node journal (gzipped, no .gz suffix)
+      6. Run deterministic parser, return formatted markdown
+
+    :param job_url: prow view URL (https://prow.ci.../view/gs/...)
+    :param step_hint: workload name (e.g. "node-density-cni") to select the right
+        openshift-qe-* artifacts directory when multiple workloads are present
+    :return: markdown RCA report string, or an error message string
+    """
+    try:
+        gcs_path = extract_gcs_path(job_url)
+    except Exception as exc:
+        return f"node-rca: could not extract GCS path from URL: {exc}"
+
+    log_folder_path, _ = get_prow_inner_artifact_files(gcs_path)
+    if not log_folder_path:
+        return "node-rca: could not find inner artifact folder (is this a valid prow URL?)."
+
+    # log_folder_path looks like "gs://bucket/.../artifacts/<log_folder>/"
+    # extract just the log_folder name
+    log_folder = log_folder_path.rstrip("/").split("/artifacts/", 1)[-1].rstrip("/")
+
+    tmp_dir = tempfile.mkdtemp(prefix="bugzooka-node-rca-")
+
+    metrics_url = find_pod_latency_file(gcs_path, log_folder, step_hint=step_hint or None)
+    if not metrics_url:
+        return (
+            "node-rca: podLatencyMeasurement JSON not found under "
+            f"artifacts/{log_folder}/openshift-qe-*. "
+            "Is this a node-density or similar perf job?"
+        )
+
+    json_path = download_pod_latency_file(metrics_url, tmp_dir)
+    if not json_path:
+        return f"node-rca: failed to download {metrics_url}."
+
+    pod_name, node_hostname, latency_ms = parse_slowest_pod(json_path)
+    if not pod_name or not node_hostname:
+        return "node-rca: could not determine slowest pod from podLatencyMeasurement JSON."
+
+    logger.info(
+        "node-rca: slowest pod=%s node=%s latency=%dms",
+        pod_name, node_hostname, latency_ms,
+    )
+
+    journal_path = download_node_journal(gcs_path, log_folder, node_hostname, tmp_dir)
+    if not journal_path:
+        return (
+            f"node-rca: could not download journal for node {node_hostname}. "
+            "Check that gather-extra artifacts are present for this run."
+        )
+
+    rca_result = analyze_node_journal(journal_path, pod_name=pod_name)
+    header = (
+        f"**Node RCA for `{pod_name}`** "
+        f"(slowest pod, {latency_ms / 1000:.1f}s podReadyLatency)\n\n"
+    )
+    return header + format_result_markdown(rca_result)
diff --git a/bugzooka/analysis/log_summarizer.py b/bugzooka/analysis/log_summarizer.py
index ed3c149..c3b9a9f 100644
--- a/bugzooka/analysis/log_summarizer.py
+++ b/bugzooka/analysis/log_summarizer.py
@@ -1,3 +1,5 @@
+import gzip
+import json
 import logging
 import os
 import re
@@ -381,6 +383,167 @@ def generate_prompt(error_list):
     return messages
 
 
+# ---------------------------------------------------------------------------
+# Node journal RCA helpers
+# ---------------------------------------------------------------------------
+
+
+def find_pod_latency_file(
+    gcs_path: str, log_folder: str, step_hint: Optional[str] = None
+) -> Optional[str]:
+    """
+    Locate podLatencyMeasurement-*.json under a prow artifact directory by
+    listing GCS at each variable path segment.
+
+    Expected layout:
+        gs://<gcs_path>/artifacts/<log_folder>/openshift-qe-*/
+            artifacts/collected-metrics-<uuid>/podLatencyMeasurement-*.json
+
+    :param gcs_path: raw GCS path without gs:// prefix
+    :param log_folder: inner log folder name (from get_prow_inner_artifact_files)
+    :param step_hint: workload name (e.g. "node-density-cni") to prefer the matching
+        openshift-qe-<step_hint> directory; falls back to all openshift-qe-* dirs
+    :return: full gs:// URL of the JSON file, or None if not found
+    """
+    base = f"gs://{gcs_path}/artifacts/{log_folder}/"
+    try:
+        top_entries = list_gcs_files(base)
+    except Exception as exc:
+        logger.error("find_pod_latency_file: cannot list %s: %s", base, exc)
+        return None
+
+    # Find openshift-qe-* step directories (GCS lists dirs with trailing /)
+    qe_dirs = [e for e in top_entries if e.endswith("/")
+               and "openshift-qe-" in gcs_basename(e.rstrip("/"))]
+    if not qe_dirs:
+        logger.info("find_pod_latency_file: no openshift-qe-* dirs under %s", base)
+        return None
+
+    # When a step_hint is given (e.g. "node-density-cni"), search matching dirs
+    # first so we pick the right workload's metrics, not an alphabetically earlier one.
+    if step_hint:
+        preferred = [d for d in qe_dirs if step_hint in gcs_basename(d.rstrip("/"))]
+        rest = [d for d in qe_dirs if d not in preferred]
+        qe_dirs = preferred + rest
+
+    for qe_dir in qe_dirs:
+        artifacts_path = qe_dir.rstrip("/") + "/artifacts/"
+        try:
+            artifacts_entries = list_gcs_files(artifacts_path)
+        except Exception:
+            continue
+
+        metrics_dirs = [
+            e for e in artifacts_entries
+            if e.endswith("/") and "collected-metrics-" in gcs_basename(e.rstrip("/"))
+        ]
+        for metrics_dir in metrics_dirs:
+            try:
+                files = list_gcs_files(metrics_dir.rstrip("/") + "/")
+            except Exception:
+                continue
+            for f in files:
+                name = gcs_basename(f.rstrip("/"))
+                if name.startswith("podLatencyMeasurement-") and name.endswith(".json"):
+                    return f.rstrip()
+
+    logger.info("find_pod_latency_file: no podLatencyMeasurement JSON found under %s", base)
+    return None
+
+
+def download_pod_latency_file(gcs_url: str, output_dir: str) -> Optional[str]:
+    """
+    Download a podLatencyMeasurement JSON from GCS.
+
+    :param gcs_url: full gs:// URL
+    :param output_dir: local directory to write the file
+    :return: local file path, or None on failure
+    """
+    try:
+        download_file_from_gcs(gcs_url, output_dir)
+        local_path = os.path.join(output_dir, gcs_basename(gcs_url))
+        if os.path.exists(local_path):
+            return local_path
+    except Exception as exc:
+        logger.error("download_pod_latency_file failed: %s", exc)
+    return None
+
+
+def parse_slowest_pod(json_path: str) -> Tuple[Optional[str], Optional[str], int]:
+    """
+    Read a podLatencyMeasurement JSON and return the pod with the highest
+    podReadyLatency.
+
+    :param json_path: local path to the JSON file
+    :return: (pod_name, node_name, latency_ms) — all None/0 on failure
+    """
+    try:
+        with open(json_path, encoding="utf-8") as fh:
+            records = json.load(fh)
+        if not records:
+            return None, None, 0
+        slowest = max(records, key=lambda r: r.get("podReadyLatency", 0))
+        return (
+            slowest.get("podName"),
+            slowest.get("nodeName"),
+            slowest.get("podReadyLatency", 0),
+        )
+    except Exception as exc:
+        logger.error("parse_slowest_pod failed for %s: %s", json_path, exc)
+        return None, None, 0
+
+
+def download_node_journal(
+    gcs_path: str,
+    log_folder: str,
+    node_hostname: str,
+    output_dir: str,
+) -> Optional[str]:
+    """
+    Download the node's systemd journal (gzipped without .gz suffix) and
+    decompress it to a plain-text file.
+
+    GCS path:
+        gs://<gcs_path>/artifacts/<log_folder>/gather-extra/artifacts/nodes/<node>/journal
+
+    :param gcs_path: raw GCS path without gs:// prefix
+    :param log_folder: inner log folder name
+    :param node_hostname: full node hostname (e.g. ip-10-0-67-181.us-west-2...)
+    :param output_dir: local directory for downloaded + decompressed files
+    :return: path to decompressed journal file, or None on failure
+    """
+    journal_url = (
+        f"gs://{gcs_path}/artifacts/{log_folder}/"
+        f"gather-extra/artifacts/nodes/{node_hostname}/journal"
+    )
+    node_dir = os.path.join(output_dir, node_hostname)
+    os.makedirs(node_dir, exist_ok=True)
+
+    try:
+        download_file_from_gcs(journal_url, node_dir)
+    except Exception as exc:
+        logger.error("download_node_journal: gsutil failed for %s: %s", journal_url, exc)
+        return None
+
+    compressed_path = os.path.join(node_dir, "journal")
+    if not os.path.exists(compressed_path):
+        logger.error("download_node_journal: journal not found at %s", compressed_path)
+        return None
+
+    decompressed_path = compressed_path + ".decompressed"
+    try:
+        with gzip.open(compressed_path, "rt", errors="replace") as f_in, \
+                open(decompressed_path, "w", encoding="utf-8") as f_out:
+            f_out.write(f_in.read())
+        return decompressed_path
+    except Exception as exc:
+        logger.error(
+            "download_node_journal: decompression failed for %s: %s",
+            compressed_path, exc,
+        )
+        return None
+
+
 def classify_failure_type(errors_list, categorization_message, is_install_issue):
     """
     Map analysis outputs to a display label for failure type.
diff --git a/bugzooka/analysis/node_log_analyzer.py b/bugzooka/analysis/node_log_analyzer.py
new file mode 100644
index 0000000..20c79e6
--- /dev/null
+++ b/bugzooka/analysis/node_log_analyzer.py
@@ -0,0 +1,503 @@
+"""
+Node journal log analyzer for Kubernetes node-density / pod startup RCA.
+
+Parses journal.trimmed / journal.folded.log / gzip-decompressed journal files
+from prow gather-extra artifacts and extracts deterministic metrics:
+  - Pod lifecycle timeline (ADD → sandbox → CNI → container start → PLEG)
+  - PLEG detection lag per container (crio StartContainer vs PLEG event)
+  - Housekeeping overruns (count, peak, time-bucketed)
+  - PLEG relist silence gaps > 2s
+  - Node-wide SLO distribution (pod_startup_latency_tracker)
+  - Peak concurrent pod activity
+
+Standalone CLI:
+    python -m bugzooka.analysis.node_log_analyzer \\
+        --log /path/to/journal --pod node-density-956
+
+BugZooka integration:
+    from bugzooka.analysis.node_log_analyzer import analyze_node_journal, format_result_markdown
+    result = analyze_node_journal("/tmp/journal.decompressed", pod_name="node-density-956")
+    print(format_result_markdown(result))
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import re
+import sys
+from pathlib import Path
+from typing import NamedTuple, Optional
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Regex patterns
+# ---------------------------------------------------------------------------
+
+_TS_RE = re.compile(
+    r"^\w{3}\s+\d+\s+(\d{2}:\d{2}:\d{2}\.\d+)\s+\S+\s+(\S+):\s*(.*)"
+)
+_CRIO_TS_RE = re.compile(r'time="[^"]*T(\d{2}:\d{2}:\d{2}\.\d+)')
+_PLEG_DATA_RE = re.compile(r'"Data":"([a-f0-9]+)"')
+_CRIO_STARTED_ID_RE = re.compile(r"containerID=([a-f0-9]+)")
+_HK_ACTUAL_RE = re.compile(r'actual="([\d.]+)')
+_SLO_RE = re.compile(r"podStartSLOduration=([\d.]+)")
+_CRIO_CREATED_ID_RE = re.compile(r"Created container ([a-f0-9]+):")
+_UID_RE = re.compile(
+    r'kubernetes\.io/projected/([a-f0-9-]+)-kube-api-access'
+)
+
+
+def _ts_to_secs(ts: str) -> float:
+    h, m, rest = ts.split(":", 2)
+    return int(h) * 3600 + int(m) * 60 + float(rest)
+
+
+def _fmt(secs: float) -> str:
+    return f"{secs:.3f}s"
+
+
+# ---------------------------------------------------------------------------
+# Result types
+# ---------------------------------------------------------------------------
+
+
+class TimelineEvent(NamedTuple):
+    timestamp: str
+    source: str
+    event: str
+
+
+class PlegLag(NamedTuple):
+    container_id: str
+    role: str               # "pause" or "app"
+    crio_started_ts: Optional[str]
+    pleg_event_ts: str
+    lag_secs: float
+
+
+class HkOverrun(NamedTuple):
+    timestamp: str
+    actual_secs: float
+
+
+class PlegGap(NamedTuple):
+    start_ts: str
+    end_ts: str
+    gap_secs: float
+
+
+class SloStats(NamedTuple):
+    count: int
+    min_secs: float
+    p50_secs: float
+    p90_secs: float
+    max_secs: float
+
+
+class NodeLogRCAResult(NamedTuple):
+    node: str
+    log_path: str
+    pod: Optional[str]
+    pod_uid: Optional[str]
+    timeline: list
+    pleg_lags: list
+    hk_overruns: list
+    pleg_gaps: list
+    slo_stats: Optional[SloStats]
+    peak_concurrency: int
+    peak_concurrency_ts: Optional[str]
+    summary: str
+
+
+# ---------------------------------------------------------------------------
+# Core parser
+# ---------------------------------------------------------------------------
+
+
+def analyze_node_journal(
+    log_path: str,
+    pod_name: Optional[str] = None,
+) -> NodeLogRCAResult:
+    """
+    Parse a node journal file and return structured RCA metrics.
+
+    :param log_path: path to journal file (plain text or gzip-decompressed)
+    :param pod_name: pod name to anchor the lifecycle timeline
+    :return: NodeLogRCAResult
+    """
+    path = Path(log_path)
+    node = path.parent.name
+
+    try:
+        with open(path, encoding="utf-8", errors="replace") as fh:
+            lines = fh.readlines()
+    except OSError as exc:
+        logger.error("Cannot read %s: %s", log_path, exc)
+        return NodeLogRCAResult(
+            node=node, log_path=log_path, pod=pod_name, pod_uid=None,
+            timeline=[], pleg_lags=[], hk_overruns=[], pleg_gaps=[],
+            slo_stats=None, peak_concurrency=0, peak_concurrency_ts=None,
+            summary=f"Could not read log file: {exc}",
+        )
+
+    pod_uid: Optional[str] = None
+    timeline_raw: list[TimelineEvent] = []
+
+    # container_id -> crio StartContainer timestamp
+    crio_start_ts: dict[str, str] = {}
+    # anchor pod PLEG events: [(ts, container_id)]
+    pleg_pod_events: list[tuple[str, str]] = []
+    # all PLEG ContainerStarted timestamps (any pod) for gap detection
+    pleg_all_ts: list[str] = []
+
+    hk_overruns: list[HkOverrun] = []
+    slo_values: list[float] = []
+    concurrency_by_sec: dict[str, int] = {}
+
+    for raw in lines:
+        line = raw.rstrip()
+        # skip folded-log section markers
+        if line in ("{{{", "}}}"):
+            continue
+
+        m = _TS_RE.match(line)
+        if not m:
+            continue
+        hms, proc, body = m.group(1), m.group(2), m.group(3)
+        sec_key = hms[:8]
+
+        # --- uid extraction (once per pod) ---
+        if pod_uid is None and pod_name and pod_name in body:
+            uid_m = _UID_RE.search(body)
+            if uid_m:
+                pod_uid = uid_m.group(1)
+
+        # --- concurrency counter ---
+        if any(k in body for k in ("SyncLoop ADD", "SyncLoop UPDATE", "ContainerStarted")):
+            concurrency_by_sec[sec_key] = concurrency_by_sec.get(sec_key, 0) + 1
+
+        is_anchor = pod_name and pod_name in body
+
+        # --- kubelet events ---
+        if "SyncLoop ADD" in body and is_anchor:
+            timeline_raw.append(TimelineEvent(hms, proc, "SyncLoop ADD"))
+        elif "SyncLoop UPDATE" in body and is_anchor:
+            pass  # too noisy for timeline
+        elif ("MountVolume started" in body or "MountVolume.SetUp succeeded" in body) and is_anchor:
+            verb = "MountVolume.SetUp succeeded" if "succeeded" in body else "MountVolume started"
+            timeline_raw.append(TimelineEvent(hms, proc, verb))
+        elif ("No sandbox for pod" in body or "No ready sandbox" in body) and is_anchor:
+            timeline_raw.append(TimelineEvent(hms, proc, "No sandbox — need new one"))
+
+        # --- crio events ---
+        if "crio" in proc:
+            ts = _crio_ts(body, hms)
+            if is_anchor:
+                if "Running pod sandbox" in body:
+                    timeline_raw.append(TimelineEvent(ts, "crio", "RunPodSandbox started"))
+                elif "Adding pod" in body and "CNI" in body:
+                    timeline_raw.append(TimelineEvent(ts, "crio", "Adding pod to CNI (multus-shim)"))
+                elif "Ran pod sandbox" in body:
+                    timeline_raw.append(TimelineEvent(ts, "crio", "Ran pod sandbox (sandbox ready)"))
+                elif "Creating container" in body:
+                    timeline_raw.append(TimelineEvent(ts, "crio", "CreateContainer started"))
+                elif "Created container" in body:
+                    cid_m = _CRIO_CREATED_ID_RE.search(body)
+                    if cid_m:
+                        cid = cid_m.group(1)
+                        timeline_raw.append(TimelineEvent(ts, "crio", f"Container {cid[:12]} created"))
+
+            if "Started container" in body:
+                cid_m = _CRIO_STARTED_ID_RE.search(body)
+                if cid_m:
+                    cid = cid_m.group(1)
+                    if cid not in crio_start_ts:
+                        crio_start_ts[cid] = ts
+                    if is_anchor:
+                        timeline_raw.append(TimelineEvent(ts, "crio", f"StartContainer succeeded ({cid[:12]})"))
+
+        # --- PLEG ContainerStarted ---
+        if "SyncLoop (PLEG)" in body and "ContainerStarted" in body:
+            pleg_all_ts.append(hms)
+            if is_anchor:
+                data_m = _PLEG_DATA_RE.search(body)
+                if data_m:
+                    cid = data_m.group(1)
+                    pleg_pod_events.append((hms, cid))
+                    timeline_raw.append(TimelineEvent(hms, proc, f"PLEG ContainerStarted ({cid[:12]})"))
+
+        # --- pod_startup_latency_tracker ---
+        if "pod_startup_latency_tracker" in body:
+            slo_m = _SLO_RE.search(body)
+            if slo_m:
+                slo_values.append(float(slo_m.group(1)))
+            if is_anchor and slo_m:
+                timeline_raw.append(TimelineEvent(
+                    hms, proc, f"pod_startup_latency_tracker SLO = {float(slo_m.group(1)):.3f}s"
+                ))
+
+        # --- housekeeping overruns ---
+        if "Housekeeping took longer" in body:
+            hk_m = _HK_ACTUAL_RE.search(body)
+            if hk_m:
+                hk_overruns.append(HkOverrun(hms, float(hk_m.group(1))))
+
+    # --- PLEG lags for anchor pod ---
+    pleg_lags: list[PlegLag] = []
+    if pleg_pod_events:
+        sandbox_id = pleg_pod_events[0][1]
+        for pleg_ts, cid in pleg_pod_events:
+            role = "pause" if cid == sandbox_id else "app"
+            start = crio_start_ts.get(cid)
+            lag = (_ts_to_secs(pleg_ts) - _ts_to_secs(start)) if start else 0.0
+            pleg_lags.append(PlegLag(cid, role, start, pleg_ts, lag))
+
+    # --- PLEG silence gaps ---
+    pleg_gaps: list[PlegGap] = sorted(
+        [
+            PlegGap(pleg_all_ts[i - 1], pleg_all_ts[i],
+                    _ts_to_secs(pleg_all_ts[i]) - _ts_to_secs(pleg_all_ts[i - 1]))
+            for i in range(1, len(pleg_all_ts))
+            if (_ts_to_secs(pleg_all_ts[i]) - _ts_to_secs(pleg_all_ts[i - 1])) > 2.0
+        ],
+        key=lambda g: g.gap_secs,
+        reverse=True,
+    )
+
+    # --- SLO stats ---
+    slo_stats: Optional[SloStats] = None
+    if slo_values:
+        slo_values.sort()
+        n = len(slo_values)
+        slo_stats = SloStats(
+            count=n,
+            min_secs=slo_values[0],
+            p50_secs=slo_values[n // 2],
+            p90_secs=slo_values[int(n * 0.9)],
+            max_secs=slo_values[-1],
+        )
+
+    # --- peak concurrency ---
+    peak_ts, peak_count = None, 0
+    for ts, cnt in concurrency_by_sec.items():
+        if cnt > peak_count:
+            peak_count, peak_ts = cnt, ts
+
+    # --- deduplicate + sort timeline ---
+    seen: set[tuple] = set()
+    timeline: list[TimelineEvent] = []
+    for ev in timeline_raw:
+        key = (ev.timestamp, ev.event)
+        if key not in seen:
+            seen.add(key)
+            timeline.append(ev)
+    timeline.sort(key=lambda e: _ts_to_secs(e.timestamp))
+
+    summary = _build_summary(
+        node=node, pod=pod_name, pleg_lags=pleg_lags,
+        hk_overruns=hk_overruns, pleg_gaps=pleg_gaps,
+        slo_stats=slo_stats, peak_count=peak_count, peak_ts=peak_ts,
+    )
+
+    return NodeLogRCAResult(
+        node=node, log_path=log_path, pod=pod_name, pod_uid=pod_uid,
+        timeline=timeline, pleg_lags=pleg_lags,
+        hk_overruns=hk_overruns, pleg_gaps=pleg_gaps,
+        slo_stats=slo_stats,
+        peak_concurrency=peak_count, peak_concurrency_ts=peak_ts,
+        summary=summary,
+    )
+
+
+def _crio_ts(body: str, fallback: str) -> str:
+    m = _CRIO_TS_RE.search(body)
+    return m.group(1) if m else fallback
+
+
+def _build_summary(
+    node, pod, pleg_lags, hk_overruns, pleg_gaps, slo_stats, peak_count, peak_ts
+) -> str:
+    parts: list[str] = []
+
+    if pod and pleg_lags:
+        app_lags = [lag for lag in pleg_lags if lag.role == "app"]
+        if app_lags:
+            worst = max(app_lags, key=lambda k: k.lag_secs)
+            parts.append(
+                f"Pod {pod} on {node}: app container PLEG detection lag = "
+                f"{_fmt(worst.lag_secs)} "
+                f"(crio started {worst.crio_started_ts}, PLEG fired {worst.pleg_event_ts})."
+            )
+    elif pod:
+        parts.append(f"Pod {pod} on {node}: no PLEG ContainerStarted events found.")
+
+    if hk_overruns:
+        peak_hk = max(hk_overruns, key=lambda h: h.actual_secs)
+        parts.append(
+            f"Housekeeping overruns: {len(hk_overruns)} total, "
+            f"peak {_fmt(peak_hk.actual_secs)} at {peak_hk.timestamp}."
+        )
+    else:
+        parts.append("No housekeeping overruns.")
+
+    if pleg_gaps:
+        worst_gap = pleg_gaps[0]
+        parts.append(
+            f"PLEG silence gaps >2s: {len(pleg_gaps)} total, "
+            f"worst {_fmt(worst_gap.gap_secs)} ({worst_gap.start_ts} → {worst_gap.end_ts})."
+        )
+
+    if slo_stats:
+        parts.append(
+            f"Node-wide SLO ({slo_stats.count} pods): "
+            f"p50={_fmt(slo_stats.p50_secs)}, "
+            f"p90={_fmt(slo_stats.p90_secs)}, "
+            f"max={_fmt(slo_stats.max_secs)}."
+        )
+
+    if peak_count:
+        parts.append(f"Peak pod concurrency: {peak_count} events/s at {peak_ts}.")
+
+    # Root cause classification
+    app_lag_max = max((j.lag_secs for j in pleg_lags if j.role == "app"), default=0.0)
+    if app_lag_max > 5:
+        if hk_overruns:
+            parts.append(
+                "Root cause: continuous PLEG relist saturation — CRI ListContainers "
+                "calls blocking under high pod-creation burst; housekeeping overruns "
+                "confirm kubelet goroutine pressure. Fix: Evented PLEG (KEP-3386)."
+            )
+        elif pleg_gaps and pleg_gaps[0].gap_secs > 30:
+            parts.append(
+                "Root cause: periodic PLEG blackouts — relist blocked on a single slow "
+                "CRI call during container churn; no housekeeping overruns. "
+                "Fix: Evented PLEG (KEP-3386)."
+            )
+        else:
+            parts.append(
+                "Root cause: PLEG detection lag under pod-creation pressure. "
+                "Fix: Evented PLEG (KEP-3386)."
+            )
+    elif pleg_gaps and pleg_gaps[0].gap_secs > 30:
+        parts.append(
+            "Significant PLEG blackout windows detected. "
+            "Evented PLEG (KEP-3386) recommended."
+        )
+
+    return " ".join(parts) if parts else f"No significant events found for node {node}."
+
+
+# ---------------------------------------------------------------------------
+# Formatting
+# ---------------------------------------------------------------------------
+
+
+def format_result_markdown(result: NodeLogRCAResult) -> str:
+    lines: list[str] = [f"## Node RCA — {result.node}", ""]
+
+    if result.pod:
+        lines += [
+            f"**Anchor pod:** `{result.pod}`",
+            f"**UID:** `{result.pod_uid or 'unknown'}`",
+            "",
+        ]
+
+    if result.timeline:
+        lines += [
+            "### Lifecycle timeline",
+            "",
+            "| Timestamp | Source | Event |",
+            "|-----------|--------|-------|",
+        ]
+        for ev in result.timeline:
+            lines.append(f"| `{ev.timestamp}` | {ev.source} | {ev.event} |")
+        lines.append("")
+
+    if result.pleg_lags:
+        lines += [
+            "### PLEG detection lag",
+            "",
+            "| Role | Container | crio Started | PLEG Event | Lag |",
+            "|------|-----------|-------------|------------|-----|",
+        ]
+        for lag in result.pleg_lags:
+            lines.append(
+                f"| {lag.role} | `{lag.container_id[:12]}` "
+                f"| {lag.crio_started_ts or '—'} | {lag.pleg_event_ts} "
+                f"| **{_fmt(lag.lag_secs)}** |"
+            )
+        lines.append("")
+
+    if result.hk_overruns:
+        peak = max(result.hk_overruns, key=lambda h: h.actual_secs)
+        lines += [
+            "### Housekeeping overruns",
+            "",
+            f"{len(result.hk_overruns)} overruns. "
+            f"Peak: {_fmt(peak.actual_secs)} at `{peak.timestamp}`.",
+            "",
+        ]
+    else:
+        lines += ["### Housekeeping overruns", "", "None detected.", ""]
+
+    if result.pleg_gaps:
+        lines += [
+            "### PLEG silence gaps > 2s",
+            "",
+            "| Start | End | Gap |",
+            "|-------|-----|-----|",
+        ]
+        for gap in result.pleg_gaps[:10]:
+            lines.append(
+                f"| `{gap.start_ts}` | `{gap.end_ts}` | **{_fmt(gap.gap_secs)}** |"
+            )
+        lines.append("")
+
+    if result.slo_stats:
+        s = result.slo_stats
+        lines += [
+            "### Node-wide pod startup SLO",
+            "",
+            f"count={s.count}  "
+            f"min={_fmt(s.min_secs)}  "
+            f"p50={_fmt(s.p50_secs)}  "
+            f"p90={_fmt(s.p90_secs)}  "
+            f"max={_fmt(s.max_secs)}",
+            "",
+        ]
+
+    lines += ["### Summary", "", result.summary, ""]
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Analyze a Kubernetes node journal for pod startup RCA."
+    )
+    parser.add_argument("--log", required=True, help="Path to journal file")
+    parser.add_argument("--pod", default=None, help="Pod name to anchor timeline")
+    parser.add_argument(
+        "--json", action="store_true", dest="as_json",
+        help="Output raw JSON instead of markdown",
+    )
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.WARNING, stream=sys.stderr)
+    result = analyze_node_journal(args.log, pod_name=args.pod)
+
+    if args.as_json:
+        import json as _json
+        print(_json.dumps(result._asdict(), indent=2, default=str))
+    else:
+        print(format_result_markdown(result))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/bugzooka/core/utils.py b/bugzooka/core/utils.py
index 23870e1..41e4083 100644
--- a/bugzooka/core/utils.py
+++ b/bugzooka/core/utils.py
@@ -96,7 +96,7 @@ def download_file_from_gcs(gcs_url, local_path):
     :param local_path: local file system path to download
     :return: None
     """
-    command = f"gsutil -m cp -r {gcs_url} {local_path}"
+    command = f"gsutil cp {gcs_url} {local_path}"
     file_name = gcs_basename(gcs_url)
     try:
         logger.info("Downloading %s...", file_name)
diff --git a/bugzooka/integrations/slack_fetcher.py b/bugzooka/integrations/slack_fetcher.py
index a59d698..1591ff9 100644
--- a/bugzooka/integrations/slack_fetcher.py
+++ b/bugzooka/integrations/slack_fetcher.py
@@ -13,6 +13,7 @@
     download_and_analyze_logs,
     filter_errors_with_llm,
     run_agent_analysis,
+    run_node_rca_analysis,  # auto-triggered on podReadyLatency_P99 regressions
 )
 from bugzooka.analysis.log_summarizer import (
     classify_failure_type,
@@ -638,6 +639,63 @@ def _process_message(self, msg, enable_inference):
         # Add job-history info in the thread after the full error log
         self._handle_job_history(thread_ts=ts, current_message=msg)
 
+        # Auto-trigger node journal RCA when orion reports a podReadyLatency_P99 regression
+        all_errors = list(errors_list or []) + list(full_errors_for_file or [])
+        if any("podReadyLatency_P99" in (e or "") for e in all_errors):
+            # Extract workload name from the first "[workload]" bracket line in errors,
+            # e.g. "[node-density-cni]" → "node-density-cni", so we search the right
+            # openshift-qe-<workload> artifacts directory instead of a different workload's.
+            import re as _re
+            _workload_hint = ""
+            for _e in all_errors:
+                _m = _re.search(r"^\s*\[([^\]]+)\]", _e or "")
+                if _m:
+                    _workload_hint = _m.group(1).strip()
+                    break
+
+            _rca_start = time.time()
+            _rca_success = False
+            _rca_error_message = None
+            _rca_error_type = None
+            try:
+                rca_url = view_url or extract_job_details(text)[0]
+                if rca_url:
+                    self.logger.info(
+                        "podReadyLatency_P99 regression detected — running node journal RCA "
+                        "(workload hint: %s)",
+                        _workload_hint or "<none>",
+                    )
+                    rca_report = run_node_rca_analysis(rca_url, step_hint=_workload_hint)
+                    self.client.chat_postMessage(
+                        channel=self.channel_id,
+                        text=":mag: *Node Journal RCA* (podReadyLatency_P99 regression)",
+                        thread_ts=ts,
+                    )
+                    self.client.files_upload_v2(
+                        channel=self.channel_id,
+                        content=rca_report,
+                        filename="node-rca.md",
+                        title="Node Journal RCA",
+                        thread_ts=ts,
+                    )
+                    _rca_success = True
+            except Exception as exc:
+                self.logger.error("node journal RCA failed: %s", exc)
+                _rca_error_message = str(exc)
+                _rca_error_type = type(exc).__name__
+            finally:
+                telemetry.emit({
+                    "command": "node_rca",
+                    "trigger_type": "automatic",
+                    "channel_id": self.channel_id,
+                    "user_id": user if user != "Unknown" else None,
+                    "success": _rca_success,
+                    "error_message": _rca_error_message,
+                    "error_type": _rca_error_type,
+                    "duration_ms": int((time.time() - _rca_start) * 1000),
+                    "retry_count": 0,
+                })
+
         if is_install_issue or not enable_inference:
             _aa_success = True
             telemetry.emit({
diff --git a/tests/test_node_log_analyzer.py b/tests/test_node_log_analyzer.py
new file mode 100644
index 0000000..6663364
--- /dev/null
+++ b/tests/test_node_log_analyzer.py
@@ -0,0 +1,181 @@
+"""
+Tests for node_log_analyzer: parser correctness and Slack message structure.
+
+The "real log" tests use the journal.folded.log checked into the repo under
+tests/fixtures/. This file is a trimmed subset of the node journal from
+ip-10-0-67-181, captured during the node-density RCA investigation, and
+contains the known node-density-956 PLEG lag scenario (17s detection lag,
+3 housekeeping overruns, root cause: PLEG relist saturation).
+"""
+
+import textwrap
+
+import pytest
+
+from bugzooka.analysis.node_log_analyzer import (
+    analyze_node_journal,
+    format_result_markdown,
+)
+
+# ---------------------------------------------------------------------------
+# Fixture: minimal synthetic journal lines for fast unit tests
+# ---------------------------------------------------------------------------
+
+_MINIMAL_JOURNAL = textwrap.dedent("""\
+    May 27 16:02:27.147175 ip-10-0-67-181 kubenswrapper[2509]: I0527 16:02:27.147102    2509 kubelet.go:2608] "SyncLoop ADD" source="api" pods=["node-density-0/node-density-956"]
+    May 27 16:02:27.620058 ip-10-0-67-181 kubenswrapper[2509]: I0527 16:02:27.619909    2509 reconciler_common.go:225] "operationExecutor.MountVolume started for volume \\"kube-api-access-gxkkp\\" (UniqueName: \\"kubernetes.io/projected/cdcd7672-d87d-434e-86ea-0e2035dcacb8-kube-api-access-gxkkp\\") pod \\"node-density-956\\"" pod="node-density-0/node-density-956"
+    May 27 16:02:27.731793 ip-10-0-67-181 kubenswrapper[2509]: I0527 16:02:27.730011    2509 operation_generator.go:614] "MountVolume.SetUp succeeded for volume \\"kube-api-access-gxkkp\\"" pod="node-density-0/node-density-956"
+    May 27 16:02:27.950642 ip-10-0-67-181 crio[2440]: time="2026-05-27T16:02:27.950581297Z" level=info msg="Running pod sandbox: node-density-0/node-density-956/POD" id=0be966e9
+    May 27 16:02:29.013534 ip-10-0-67-181 crio[2440]: time="2026-05-27T16:02:29.013490793Z" level=info msg="Adding pod node-density-0_node-density-956 to CNI network \\"multus-cni-network\\" (type=multus-shim)"
+    May 27 16:02:31.769676 ip-10-0-67-181 crio[2440]: time="2026-05-27T16:02:31.769588511Z" level=info msg="Ran pod sandbox 337b3f3c6d63749cd0091d92e825729e9f818f74c626d58fe72978d0c2ea3320 with infra container: node-density-0/node-density-956/POD"
+    May 27 16:02:33.451890 ip-10-0-67-181 crio[2440]: time="2026-05-27T16:02:33.43583158Z" level=info msg="Creating container: node-density-0/node-density-956/node-density"
+    May 27 16:02:35.024982 ip-10-0-67-181 crio[2440]: time="2026-05-27T16:02:35.014226005Z" level=info msg="Created container 75ef25174d7d464246c9cb27d850db42604394b920e6f66b56a94ab3d9c5c983: node-density-0/node-density-956/node-density"
+    May 27 16:02:35.291348 ip-10-0-67-181 crio[2440]: time="2026-05-27T16:02:35.269681149Z" level=info msg="Started container" PID=133675 containerID=75ef25174d7d464246c9cb27d850db42604394b920e6f66b56a94ab3d9c5c983 description=node-density-0/node-density-956/node-density sandboxID=337b3f3c6d63749cd0091d92e825729e9f818f74c626d58fe72978d0c2ea3320
+    May 27 16:02:35.669260 ip-10-0-67-181 kubenswrapper[2509]: I0527 16:02:35.669208    2509 kubelet.go:2639] "SyncLoop (PLEG): event for pod" pod="node-density-0/node-density-956" event={"ID":"cdcd7672-d87d-434e-86ea-0e2035dcacb8","Type":"ContainerStarted","Data":"337b3f3c6d63749cd0091d92e825729e9f818f74c626d58fe72978d0c2ea3320"}
+    May 27 16:02:37.208232 ip-10-0-67-181 kubenswrapper[2509]: E0527 16:02:37.198170    2509 kubelet.go:2711] "Housekeeping took longer than expected" err="housekeeping took too long" expected="1s" actual="1.534s"
+    May 27 16:02:39.093457 ip-10-0-67-181 kubenswrapper[2509]: E0527 16:02:39.093410    2509 kubelet.go:2711] "Housekeeping took longer than expected" err="housekeeping took too long" expected="1s" actual="1.01s"
+    May 27 16:02:52.403214 ip-10-0-67-181 kubenswrapper[2509]: I0527 16:02:52.403189    2509 kubelet.go:2639] "SyncLoop (PLEG): event for pod" pod="node-density-0/node-density-956" event={"ID":"cdcd7672-d87d-434e-86ea-0e2035dcacb8","Type":"ContainerStarted","Data":"75ef25174d7d464246c9cb27d850db42604394b920e6f66b56a94ab3d9c5c983"}
+    May 27 16:02:52.388631 ip-10-0-67-181 kubenswrapper[2509]: I0527 16:02:52.388614    2509 pod_startup_latency_tracker.go:148] "Observed pod startup duration" pod="node-density-0/node-density-956" podStartSLOduration=25.388 podStartE2EDuration="25.388s" totalImagesPullingTime="0s" totalInitContainerRuntime="0s" isStatefulPod=false podCreationTimestamp="2026-05-27 16:02:27 +0000 UTC"
+""")
+
+
+@pytest.fixture
+def minimal_journal(tmp_path):
+    """Write the minimal journal to a temp file and return its path."""
+    node_dir = tmp_path / "ip-10-0-67-181.us-west-2.compute.internal"
+    node_dir.mkdir()
+    journal = node_dir / "journal.decompressed"
+    journal.write_text(_MINIMAL_JOURNAL, encoding="utf-8")
+    return str(journal)
+
+
+# ---------------------------------------------------------------------------
+# Parser unit tests
+# ---------------------------------------------------------------------------
+
+
+class TestAnalyzeNodeJournal:
+    def test_pod_uid_extracted(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        assert result.pod_uid == "cdcd7672-d87d-434e-86ea-0e2035dcacb8"
+
+    def test_timeline_contains_key_events(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        event_names = [e.event for e in result.timeline]
+        assert "SyncLoop ADD" in event_names
+        assert "MountVolume started" in event_names
+        assert "MountVolume.SetUp succeeded" in event_names
+        assert "RunPodSandbox started" in event_names
+        assert "Ran pod sandbox (sandbox ready)" in event_names
+        assert "CreateContainer started" in event_names
+        assert any("75ef25174d7d" in e for e in event_names), "app container start missing"
+        assert any("PLEG ContainerStarted" in e for e in event_names)
+
+    def test_timeline_is_sorted(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        timestamps = [e.timestamp for e in result.timeline]
+        from bugzooka.analysis.node_log_analyzer import _ts_to_secs
+        secs = [_ts_to_secs(t) for t in timestamps]
+        assert secs == sorted(secs), "timeline events not in chronological order"
+
+    def test_pleg_lag_app_container(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        app_lags = [lag for lag in result.pleg_lags if lag.role == "app"]
+        assert len(app_lags) == 1
+        lag = app_lags[0]
+        # crio started at 16:02:35.269..., PLEG fired at 16:02:52.403...  → ~17.13s
+        assert lag.lag_secs == pytest.approx(17.13, abs=0.1)
+        assert lag.container_id.startswith("75ef25174d7d")
+
+    def test_pleg_lag_pause_container_zero(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        pause_lags = [lag for lag in result.pleg_lags if lag.role == "pause"]
+        assert len(pause_lags) == 1
+        # no crio StartContainer for the pause container in this log → lag = 0
+        assert pause_lags[0].lag_secs == 0.0
+
+    def test_housekeeping_overruns(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        assert len(result.hk_overruns) == 2
+        peak = max(result.hk_overruns, key=lambda h: h.actual_secs)
+        assert peak.actual_secs == pytest.approx(1.534)
+
+    def test_pleg_gaps(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        # gap between the two PLEG events: 52.403 - 35.669 = ~16.73s
+        assert len(result.pleg_gaps) == 1
+        assert result.pleg_gaps[0].gap_secs == pytest.approx(16.73, abs=0.1)
+
+    def test_slo_stats(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        assert result.slo_stats is not None
+        assert result.slo_stats.count == 1
+        assert result.slo_stats.p50_secs == pytest.approx(25.388, abs=0.01)
+
+    def test_no_pod_name_still_parses(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal)
+        assert result.pod is None
+        assert result.timeline == []
+        assert result.slo_stats is not None  # SLO parsed regardless of anchor pod
+
+    def test_missing_file_returns_error_result(self, tmp_path):
+        result = analyze_node_journal(str(tmp_path / "nonexistent" / "journal"))
+        assert "Could not read" in result.summary
+
+
+# ---------------------------------------------------------------------------
+# format_result_markdown output structure
+# ---------------------------------------------------------------------------
+
+
+class TestFormatResultMarkdown:
+    def test_sections_present(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        md = format_result_markdown(result)
+
+        assert "## Node RCA —" in md
+        assert "**Anchor pod:** `node-density-956`" in md
+        assert "### Lifecycle timeline" in md
+        assert "### PLEG detection lag" in md
+        assert "### Housekeeping overruns" in md
+        assert "### PLEG silence gaps > 2s" in md
+        assert "### Node-wide pod startup SLO" in md
+        assert "### Summary" in md
+
+    def test_pleg_lag_shown_in_table(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        md = format_result_markdown(result)
+        assert "app" in md
+        assert "17." in md  # ~17s lag
+
+    def test_hk_overrun_count_shown(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        md = format_result_markdown(result)
+        assert "2 overruns" in md
+        assert "1.534s" in md
+
+    def test_root_cause_in_summary(self, minimal_journal):
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        md = format_result_markdown(result)
+        assert "PLEG relist saturation" in md
+        assert "KEP-3386" in md
+
+    def test_slack_file_content(self, minimal_journal):
+        """
+        Verify the markdown content uploaded as node-rca.md to Slack.
+
+        The RCA is posted as a file attachment (files_upload_v2) rather than
+        a block message, so there is no block structure to assert — only the
+        file content that users open in Slack.
+        """
+        result = analyze_node_journal(minimal_journal, pod_name="node-density-956")
+        report = format_result_markdown(result)
+
+        # Key structural assertions — what the user reads when opening node-rca.md
+        assert "## Node RCA —" in report
+        assert "node-density-956" in report
+        assert "### PLEG detection lag" in report
+        assert "17." in report          # ~17s lag visible in table
+        assert "2 overruns" in report   # housekeeping count
+        assert "PLEG relist saturation" in report
+        assert "KEP-3386" in report
diff --git a/tests/test_slack_fetcher.py b/tests/test_slack_fetcher.py
index 0ef15c4..db7ce95 100644
--- a/tests/test_slack_fetcher.py
+++ b/tests/test_slack_fetcher.py
@@ -22,6 +22,31 @@
     None,
 )
 
+MOCK_ANALYSIS_RESULT_POD_LATENCY = (
+    [
+        "\n[node-density]",
+        "  podReadyLatency_P99: +45.23%",
+        "  Changepoint at: 5.0.0-0.nightly-2026-05-27-134409",
+    ],
+    "test phase: openshift-qe-node-density orion regression",
+    False,
+    False,
+    "openshift-qe-node-density",
+    None,
+)
+
+MOCK_NODE_RCA_REPORT = """\
+## Node RCA — ip-10-0-67-181.us-west-2.compute.internal
+
+**Anchor pod:** `node-density-956`
+**UID:** `cdcd7672-d87d-434e-86ea-0e2035dcacb8`
+
+### Summary
+
+Pod node-density-956: app container PLEG detection lag = 17.134s. \
+Root cause: continuous PLEG relist saturation. Fix: Evented PLEG (KEP-3386).\
+"""
+
 
 def run_slack_fetcher_test(
     test_messages,