Add performance benchmark and soak validation

.gitignore

@@ -10,6 +10,8 @@ htmlcov/
 venv/
 .env
 .runtime/
+benchmark_results_*.json
+benchmark_summary_*.md
 
 node_modules/
 frontend/node_modules/

README.md

@@ -74,6 +74,7 @@ or PCAP artifacts.
 | Live Ring Buffer | Foundation step | Recent packet, flow, alert, and Expert summaries are held in per-category bounded memory with eviction/query metrics and Ops Snapshot visibility. | Stored and returned records are redacted summaries; raw packet payloads and credentials are excluded. |
 | Batch Persistence / Storage Backpressure | Foundation step | Redacted packet, alert, and flow records use bounded batches with queue health, write latency, retry, backlog, failure, and Ops Snapshot visibility. | Audit and report exports remain synchronous; this is not a distributed storage engine or the complete performance pipeline. |
 | WebSocket Event Aggregator | Foundation step | Realtime packet/alert batching, slow-client protection, WebSocket pressure metrics, and Ops Snapshot visibility are tested. | Realtime delivery batching only. |
+| Performance Benchmark Suite | Validated foundation | Synthetic stage benchmarks, a bounded full-pipeline soak mode, JSON/Markdown reports, and a CI-safe structural smoke test cover the current pipeline. | No live capture, external network, admin privileges, or production throughput claims; see [Performance Validation](docs/PERFORMANCE_VALIDATION.md). |
 | Demo and operational QA | Ready | Token-safe demo and Agent script behavior is tested. | Demo launchers and status commands do not print raw tokens. |
 | Windows release packaging | Validated path | Desktop smoke, version consistency, and release workflow checks run in CI. | Versioned artifacts include SHA256 checksums. |
 | Linux desktop packaging | Staged | Build workflow exists; native production validation remains pending. | Publish only after native smoke and release QA. |
@@ -253,6 +254,7 @@ and surfaced in Ops Snapshot metrics.
 
 - [Agent Mode](docs/AGENT_MODE.md)
 - [Performance Pipeline](docs/PERFORMANCE_PIPELINE.md)
+- [Performance Validation](docs/PERFORMANCE_VALIDATION.md)
 - [Flow Analysis And Protocol Intelligence](docs/FLOW_ANALYSIS.md)
 - [Deep Packet Inspection](docs/DEEP_PACKET_INSPECTION.md)
 - [Agent Operational QA Checklist](docs/AGENT_QA_CHECKLIST.md)

benchmarks/README.md

@@ -0,0 +1,54 @@
+# Performance Benchmarks
+
+This directory validates NetBotPro's bounded performance pipeline with local,
+synthetic packet summaries. The tools do not capture traffic, open network
+connections, require Administrator privileges, or store raw payloads.
+
+## Quick smoke test
+
+```powershell
+python benchmarks/soak_test_pipeline.py `
+  --duration-sec 10 `
+  --events-per-sec 200 `
+  --flows 20 `
+  --ci-safe `
+  --output .runtime/benchmarks/smoke
+```
+
+The command writes:
+
+- `benchmark_results.json`: structured configuration, resource, stage, and
+  operational-health metrics.
+- `benchmark_summary.md`: a compact human-readable interpretation.
+
+Both outputs pass through central redaction before they are written.
+
+## Stage benchmarks
+
+Run a focused benchmark with the same common CLI options:
+
+```powershell
+python benchmarks/benchmark_packet_pipeline.py --duration-sec 10 --events-per-sec 500
+python benchmarks/benchmark_worker_pool.py --duration-sec 10 --events-per-sec 500 --flows 50
+python benchmarks/benchmark_websocket_aggregator.py --duration-sec 10 --events-per-sec 500
+python benchmarks/benchmark_batch_persistence.py --duration-sec 10 --events-per-sec 500
+python benchmarks/benchmark_live_ring_buffer.py --duration-sec 10 --events-per-sec 500
+```
+
+Available options include `--duration-sec`, `--events-per-sec`, `--flows`,
+`--packet-rate`, `--alert-rate`, `--output`, `--json`, `--markdown`,
+`--profile`, `--no-live-capture`, and `--ci-safe`. The `ci-safe` profile caps
+duration and event rate. Live capture remains disabled regardless of profile.
+
+## Longer local soak
+
+The local profile defaults to five minutes when duration is omitted:
+
+```powershell
+python benchmarks/soak_test_pipeline.py --profile local --events-per-sec 1000 --flows 100
+```
+
+Results are machine-dependent. They validate bounded behavior, observable
+drops, ordering, report generation, and resource trends; they are not a
+production capacity guarantee. Use authorized workload traces separately when
+sizing a deployment.

benchmarks/__init__.py

@@ -0,0 +1 @@
+"""Safe synthetic performance validation tools for NetBotPro."""

benchmarks/benchmark_batch_persistence.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import argparse
+import sys
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from backend.app.services.batch_persistence import BatchPersistenceWriter
+from benchmarks.benchmark_report import (
+    BenchmarkConfig,
+    add_common_arguments,
+    config_from_args,
+    synthetic_alert,
+    synthetic_packet,
+)
+
+
+def run_persistence_benchmark(config: BenchmarkConfig) -> dict[str, Any]:
+    config = config.normalized()
+    event_count = max(1, min(int(config.duration_sec * config.events_per_sec), 50_000))
+    written = 0
+    batches = 0
+    lock = threading.Lock()
+
+    def write_batch(grouped: dict[str, list[dict[str, Any]]]) -> None:
+        nonlocal written, batches
+        with lock:
+            written += sum(len(rows) for rows in grouped.values())
+            batches += 1
+
+    writer = BatchPersistenceWriter(
+        write_batch,
+        enabled=True,
+        queue_max=max(100, min(event_count, 5000)),
+        overflow_policy="drop_oldest",
+        retry_max=1,
+        retry_backoff_ms=1,
+        batch_sizes={
+            "packet_record": 100,
+            "flow_record": 50,
+            "alert_record": 25,
+        },
+        flush_ms={
+            "packet_record": 20,
+            "flow_record": 20,
+            "alert_record": 20,
+        },
+    )
+    try:
+        for sequence in range(event_count):
+            packet = synthetic_packet(sequence, config.flows)
+            writer.enqueue("packet_record", packet, source="synthetic_benchmark")
+            if sequence % 10 == 0:
+                writer.enqueue(
+                    "flow_record",
+                    {"flow_id": packet["flow_key"], "packets": sequence + 1},
+                    source="synthetic_benchmark",
+                )
+            if (
+                config.alert_rate
+                and sequence % max(1, config.events_per_sec // config.alert_rate) == 0
+            ):
+                writer.enqueue(
+                    "alert_record",
+                    synthetic_alert(sequence, packet),
+                    source="synthetic_benchmark",
+                )
+        flushed = writer.flush(10.0)
+        deadline = time.monotonic() + 10.0
+        while time.monotonic() < deadline:
+            metrics = writer.metrics()
+            if metrics["queue_depth"] == 0:
+                break
+            time.sleep(0.01)
+        metrics = writer.metrics()
+    finally:
+        writer.close(10.0)
+    metrics.update(
+        {
+            "sink_events_written_total": written,
+            "sink_batches_total": batches,
+            "manual_flush_completed": flushed,
+            "bounded": metrics["queue_depth"] <= metrics["queue_max"],
+            "write_latency_ms_avg": metrics["write_latency_avg_ms"],
+            "write_latency_ms_p95": metrics["write_latency_p95_ms"],
+        }
+    )
+    return metrics
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Benchmark batch persistence.")
+    add_common_arguments(parser)
+    print(run_persistence_benchmark(config_from_args(parser.parse_args())))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/benchmark_live_ring_buffer.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import argparse
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from backend.app.services.live_ring_buffer import DEFAULT_CAPACITIES, LiveRingBuffer
+from benchmarks.benchmark_report import (
+    BenchmarkConfig,
+    add_common_arguments,
+    config_from_args,
+    percentile,
+    synthetic_packet,
+)
+
+
+def run_live_ring_benchmark(config: BenchmarkConfig) -> dict[str, Any]:
+    config = config.normalized()
+    event_count = max(1, min(int(config.duration_sec * config.events_per_sec), 50_000))
+    packet_capacity = max(10, min(event_count // 2 or 1, 5000))
+    capacities = {category: 10 for category in DEFAULT_CAPACITIES}
+    capacities["packet"] = packet_capacity
+    ring = LiveRingBuffer(
+        capacities=capacities,
+        default_query_limit=50,
+        max_query_limit=100,
+    )
+    append_latencies: list[float] = []
+    query_latencies: list[float] = []
+    started = time.perf_counter()
+    for sequence in range(event_count):
+        packet = synthetic_packet(sequence, config.flows)
+        item_started = time.perf_counter()
+        ring.append(
+            "packet",
+            packet,
+            flow_key=packet["flow_key"],
+            source="synthetic_benchmark",
+        )
+        append_latencies.append((time.perf_counter() - item_started) * 1000.0)
+        if sequence % 100 == 0:
+            query_started = time.perf_counter()
+            ring.query("packet", limit=50)
+            query_latencies.append((time.perf_counter() - query_started) * 1000.0)
+    ring.query("packet", limit=1000)
+    elapsed = max(time.perf_counter() - started, 0.000001)
+    metrics = ring.metrics()
+    metrics.update(
+        {
+            "benchmark_events_total": event_count,
+            "benchmark_throughput_events_sec": round(event_count / elapsed, 2),
+            "append_latency_ms_avg": round(
+                sum(append_latencies) / len(append_latencies), 3
+            ),
+            "append_latency_ms_p95": percentile(append_latencies, 0.95),
+            "query_latency_ms_avg": (
+                round(sum(query_latencies) / len(query_latencies), 3)
+                if query_latencies
+                else 0.0
+            ),
+            "query_latency_ms_p95": percentile(query_latencies, 0.95),
+            "bounded": metrics["categories"]["packet"]["records"]
+            <= metrics["categories"]["packet"]["capacity"],
+        }
+    )
+    return metrics
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Benchmark Live Ring Buffer.")
+    add_common_arguments(parser)
+    print(run_live_ring_benchmark(config_from_args(parser.parse_args())))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/benchmark_packet_pipeline.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from backend.app.services.packet_queue import BoundedPacketQueue
+from benchmarks.benchmark_report import (
+    BenchmarkConfig,
+    add_common_arguments,
+    config_from_args,
+    percentile,
+    synthetic_packet,
+)
+
+
+def run_packet_queue_benchmark(config: BenchmarkConfig) -> dict[str, Any]:
+    config = config.normalized()
+    event_count = max(1, min(int(config.duration_sec * config.packet_rate), 50_000))
+    queue_max = max(10, min(event_count // 2 or 1, 5000))
+    packet_queue = BoundedPacketQueue(max_size=queue_max, overflow_policy="drop_oldest")
+    latencies_ms: list[float] = []
+    started = time.perf_counter()
+    queue_logger = logging.getLogger("backend.app.services.packet_queue")
+    previous_disabled = queue_logger.disabled
+    queue_logger.disabled = True
+    try:
+        for sequence in range(event_count):
+            item_started = time.perf_counter()
+            packet_queue.put(synthetic_packet(sequence, config.flows))
+            latencies_ms.append((time.perf_counter() - item_started) * 1000.0)
+    finally:
+        queue_logger.disabled = previous_disabled
+    while not packet_queue.empty():
+        packet_queue.get(timeout=0.1)
+        packet_queue.task_done()
+    elapsed = max(time.perf_counter() - started, 0.000001)
+    metrics = packet_queue.stats(worker_alive=True)
+    metrics.update(
+        {
+            "benchmark_events_total": event_count,
+            "benchmark_throughput_events_sec": round(event_count / elapsed, 2),
+            "intake_latency_ms_avg": round(sum(latencies_ms) / len(latencies_ms), 3),
+            "intake_latency_ms_p95": percentile(latencies_ms, 0.95),
+            "intake_latency_ms_max": round(max(latencies_ms), 3),
+            "bounded": metrics["high_water_mark"] <= metrics["max_size"],
+        }
+    )
+    return metrics
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Benchmark bounded packet intake.")
+    add_common_arguments(parser)
+    config = config_from_args(parser.parse_args())
+    print(run_packet_queue_benchmark(config))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())