hh-build-report - zygaena/hammerhead

root / base / usr / src / tools / scripts / hh-build-report
hh-build-report Plain Text 1211 lines 47.7 KB
#!/usr/bin/python3
#
# CDDL HEADER START
#
# The contents of this file are subject to the terms of the
# Common Development and Distribution License (the "License").
# You may not use this file except in compliance with the License.
#
# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
# or http://www.opensolaris.org/os/licensing.
# See the License for the specific language governing permissions
# and limitations under the License.
#
# When distributing Covered Code, include this CDDL HEADER in each
# file and include the License file at usr/src/OPENSOLARIS.LICENSE.
# If applicable, add the following below this CDDL HEADER, with the
# fields enclosed by brackets "[]" replaced with your own identifying
# information: Portions Copyright [yyyy] [name of copyright owner]
#
# CDDL HEADER END
#
# Copyright 2026 Zygaena Project - Hammerhead Build System
#
# hh-build-report: Comprehensive build analysis and reporting tool
#
# Generates both human-readable (build-report.txt) and machine-parseable
# (build-report.json) reports after each build.  Tracks trends across
# builds via build-history.json.
#
# Usage:
#   hh-build-report --log-dir <dir> --proto <dir> --workspace <dir>
#                   [--history <file>] [--previous <dir>]
#
# Designed to be called automatically by hh-build at the end of each
# build, or run standalone for ad-hoc analysis.
#

import argparse
import json
import os
import re
import stat
import subprocess
import sys
import time
from collections import defaultdict, OrderedDict
from datetime import datetime
from pathlib import Path

# ---------------------------------------------------------------------------
# Critical files that should be present for a bootable system
# ---------------------------------------------------------------------------
CRITICAL_FILES = [
    # Kernel (Hammerhead: all kernel modules in /kernel/, flat)
    ("kernel/unix", "kernel image"),
    ("kernel/genunix", "generic kernel"),
    # Runtime linker
    ("usr/lib/ld.so.1", "runtime linker"),
    # Core libraries
    ("usr/lib/libc.so.1", "C library"),
    ("usr/lib/libdl.so.1", "dynamic linking"),
    ("usr/lib/libnsl.so.1", "name services"),
    ("usr/lib/libsocket.so.1", "sockets"),
    ("usr/lib/libm.so.2", "math library"),
    ("usr/lib/libpthread.so.1", "POSIX threads"),
    ("usr/lib/librt.so.1", "realtime"),
    ("usr/lib/libscf.so.1", "SMF client"),
    ("usr/lib/libuutil.so.1", "userland utility"),
    ("usr/lib/libzfs.so.1", "ZFS library"),
    ("usr/lib/libcrypto.so", "LibreSSL crypto"),
    ("usr/lib/libssl.so", "LibreSSL SSL"),
    ("usr/lib/libnvpair.so.1", "name-value pairs"),
    ("usr/lib/libumem.so.1", "memory allocator"),
    ("usr/lib/libsec.so.1", "security"),
    ("usr/lib/libdevinfo.so.1", "device info"),
    ("usr/lib/libgen.so.1", "general"),
    ("usr/lib/libcmdutils.so.1", "command utilities"),
    # Init (zyginit replaced SMF — Phase E v2 complete 2026-05-12)
    ("sbin/init", "init process (symlink → zyginit)"),
    ("sbin/zyginit", "zyginit PID 1 daemon"),
    # Key kernel modules (Hammerhead: flattened paths)
    ("kernel/misc/ctf", "CTF support"),
    ("kernel/fs/zfs", "ZFS filesystem"),
    ("kernel/fs/tmpfs", "tmpfs"),
    ("kernel/fs/procfs", "procfs"),
    ("kernel/fs/dev", "devfs"),
    ("kernel/drv/sd", "SCSI disk driver"),
    ("kernel/drv/e1000g", "Intel NIC driver"),
    ("kernel/misc/virtio", "virtio support"),
    # Key commands — /bin (boot-essential)
    ("bin/ls", "list files"),
    ("bin/cat", "concatenate"),
    ("bin/echo", "echo"),
    ("bin/sh", "Bourne shell"),
    ("bin/ksh93", "Korn shell"),
    ("bin/zsh", "Z shell"),
    ("bin/cp", "copy"),
    ("bin/mv", "move"),
    ("bin/rm", "remove"),
    ("bin/mkdir", "make directory"),
    ("bin/chmod", "change mode"),
    ("bin/chown", "change owner"),
    ("bin/grep", "pattern match"),
    ("bin/sed", "stream editor"),
    ("bin/awk", "pattern processing"),
    ("bin/su", "switch user"),
    ("bin/ps", "process status"),
    ("bin/df", "disk free"),
    # Key commands — /usr/bin
    ("usr/bin/login", "login"),
    ("usr/bin/passwd", "change password"),
    ("usr/bin/kill", "signal process"),
    # Key commands — /sbin
    ("sbin/mount", "mount filesystem"),
    ("sbin/umount", "unmount filesystem"),
    ("sbin/reboot", "reboot"),
    ("sbin/halt", "halt"),
    # Key commands — /usr/sbin
    ("usr/sbin/zfs", "ZFS admin"),
    ("usr/sbin/zpool", "ZFS pool admin"),
    ("sbin/zygctl", "zyginit admin CLI"),
    ("usr/sbin/ifconfig", "network interface config"),
    ("usr/sbin/ipadm", "IP address admin"),
    ("usr/sbin/dladm", "datalink admin"),
    ("usr/sbin/devfsadm", "device filesystem admin"),
    ("usr/sbin/modload", "load kernel module"),
    ("usr/sbin/modinfo", "module info"),
    ("usr/sbin/bootadm", "boot admin"),
    ("usr/sbin/shutdown", "shutdown"),
]

# ---------------------------------------------------------------------------
# Error pattern matching
# ---------------------------------------------------------------------------
# Patterns to identify root causes when looking backwards from Error 2
ERROR_PATTERNS = [
    ("undefined_reference", re.compile(r"undefined reference to [`'](.+?)'")),
    ("cannot_find_lib", re.compile(r"cannot find -l(\S+)")),
    ("missing_header", re.compile(r"fatal error:\s+(\S+\.h).*[Nn]o such file")),
    ("missing_source", re.compile(r"No rule to make target ['`](\S+\.[cSsh])'")),
    ("compile_error", re.compile(r"error:.*(?!ld returned)")),
    ("ctf_error", re.compile(r"ctf(?:convert|merge):\s+(.+)")),
    ("permission_denied", re.compile(r"[Pp]ermission denied")),
    ("svc_configd_crash", re.compile(r"svc\.configd.*(?:died|core|crash|broken)")),
    ("recipe_failed", re.compile(r"\*\*\*.*(?:Error|Signal)")),
]


class BuildReport:
    """Comprehensive build analysis and reporting."""

    def __init__(self, log_dir, proto_dir, workspace, history_file=None,
                 previous_dir=None):
        self.log_dir = Path(log_dir)
        self.proto_dir = Path(proto_dir)
        self.workspace = Path(workspace)
        self.history_file = Path(history_file) if history_file else \
            self.workspace / "log" / "build-history.json"
        self.previous_dir = Path(previous_dir) if previous_dir else None

        # Find the build log
        self.log_file = None
        for name in ("build.log", "nightly.log"):
            p = self.log_dir / name
            if p.exists():
                self.log_file = p
                break

        # Find build-info if present, fall back to log parsing
        self.build_info = self._read_build_info()
        # Try to detect build status from log if not in build-info
        if self.build_info.get("build_ok") == "unknown" and self.log_file:
            self._detect_build_status()

        # Results
        self.proto = {}
        self.errors = {}
        self.error_analysis = {}
        self.timing = {}
        self.delta = {}

    # ------------------------------------------------------------------
    # Build info
    # ------------------------------------------------------------------
    def _read_build_info(self):
        """Read build-info metadata file if present."""
        info_file = self.log_dir / "build-info"
        info = {
            "commit": "unknown",
            "commit_short": "unknown",
            "branch": "unknown",
            "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "host": "unknown",
            "user": "unknown",
            "workspace": str(self.workspace),
            "build_ok": "unknown",
        }
        if info_file.exists():
            for line in info_file.read_text().splitlines():
                line = line.strip()
                if "=" in line:
                    key, _, val = line.partition("=")
                    info[key.strip()] = val.strip()
        else:
            # Try to get info from git
            try:
                info["commit"] = subprocess.check_output(
                    ["git", "-C", str(self.workspace), "rev-parse", "HEAD"],
                    stderr=subprocess.DEVNULL).decode().strip()
                info["commit_short"] = info["commit"][:10]
                info["branch"] = subprocess.check_output(
                    ["git", "-C", str(self.workspace), "rev-parse",
                     "--abbrev-ref", "HEAD"],
                    stderr=subprocess.DEVNULL).decode().strip()
            except (subprocess.CalledProcessError, FileNotFoundError):
                pass
        return info

    def _detect_build_status(self):
        """Try to determine build status from the log file tail."""
        try:
            with open(str(self.log_file), "r", errors="replace") as f:
                # Read last 50 lines
                lines = f.readlines()
                tail = lines[-50:] if len(lines) >= 50 else lines
                for line in tail:
                    if "build completed" in line.lower():
                        self.build_info["build_ok"] = "y"
                        return
                    if "build failed" in line.lower():
                        self.build_info["build_ok"] = "n"
                        return
                # If we find "Completed" status
                for line in tail:
                    if "Completed" in line:
                        self.build_info["build_ok"] = "y"
                        return
        except IOError:
            pass

    # ------------------------------------------------------------------
    # Proto area analysis
    # ------------------------------------------------------------------
    def analyze_proto(self):
        """Walk the proto area and categorize all files."""
        categories = {
            "executables": {"count": 0, "size": 0, "files": []},
            "shared_libraries": {"count": 0, "size": 0, "files": []},
            "kernel_modules": {"count": 0, "size": 0, "files": []},
            "static_libraries": {"count": 0, "size": 0, "files": []},
            "headers": {"count": 0, "size": 0, "files": []},
            "man_pages": {"count": 0, "size": 0, "files": []},
            "config_files": {"count": 0, "size": 0, "files": []},
            "symlinks": {"count": 0, "files": []},
            "directories": {"count": 0},
            "other": {"count": 0, "size": 0, "files": []},
        }
        total_size = 0
        total_files = 0

        if not self.proto_dir.exists():
            self.proto = {
                "total_size": 0, "total_files": 0,
                "categories": categories,
                "critical_files": {"present": [], "missing": [
                    (f, d) for f, d in CRITICAL_FILES]},
            }
            return

        proto_str = str(self.proto_dir)

        # Hammerhead: followlinks=True to count files inside symlinked dirs,
        # but prune self-referential symlinks (e.g. lib/64 -> .) to avoid
        # infinite loops.
        seen_real = set()
        for root, dirs, files in os.walk(proto_str, followlinks=True):
            real_root = os.path.realpath(root)
            if real_root in seen_real:
                dirs.clear()
                continue
            seen_real.add(real_root)

            rel_root = os.path.relpath(root, proto_str)
            if rel_root == ".":
                rel_root = ""

            categories["directories"]["count"] += len(dirs)

            for fname in files:
                fpath = os.path.join(root, fname)
                rel_path = os.path.join(rel_root, fname) if rel_root else fname

                # Handle symlinks first
                if os.path.islink(fpath):
                    categories["symlinks"]["count"] += 1
                    categories["symlinks"]["files"].append(rel_path)
                    total_files += 1
                    continue

                try:
                    fstat = os.stat(fpath)
                    fsize = fstat.st_size
                except OSError:
                    continue

                total_size += fsize
                total_files += 1

                cat = self._categorize_file(rel_path, fpath, fsize)
                categories[cat]["count"] += 1
                categories[cat]["size"] = categories[cat].get("size", 0) + fsize
                # Only store file list for categories with manageable counts
                if cat in ("executables", "shared_libraries",
                           "kernel_modules", "static_libraries"):
                    categories[cat]["files"].append(rel_path)
                elif cat == "headers":
                    # Store count but not full list (too many)
                    pass
                elif cat == "man_pages":
                    pass
                else:
                    if categories[cat]["count"] <= 500:
                        categories[cat]["files"].append(rel_path)

        # Check critical files
        present = []
        missing = []
        for fpath, description in CRITICAL_FILES:
            full = self.proto_dir / fpath
            if full.exists() or full.is_symlink():
                size = 0
                try:
                    size = os.stat(str(full)).st_size
                except OSError:
                    pass
                present.append((fpath, description, size))
            else:
                missing.append((fpath, description))

        # Don't store individual file lists in JSON for large categories
        # (headers, man_pages, other) — just counts and sizes
        for cat in ("headers", "man_pages", "other", "config_files",
                    "symlinks"):
            if len(categories[cat].get("files", [])) > 200:
                categories[cat]["files"] = \
                    categories[cat]["files"][:20] + ["... truncated ..."]

        self.proto = {
            "total_size": total_size,
            "total_files": total_files,
            "categories": categories,
            "critical_files": {
                "present": present,
                "missing": missing,
            },
        }

    def _categorize_file(self, rel_path, abs_path, size):
        """Determine category for a file based on path and content."""
        parts = rel_path.split("/")

        # Kernel modules
        if (rel_path.startswith("kernel/") or
                (len(parts) >= 3 and parts[1] == "kernel")):
            # platform/i86pc/kernel/...
            return "kernel_modules"

        # Shared libraries
        if re.search(r'\.so(\.\d+)*$', rel_path):
            return "shared_libraries"

        # Static libraries
        if rel_path.endswith(".a"):
            return "static_libraries"

        # Headers
        if rel_path.endswith(".h") or "/include/" in rel_path:
            return "headers"

        # Man pages
        if "/man/" in rel_path or rel_path.startswith("usr/share/man/"):
            return "man_pages"

        # Config files
        if rel_path.startswith("etc/"):
            return "config_files"

        # Executables — check if in bin/sbin directories
        exe_dirs = ("bin/", "sbin/", "usr/bin/", "usr/sbin/",
                    "usr/xpg4/bin/", "usr/xpg6/bin/", "usr/ccs/bin/",
                    "usr/lib/isaexec")
        for d in exe_dirs:
            if rel_path.startswith(d) or rel_path == d.rstrip("/"):
                return "executables"

        # Check ELF magic for files in lib directories that aren't .so
        if size >= 4 and any(rel_path.startswith(p) for p in
                            ("usr/lib/", "lib/", "usr/sbin/", "sbin/")):
            try:
                with open(abs_path, "rb") as f:
                    magic = f.read(4)
                if magic == b"\x7fELF":
                    # ELF file in lib directory — likely a helper binary
                    if "/lib/" in rel_path and not rel_path.endswith(".so"):
                        return "executables"
            except (IOError, PermissionError):
                pass

        return "other"

    # ------------------------------------------------------------------
    # Build log analysis
    # ------------------------------------------------------------------
    def analyze_log(self):
        """Parse the build log for errors, timing, and phases."""
        if not self.log_file or not self.log_file.exists():
            self.errors = {
                "collect2": {"count": 0, "details": []},
                "error2": {"count": 0, "details": []},
                "ctf": {"count": 0, "details": []},
                "warnings_total": 0,
            }
            self.timing = {"total_seconds": 0, "formatted": "unknown"}
            self.error_analysis = {"root_causes": [], "cascades": []}
            return

        collect2_errors = []
        error2_all = []       # All Error lines (raw count)
        error2_leaf = []      # Only leaf-level failures (deduplicated)
        ctf_errors = []
        warning_count = 0
        build_time_str = ""
        build_started = ""
        build_completed = ""

        # Track current directory for context
        dir_stack = []
        current_dir = ""

        # Buffer recent lines for context around errors
        context_buffer = []
        CONTEXT_SIZE = 15

        # Track which directories failed and at what make depth
        failed_dirs = set()
        # Track directories that reported failures in subdirectories
        # (these are cascade propagations, not root failures)
        cascade_dirs = set()
        # Track "cannot find -l" errors to detect cascades
        missing_libs = defaultdict(list)

        log_path = str(self.log_file)
        try:
            with open(log_path, "r", errors="replace") as f:
                for line_num, line in enumerate(f, 1):
                    line_stripped = line.rstrip()

                    # Maintain context buffer
                    context_buffer.append(line_stripped)
                    if len(context_buffer) > CONTEXT_SIZE:
                        context_buffer.pop(0)

                    # Track directory changes via enter/leave stack
                    m = re.match(
                        r"gmake\[(\d+)\]: Entering directory '(.+)'", line)
                    if m:
                        depth = int(m.group(1))
                        entered_dir = m.group(2)
                        ws = str(self.workspace)
                        if entered_dir.startswith(ws):
                            entered_dir = entered_dir[len(ws):].lstrip("/")
                        current_dir = entered_dir
                        # Track on stack by depth
                        while len(dir_stack) >= depth:
                            dir_stack.pop()
                        dir_stack.append(entered_dir)
                        continue

                    m = re.match(
                        r"gmake\[(\d+)\]: Leaving directory '(.+)'", line)
                    if m:
                        depth = int(m.group(1))
                        while len(dir_stack) >= depth:
                            dir_stack.pop()
                        # Restore current_dir to parent
                        current_dir = dir_stack[-1] if dir_stack else ""
                        continue

                    # Build timing
                    m = re.match(r"real\s+(\d+):(\d+):(\d+)", line)
                    if m:
                        h, mn, s = int(m.group(1)), int(m.group(2)), \
                            int(m.group(3))
                        total_s = h * 3600 + mn * 60 + s
                        self.timing = {
                            "total_seconds": total_s,
                            "formatted": f"{h}h{mn:02d}m{s:02d}s",
                        }

                    # Build start/complete timestamps
                    if "build started:" in line:
                        build_started = line_stripped
                    if "build completed:" in line:
                        build_completed = line_stripped

                    # Count warnings (rough — not all are real warnings)
                    if ": warning:" in line and "ignoring old recipe" not in \
                            line:
                        warning_count += 1

                    # Collect2 errors (linker failures)
                    if "collect2: error" in line:
                        detail = {
                            "line": line_num,
                            "directory": current_dir,
                            "context": list(context_buffer[-8:]),
                        }
                        # Extract the target from context
                        for ctx_line in reversed(context_buffer[:-1]):
                            if "-o " in ctx_line:
                                m2 = re.search(r"-o\s+(\S+)", ctx_line)
                                if m2:
                                    detail["target"] = m2.group(1)
                                    break
                        collect2_errors.append(detail)
                        failed_dirs.add(current_dir)

                    # CTF errors
                    if re.search(r"ctf(convert|merge):.*(?:error|failed|"
                                 r"cannot)", line, re.I):
                        ctf_errors.append({
                            "line": line_num,
                            "directory": current_dir,
                            "message": line_stripped[:200],
                        })

                    # Make errors (*** [...] Error N)
                    m = re.match(
                        r"gmake(?:\[(\d+)\])?: \*\*\* "
                        r"\[(.+?)(?::(\d+))?: (.+?)\] Error (\d+)", line)
                    if not m:
                        m = re.match(
                            r"gmake(?:\[(\d+)\])?: \*\*\* "
                            r"\[(.+?)\] Error (\d+)", line)
                    if m:
                        error2_all.append({
                            "line": line_num,
                            "directory": current_dir,
                            "message": line_stripped[:200],
                        })

                        # Determine if this is a leaf failure or cascade
                        # A cascade is when the error target is a
                        # subdirectory name (make dispatching to child)
                        is_cascade = False
                        target = m.group(m.lastindex - 1) if \
                            m.lastindex and m.lastindex > 1 else ""
                        # If any child of current_dir already failed,
                        # this is likely a cascade propagation
                        for fd in failed_dirs:
                            if fd.startswith(current_dir + "/") and \
                                    fd != current_dir:
                                is_cascade = True
                                cascade_dirs.add(current_dir)
                                break

                        if not is_cascade:
                            detail = {
                                "line": line_num,
                                "directory": current_dir,
                                "message": line_stripped[:200],
                            }
                            root_cause = self._find_root_cause(
                                context_buffer[:-1])
                            if root_cause:
                                detail["root_cause"] = root_cause
                            error2_leaf.append(detail)

                        failed_dirs.add(current_dir)

                        # Track missing library references for cascade
                        for ctx in context_buffer:
                            m2 = re.search(r"cannot find -l(\S+)", ctx)
                            if m2:
                                missing_libs[m2.group(1)].append(current_dir)

        except IOError as e:
            print(f"Warning: Could not read log file: {e}", file=sys.stderr)

        self.errors = {
            "collect2": {"count": len(collect2_errors),
                         "details": collect2_errors},
            "error2": {"count": len(error2_leaf),
                       "count_raw": len(error2_all),
                       "details": error2_leaf},
            "ctf": {"count": len(ctf_errors), "details": ctf_errors},
            "warnings_total": warning_count,
        }

        if "total_seconds" not in self.timing:
            self.timing = {"total_seconds": 0, "formatted": "unknown"}

        # Post-process: find leaf failures (dirs with no child also failing)
        leaf_dirs = set()
        for d in failed_dirs:
            if not d:
                continue
            is_parent = False
            for other in failed_dirs:
                if other != d and other.startswith(d + "/"):
                    is_parent = True
                    break
            if not is_parent:
                leaf_dirs.add(d)

        self.errors["failed_dirs_leaf"] = len(leaf_dirs)
        self.errors["failed_dirs_total"] = len(failed_dirs)

        # Perform error analysis
        self._analyze_error_patterns(error2_leaf, missing_libs, leaf_dirs)

    def _find_root_cause(self, context_lines):
        """Look through context lines to identify the root cause of an error."""
        # Search from most recent backwards
        for line in reversed(context_lines):
            for category, pattern in ERROR_PATTERNS:
                if pattern.search(line):
                    return {
                        "category": category,
                        "detail": line.strip()[:200],
                    }
        return None

    def _analyze_error_patterns(self, error2_errors, missing_libs,
                                failed_dirs):
        """Group errors by root cause and detect cascades."""
        # Group by root cause category
        cause_groups = defaultdict(list)
        uncategorized = []

        for err in error2_errors:
            rc = err.get("root_cause")
            if rc:
                cause_groups[rc["category"]].append(err)
            else:
                uncategorized.append(err)

        # Build root causes summary
        root_causes = []
        for category, errors in sorted(cause_groups.items(),
                                       key=lambda x: -len(x[1])):
            dirs = list(set(e["directory"] for e in errors if e["directory"]))
            # Shorten directory paths
            short_dirs = [self._shorten_dir(d) for d in dirs[:10]]
            root_causes.append({
                "category": category,
                "count": len(errors),
                "directories": short_dirs,
                "sample": errors[0].get("root_cause", {}).get("detail",
                                                              "")[:150],
            })

        if uncategorized:
            dirs = list(set(e["directory"] for e in uncategorized
                            if e["directory"]))
            short_dirs = [self._shorten_dir(d) for d in dirs[:10]]
            root_causes.append({
                "category": "uncategorized",
                "count": len(uncategorized),
                "directories": short_dirs,
                "sample": uncategorized[0].get("message", "")[:150],
            })

        # Detect cascades: if a library is missing and multiple dirs fail
        cascades = []
        for lib, dirs in sorted(missing_libs.items(),
                                key=lambda x: -len(x[1])):
            if len(dirs) >= 2:
                short_dirs = [self._shorten_dir(d) for d in dirs[:10]]
                cascades.append({
                    "missing_library": f"lib{lib}",
                    "affected_count": len(dirs),
                    "affected_directories": short_dirs,
                })

        # Also detect parent directory cascades
        parent_failures = defaultdict(list)
        for d in failed_dirs:
            parts = d.split("/")
            # Group by top-level subdirectory (e.g., usr/src/cmd/foo)
            if len(parts) >= 4:
                parent = "/".join(parts[:4])
                parent_failures[parent].append(d)

        for parent, children in sorted(parent_failures.items(),
                                       key=lambda x: -len(x[1])):
            if len(children) >= 3:
                cascades.append({
                    "parent_directory": self._shorten_dir(parent),
                    "affected_count": len(children),
                    "affected_directories": [
                        self._shorten_dir(c) for c in children[:8]],
                })

        self.error_analysis = {
            "root_causes": root_causes,
            "cascades": cascades,
            "failed_directories": [self._shorten_dir(d)
                                   for d in sorted(failed_dirs) if d],
        }

    def _shorten_dir(self, d):
        """Shorten directory path for readability."""
        # Remove common prefixes
        for prefix in ("usr/src/", "base/usr/src/",
                       "/usr/src/hammerhead/base/usr/src/",
                       "/usr/share/src/hammerhead/base/usr/src/"):
            if d.startswith(prefix):
                return d[len(prefix):]
        return d

    # ------------------------------------------------------------------
    # Delta comparison
    # ------------------------------------------------------------------
    def compute_delta(self):
        """Compare against previous build report."""
        prev_json = None

        # Find previous build report
        if self.previous_dir:
            prev_path = Path(self.previous_dir) / "build-report.json"
            if prev_path.exists():
                try:
                    prev_json = json.loads(prev_path.read_text())
                except (json.JSONDecodeError, IOError):
                    pass

        # Auto-detect from latest symlink
        if not prev_json:
            latest = self.workspace / "log" / "latest"
            if latest.is_symlink():
                target = latest.resolve()
                if target != self.log_dir.resolve():
                    prev_path = target / "build-report.json"
                    if prev_path.exists():
                        try:
                            prev_json = json.loads(prev_path.read_text())
                        except (json.JSONDecodeError, IOError):
                            pass

        if not prev_json:
            self.delta = {"available": False}
            return

        prev_info = prev_json.get("build_info", {})
        prev_proto = prev_json.get("proto", {})
        prev_errors = prev_json.get("errors", {})

        # Error deltas
        error_delta = {}
        for key in ("collect2", "ctf"):
            curr = self.errors.get(key, {}).get("count", 0)
            prev = prev_errors.get(key, {}).get("count", 0)
            error_delta[key] = {
                "previous": prev, "current": curr, "change": curr - prev
            }
        # Use leaf failed dirs as the primary error metric
        curr_leaf = self.errors.get("failed_dirs_leaf", 0)
        prev_leaf = prev_errors.get("failed_dirs_leaf", 0)
        error_delta["failed_dirs"] = {
            "previous": prev_leaf, "current": curr_leaf,
            "change": curr_leaf - prev_leaf,
        }

        # Proto size delta
        curr_size = self.proto.get("total_size", 0)
        prev_size = prev_proto.get("total_size", 0)

        curr_files = self.proto.get("total_files", 0)
        prev_files = prev_proto.get("total_files", 0)

        # File list comparison for key categories
        added_files = []
        removed_files = []
        for cat in ("executables", "shared_libraries", "kernel_modules"):
            curr_set = set(self.proto.get("categories", {}).get(
                cat, {}).get("files", []))
            prev_set = set(prev_proto.get("categories", {}).get(
                cat, {}).get("files", []))
            for f in sorted(curr_set - prev_set):
                added_files.append({"file": f, "category": cat})
            for f in sorted(prev_set - curr_set):
                removed_files.append({"file": f, "category": cat})

        # Critical files delta
        curr_missing = set(f for f, _ in
                           self.proto.get("critical_files", {}).get(
                               "missing", []))
        prev_missing = set(f for f, _ in
                           prev_proto.get("critical_files", {}).get(
                               "missing", []))
        newly_present = sorted(prev_missing - curr_missing)
        newly_missing = sorted(curr_missing - prev_missing)

        self.delta = {
            "available": True,
            "previous_build": prev_info.get("commit_short", "unknown"),
            "previous_date": prev_info.get("date", "unknown"),
            "errors": error_delta,
            "proto": {
                "size_change": curr_size - prev_size,
                "file_count_change": curr_files - prev_files,
                "previous_size": prev_size,
                "current_size": curr_size,
            },
            "files_added": added_files[:50],
            "files_removed": removed_files[:50],
            "files_added_count": len(added_files),
            "files_removed_count": len(removed_files),
            "critical_files_fixed": newly_present,
            "critical_files_regressed": newly_missing,
        }

    # ------------------------------------------------------------------
    # Report generation
    # ------------------------------------------------------------------
    def generate_json(self):
        """Generate the JSON report."""
        report = OrderedDict([
            ("report_version", 1),
            ("generated", datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
            ("build_info", self.build_info),
            ("timing", self.timing),
            ("proto", self.proto),
            ("errors", self.errors),
            ("error_analysis", self.error_analysis),
            ("delta", self.delta),
        ])

        # Remove full file lists from error details to keep JSON manageable
        # Keep first 20 of each for reference
        for key in ("collect2", "error2", "ctf"):
            details = report["errors"][key].get("details", [])
            if len(details) > 30:
                report["errors"][key]["details"] = details[:30]
                report["errors"][key]["details_truncated"] = True
                report["errors"][key]["details_total"] = len(details)

        return report

    def generate_text(self):
        """Generate the human-readable text report."""
        lines = []

        def add(s=""):
            lines.append(s)

        def add_sep(char="=", width=72):
            lines.append(char * width)

        def fmt_size(n):
            """Format bytes to human-readable."""
            if n >= 1024 * 1024 * 1024:
                return f"{n / (1024**3):.1f} GB"
            elif n >= 1024 * 1024:
                return f"{n / (1024**2):.0f} MB"
            elif n >= 1024:
                return f"{n / 1024:.0f} KB"
            return f"{n} B"

        def fmt_change(n, unit=""):
            """Format a numeric change with +/- prefix."""
            if n > 0:
                return f"+{n}{unit}"
            elif n < 0:
                return f"{n}{unit}"
            return f"0{unit}"

        # Header
        add_sep()
        add("HAMMERHEAD BUILD REPORT")
        add_sep()
        add()
        add(f"  Version:   {self.build_info.get('hammerhead_version', '?')}")
        add(f"  Commit:    {self.build_info.get('commit_short', '?')} "
            f"({self.build_info.get('branch', '?')})")
        add(f"  Date:      {self.build_info.get('date', '?')}")
        add(f"  Host:      {self.build_info.get('host', '?')}")
        add(f"  Time:      {self.timing.get('formatted', '?')}")
        status = "PASS" if self.build_info.get("build_ok") == "y" else \
            "FAIL" if self.build_info.get("build_ok") == "n" else \
            self.build_info.get("build_ok", "?").upper()
        add(f"  Status:    {status}")
        add()

        # Error summary (prominent)
        add_sep("-")
        add("BUILD ERRORS")
        add_sep("-")
        c2 = self.errors.get("collect2", {}).get("count", 0)
        e2_raw = self.errors.get("error2", {}).get("count_raw", 0)
        ctf = self.errors.get("ctf", {}).get("count", 0)
        warn = self.errors.get("warnings_total", 0)
        n_leaf = self.errors.get("failed_dirs_leaf", 0)
        n_total = self.errors.get("failed_dirs_total", 0)

        add(f"  Linker errors (collect2):  {c2}")
        add(f"  Failed directories:        {n_leaf}  "
            f"({n_total} incl. parents, {e2_raw} raw error lines)")
        add(f"  CTF errors:                {ctf}")
        add(f"  Compiler warnings:         {warn}")
        add()

        # Delta section (if available)
        if self.delta.get("available"):
            add_sep("-")
            add(f"DELTA (vs {self.delta.get('previous_build', '?')})")
            add_sep("-")
            for key, label in [("collect2", "Linker errors"),
                                ("failed_dirs", "Failed dirs"),
                                ("ctf", "CTF errors")]:
                d = self.delta.get("errors", {}).get(key, {})
                prev = d.get("previous", 0)
                curr = d.get("current", 0)
                change = d.get("change", 0)
                marker = "  " if change == 0 else \
                    " *" if change < 0 else " !"
                add(f"  {label:.<30} {prev:>4} -> {curr:>4}  "
                    f"({fmt_change(change)}){marker}")

            pd = self.delta.get("proto", {})
            size_change = pd.get("size_change", 0)
            add(f"  {'Proto size':.<30} "
                f"{fmt_size(pd.get('previous_size', 0)):>4} -> "
                f"{fmt_size(pd.get('current_size', 0)):>4}  "
                f"({fmt_change(size_change // (1024*1024), ' MB')})")

            fc = pd.get("file_count_change", 0)
            add(f"  {'File count':.<30} {fmt_change(fc)}")

            # Newly fixed / regressed critical files
            fixed = self.delta.get("critical_files_fixed", [])
            regressed = self.delta.get("critical_files_regressed", [])
            if fixed:
                add()
                add(f"  Critical files FIXED ({len(fixed)}):")
                for f in fixed[:10]:
                    add(f"    [+] {f}")
            if regressed:
                add()
                add(f"  Critical files REGRESSED ({len(regressed)}):")
                for f in regressed[:10]:
                    add(f"    [-] {f}")

            # Added/removed files
            added = self.delta.get("files_added", [])
            removed = self.delta.get("files_removed", [])
            add_count = self.delta.get("files_added_count", 0)
            rem_count = self.delta.get("files_removed_count", 0)
            if added:
                add()
                add(f"  New files ({add_count}):")
                for entry in added[:15]:
                    add(f"    + {entry['file']}")
                if add_count > 15:
                    add(f"    ... and {add_count - 15} more")
            if removed:
                add()
                add(f"  Removed files ({rem_count}):")
                for entry in removed[:15]:
                    add(f"    - {entry['file']}")
                if rem_count > 15:
                    add(f"    ... and {rem_count - 15} more")
            add()

        # Proto area summary
        add_sep("-")
        add("PROTO AREA")
        add_sep("-")
        add(f"  Total size:  {fmt_size(self.proto.get('total_size', 0))}")
        add(f"  Total files: {self.proto.get('total_files', 0):,}")
        add()

        cats = self.proto.get("categories", {})
        cat_order = [
            ("executables", "Executables"),
            ("shared_libraries", "Shared libraries"),
            ("kernel_modules", "Kernel modules"),
            ("static_libraries", "Static libraries"),
            ("headers", "Headers"),
            ("man_pages", "Man pages"),
            ("config_files", "Config files"),
            ("symlinks", "Symlinks"),
            ("directories", "Directories"),
            ("other", "Other files"),
        ]
        for key, label in cat_order:
            cat = cats.get(key, {})
            count = cat.get("count", 0)
            size = cat.get("size", 0)
            if count > 0:
                if size:
                    add(f"  {label:.<25} {count:>6}  ({fmt_size(size)})")
                else:
                    add(f"  {label:.<25} {count:>6}")
        add()

        # Critical files
        add_sep("-")
        add("CRITICAL FILES")
        add_sep("-")
        present = self.proto.get("critical_files", {}).get("present", [])
        missing = self.proto.get("critical_files", {}).get("missing", [])

        add(f"  Present: {len(present)}/{len(present) + len(missing)}")
        add()

        if missing:
            add("  MISSING:")
            for fpath, desc in missing:
                add(f"    [!!] {fpath:50s} ({desc})")
            add()

        if present:
            add("  Present:")
            for item in present:
                fpath, desc = item[0], item[1]
                size = item[2] if len(item) > 2 else 0
                size_str = f" ({fmt_size(size)})" if size else ""
                add(f"    [OK] {fpath:50s} {desc}{size_str}")
            add()

        # Error analysis
        if self.error_analysis.get("root_causes"):
            add_sep("-")
            add("ERROR ANALYSIS")
            add_sep("-")
            add()
            add("  Root causes:")
            for i, rc in enumerate(self.error_analysis["root_causes"], 1):
                cat = rc["category"].replace("_", " ").title()
                add(f"    {i:2d}. {cat} ({rc['count']} errors)")
                if rc.get("sample"):
                    # Truncate long samples
                    sample = rc["sample"][:100]
                    add(f"        Example: {sample}")
                if rc.get("directories"):
                    dirs_str = ", ".join(rc["directories"][:5])
                    if len(rc["directories"]) > 5:
                        dirs_str += f" +{len(rc['directories']) - 5} more"
                    add(f"        Dirs: {dirs_str}")
                add()

        if self.error_analysis.get("cascades"):
            add("  Dependency cascades:")
            for cascade in self.error_analysis["cascades"]:
                if "missing_library" in cascade:
                    add(f"    - {cascade['missing_library']} not found -> "
                        f"{cascade['affected_count']} directories affected")
                elif "parent_directory" in cascade:
                    add(f"    - {cascade['parent_directory']} -> "
                        f"{cascade['affected_count']} sub-failures")
            add()

        # Failed directories listing
        failed = self.error_analysis.get("failed_directories", [])
        if failed:
            add_sep("-")
            add(f"FAILED DIRECTORIES ({len(failed)})")
            add_sep("-")
            for d in failed:
                add(f"    {d}")
            add()

        add_sep()
        add(f"Report generated: "
            f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        add(f"Log directory: {self.log_dir}")
        add_sep()

        return "\n".join(lines) + "\n"

    # ------------------------------------------------------------------
    # History tracking
    # ------------------------------------------------------------------
    def update_history(self):
        """Append a summary entry to the build history file."""
        entry = {
            "hammerhead_version": self.build_info.get(
                "hammerhead_version", "unknown"),
            "commit": self.build_info.get("commit_short", "unknown"),
            "branch": self.build_info.get("branch", "unknown"),
            "date": self.build_info.get("date", "unknown"),
            "build_ok": self.build_info.get("build_ok", "unknown"),
            "build_time": self.timing.get("formatted", "unknown"),
            "build_time_seconds": self.timing.get("total_seconds", 0),
            "collect2_errors": self.errors.get("collect2", {}).get("count", 0),
            "failed_dirs_leaf": self.errors.get("failed_dirs_leaf", 0),
            "failed_dirs_total": self.errors.get("failed_dirs_total", 0),
            "ctf_errors": self.errors.get("ctf", {}).get("count", 0),
            "warnings": self.errors.get("warnings_total", 0),
            "proto_size": self.proto.get("total_size", 0),
            "proto_files": self.proto.get("total_files", 0),
            "executables": self.proto.get("categories", {}).get(
                "executables", {}).get("count", 0),
            "shared_libraries": self.proto.get("categories", {}).get(
                "shared_libraries", {}).get("count", 0),
            "kernel_modules": self.proto.get("categories", {}).get(
                "kernel_modules", {}).get("count", 0),
            "critical_present": len(self.proto.get("critical_files", {}).get(
                "present", [])),
            "critical_missing": len(self.proto.get("critical_files", {}).get(
                "missing", [])),
            "log_dir": str(self.log_dir),
        }

        history = []
        if self.history_file.exists():
            try:
                history = json.loads(self.history_file.read_text())
                if not isinstance(history, list):
                    history = []
            except (json.JSONDecodeError, IOError):
                history = []

        history.append(entry)

        try:
            self.history_file.parent.mkdir(parents=True, exist_ok=True)
            self.history_file.write_text(
                json.dumps(history, indent=2) + "\n")
        except IOError as e:
            print(f"Warning: Could not write history: {e}", file=sys.stderr)

    # ------------------------------------------------------------------
    # Main execution
    # ------------------------------------------------------------------
    def run(self):
        """Run the full analysis and generate reports."""
        print("Analyzing proto area...", file=sys.stderr)
        self.analyze_proto()

        print("Analyzing build log...", file=sys.stderr)
        self.analyze_log()

        print("Computing delta...", file=sys.stderr)
        self.compute_delta()

        # Generate reports
        json_report = self.generate_json()
        text_report = self.generate_text()

        # Write reports
        json_path = self.log_dir / "build-report.json"
        text_path = self.log_dir / "build-report.txt"

        try:
            json_path.write_text(json.dumps(json_report, indent=2,
                                            default=str) + "\n")
            print(f"JSON report: {json_path}", file=sys.stderr)
        except IOError as e:
            print(f"Error writing JSON report: {e}", file=sys.stderr)

        try:
            text_path.write_text(text_report)
            print(f"Text report: {text_path}", file=sys.stderr)
        except IOError as e:
            print(f"Error writing text report: {e}", file=sys.stderr)

        # Update history
        print("Updating build history...", file=sys.stderr)
        self.update_history()

        # Print text report to stdout as well
        print(text_report)

        return 0 if self.errors.get("collect2", {}).get("count", 0) == 0 \
            else 1


def main():
    parser = argparse.ArgumentParser(
        description="Hammerhead build analysis and reporting tool")
    parser.add_argument("--log-dir", required=True,
                        help="Build log directory (contains build.log)")
    parser.add_argument("--proto", required=True,
                        help="Proto area directory (e.g., proto/root_amd64)")
    parser.add_argument("--workspace", required=True,
                        help="Workspace root directory")
    parser.add_argument("--history",
                        help="Path to build history JSON file "
                             "(default: <workspace>/log/build-history.json)")
    parser.add_argument("--previous",
                        help="Path to previous build's log directory "
                             "for delta comparison (auto-detected if omitted)")
    args = parser.parse_args()

    report = BuildReport(
        log_dir=args.log_dir,
        proto_dir=args.proto,
        workspace=args.workspace,
        history_file=args.history,
        previous_dir=args.previous,
    )

    sys.exit(report.run())


if __name__ == "__main__":
    main()