## Day 21 — Audit log в SQLite + replay (v0.4.0)

CHANGELOG.md

@@ -2,6 +2,25 @@
 
 All notable changes to CyberAI are documented here.
 
+## [0.4.0] - 2026-06-12
+
+### Accelerated & Observable — Week 3
+
+Week 3 turns the working pipeline into a fast, cost-aware and auditable one.
+
+### Added
+- Async pipeline: `AsyncOrchestrator`, async DNS / subdomain enum, batched
+  async CVE lookups with a sync-vs-async no-regression benchmark gate.
+- Cost tracking: `CostTracker` + `TokenUsage`, per-model pricing, CLI cost
+  summary, `BudgetExceeded` hard cap via `max_cost_usd`.
+- Anthropic prompt caching (`cache_control`) with cache-aware pricing.
+- Native LLM tool calling: Tool→OpenAI/Anthropic spec converters, `call_tools`
+  returning structured `LLMResponse`, provider-aware tool-result threading.
+- Structured outputs: `structured_call` (OpenAI `json_schema` / Anthropic
+  forced tool), Pydantic `ReportSection`, HackerOne-compatible export.
+- Observability: SQLite-backed audit log, full session export/import
+  (`to_json` / `from_json`), and `cyberai replay <session_id>`.
+
 ## [0.3.0] - 2026-06-02
 
 ### Hardening — Week 2 complete

cyberai/__main__.py

@@ -56,7 +56,13 @@ def scan(
     console.print("[yellow]→[/yellow] Starting pipeline...")
     session = orchestrator.run(target, authorized_scope=list(scope))
 
+    from cyberai.cli.replay import save_session
+
+    saved = save_session(session, config.output_dir)
     console.print(f"\n[green]✓[/green] Done. Findings: {len(session.findings)}")
+    console.print(
+        f"[dim]Session saved: {saved} (replay with: cyberai replay {session.session_id})[/dim]"
+    )
 
     from cyberai.core.cost_tracker import format_summary
 
@@ -67,6 +73,16 @@ def scan(
         console.print(f"  {key}: {value}")
 
 
+@cli.command()
+@click.argument("session_id")
+def replay(session_id: str) -> None:
+    """Reload SESSION_ID, re-run in dry-run mode and diff the phases."""
+    from cyberai.cli.replay import run_replay
+
+    config = CyberAIConfig.from_env()
+    raise SystemExit(run_replay(session_id, config))
+
+
 @cli.command()
 def status() -> None:
     """Show CyberAI status and config."""

cyberai/cli/replay.py

@@ -0,0 +1,96 @@
+"""Session replay (day 21): reload a saved ScanSession and re-run it.
+
+The saved session JSON (written by `cyberai scan`) is reloaded, the pipeline
+is re-run in dry-run mode against the same target, and the replayed phases
+are diffed against the originals. Replay is observability: same input ->
+same deterministic pipeline shape.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from rich.console import Console
+from rich.table import Table
+
+from cyberai.core.config import CyberAIConfig
+from cyberai.core.orchestrator import Orchestrator
+from cyberai.core.scan_session import ScanSession
+
+console = Console()
+
+
+def _session_path(output_dir: Path, session_id: str) -> Path:
+    return output_dir / f"session_{session_id}.json"
+
+
+def load_session(output_dir: Path, session_id: str) -> Optional[ScanSession]:
+    """Load a saved session by id; None if the file is missing."""
+    path = _session_path(output_dir, session_id)
+    if not path.exists():
+        return None
+    return ScanSession.from_json(path.read_text())
+
+
+def diff_phases(original: ScanSession, replayed: ScanSession) -> List[Dict[str, Any]]:
+    """Compare phase success between the original and replayed sessions."""
+    orig = {p.phase.value: p.success for p in original.phases}
+    new = {p.phase.value: p.success for p in replayed.phases}
+    rows: List[Dict[str, Any]] = []
+    for phase in sorted(set(orig) | set(new)):
+        o = orig.get(phase)
+        n = new.get(phase)
+        rows.append(
+            {
+                "phase": phase,
+                "original": o,
+                "replayed": n,
+                "match": o == n,
+            }
+        )
+    return rows
+
+
+def run_replay(session_id: str, config: Optional[CyberAIConfig] = None) -> int:
+    """Reload, re-run (dry-run) and diff a session. Returns process exit code."""
+    config = config or CyberAIConfig.from_env()
+    original = load_session(config.output_dir, session_id)
+    if original is None:
+        console.print(
+            f"[red]✗[/red] No saved session [bold]{session_id}[/bold] in {config.output_dir}"
+        )
+        return 1
+
+    console.print(
+        f"[yellow]→[/yellow] Replaying session [bold]{session_id}[/bold] "
+        f"(target: {original.target})"
+    )
+    orchestrator = Orchestrator(config=config, dry_run=True)
+    replayed = orchestrator.run(original.target, authorized_scope=list(original.authorized_scope))
+
+    rows = diff_phases(original, replayed)
+    table = Table(title=f"Replay diff — {session_id}", style="cyan")
+    table.add_column("Phase", style="bold")
+    table.add_column("Original", justify="center")
+    table.add_column("Replayed", justify="center")
+    table.add_column("Match", justify="center")
+    for r in rows:
+        mark = "[green]✓[/green]" if r["match"] else "[red]✗[/red]"
+        table.add_row(r["phase"], str(r["original"]), str(r["replayed"]), mark)
+    console.print(table)
+
+    all_match = all(r["match"] for r in rows)
+    if all_match:
+        console.print("[green]✓[/green] Replay deterministic — phases match.")
+        return 0
+    console.print("[red]✗[/red] Replay mismatch — pipeline not deterministic.")
+    return 2
+
+
+def save_session(session: ScanSession, output_dir: Path) -> Path:
+    """Persist a session as session_<id>.json for later replay."""
+    output_dir.mkdir(parents=True, exist_ok=True)
+    path = _session_path(output_dir, session.session_id)
+    path.write_text(session.to_json())
+    return path

cyberai/core/knowledge_base.py

@@ -60,6 +60,14 @@ def keys(self) -> List[str]:
     def snapshot(self) -> Dict[str, Any]:
         return {k: v.value for k, v in self._store.items()}
 
+    @classmethod
+    def from_snapshot(cls, data: Dict[str, Any]) -> "KnowledgeBase":
+        """Rebuild a KB from a snapshot() dict (agent/tags/ts not restored)."""
+        kb = cls()
+        for key, value in (data or {}).items():
+            kb.set(key, value, agent="replay")
+        return kb
+
     def history(self) -> List[Dict]:
         return [{"key": e.key, "agent": e.agent, "timestamp": e.timestamp} for e in self._history]
 

cyberai/core/logger.py

@@ -1,6 +1,7 @@
-from typing import Any
+from typing import Any, List, Dict, Optional
 import logging
 import json
+import sqlite3
 from datetime import datetime, timezone
 from pathlib import Path
 from rich.console import Console
@@ -47,19 +48,95 @@ def format(self, record: logging.LogRecord) -> str:
 
 
 class AuditLogger:
-    """Wrapper for structured pentest audit logging"""
+    """Wrapper for structured pentest audit logging.
 
-    def __init__(self, session_id: str, output_dir: str = "reports/"):
+    Always writes a JSONL trail. When `db_path` is given, every event is
+    also appended to an append-only SQLite `audit_events` table, enabling
+    queryable audit and session replay (day 21). db_path=None keeps the
+    legacy JSONL-only behaviour (no regression).
+    """
+
+    def __init__(
+        self,
+        session_id: str,
+        output_dir: str = "reports/",
+        db_path: Optional[str] = None,
+    ):
         log_path = f"{output_dir}/audit_{session_id}.jsonl"
         self.logger = get_logger(f"cyberai.audit.{session_id}", log_path)
         self.session_id = session_id
+        self.db_path = db_path
+        if db_path:
+            Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+            self._init_db()
+
+    def _init_db(self) -> None:
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS audit_events (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    session_id TEXT NOT NULL,
+                    agent TEXT NOT NULL,
+                    action TEXT NOT NULL,
+                    inputs_json TEXT,
+                    outputs_json TEXT,
+                    timestamp TEXT NOT NULL
+                )
+                """
+            )
+
+    def _db_append(
+        self,
+        agent: str,
+        action: str,
+        inputs: Any = None,
+        outputs: Any = None,
+    ) -> None:
+        if not self.db_path:
+            return
+        inputs_json = json.dumps(inputs, default=str) if inputs is not None else None
+        outputs_json = json.dumps(outputs, default=str) if outputs is not None else None
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute(
+                "INSERT INTO audit_events "
+                "(session_id, agent, action, inputs_json, outputs_json, timestamp) "
+                "VALUES (?, ?, ?, ?, ?, ?)",
+                (
+                    self.session_id,
+                    agent,
+                    action,
+                    inputs_json,
+                    outputs_json,
+                    datetime.now(timezone.utc).isoformat(),
+                ),
+            )
+
+    def read_events(self, session_id: Optional[str] = None) -> List[Dict[str, Any]]:
+        """Read audit events (all, or for one session) ordered by id.
+
+        Returns [] when no SQLite backend is configured.
+        """
+        if not self.db_path:
+            return []
+        sid = session_id or self.session_id
+        with sqlite3.connect(self.db_path) as conn:
+            conn.row_factory = sqlite3.Row
+            rows = conn.execute(
+                "SELECT * FROM audit_events WHERE session_id = ? ORDER BY id",
+                (sid,),
+            ).fetchall()
+        return [dict(r) for r in rows]
 
     def agent_action(self, agent: str, action: str, data: Any = None):
         extra = {"agent": agent, "data": data}
         self.logger.info(f"[{agent}] {action}", extra=extra)
+        self._db_append(agent, action, inputs=data)
 
     def finding(self, agent: str, title: str, severity: str):
         self.logger.warning(f"[FINDING][{severity}] {title}", extra={"agent": agent})
+        self._db_append(agent, "finding", outputs={"title": title, "severity": severity})
 
     def error(self, agent: str, msg: str):
         self.logger.error(f"[{agent}] {msg}", extra={"agent": agent})
+        self._db_append(agent, "error", outputs={"message": msg})

cyberai/core/scan_session.py

@@ -13,8 +13,9 @@
 
 from __future__ import annotations
 
+import json
 import uuid
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from enum import Enum
 from typing import Any, Dict, List, Optional
@@ -230,6 +231,53 @@ def summary(self) -> Dict[str, Any]:
             "kb_keys": list(self.kb.keys()),
         }
 
+    # ── full serialization for replay (day 21) ────────────────────────
+
+    def to_json(self, indent: int = 2) -> str:
+        """Full session export including KB values, findings and phases.
+
+        Non-JSON-native values fall back to str(). Restorable via from_json().
+        """
+        payload = {
+            "session_id": self.session_id,
+            "target": self.target,
+            "state": self.state.value,
+            "created_at": self.created_at,
+            "started_at": self.started_at,
+            "ended_at": self.ended_at,
+            "authorized_scope": list(self.authorized_scope),
+            "errors": list(self.errors),
+            "findings": [_finding_to_dict(f) for f in self.findings],
+            "phases": [_phase_to_dict(p) for p in self.phases],
+            "kb": self.kb.snapshot(),
+        }
+        return json.dumps(payload, indent=indent, default=str)
+
+    @classmethod
+    def from_json(cls, raw: str) -> "ScanSession":
+        """Rebuild a ScanSession from to_json() output.
+
+        Findings/phases are restored as dataclasses; KB values are restored
+        verbatim from the snapshot. Timestamps and ids are preserved.
+        """
+        data = json.loads(raw)
+        session = cls(
+            target=data["target"],
+            session_id=data.get("session_id", str(uuid.uuid4())[:8]),
+        )
+        session.state = ScanState(data.get("state", "created"))
+        session.created_at = data.get("created_at", session.created_at)
+        session.started_at = data.get("started_at")
+        session.ended_at = data.get("ended_at")
+        session.authorized_scope = list(data.get("authorized_scope", []))
+        session.errors = list(data.get("errors", []))
+        session.kb = KnowledgeBase.from_snapshot(data.get("kb", {}))
+        for fd in data.get("findings", []):
+            session.findings.append(_finding_from_dict(fd))
+        for pd in data.get("phases", []):
+            session.phases.append(_phase_from_dict(pd))
+        return session
+
     def __repr__(self) -> str:
         return (
             f"ScanSession(id={self.session_id}, "
@@ -261,3 +309,36 @@ def _phase_summary(p: PhaseResult) -> Dict[str, Any]:
         "duration_s": p.duration_s,
         "error": p.error,
     }
+
+
+# ── (de)serialization helpers for replay (day 21) ─────────────────────
+
+
+def _finding_to_dict(f: "Finding") -> Dict[str, Any]:
+    d = asdict(f)
+    if isinstance(d.get("severity"), Severity):
+        d["severity"] = f.severity.value
+    elif hasattr(f.severity, "value"):
+        d["severity"] = f.severity.value
+    return d
+
+
+def _finding_from_dict(d: Dict[str, Any]) -> "Finding":
+    data = dict(d)
+    sev = data.get("severity", "INFO")
+    data["severity"] = sev if isinstance(sev, Severity) else Severity(str(sev).upper())
+    return Finding(**data)
+
+
+def _phase_to_dict(p: "PhaseResult") -> Dict[str, Any]:
+    d = asdict(p)
+    if hasattr(p.phase, "value"):
+        d["phase"] = p.phase.value
+    return d
+
+
+def _phase_from_dict(d: Dict[str, Any]) -> "PhaseResult":
+    data = dict(d)
+    ph = data.get("phase")
+    data["phase"] = ph if isinstance(ph, ScanPhase) else ScanPhase(str(ph))
+    return PhaseResult(**data)

cyberai/version.py

@@ -1,3 +1,3 @@
-__version__ = "0.3.0"
+__version__ = "0.4.0"
 __author__ = "evkir"
 __description__ = "CyberAI — AI-native multi-agent pentest platform"