parse_logs.py

#!/usr/bin/env python3
"""PentesterPro log parser with redundancy/debug diagnostics.

Supports plain text `run_log.txt` and JSONL decision logs.
"""

from __future__ import annotations

import argparse
import io
import json
import re
from contextlib import redirect_stdout
from collections import Counter, defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional


# ---------- File type ----------

def detect_file_type(path: Path) -> str:
    try:
        first = path.read_text(encoding="utf-8", errors="ignore").splitlines()[0].strip()
        if first.startswith("{") and first.endswith("}"):
            return "jsonl"
    except Exception:
        pass
    return "text"


# ---------- Text log parsing ----------


@dataclass
class TextLogData:
    tasks: Dict[str, List[str]]
    starts_by_task: Counter
    starts_by_url: Counter
    starts_by_type: Counter
    scheduled_by_type: Counter
    processed_by_type: Counter
    intake_accept_by_type: Counter
    intake_accept_by_url: Counter
    intake_drop_scope: Counter
    intake_drop_frontier: Counter
    intake_drop_saturation: Counter
    invalidated: Counter
    skipped_role_sig: Counter
    skipped_global_sat: Counter
    skipped_cross_page: Counter
    skipped_duplicate_submit: Counter
    coalesced_duplicate_submit: Counter
    planned_zero: int
    worker_failed: int
    worker_crashed: int
    transitions: List[tuple]
    executed_action_sig: Counter
    repeated_exec_by_task_sig: Counter
    raw_lines: List[str]


def parse_text_logs(path: Path) -> TextLogData:
    tasks: Dict[str, List[str]] = defaultdict(list)
    starts_by_task: Counter = Counter()
    starts_by_url: Counter = Counter()
    starts_by_type: Counter = Counter()
    scheduled_by_type: Counter = Counter()
    processed_by_type: Counter = Counter()
    intake_accept_by_type: Counter = Counter()
    intake_accept_by_url: Counter = Counter()
    intake_drop_scope: Counter = Counter()
    intake_drop_frontier: Counter = Counter()
    intake_drop_saturation: Counter = Counter()
    invalidated: Counter = Counter()
    skipped_role_sig: Counter = Counter()
    skipped_global_sat: Counter = Counter()
    skipped_cross_page: Counter = Counter()
    skipped_duplicate_submit: Counter = Counter()
    coalesced_duplicate_submit: Counter = Counter()
    transitions: List[tuple] = []
    executed_action_sig: Counter = Counter()
    exec_by_task_sig: Dict[str, Counter] = defaultdict(Counter)
    repeated_exec_by_task_sig: Counter = Counter()

    planned_zero = 0
    worker_failed = 0
    worker_crashed = 0

    current_task: Optional[str] = None
    raw_lines: List[str] = []

    re_task_start = re.compile(r"Starting Task ([a-f0-9-]+) \(([^)]+)\) on (https?://\S+)")
    re_schedule = re.compile(r"SCHEDULE: [a-f0-9-]+ \(([^)]+)\)")
    re_processed = re.compile(r"Processed discovery: [a-f0-9-]+ \(([^)]+)\)")
    re_intake_accept = re.compile(r"INTAKE ACCEPT: (.*?) \(([^)]+)\)")
    re_scope_drop = re.compile(r"SCOPE_DROP: (.*?) filtered")
    re_frontier_drop = re.compile(r"FRONTIER DROP: (.*?) already known")
    re_sat_drop = re.compile(r"SATURATION_DROP: (.*?) is saturated")
    re_invalidated = re.compile(r"Action invalidated: ([a-f0-9]+)")
    re_skip_role = re.compile(r"Skipping role-saturated(?: planned)? signature: ([a-f0-9]+)")
    re_skip_global = re.compile(r"Skipping saturated global nav action: ([a-f0-9]+)")
    re_skip_cross = re.compile(r"Skipping executed cross-page action: ([a-f0-9]+)")
    re_skip_dup_submit = re.compile(r"Skipping duplicate submit intent: ([a-f0-9]+)")
    re_coalesce_dup_submit = re.compile(r"Coalesced duplicate submit action: ([a-f0-9]+)")
    re_exec_action = re.compile(r"Executing planned action: ([a-f0-9]+)")
    re_transition = re.compile(r"State Transition detected \((.*?) -> (.*?) \| ([a-f0-9]+) -> ([a-f0-9]+)\)")

    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
        raw_lines.append(line)

        m = re_task_start.search(line)
        if m:
            task_id, entity_type, url = m.group(1), m.group(2), m.group(3)
            current_task = task_id
            starts_by_task[task_id] += 1
            starts_by_url[url] += 1
            starts_by_type[entity_type] += 1

        if current_task:
            tasks[current_task].append(line)

        m = re_schedule.search(line)
        if m:
            scheduled_by_type[m.group(1).upper()] += 1

        m = re_processed.search(line)
        if m:
            processed_by_type[m.group(1).upper()] += 1

        m = re_intake_accept.search(line)
        if m:
            url, etype = m.group(1), m.group(2)
            intake_accept_by_type[etype.upper()] += 1
            intake_accept_by_url[url] += 1

        m = re_scope_drop.search(line)
        if m:
            intake_drop_scope[m.group(1)] += 1

        m = re_frontier_drop.search(line)
        if m:
            intake_drop_frontier[m.group(1)] += 1

        m = re_sat_drop.search(line)
        if m:
            intake_drop_saturation[m.group(1)] += 1

        m = re_invalidated.search(line)
        if m:
            invalidated[m.group(1)] += 1

        m = re_skip_role.search(line)
        if m:
            skipped_role_sig[m.group(1)] += 1

        m = re_skip_global.search(line)
        if m:
            skipped_global_sat[m.group(1)] += 1

        m = re_skip_cross.search(line)
        if m:
            skipped_cross_page[m.group(1)] += 1

        m = re_skip_dup_submit.search(line)
        if m:
            skipped_duplicate_submit[m.group(1)] += 1

        m = re_coalesce_dup_submit.search(line)
        if m:
            coalesced_duplicate_submit[m.group(1)] += 1

        m = re_exec_action.search(line)
        if m:
            sig = m.group(1)
            executed_action_sig[sig] += 1
            if current_task:
                exec_by_task_sig[current_task][sig] += 1

        m = re_transition.search(line)
        if m:
            transitions.append((m.group(1), m.group(2), m.group(3), m.group(4)))

        if "Frontier Freeze: Planned 0 actions" in line:
            planned_zero += 1
        if "Task for " in line and " FAILED" in line:
            worker_failed += 1
        if "Worker Execution crashed" in line:
            worker_crashed += 1

    for task_id, by_sig in exec_by_task_sig.items():
        for sig, count in by_sig.items():
            if count > 1:
                repeated_exec_by_task_sig[f"{task_id}:{sig}"] = count

    return TextLogData(
        tasks=tasks,
        starts_by_task=starts_by_task,
        starts_by_url=starts_by_url,
        starts_by_type=starts_by_type,
        scheduled_by_type=scheduled_by_type,
        processed_by_type=processed_by_type,
        intake_accept_by_type=intake_accept_by_type,
        intake_accept_by_url=intake_accept_by_url,
        intake_drop_scope=intake_drop_scope,
        intake_drop_frontier=intake_drop_frontier,
        intake_drop_saturation=intake_drop_saturation,
        invalidated=invalidated,
        skipped_role_sig=skipped_role_sig,
        skipped_global_sat=skipped_global_sat,
        skipped_cross_page=skipped_cross_page,
        skipped_duplicate_submit=skipped_duplicate_submit,
        coalesced_duplicate_submit=coalesced_duplicate_submit,
        planned_zero=planned_zero,
        worker_failed=worker_failed,
        worker_crashed=worker_crashed,
        transitions=transitions,
        executed_action_sig=executed_action_sig,
        repeated_exec_by_task_sig=repeated_exec_by_task_sig,
        raw_lines=raw_lines,
    )


# ---------- JSONL parsing ----------


def parse_jsonl_logs(path: Path) -> Dict[str, Any]:
    out: Dict[str, Any] = {
        "count": 0,
        "category": Counter(),
        "decision": Counter(),
        "reason": Counter(),
        "url": Counter(),
        "events": [],
    }
    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except Exception:
            continue
        out["count"] += 1
        cat = msg.get("category") or "UNKNOWN"
        dec = msg.get("decision") or "UNKNOWN"
        proof = msg.get("proof") or {}
        ctx = msg.get("context") or {}
        url = (ctx.get("url") or proof.get("url") or "").strip()
        reason = (proof.get("reason") or "").strip()

        out["category"][str(cat)] += 1
        out["decision"][str(dec)] += 1
        if reason:
            out["reason"][reason] += 1
        if url:
            out["url"][url] += 1
        out["events"].append(msg)

    return out


# ---------- Printing helpers ----------


def _print_counter(title: str, c: Counter, top: int = 20, min_count: int = 1) -> None:
    print(f"\n{title}")
    print("-" * len(title))
    shown = 0
    for k, v in c.most_common():
        if v < min_count:
            continue
        print(f"{v:>5}  {k}")
        shown += 1
        if shown >= top:
            break
    if shown == 0:
        print("(none)")


def print_text_summary(d: TextLogData) -> None:
    print("\n=== Text Log Summary ===")
    print(f"Tasks seen: {len(d.tasks)}")
    print(f"Task starts: {sum(d.starts_by_task.values())}")
    print(f"Unique started URLs: {len(d.starts_by_url)}")
    print(f"Transitions: {len(d.transitions)}")
    print(f"Invalidated actions: {sum(d.invalidated.values())}")
    print(f"Skipped role signatures: {sum(d.skipped_role_sig.values())}")
    print(f"Skipped global-saturated: {sum(d.skipped_global_sat.values())}")
    print(f"Skipped cross-page: {sum(d.skipped_cross_page.values())}")
    print(f"Coalesced duplicate submits: {sum(d.coalesced_duplicate_submit.values())}")
    print(f"Skipped duplicate submit executes: {sum(d.skipped_duplicate_submit.values())}")
    print(f"Repeated action executes (task+sig): {sum(1 for _ in d.repeated_exec_by_task_sig.items())}")
    print(f"Planned 0 actions: {d.planned_zero}")
    print(f"Worker failed tasks: {d.worker_failed}")
    print(f"Worker crashes: {d.worker_crashed}")


def print_jsonl_summary(d: Dict[str, Any]) -> None:
    print("\n=== JSONL Summary ===")
    print(f"Events: {d['count']}")
    _print_counter("By Category", d["category"], top=20)
    _print_counter("By Decision", d["decision"], top=20)
    _print_counter("By Reason", d["reason"], top=20)


def print_redundancy_report(d: TextLogData, top: int, min_count: int) -> None:
    print("\n=== Redundancy Report ===")
    _print_counter("Repeated URL Starts", d.starts_by_url, top=top, min_count=min_count)
    _print_counter("Repeated Task Starts", d.starts_by_task, top=top, min_count=min_count)
    _print_counter("Repeated Executed Actions (task:sig)", d.repeated_exec_by_task_sig, top=top, min_count=min_count)
    _print_counter("Invalidated Signatures", d.invalidated, top=top, min_count=min_count)
    _print_counter("Role Signature Skips", d.skipped_role_sig, top=top, min_count=min_count)
    _print_counter("Frontier Drop URLs", d.intake_drop_frontier, top=top, min_count=min_count)


def print_submit_report(d: TextLogData, top: int, min_count: int) -> None:
    print("\n=== Submit Interaction Report ===")
    _print_counter("Coalesced Duplicate Submit Signatures", d.coalesced_duplicate_submit, top=top, min_count=min_count)
    _print_counter("Skipped Duplicate Submit Signatures", d.skipped_duplicate_submit, top=top, min_count=min_count)
    _print_counter("Executed Action Signatures", d.executed_action_sig, top=top, min_count=min_count)


def print_pipeline_report(d: TextLogData, top: int, min_count: int) -> None:
    print("\n=== Pipeline Report ===")
    _print_counter("Starts By EntityType", d.starts_by_type, top=top, min_count=min_count)
    _print_counter("Scheduled By EntityType", d.scheduled_by_type, top=top, min_count=min_count)
    _print_counter("Processed Discoveries", d.processed_by_type, top=top, min_count=min_count)
    _print_counter("Intake Accept By Type", d.intake_accept_by_type, top=top, min_count=min_count)


def _delta_counter(current: Counter, baseline: Counter) -> Counter:
    keys = set(current.keys()) | set(baseline.keys())
    out: Counter = Counter()
    for k in keys:
        out[k] = int(current.get(k, 0)) - int(baseline.get(k, 0))
    return out


def _counter_to_delta_dict(current: Counter, baseline: Counter, min_abs_delta: int = 1) -> Dict[str, int]:
    delta = _delta_counter(current, baseline)
    items = sorted(delta.items(), key=lambda kv: abs(kv[1]), reverse=True)
    out: Dict[str, int] = {}
    for k, v in items:
        if abs(v) < min_abs_delta:
            continue
        out[str(k)] = int(v)
    return out


def _print_delta_counter(title: str, delta: Counter, top: int = 20, min_abs_delta: int = 1) -> None:
    print(f"\n{title}")
    print("-" * len(title))
    items = sorted(delta.items(), key=lambda kv: abs(kv[1]), reverse=True)
    shown = 0
    for k, v in items:
        if abs(v) < min_abs_delta:
            continue
        sign = "+" if v >= 0 else ""
        print(f"{sign}{v:>4}  {k}")
        shown += 1
        if shown >= top:
            break
    if shown == 0:
        print("(no changes)")


def print_text_compare(current: TextLogData, baseline: TextLogData, top: int, min_abs_delta: int) -> None:
    print("\n=== Text Log Compare (Current - Baseline) ===")

    scalar_rows = [
        ("tasks_seen", len(current.tasks), len(baseline.tasks)),
        ("task_starts", sum(current.starts_by_task.values()), sum(baseline.starts_by_task.values())),
        ("unique_started_urls", len(current.starts_by_url), len(baseline.starts_by_url)),
        ("transitions", len(current.transitions), len(baseline.transitions)),
        ("invalidated_total", sum(current.invalidated.values()), sum(baseline.invalidated.values())),
        ("skipped_role_sig_total", sum(current.skipped_role_sig.values()), sum(baseline.skipped_role_sig.values())),
        ("skipped_global_sat_total", sum(current.skipped_global_sat.values()), sum(baseline.skipped_global_sat.values())),
        ("skipped_cross_page_total", sum(current.skipped_cross_page.values()), sum(baseline.skipped_cross_page.values())),
        ("coalesced_duplicate_submit_total", sum(current.coalesced_duplicate_submit.values()), sum(baseline.coalesced_duplicate_submit.values())),
        ("skipped_duplicate_submit_total", sum(current.skipped_duplicate_submit.values()), sum(baseline.skipped_duplicate_submit.values())),
        ("planned_zero", current.planned_zero, baseline.planned_zero),
        ("worker_failed", current.worker_failed, baseline.worker_failed),
        ("worker_crashed", current.worker_crashed, baseline.worker_crashed),
    ]
    for name, cur, base in scalar_rows:
        delta = cur - base
        sign = "+" if delta >= 0 else ""
        print(f"{name:28} {cur:>6}  (base {base:>6}, delta {sign}{delta})")

    _print_delta_counter(
        "Delta: Started URLs",
        _delta_counter(current.starts_by_url, baseline.starts_by_url),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Invalidated Signatures",
        _delta_counter(current.invalidated, baseline.invalidated),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Role Signature Skips",
        _delta_counter(current.skipped_role_sig, baseline.skipped_role_sig),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Coalesced Duplicate Submit Signatures",
        _delta_counter(current.coalesced_duplicate_submit, baseline.coalesced_duplicate_submit),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Skipped Duplicate Submit Signatures",
        _delta_counter(current.skipped_duplicate_submit, baseline.skipped_duplicate_submit),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Scheduled Entity Types",
        _delta_counter(current.scheduled_by_type, baseline.scheduled_by_type),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Frontier Drop URLs",
        _delta_counter(current.intake_drop_frontier, baseline.intake_drop_frontier),
        top=top,
        min_abs_delta=min_abs_delta,
    )


def print_jsonl_compare(current: Dict[str, Any], baseline: Dict[str, Any], top: int, min_abs_delta: int) -> None:
    print("\n=== JSONL Compare (Current - Baseline) ===")
    scalar_rows = [
        ("events", current["count"], baseline["count"]),
        ("unique_categories", len(current["category"]), len(baseline["category"])),
        ("unique_decisions", len(current["decision"]), len(baseline["decision"])),
        ("unique_urls", len(current["url"]), len(baseline["url"])),
    ]
    for name, cur, base in scalar_rows:
        delta = cur - base
        sign = "+" if delta >= 0 else ""
        print(f"{name:28} {cur:>6}  (base {base:>6}, delta {sign}{delta})")

    _print_delta_counter(
        "Delta: Categories",
        _delta_counter(current["category"], baseline["category"]),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Decisions",
        _delta_counter(current["decision"], baseline["decision"]),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: Reasons",
        _delta_counter(current["reason"], baseline["reason"]),
        top=top,
        min_abs_delta=min_abs_delta,
    )
    _print_delta_counter(
        "Delta: URLs",
        _delta_counter(current["url"], baseline["url"]),
        top=top,
        min_abs_delta=min_abs_delta,
    )


def build_text_compare_json(current: TextLogData, baseline: TextLogData, min_abs_delta: int) -> Dict[str, Any]:
    cur_scalars = {
        "tasks_seen": len(current.tasks),
        "task_starts": sum(current.starts_by_task.values()),
        "unique_started_urls": len(current.starts_by_url),
        "transitions": len(current.transitions),
        "invalidated_total": sum(current.invalidated.values()),
        "skipped_role_sig_total": sum(current.skipped_role_sig.values()),
        "skipped_global_sat_total": sum(current.skipped_global_sat.values()),
        "skipped_cross_page_total": sum(current.skipped_cross_page.values()),
        "coalesced_duplicate_submit_total": sum(current.coalesced_duplicate_submit.values()),
        "skipped_duplicate_submit_total": sum(current.skipped_duplicate_submit.values()),
        "planned_zero": current.planned_zero,
        "worker_failed": current.worker_failed,
        "worker_crashed": current.worker_crashed,
    }
    base_scalars = {
        "tasks_seen": len(baseline.tasks),
        "task_starts": sum(baseline.starts_by_task.values()),
        "unique_started_urls": len(baseline.starts_by_url),
        "transitions": len(baseline.transitions),
        "invalidated_total": sum(baseline.invalidated.values()),
        "skipped_role_sig_total": sum(baseline.skipped_role_sig.values()),
        "skipped_global_sat_total": sum(baseline.skipped_global_sat.values()),
        "skipped_cross_page_total": sum(baseline.skipped_cross_page.values()),
        "coalesced_duplicate_submit_total": sum(baseline.coalesced_duplicate_submit.values()),
        "skipped_duplicate_submit_total": sum(baseline.skipped_duplicate_submit.values()),
        "planned_zero": baseline.planned_zero,
        "worker_failed": baseline.worker_failed,
        "worker_crashed": baseline.worker_crashed,
    }
    scalar_delta = {k: int(cur_scalars[k] - base_scalars[k]) for k in cur_scalars.keys()}
    return {
        "mode": "compare",
        "type": "text",
        "scalar": {
            "current": cur_scalars,
            "baseline": base_scalars,
            "delta": scalar_delta,
        },
        "delta": {
            "started_urls": _counter_to_delta_dict(current.starts_by_url, baseline.starts_by_url, min_abs_delta),
            "invalidated_signatures": _counter_to_delta_dict(current.invalidated, baseline.invalidated, min_abs_delta),
            "role_signature_skips": _counter_to_delta_dict(current.skipped_role_sig, baseline.skipped_role_sig, min_abs_delta),
            "coalesced_duplicate_submit": _counter_to_delta_dict(current.coalesced_duplicate_submit, baseline.coalesced_duplicate_submit, min_abs_delta),
            "skipped_duplicate_submit": _counter_to_delta_dict(current.skipped_duplicate_submit, baseline.skipped_duplicate_submit, min_abs_delta),
            "scheduled_entity_types": _counter_to_delta_dict(current.scheduled_by_type, baseline.scheduled_by_type, min_abs_delta),
            "frontier_drop_urls": _counter_to_delta_dict(current.intake_drop_frontier, baseline.intake_drop_frontier, min_abs_delta),
        },
    }


def build_jsonl_compare_json(current: Dict[str, Any], baseline: Dict[str, Any], min_abs_delta: int) -> Dict[str, Any]:
    cur_scalars = {
        "events": int(current["count"]),
        "unique_categories": len(current["category"]),
        "unique_decisions": len(current["decision"]),
        "unique_urls": len(current["url"]),
    }
    base_scalars = {
        "events": int(baseline["count"]),
        "unique_categories": len(baseline["category"]),
        "unique_decisions": len(baseline["decision"]),
        "unique_urls": len(baseline["url"]),
    }
    scalar_delta = {k: int(cur_scalars[k] - base_scalars[k]) for k in cur_scalars.keys()}
    return {
        "mode": "compare",
        "type": "jsonl",
        "scalar": {
            "current": cur_scalars,
            "baseline": base_scalars,
            "delta": scalar_delta,
        },
        "delta": {
            "categories": _counter_to_delta_dict(current["category"], baseline["category"], min_abs_delta),
            "decisions": _counter_to_delta_dict(current["decision"], baseline["decision"], min_abs_delta),
            "reasons": _counter_to_delta_dict(current["reason"], baseline["reason"], min_abs_delta),
            "urls": _counter_to_delta_dict(current["url"], baseline["url"], min_abs_delta),
        },
    }


def filter_lines(lines: Iterable[str], contains: Optional[str], regex: Optional[str], ignore_case: bool) -> List[str]:
    out: List[str] = []
    flags = re.IGNORECASE if ignore_case else 0
    cre = re.compile(regex, flags) if regex else None
    for ln in lines:
        if contains and ((contains.lower() not in ln.lower()) if ignore_case else (contains not in ln)):
            continue
        if cre and not cre.search(ln):
            continue
        out.append(ln)
    return out


def grep_lines(lines: Iterable[str], pattern: str, ignore_case: bool, use_regex: bool) -> List[str]:
    out: List[str] = []
    if use_regex:
        flags = re.IGNORECASE if ignore_case else 0
        cre = re.compile(pattern, flags)
        for ln in lines:
            if cre.search(ln):
                out.append(ln)
        return out
    needle = pattern.lower() if ignore_case else pattern
    for ln in lines:
        hay = ln.lower() if ignore_case else ln
        if needle in hay:
            out.append(ln)
    return out


def highlight_line(line: str, pattern: str, ignore_case: bool, use_regex: bool) -> str:
    if use_regex:
        flags = re.IGNORECASE if ignore_case else 0
        cre = re.compile(pattern, flags)
        return cre.sub(lambda m: f"\033[1;33m{m.group(0)}\033[0m", line)
    flags = re.IGNORECASE if ignore_case else 0
    cre = re.compile(re.escape(pattern), flags)
    return cre.sub(lambda m: f"\033[1;33m{m.group(0)}\033[0m", line)


def main() -> None:
    p = argparse.ArgumentParser(description="PentesterPro log parser")
    p.add_argument("file", help="Path to log file")

    # generic outputs
    p.add_argument("--summary", action="store_true", help="Show high-level summary")
    p.add_argument("--all", action="store_true", help="Show summary + redundancy + pipeline")
    p.add_argument("--json", action="store_true", help="Emit machine-readable JSON summary")
    p.add_argument("--compare", help="Compare current log with baseline log path")
    p.add_argument("--compare-json", action="store_true", help="Emit machine-readable JSON diff (requires --compare)")

    # diagnostics
    p.add_argument("--redundancy", action="store_true", help="Show redundancy-focused report")
    p.add_argument("--pipeline", action="store_true", help="Show scheduling/intake pipeline counters")
    p.add_argument("--top-urls", action="store_true", help="Top started URLs")
    p.add_argument("--top-tasks", action="store_true", help="Top started task IDs")
    p.add_argument("--top-invalidated", action="store_true", help="Top invalidated action signatures")
    p.add_argument("--top-role-skips", action="store_true", help="Top role-signature skip counts")
    p.add_argument("--top-duplicate-exec", action="store_true", help="Top repeated executed action signatures per task")
    p.add_argument("--top-frontier-drops", action="store_true", help="Top frontier-drop URLs")
    p.add_argument("--top-intake-accept", action="store_true", help="Top intake accepted URLs")
    p.add_argument("--submit-report", action="store_true", help="Show submit interaction diagnostics")
    p.add_argument("--planned-zero", action="store_true", help="Print planned-zero-actions count")
    p.add_argument("--failures", action="store_true", help="Print worker failed/crash counters")

    # filters
    p.add_argument("--task", help="Show lines for task id (prefix/substring)")
    p.add_argument("--url", help="Show lines containing URL substring")
    p.add_argument("--signature", help="Show lines for action signature substring")
    p.add_argument("--contains", help="Filter raw lines by plain substring")
    p.add_argument("--regex", help="Filter raw lines by regex")
    p.add_argument("--grep", help="Search pattern over logs/report output")
    p.add_argument("--find", help="Alias of --grep")
    p.add_argument("--grep-target", choices=["raw", "report", "both"], default="raw", help="Where --grep/--find is applied")
    p.add_argument("--grep-regex", action="store_true", help="Treat --grep/--find as regex")
    p.add_argument("--highlight", action="store_true", help="Highlight grep matches")
    p.add_argument("--ignore-case", action="store_true", help="Case-insensitive line filters")
    p.add_argument("--tail", type=int, default=0, help="Tail N filtered lines")
    p.add_argument("--head", type=int, default=0, help="Head N filtered lines")

    # formatting
    p.add_argument("--top", type=int, default=20, help="Top N entries")
    p.add_argument("--min-count", type=int, default=1, help="Minimum count threshold")
    p.add_argument("--min-abs-delta", type=int, default=1, help="Minimum absolute delta for compare mode")

    args = p.parse_args()
    grep_pattern = args.grep or args.find
    path = Path(args.file)
    if not path.exists():
        raise SystemExit(f"File not found: {path}")

    ftype = detect_file_type(path)
    baseline_path = Path(args.compare).expanduser().resolve() if args.compare else None
    if baseline_path and not baseline_path.exists():
        raise SystemExit(f"Baseline file not found: {baseline_path}")
    baseline_type = detect_file_type(baseline_path) if baseline_path else None
    if baseline_type and baseline_type != ftype:
        raise SystemExit(
            f"File type mismatch: current={ftype}, baseline={baseline_type}. Compare requires same log type."
        )

    if ftype == "jsonl":
        data = parse_jsonl_logs(path)
        baseline_data = parse_jsonl_logs(baseline_path) if baseline_path else None
        if args.compare_json and not baseline_data:
            raise SystemExit("--compare-json requires --compare <baseline_log>")
        if args.compare_json and baseline_data:
            print(json.dumps(build_jsonl_compare_json(data, baseline_data, args.min_abs_delta), indent=2))
            return
        if baseline_data:
            buf = io.StringIO()
            with redirect_stdout(buf):
                print_jsonl_compare(data, baseline_data, top=args.top, min_abs_delta=args.min_abs_delta)
            out = buf.getvalue()
            if grep_pattern and args.grep_target in ("report", "both"):
                lines = grep_lines(out.splitlines(), grep_pattern, args.ignore_case, args.grep_regex)
                for ln in lines:
                    print(highlight_line(ln, grep_pattern, args.ignore_case, args.grep_regex) if args.highlight else ln)
            else:
                print(out, end="")
            if grep_pattern and args.grep_target in ("raw", "both"):
                raw = grep_lines(path.read_text(encoding="utf-8", errors="ignore").splitlines(), grep_pattern, args.ignore_case, args.grep_regex)
                print("\nRaw Grep Matches")
                print("----------------")
                for ln in raw:
                    print(highlight_line(ln, grep_pattern, args.ignore_case, args.grep_regex) if args.highlight else ln)
                if not raw:
                    print("(none)")
            return
        if args.json:
            serializable = {
                "type": "jsonl",
                "count": data["count"],
                "category": dict(data["category"]),
                "decision": dict(data["decision"]),
                "reason": dict(data["reason"]),
                "url": dict(data["url"]),
            }
            print(json.dumps(serializable, indent=2))
            return

        if args.all or args.summary or not any(
            [
                args.redundancy,
                args.pipeline,
                args.top_urls,
                args.top_tasks,
                args.top_invalidated,
                args.top_role_skips,
                args.top_duplicate_exec,
                args.top_frontier_drops,
                args.top_intake_accept,
                args.submit_report,
                args.planned_zero,
                args.failures,
                args.task,
                args.url,
                args.signature,
                args.contains,
                args.regex,
                grep_pattern,
            ]
        ):
            buf = io.StringIO()
            with redirect_stdout(buf):
                print_jsonl_summary(data)
                if args.top_urls:
                    _print_counter("Top URLs", data["url"], top=args.top, min_count=args.min_count)
            out = buf.getvalue()
            if grep_pattern and args.grep_target in ("report", "both"):
                lines = grep_lines(out.splitlines(), grep_pattern, args.ignore_case, args.grep_regex)
                for ln in lines:
                    print(highlight_line(ln, grep_pattern, args.ignore_case, args.grep_regex) if args.highlight else ln)
            else:
                print(out, end="")

        if args.top_urls and not (args.all or args.summary):
            _print_counter("Top URLs", data["url"], top=args.top, min_count=args.min_count)
        if args.top_frontier_drops or args.top_invalidated or args.top_role_skips or args.top_tasks:
            print("\nThese counters are text-log specific.")
        if grep_pattern and args.grep_target in ("raw", "both"):
            raw = grep_lines(path.read_text(encoding="utf-8", errors="ignore").splitlines(), grep_pattern, args.ignore_case, args.grep_regex)
            print("\nRaw Grep Matches")
            print("----------------")
            for ln in raw:
                print(highlight_line(ln, grep_pattern, args.ignore_case, args.grep_regex) if args.highlight else ln)
            if not raw:
                print("(none)")
        return

    # text log
    d = parse_text_logs(path)
    baseline_text = parse_text_logs(baseline_path) if baseline_path else None
    if args.compare_json and not baseline_text:
        raise SystemExit("--compare-json requires --compare <baseline_log>")
    if args.compare_json and baseline_text:
        print(json.dumps(build_text_compare_json(d, baseline_text, args.min_abs_delta), indent=2))
        return
    if baseline_text:
        print_text_compare(d, baseline_text, top=args.top, min_abs_delta=args.min_abs_delta)
        return

    if args.json:
        serializable = {
            "type": "text",
            "tasks": len(d.tasks),
            "task_starts": sum(d.starts_by_task.values()),
            "starts_by_url": dict(d.starts_by_url),
            "starts_by_task": dict(d.starts_by_task),
            "starts_by_type": dict(d.starts_by_type),
            "scheduled_by_type": dict(d.scheduled_by_type),
            "processed_by_type": dict(d.processed_by_type),
            "intake_accept_by_type": dict(d.intake_accept_by_type),
            "intake_accept_by_url": dict(d.intake_accept_by_url),
            "invalidated": dict(d.invalidated),
            "skipped_role_sig": dict(d.skipped_role_sig),
            "skipped_global_sat": dict(d.skipped_global_sat),
            "skipped_cross_page": dict(d.skipped_cross_page),
            "coalesced_duplicate_submit": dict(d.coalesced_duplicate_submit),
            "skipped_duplicate_submit": dict(d.skipped_duplicate_submit),
            "repeated_exec_by_task_sig": dict(d.repeated_exec_by_task_sig),
            "planned_zero": d.planned_zero,
            "worker_failed": d.worker_failed,
            "worker_crashed": d.worker_crashed,
            "transitions": len(d.transitions),
        }
        print(json.dumps(serializable, indent=2))
        return

    run_default = not any(
        [
            args.all,
            args.summary,
            args.redundancy,
            args.pipeline,
            args.top_urls,
            args.top_tasks,
            args.top_invalidated,
            args.top_role_skips,
            args.top_duplicate_exec,
            args.top_frontier_drops,
            args.top_intake_accept,
            args.submit_report,
            args.planned_zero,
            args.failures,
            args.task,
            args.url,
            args.signature,
            args.contains,
            args.regex,
            grep_pattern,
        ]
    )
    report_buf = io.StringIO()
    with redirect_stdout(report_buf):
        if args.all or args.summary or run_default:
            print_text_summary(d)

        if args.all or args.redundancy:
            print_redundancy_report(d, top=args.top, min_count=args.min_count)

        if args.all or args.pipeline:
            print_pipeline_report(d, top=args.top, min_count=args.min_count)

        if args.all or args.submit_report:
            print_submit_report(d, top=args.top, min_count=args.min_count)

        if args.top_urls:
            _print_counter("Top Started URLs", d.starts_by_url, top=args.top, min_count=args.min_count)
        if args.top_tasks:
            _print_counter("Top Started Tasks", d.starts_by_task, top=args.top, min_count=args.min_count)
        if args.top_invalidated:
            _print_counter("Top Invalidated Signatures", d.invalidated, top=args.top, min_count=args.min_count)
        if args.top_role_skips:
            _print_counter("Top Role Signature Skips", d.skipped_role_sig, top=args.top, min_count=args.min_count)
        if args.top_duplicate_exec:
            _print_counter("Top Repeated Executed Actions (task:sig)", d.repeated_exec_by_task_sig, top=args.top, min_count=args.min_count)
        if args.top_frontier_drops:
            _print_counter("Top Frontier Drop URLs", d.intake_drop_frontier, top=args.top, min_count=args.min_count)
        if args.top_intake_accept:
            _print_counter("Top Intake Accepted URLs", d.intake_accept_by_url, top=args.top, min_count=args.min_count)

        if args.planned_zero:
            print(f"\nPlanned 0 actions count: {d.planned_zero}")
        if args.failures:
            print(f"\nWorker failed tasks: {d.worker_failed}")
            print(f"Worker crashes: {d.worker_crashed}")

    report_out = report_buf.getvalue()
    if grep_pattern and args.grep_target in ("report", "both"):
        report_lines = grep_lines(report_out.splitlines(), grep_pattern, args.ignore_case, args.grep_regex)
        print("\nReport Grep Matches")
        print("-------------------")
        for ln in report_lines:
            print(highlight_line(ln, grep_pattern, args.ignore_case, args.grep_regex) if args.highlight else ln)
        if not report_lines:
            print("(none)")
    else:
        print(report_out, end="")

    # line-oriented filters / raw grep
    filtered = d.raw_lines
    if args.task:
        filtered = [ln for ln in filtered if args.task in ln]
    if args.url:
        filtered = [ln for ln in filtered if args.url in ln]
    if args.signature:
        filtered = [ln for ln in filtered if args.signature in ln]
    filtered = filter_lines(filtered, args.contains, args.regex, args.ignore_case)
    grep_raw_enabled = bool(grep_pattern and args.grep_target in ("raw", "both"))
    if grep_raw_enabled:
        filtered = grep_lines(filtered, grep_pattern, args.ignore_case, args.grep_regex)

    if any([args.task, args.url, args.signature, args.contains, args.regex, grep_raw_enabled]):
        if args.head > 0:
            filtered = filtered[: args.head]
        if args.tail > 0:
            filtered = filtered[-args.tail :]
        print("\nFiltered Lines")
        print("--------------")
        for ln in filtered:
            print(highlight_line(ln, grep_pattern, args.ignore_case, args.grep_regex) if (grep_pattern and args.highlight) else ln)
        if not filtered:
            print("(none)")


if __name__ == "__main__":
    main()