feat(reports): store and browse report artifacts

B001 change caused 'unsupported operand type(s) for -: list and set' during a
full consolidated scan: _db_complete_blocklist returned split_cidr_versions()
output, which is tuple[list, list], but write_blocklist_report()/blocklist_delta()
expect set operands. Return set(v4), set(v6) to match cumulative_blocklist's
contract. Reproduced the exact report path with a populated ledger and
verified it no longer crashes.

REPORTS.md

@@ -0,0 +1,142 @@
+# Report Inventory
+
+This inventory records the current report surface before moving report reads
+or writes into SQLite. Generated paths are rooted at `REPORT_DIR` (normally
+`reports/`). The SQLite tables listed below already exist in schema v1 unless
+noted otherwise.
+
+## Artifacts
+
+| Area | Producer and formats | Current role and consumers | Classification |
+| --- | --- | --- | --- |
+| `copyparty/` | `Analyzer.write_report()` creates timestamped `.txt`; `write_data()` creates per-IP CSV/JSON, denied-path CSV, flagged/banned IP TXT, `requests.csv`, `authed_users.json`, and `mikrotik-block.rsc`. | Reports API serves top-level text reports. `requests.csv` is imported by `app.db.backfill`; per-run data supports inspection and legacy workflows. Retention prunes generated files. | Request rows and per-scan IP stats are canonical in `requests` and `ip_stats`; rendered summaries and exports are derived. |
+| `npm/` | Same analyzer outputs, plus user-agent and host summary CSV files. | Same consumers as copyparty. `requests.csv` is a legacy backfill input; retention prunes generated files. | Request rows and per-scan IP stats are canonical in `requests` and `ip_stats`; summaries and exports are derived. |
+| `app/data/` | Consolidated scan writes `blocklist_ipv4.txt` and `blocklist_ipv6.txt`. | Blocklist UI/API reads these files; scan history reconstruction and firewall sync use them as current/compatibility inputs. Retention explicitly protects them. | Operational compatibility exports. Canonical blocked targets and events belong in `blocklist_entries` and `blocklist_events`; keep files until every consumer uses the DB and git-history contract is replaced. |
+| `app/` timestamped JSON | `write_blocklist_report()` emits scan/firewall change reports when the router is reachable. | Human-readable audit trail containing run delta, cumulative targets, router changes, dry-run state, and errors; retained under report retention. | Derived audit artifact. Preserve structured fields and source scan association in the DB before considering file retirement. |
+| `mikrotik/` | `MikrotikFirewall.save_state_report()` writes timestamped JSON snapshots and report data; latest state is also written to `data/mikrotik/*.json`. | Firewall UI/API displays current state and snapshot history. `app.db.backfill` currently imports only the latest `data/mikrotik` snapshot; runtime snapshot persistence uses `firewall_snapshots`, `firewall_address_list_entries`, and `firewall_rule_entries`. | Firewall rows and snapshots are canonical in the DB once persisted. Keep JSON exports during migration and retain any git-diff inputs required by deployment/audit. |
+
+## Consumers
+
+- `GET /api/reports/<source>` lists top-level files in the selected report
+  directory; `GET /api/reports/<source>/<path>` reads a file below it.
+- The reports page consumes those endpoints. Scan status and dashboard helpers
+  select the newest source and MikroTik report files for display.
+- `/api/blocklist` reads the consolidated IPv4/IPv6 text files. Scan history
+  reconstruction and firewall sync also consume those files alongside the
+  SQLite ledger and git history.
+- RouterOS `.rsc` exports and firewall synchronization consume generated
+  targets; these are operational outputs, not disposable presentation files.
+- `app.db.backfill` imports per-source `requests.csv`, blocklist history, and
+  the latest common MikroTik JSON snapshot. It does not currently import all
+  timestamped reports or all historical MikroTik snapshots.
+- Report retention removes old generated artifacts after consolidated scans,
+  subject to configured age/file-count limits; the consolidated blocklist
+  files are protected.
+
+## Restructuring Implications
+
+- Keep `requests`, `ip_stats`, blocklist tables, and firewall snapshot tables
+  as the queryable source of truth for facts they already represent. Avoid
+  copying those facts into a generic report blob.
+- A future report-artifact record should store metadata and structured fields
+  that have no canonical home yet. Rendered text/CSV/JSON should be generated
+  from those records or remain an explicit export.
+- Preserve source paths, generated timestamps, scan IDs, and hashes during
+  import so DB records can be traced back to original files.
+- Do not retire blocklist exports, RouterOS scripts, or source logs until
+  their consumers and audit requirements have explicit replacements.
+
+## Database and API Design
+
+### Canonical Data
+
+Continue treating normalized domain tables as canonical for facts they
+already represent:
+
+- `requests` for parsed request events.
+- `ip_stats` for per-IP/per-scan aggregates.
+- `scans` and `scans_sources` for run status, parameters, and source counts.
+- `blocklist_entries`, `blocklist_reasons`, and `blocklist_events` for block
+  decisions and history.
+- `firewall_snapshots`, `firewall_address_list_entries`, and
+  `firewall_rule_entries` for router state.
+
+Add a `report_artifacts` table only for report-specific summaries that cannot
+be reconstructed from those canonical rows. Proposed columns:
+
+| Column | Purpose |
+| --- | --- |
+| `id` | Integer primary key. |
+| `source` | `copyparty`, `npm`, `app`, or `mikrotik`. Keep as text because `log_sources` currently covers only request sources. |
+| `scan_id` | Nullable link to the generating scan; legacy imports may not have a known scan. |
+| `report_type` | Stable type such as `analyzer_summary`, `blocklist_change`, or `firewall_snapshot_summary`. |
+| `generated_at` | Original generation timestamp when known; otherwise import time plus an explicit unknown/original-time flag in metadata. |
+| `schema_version` | Version of the structured payload shape. |
+| `payload_json` | Structured report-specific summary, not a duplicate of requests or raw logs. |
+| `original_path` | Relative source path for provenance and legacy lookup. |
+| `content_sha256` | Hash of imported source content for integrity and idempotent import. |
+| `created_at` | Database import/creation timestamp. |
+
+Use a unique index on `(original_path, content_sha256)` for idempotent file
+imports. Keep payload schemas type-specific and versioned. Do not copy
+`requests.csv`, `ips.json`, firewall rows, or blocklists wholesale into
+`payload_json`; query their canonical tables and generate exports from those
+tables. If a rendered text report is retained as a database artifact, store
+its report-specific structured inputs and render text on demand rather than
+using opaque text as the only source of truth.
+
+### API Transition Contract
+
+- Add `GET /api/reports` with optional `source`, `type`, `since`, `until`,
+  `limit`, and opaque `cursor` filters. Return a stable list of artifact
+  metadata: `id`, `source`, `scan_id`, `type`, `generated_at`, and available
+  export formats. Enforce a bounded default and maximum page size.
+- Add `GET /api/report-artifacts/<artifact_id>` for the structured payload and
+  provenance metadata. Use a distinct path because `/api/reports/<source>`
+  already owns the one-segment route shape.
+- Add `GET /api/report-artifacts/<artifact_id>/export` for supported report types and formats
+  (initially JSON and, where useful, CSV/TXT). Generate exports from canonical
+  tables and versioned artifact payloads; do not accept filesystem paths from
+  a database-backed export request.
+- Preserve the current source/path routes as a read-only compatibility
+  fallback while files remain. The reports page should consume DB metadata
+  first, then offer legacy files when no imported artifact exists. Return
+  stable JSON errors for unknown IDs, unsupported formats, and unavailable
+  DB state.
+- Keep ordering deterministic (`generated_at DESC`, then `id DESC`), validate
+  filter values, and use cursor pagination for large histories. Do not expose
+  absolute server paths in list responses.
+
+### Sensitive and Large Content
+
+Do not include raw request lines or full request histories in report payloads;
+those belong in the dedicated log-storage design and require separate access
+and retention controls. Keep high-volume CSV exports generated on demand. Treat
+IP addresses, paths/query strings, user names, user agents, and firewall state
+as sensitive operational data: the existing web server must remain
+localhost-only, payloads should include only fields needed by the report, and
+new endpoints must not expose raw log data by default. Backups and exported
+files inherit the same access and retention requirements as the database.
+
+## Migration Implementation Status
+
+- Schema v2 adds `report_artifacts` with source, scan link, report type,
+  generation time, payload version, original relative path, and SHA-256 hash.
+- `python -m app.db.backfill --dry-run --reports-only` inventories supported
+  report files by source and format. The importer is idempotent, preserves
+  malformed JSON as text with a parse error, reports unreadable/oversized
+  files, and never modifies source files. It skips exports owned by canonical
+  tables (for example `requests.csv` and current blocklists).
+- New analyzer text summaries and consolidated blocklist change reports are
+  written to SQLite and their existing files. Firewall snapshots continue to
+  be persisted in the normalized firewall tables.
+- The Reports page lists database artifacts first, supports cursor pagination
+  and JSON/CSV/TXT downloads, and retains the existing file-backed routes as a
+  fallback for unmigrated files.
+- Validation covers idempotency, malformed/oversized inputs, source/format
+  counts, backup/restore, dual-write payloads, and API fallback. A workspace
+  dry-run found 76 files: 60 eligible, 16 canonical exports skipped, and none
+  unmappable at the time of the check.
+- No report files have been retired. Keep this gate closed until every listed
+  consumer has moved or has an explicit compatibility requirement, and a
+  production backup/restore and parity review has been approved.

app/analyzer.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import os
 import re
 from collections import Counter
@@ -336,6 +337,7 @@ class Analyzer:
 
         self.write_data()
         report_path = self.write_report()
+        self.persist_report_artifact(report_path, db_scan_id)
         print(f"data files:     {self.data_dir}")
         print(f"summary report: {report_path}")
 
@@ -349,3 +351,28 @@ class Analyzer:
             result.update(rsc=rsc, blocked=n_blocked,
                           whitelisted=n_whitelisted)
         return result
+
+    def persist_report_artifact(self, path: str,
+                                scan_id: int | None = None) -> None:
+        """Store the rendered summary in SQLite while keeping its file export."""
+        try:
+            from .db import get_connection
+            from .db import repositories as repo
+            with open(path, "rb") as file:
+                content = file.read()
+            stat = os.stat(path)
+            relative = os.path.relpath(path, self.report_dir).replace(os.sep, "/")
+            repo.insert_report_artifact(
+                get_connection(), source=self.name,
+                report_type="analyzer_summary",
+                generated_at=datetime.fromtimestamp(
+                    stat.st_mtime).astimezone().isoformat(timespec="seconds"),
+                payload={"format": "text",
+                         "content": content.decode("utf-8", errors="replace")},
+                original_path=f"{self.name}/{relative}",
+                content_sha256=hashlib.sha256(content).hexdigest(),
+                scan_id=scan_id,
+            )
+        except Exception as exc:
+            if not self.quiet:
+                print(f"WARN: report artifact DB write skipped ({exc})")

app/db/backfill.py

@@ -19,19 +19,30 @@ from __future__ import annotations
 
 import argparse
 import csv
+import hashlib
 import json
 import os
+import re
 import sys
 from datetime import datetime, timezone
 
 from ..utils.constants import (
+    APP_REPORT_DIR,
     APP_REPORT_DATA_DIR,
     DATA_DIR,
     MIKROTIK_DATA_DIR,
+    MIKROTIK_REPORT_DIR,
     NPM_REPORT_DIR,
     COPYPARTY_REPORT_DIR,
 )
 
+MAX_REPORT_ARTIFACT_BYTES = 10 * 1024 * 1024
+CANONICAL_REPORT_EXPORTS = {
+    "requests.csv", "ips.json", "ips_ipv4.csv", "ips_ipv6.csv",
+    "blocked_ipv4.txt", "blocked_ipv6.txt", "address_list_ipv4.json",
+    "address_list_ipv6.json", "firewall_ipv4.json", "firewall_ipv6.json",
+}
+
 
 def utcnow() -> str:
     return datetime.now(timezone.utc).isoformat(timespec="seconds")
@@ -92,7 +103,7 @@ def backfill_firewall(conn, dry_run: bool = False) -> dict:
         "SELECT COUNT(*) FROM firewall_snapshots").fetchone()[0]
     if n:
         return {"firewall": {"dry_run": dry_run, "snapshot": "exists",
-                              "skipped": True}}
+                             "skipped": True}}
     if dry_run:
         return {"firewall": {"dry_run": True, "snapshot": None}}
     sid = repo.save_firewall_snapshot(
@@ -161,6 +172,152 @@ def backfill_requests(conn, dry_run: bool = False) -> dict:
     return {"requests": {"imported": total, "dry_run": dry_run}}
 
 
+def backfill_report_artifacts(conn, dry_run: bool = False,
+                              roots: dict[str, str] | None = None) -> dict:
+    """Import supported report files as versioned, hash-addressed artifacts.
+
+    Canonical event exports and current firewall/blocklist data are skipped;
+    those belong in their normalized tables. Original files are never changed
+    or removed. Oversized or unreadable files are reported for manual review.
+    """
+    from . import repositories as repo
+
+    roots = roots or {
+        "copyparty": COPYPARTY_REPORT_DIR,
+        "npm": NPM_REPORT_DIR,
+        "app": APP_REPORT_DIR,
+        "mikrotik": MIKROTIK_REPORT_DIR,
+    }
+    result = {
+        "discovered": 0, "imported": 0, "would_import": 0,
+        "already_imported": 0, "skipped_canonical": [], "unmapped": [],
+        "by_source": {}, "by_format": {}, "dry_run": dry_run,
+    }
+
+    def count(bucket: str, key: str, field: str) -> None:
+        stats = result[bucket].setdefault(key, {})
+        stats[field] = stats.get(field, 0) + 1
+
+    for source, root in roots.items():
+        if not os.path.isdir(root):
+            continue
+        for directory, _, filenames in os.walk(root):
+            for filename in sorted(filenames):
+                path = os.path.join(directory, filename)
+                suffix = os.path.splitext(filename)[1].lower()
+                if suffix not in (".txt", ".json", ".csv", ".rsc"):
+                    continue
+                relative = os.path.relpath(path, root).replace(os.sep, "/")
+                result["discovered"] += 1
+                report_format = suffix.lstrip(".")
+                count("by_source", source, "discovered")
+                count("by_format", report_format, "discovered")
+                if filename.lower() in CANONICAL_REPORT_EXPORTS:
+                    result["skipped_canonical"].append(
+                        {"source": source, "path": relative})
+                    count("by_source", source, "skipped_canonical")
+                    count("by_format", report_format, "skipped_canonical")
+                    continue
+
+                try:
+                    size = os.path.getsize(path)
+                    if size > MAX_REPORT_ARTIFACT_BYTES:
+                        raise ValueError(
+                            f"file exceeds {MAX_REPORT_ARTIFACT_BYTES} bytes")
+                    with open(path, "rb") as file:
+                        raw = file.read()
+                    text = raw.decode("utf-8", errors="replace")
+                    payload = _report_payload(suffix, text)
+                    digest = hashlib.sha256(raw).hexdigest()
+                    modified = datetime.fromtimestamp(
+                        os.path.getmtime(path), timezone.utc)
+                    generated_at = _report_generated_at(filename, modified)
+                except (OSError, ValueError) as exc:
+                    result["unmapped"].append(
+                        {"source": source, "path": relative,
+                         "reason": str(exc)})
+                    count("by_source", source, "unmapped")
+                    count("by_format", report_format, "unmapped")
+                    continue
+
+                original_path = f"{source}/{relative}"
+                exists = conn.execute(
+                    "SELECT 1 FROM report_artifacts "
+                    "WHERE original_path=? AND content_sha256=?",
+                    (original_path, digest),
+                ).fetchone()
+                if exists:
+                    result["already_imported"] += 1
+                    count("by_source", source, "already_imported")
+                    count("by_format", report_format, "already_imported")
+                    continue
+                if dry_run:
+                    result["would_import"] += 1
+                    count("by_source", source, "would_import")
+                    count("by_format", report_format, "would_import")
+                    continue
+
+                try:
+                    repo.insert_report_artifact(
+                        conn, source=source,
+                        report_type=_report_type(suffix),
+                        generated_at=generated_at, payload=payload,
+                        original_path=original_path,
+                        content_sha256=digest,
+                    )
+                    result["imported"] += 1
+                    count("by_source", source, "imported")
+                    count("by_format", report_format, "imported")
+                except (OSError, ValueError, TypeError) as exc:
+                    result["unmapped"].append(
+                        {"source": source, "path": relative,
+                         "reason": str(exc)})
+                    count("by_source", source, "unmapped")
+                    count("by_format", report_format, "unmapped")
+    return {"reports": result}
+
+
+def _report_payload(suffix: str, text: str) -> dict:
+    if suffix == ".json":
+        try:
+            return {"format": "json", "data": json.loads(text)}
+        except ValueError as exc:
+            return {"format": "json", "content": text,
+                    "parse_error": str(exc)}
+    if suffix == ".csv":
+        try:
+            reader = csv.DictReader(text.splitlines())
+            return {"format": "csv", "columns": reader.fieldnames or [],
+                    "rows": list(reader)}
+        except csv.Error as exc:
+            return {"format": "csv", "content": text,
+                    "parse_error": str(exc)}
+    return {"format": "text", "content": text}
+
+
+def _report_type(suffix: str) -> str:
+    return {".json": "json_report", ".csv": "csv_report",
+            ".rsc": "routeros_script"}.get(suffix, "text_report")
+
+
+def _report_generated_at(filename: str, modified: datetime) -> str:
+    match = re.search(
+        r"(?P<date>\d{4}-\d{2}-\d{2})(?:[-_T](?P<hour>\d{2})[-_:]"
+        r"(?P<minute>\d{2})[-_:](?P<second>\d{2}))?", filename)
+    if not match:
+        return modified.isoformat(timespec="seconds")
+    values = match.groupdict()
+    hour = int(values["hour"] or 0)
+    minute = int(values["minute"] or 0)
+    second = int(values["second"] or 0)
+    try:
+        parsed = datetime.strptime(values["date"], "%Y-%m-%d").replace(
+            hour=hour, minute=minute, second=second, tzinfo=timezone.utc)
+    except ValueError:
+        return modified.isoformat(timespec="seconds")
+    return parsed.isoformat(timespec="seconds")
+
+
 def _int_or_none(v):
     try:
         if v in (None, "", "-"):
@@ -181,14 +338,22 @@ def main(argv: list[str] | None = None) -> None:
                                 description="Backfill legacy runtime data into SQLite.")
     p.add_argument("--dry-run", action="store_true",
                    help="report what would be imported without writing")
+    p.add_argument("--reports-only", action="store_true",
+                   help="import legacy report artifacts only")
     opts = p.parse_args(argv)
 
     from . import get_connection
     conn = get_connection()
     print(f"== backfill (dry_run={opts.dry_run}) ==")
+    if opts.reports_only:
+        print("reports :", backfill_report_artifacts(
+            conn, dry_run=opts.dry_run))
+        print("== done ==")
+        return
     print("blocklist:", backfill_blocklist(conn, dry_run=opts.dry_run))
     print("firewall :", backfill_firewall(conn, dry_run=opts.dry_run))
     print("requests :", backfill_requests(conn, dry_run=opts.dry_run))
+    print("reports  :", backfill_report_artifacts(conn, dry_run=opts.dry_run))
     print("== done ==")
 
 

app/db/repositories.py

@@ -10,8 +10,10 @@ from __future__ import annotations
 
 import json
 import sqlite3
+import base64
 from contextlib import contextmanager
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
+import ipaddress
 from typing import Any, Iterator, Sequence
 
 from ..utils.models import IPStats
@@ -104,7 +106,8 @@ def _parse_ts(ts: str) -> tuple[str, float | None]:
     cleaned = ts.strip()
     # Accept a trailing timezone like +HH:MM or Z; fall back to naive parse.
     try:
-        epoch = datetime.fromisoformat(cleaned.replace("Z", "+00:00")).timestamp()
+        epoch = datetime.fromisoformat(
+            cleaned.replace("Z", "+00:00")).timestamp()
     except ValueError:
         try:
             epoch = datetime.fromisoformat(cleaned).timestamp()
@@ -231,6 +234,109 @@ def count_scans(conn: sqlite3.Connection) -> int:
     return int(conn.execute("SELECT COUNT(*) FROM scans").fetchone()[0])
 
 
+# ---------------------------------------------------------------------------
+# Report artifacts
+# ---------------------------------------------------------------------------
+
+def insert_report_artifact(conn: sqlite3.Connection, *, source: str,
+                           report_type: str, generated_at: str,
+                           payload: Any, original_path: str,
+                           content_sha256: str, scan_id: int | None = None,
+                           schema_version: int = 1) -> int:
+    """Insert a structured report artifact, idempotent by path and hash."""
+    with transaction(conn):
+        conn.execute(
+            "INSERT OR IGNORE INTO report_artifacts "
+            "(source, scan_id, report_type, generated_at, schema_version, "
+            "payload_json, original_path, content_sha256, created_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (source, scan_id, report_type, generated_at, schema_version,
+             json.dumps(payload, ensure_ascii=False), original_path,
+             content_sha256, utcnow()),
+        )
+        row = conn.execute(
+            "SELECT id FROM report_artifacts "
+            "WHERE original_path=? AND content_sha256=?",
+            (original_path, content_sha256),
+        ).fetchone()
+    return int(row["id"])
+
+
+def list_report_artifacts(conn: sqlite3.Connection, *,
+                          source: str | None = None,
+                          report_type: str | None = None,
+                          since: str | None = None,
+                          until: str | None = None,
+                          limit: int = 50,
+                          cursor: str | None = None) -> dict[str, Any]:
+    """List report metadata using bounded keyset pagination."""
+    limit = min(max(int(limit), 1), 200)
+    clauses: list[str] = []
+    args: list[Any] = []
+    if source:
+        clauses.append("source=?")
+        args.append(source)
+    if report_type:
+        clauses.append("report_type=?")
+        args.append(report_type)
+    if since:
+        clauses.append("generated_at>=?")
+        args.append(since)
+    if until:
+        clauses.append("generated_at<=?")
+        args.append(until)
+    if cursor:
+        generated_at, artifact_id = _decode_report_cursor(cursor)
+        clauses.append("(generated_at<? OR (generated_at=? AND id<?))")
+        args.extend((generated_at, generated_at, artifact_id))
+
+    sql = (
+        "SELECT id, source, scan_id, report_type, generated_at, "
+        "schema_version, original_path, content_sha256 "
+        "FROM report_artifacts"
+    )
+    if clauses:
+        sql += " WHERE " + " AND ".join(clauses)
+    sql += " ORDER BY generated_at DESC, id DESC LIMIT ?"
+    rows = [dict(row) for row in conn.execute(sql, (*args, limit + 1))]
+    has_more = len(rows) > limit
+    rows = rows[:limit]
+    next_cursor = (
+        _encode_report_cursor(rows[-1]["generated_at"], rows[-1]["id"])
+        if has_more and rows else None
+    )
+    return {"items": rows, "next_cursor": next_cursor}
+
+
+def get_report_artifact(conn: sqlite3.Connection,
+                        artifact_id: int) -> dict[str, Any] | None:
+    """Fetch one artifact and decode its structured payload."""
+    row = conn.execute(
+        "SELECT id, source, scan_id, report_type, generated_at, "
+        "schema_version, payload_json, original_path, content_sha256, created_at "
+        "FROM report_artifacts WHERE id=?", (artifact_id,)).fetchone()
+    if row is None:
+        return None
+    artifact = dict(row)
+    artifact["payload"] = json.loads(artifact.pop("payload_json"))
+    return artifact
+
+
+def _encode_report_cursor(generated_at: str, artifact_id: int) -> str:
+    raw = f"{generated_at}\n{artifact_id}".encode("utf-8")
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+
+def _decode_report_cursor(cursor: str) -> tuple[str, int]:
+    try:
+        padded = cursor + "=" * (-len(cursor) % 4)
+        generated_at, artifact_id = base64.urlsafe_b64decode(
+            padded.encode("ascii")).decode("utf-8").split("\n", 1)
+        return generated_at, int(artifact_id)
+    except (ValueError, UnicodeError) as exc:
+        raise ValueError("invalid report cursor") from exc
+
+
 def correlate_across_sources(conn: sqlite3.Connection,
                              limit: int = 100) -> list[dict[str, Any]]:
     """IPs seen in more than one log source, with per-source counts."""
@@ -339,7 +445,7 @@ def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
     with transaction(conn):
         for cidr in to_add:
             _block_entry_inner(conn, cidr, added_by=added_by, last_seen=now,
-                              reasons=reasons)
+                               reasons=reasons)
         # Refresh last_seen on entries that are still desired so the ledger
         # reflects recency for existing rows too.
         conn.executemany(
@@ -348,6 +454,52 @@ def record_blocklist(conn: sqlite3.Connection, cidrs: Sequence[str], *,
     return {"added": to_add, "active": len(before) + len(to_add)}
 
 
+def prune_old_single_ip_blocks(conn: sqlite3.Connection, *,
+                               older_than_days: int = 30,
+                               actor: str | None = "retention") -> list[str]:
+    """Remove active single-host entries older than ``older_than_days``.
+
+    Keeps CIDR ranges/subnets intact. Host entries are detected via
+    ``ipaddress.ip_network(..., strict=False)`` and only removed when the
+    network is exactly one address (/32 for IPv4, /128 for IPv6).
+    """
+    cutoff = datetime.now(timezone.utc) - \
+        timedelta(days=max(0, older_than_days))
+    rows = conn.execute(
+        "SELECT cidr, added_at, last_seen FROM blocklist_entries "
+        "WHERE status='active'").fetchall()
+
+    to_remove: list[str] = []
+    for row in rows:
+        cidr = str(row["cidr"])
+        if not _is_single_host_cidr(cidr):
+            continue
+        seen = _parse_iso_ts(
+            row["last_seen"]) or _parse_iso_ts(row["added_at"])
+        if seen is None:
+            continue
+        if seen < cutoff:
+            to_remove.append(cidr)
+
+    if not to_remove:
+        return []
+
+    now = utcnow()
+    with transaction(conn):
+        conn.executemany(
+            "UPDATE blocklist_entries SET status='removed', removed_at=? "
+            "WHERE cidr=? AND status='active'",
+            [(now, cidr) for cidr in to_remove],
+        )
+        conn.executemany(
+            "INSERT INTO blocklist_events (action, cidr, actor, fired_at, detail) "
+            "VALUES ('remove', ?, ?, ?, ?)",
+            [(cidr, actor, now, f"single-ip-retention>{older_than_days}d")
+             for cidr in to_remove],
+        )
+    return sorted(to_remove)
+
+
 def sync_blocklist(conn: sqlite3.Connection, desired_cidrs: Sequence[str], *,
                    added_by: str | None = None) -> dict[str, Any]:
     """Reconcile active blocklist to ``desired_cidrs``.
@@ -402,6 +554,26 @@ def _cidr_version(cidr: str) -> int:
         return 4 if ":" not in str(cidr) else 6
 
 
+def _is_single_host_cidr(cidr: str) -> bool:
+    try:
+        net = ipaddress.ip_network(cidr, strict=False)
+        return net.num_addresses == 1
+    except ValueError:
+        return False
+
+
+def _parse_iso_ts(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
 # ---------------------------------------------------------------------------
 # Whitelists
 # ---------------------------------------------------------------------------

app/db/schema.py

@@ -14,7 +14,7 @@ import sqlite3
 from typing import Iterable
 
 #: Latest schema version. Bump by one whenever a new migration is appended.
-SCHEMA_VERSION = 1
+SCHEMA_VERSION = 2
 
 #: Ordered migrations: index -> list of DDL statements. A migration N upgrades a
 #: database from user_version N-1 to N. Never edit an older migration in place;
@@ -31,14 +31,29 @@ MIGRATIONS: dict[int, list[str]] = {
             log_path TEXT,
             enabled INTEGER NOT NULL DEFAULT 1
         )
-        """,
         """
-        INSERT OR IGNORE INTO log_sources (name, log_path)
+    ],
-        VALUES ('copyparty', NULL), ('npm', NULL)
+
+    2: [
+        """
+        CREATE TABLE IF NOT EXISTS report_artifacts (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            source TEXT NOT NULL,
+            scan_id INTEGER,
+            report_type TEXT NOT NULL,
+            generated_at TEXT NOT NULL,
+            schema_version INTEGER NOT NULL DEFAULT 1,
+            payload_json TEXT NOT NULL,
+            original_path TEXT NOT NULL,
+            content_sha256 TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            UNIQUE (original_path, content_sha256)
+        )
         """,
-        # ------------------------------------------------------------------
+        "CREATE INDEX IF NOT EXISTS idx_report_artifacts_source_time "
-        # requests — one row per parsed request; insert-only
+        "ON report_artifacts (source, generated_at DESC, id DESC)",
-        # ------------------------------------------------------------------
+        "CREATE INDEX IF NOT EXISTS idx_report_artifacts_type_time "
+        "ON report_artifacts (report_type, generated_at DESC, id DESC)",
         """
         CREATE TABLE IF NOT EXISTS requests (
             id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -261,7 +276,8 @@ def apply_migrations(conn: sqlite3.Connection) -> None:
     for version in range(current + 1, SCHEMA_VERSION + 1):
         statements = MIGRATIONS.get(version)
         if statements is None:
-            raise RuntimeError(f"missing migration for schema version {version}")
+            raise RuntimeError(
+                f"missing migration for schema version {version}")
         with conn:
             for ddl in statements:
                 conn.execute(ddl)

app/scan/__init__.py

@@ -41,7 +41,8 @@ from ..utils.whois import consolidate_by_subnet
 
 
 def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
-                          desired_v6: list[str], added_by: str | None = None) -> None:
+                          desired_v6: list[str],
+                          added_by: str | None = None) -> dict | None:
     """Best-effort grow-only sync of block targets into the SQLite ledger.
 
     Uses ``record_blocklist`` (add-only) so the ledger is the authoritative
@@ -53,8 +54,13 @@ def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
         from ..db import get_connection
         from ..db import repositories as repo
         conn = get_connection()
-        return repo.record_blocklist(
+        result = repo.record_blocklist(
             conn, list(desired_v4) + list(desired_v6), added_by=added_by)
+        # Retention policy: expire stale single-host entries after 30 days
+        # while keeping CIDR ranges/subnets intact.
+        pruned = repo.prune_old_single_ip_blocks(conn, older_than_days=30)
+        result["pruned_single_ips"] = pruned
+        return result
     except Exception as exc:  # never break the scan for a DB issue
         print(f"WARN: could not sync blocklist to db ({exc})")
         return None
@@ -63,10 +69,11 @@ def _sync_blocklist_to_db(connected: bool, desired_v4: list[str],
 def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]):
     """Complete cumulative blocklist from the DB ledger (B001).
 
-    Returns ``(complete_v4, complete_v6)`` derived from the active blocklist
+    Returns ``(complete_v4, complete_v6)`` as **sets** (same contract as
-    ledger plus this scan's newly desired targets (in case the DB write for
+    ``cumulative_blocklist``) derived from the active blocklist ledger plus
-    this scan has not run yet). Falls back to the git-history reconstruction
+    this scan's newly desired targets (in case the DB write for this scan has
-    when the DB is unavailable, so nothing regresses.
+    not run yet). Falls back to the git-history reconstruction when the DB is
+    unavailable, so nothing regresses.
     """
     try:
         from ..db import get_connection
@@ -77,7 +84,7 @@ def _db_complete_blocklist(desired_v4: list[str], desired_v6: list[str]):
         active = []
     union = set(active) | set(desired_v4) | set(desired_v6)
     v4, v6 = split_cidr_versions(union)
-    return v4, v6
+    return set(v4), set(v6)
 
 
 @dataclass
@@ -296,7 +303,8 @@ def run_scan(params: ScanParams) -> ScanResult:
             complete_v4=complete_v4, complete_v6=complete_v6,
             firewall_changes=firewall_changes,
             dry_run=params.dry_run,
-            errors=errors)
+            errors=errors,
+            scan_id=db_scan_id)
         print(f"blocklist report: {blocklist_report_path}")
 
     # --- Step 8: prune old reports (retention) ------------------------------

app/templates/reports.html

@@ -1,5 +1,4 @@
-{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{%
+{% extends "base.html" %} {% block title %}Reports — Intrusion Detection{% endblock %} {% block content %}
-endblock %} {% block content %}
 <h1>Reports</h1>
 <p class="muted">Browse the timestamped reports written by each scanner.</p>
 
@@ -21,38 +20,87 @@ endblock %} {% block content %}
 
 <div class="panel">
   <div class="panel-head"><h2 id="content-title">Report content</h2></div>
-  <pre class="content" id="content">
+  <pre class="content" id="content">Select a report on the left to view it.</pre>
-Select a report on the left to view it.</pre
-  >
 </div>
 {% endblock %} {% block scripts %}
 <script>
-  async function loadFiles() {
+  let nextCursor = null;
+
+  function addReportLink(list, label, onClick) {
+    const link = document.createElement("a");
+    link.href = "#";
+    link.textContent = label;
+    link.addEventListener("click", (event) => {
+      event.preventDefault();
+      onClick();
+    });
+    list.append(link);
+  }
+
+  async function loadFiles(append = false) {
     const source = document.getElementById("source").value;
     const list = document.getElementById("files");
-    list.innerHTML = '<div class="muted">Loading…</div>';
+    if (!append) {
+      nextCursor = null;
+      list.innerHTML = '<div class="muted">Loading…</div>';
+    } else {
+      document.getElementById("load-more")?.remove();
+    }
+
+    let importedPaths = new Set();
     try {
-      const data = await api(`/api/reports/${source}`);
+      const query = new URLSearchParams({ source, limit: "50" });
-      list.innerHTML = "";
+      if (append && nextCursor) query.set("cursor", nextCursor);
-      const files = data.files || [];
+      const data = await api(`/api/reports?${query}`);
-      if (!files.length) {
+      if (!append) list.innerHTML = "";
-        list.innerHTML = '<div class="muted">No reports in this source.</div>';
+      for (const artifact of data.items || []) {
-        return;
+        importedPaths.add(artifact.original_path);
-      }
+        addReportLink(list, `${artifact.generated_at} | ${artifact.report_type} | ${artifact.original_path}`, () =>
-      for (const name of files) {
+          openArtifact(artifact.id),
-        list.append(
-          el("a", {
-            href: "#",
-            text: name,
-            onclick: (ev) => {
-              ev.preventDefault();
-              openReport(name);
-            },
-          }),
         );
       }
+      nextCursor = data.next_cursor;
     } catch (e) {
-      list.innerHTML = `<div class="muted">${escapeHtml(e.message)}</div>`;
+      if (append) {
+        list.append(el("div", { className: "muted", text: e.message }));
+        return;
+      }
+    }
+
+    if (!append) {
+      try {
+        const data = await api(`/api/reports/${source}`);
+        for (const name of data.files || []) {
+          if (importedPaths.has(`${source}/${name}`)) continue;
+          addReportLink(list, `${name} | legacy file`, () => openReport(name));
+        }
+      } catch (e) {
+        if (!list.childElementCount) list.append(el("div", { className: "muted", text: e.message }));
+      }
+    }
+
+    if (nextCursor) {
+      const button = document.createElement("button");
+      button.id = "load-more";
+      button.className = "btn";
+      button.textContent = "Load more";
+      button.addEventListener("click", () => loadFiles(true));
+      list.append(button);
+    }
+    if (!list.childElementCount) list.innerHTML = '<div class="muted">No reports in this source.</div>';
+  }
+
+  async function openArtifact(id) {
+    document.getElementById("content-title").textContent = "Database report";
+    const pre = document.getElementById("content");
+    pre.textContent = "Loading…";
+    try {
+      const data = await api(`/api/report-artifacts/${id}`);
+      document.getElementById("content-title").textContent = data.original_path;
+      const payload = data.payload || {};
+      pre.textContent = payload.format === "text" ? payload.content || "" : JSON.stringify(payload, null, 2);
+    } catch (e) {
+      pre.textContent = `Error: ${e.message}`;
     }
   }
 
@@ -69,8 +117,8 @@ Select a report on the left to view it.</pre
     }
   }
 
-  document.getElementById("refresh-btn").addEventListener("click", loadFiles);
+  document.getElementById("refresh-btn").addEventListener("click", () => loadFiles());
-  document.getElementById("source").addEventListener("change", loadFiles);
+  document.getElementById("source").addEventListener("change", () => loadFiles());
   loadFiles();
 </script>
 {% endblock %}

app/utils/blocklist_history.py

@@ -20,6 +20,7 @@ import re
 import subprocess
 from datetime import datetime
 from typing import Any, Iterable
+import hashlib
 
 from .files import write_json
 from .reporting import timestamp_name
@@ -151,6 +152,7 @@ def write_blocklist_report(report_dir: str,
                            firewall_changes: dict | None = None,
                            dry_run: bool = False,
                            errors: Iterable[str] = (),
+                           scan_id: int | None = None,
                            ) -> str:
     """Write the timestamped blocklist report to ``<report_dir>/app/``.
 
@@ -183,4 +185,18 @@ def write_blocklist_report(report_dir: str,
     os.makedirs(app_dir, exist_ok=True)
     path = os.path.join(app_dir, f"{timestamp_name()}.json")
     write_json(path, payload)
+    try:
+        from ..db import get_connection
+        from ..db import repositories as repo
+        with open(path, "rb") as file:
+            content = file.read()
+        repo.insert_report_artifact(
+            get_connection(), source="app", report_type="blocklist_change",
+            generated_at=payload["generated_at"], payload=payload,
+            original_path=f"app/{os.path.basename(path)}",
+            content_sha256=hashlib.sha256(content).hexdigest(),
+            scan_id=scan_id,
+        )
+    except Exception as exc:
+        print(f"WARN: blocklist report DB write skipped ({exc})")
     return path

app/web/routes.py

@@ -3,9 +3,13 @@
 
 from __future__ import annotations
 
+import csv
+import io
+import json
 import os
 
-from flask import Flask, abort, jsonify, render_template, request
+from flask import Flask, Response, abort, jsonify, render_template, request
+from werkzeug.utils import secure_filename
 
 from ..utils.constants import (
     APP_REPORT_DATA_DIR,
@@ -154,6 +158,84 @@ def register_routes(app: Flask,
 
     # --- reports -----------------------------------------------------------
 
+    @app.get("/api/reports")
+    def list_report_artifacts():
+        source = (request.args.get("source") or "").strip() or None
+        if source is not None and source not in report_sources:
+            return jsonify({"error": "unknown report source"}), 400
+        limit = request.args.get("limit", default=50, type=int)
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            result = repo.list_report_artifacts(
+                get_connection(), source=source,
+                report_type=(request.args.get("type") or "").strip() or None,
+                since=(request.args.get("since") or "").strip() or None,
+                until=(request.args.get("until") or "").strip() or None,
+                limit=limit,
+                cursor=(request.args.get("cursor") or "").strip() or None,
+            )
+            return jsonify(result)
+        except ValueError as exc:
+            return jsonify({"error": str(exc)}), 400
+        except Exception as exc:
+            return jsonify({"error": f"report database unavailable: {exc}"}), 503
+
+    @app.get("/api/report-artifacts/<int:artifact_id>")
+    def get_report_artifact(artifact_id: int):
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            artifact = repo.get_report_artifact(get_connection(), artifact_id)
+        except Exception as exc:
+            return jsonify({"error": f"report database unavailable: {exc}"}), 503
+        if artifact is None:
+            abort(404, "report artifact not found")
+        return jsonify(artifact)
+
+    @app.get("/api/report-artifacts/<int:artifact_id>/export")
+    def export_report_artifact(artifact_id: int):
+        try:
+            from ..db import get_connection
+            from ..db import repositories as repo
+            artifact = repo.get_report_artifact(get_connection(), artifact_id)
+        except Exception as exc:
+            return jsonify({"error": f"report database unavailable: {exc}"}), 503
+        if artifact is None:
+            abort(404, "report artifact not found")
+
+        payload = artifact["payload"]
+        export_format = (request.args.get("format") or "json").lower()
+        filename = secure_filename(os.path.basename(
+            artifact["original_path"])) or f"report-{artifact_id}"
+        if export_format == "json":
+            content = json.dumps(payload, ensure_ascii=False, indent=2)
+            mimetype = "application/json"
+            extension = ".json"
+        elif export_format == "txt" and payload.get("format") == "text":
+            content = payload.get("content", "")
+            mimetype = "text/plain; charset=utf-8"
+            extension = ".txt"
+        elif export_format == "csv" and payload.get("format") == "csv":
+            output = io.StringIO(newline="")
+            writer = csv.DictWriter(
+                output, fieldnames=payload.get("columns", []),
+                extrasaction="ignore")
+            writer.writeheader()
+            writer.writerows(payload.get("rows", []))
+            content = output.getvalue()
+            mimetype = "text/csv; charset=utf-8"
+            extension = ".csv"
+        else:
+            return jsonify({"error": "unsupported export format"}), 406
+
+        download_name = (filename if filename.lower().endswith(extension)
+                         else f"{filename}{extension}")
+        response = Response(content, mimetype=mimetype)
+        response.headers["Content-Disposition"] = (
+            f'attachment; filename="{download_name}"')
+        return response
+
     @app.get("/api/reports/<source>")
     def list_reports(source):
         base = report_sources.get(source)

tests/test_firewall_sync.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Tests for evidence-first firewall sync behavior."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from app.firewall import MikrotikFirewall
+
+
+class _FakeFirewall(MikrotikFirewall):
+    def __init__(self):
+        super().__init__(router_ip="0.0.0.0", username="x", password="x")
+        self.calls: list[str] = []
+
+    def get_address_list_ipv4(self, name: str) -> set[str]:
+        self.calls.append(f"v4:{name}")
+        if name == "Blocked":
+            return {"9.9.9.9/32"}
+        if name == "Init7":
+            return {"77.74.80.0/21"}
+        return set()
+
+    def get_address_list_ipv6(self, name: str) -> set[str]:
+        self.calls.append(f"v6:{name}")
+        if name == "Blocked":
+            return set()
+        if name == "Init7":
+            return {"2001:1620::/30"}
+        return set()
+
+
+class FirewallSyncEvidenceTest(unittest.TestCase):
+    def test_sync_loads_router_status_before_computing_changes(self):
+        fw = _FakeFirewall()
+        # type: ignore[assignment]
+        fw.save_state_report = lambda *args, **kwargs: "noop"
+
+        with patch("app.firewall.cumulative_blocklist",
+                   return_value=({"8.8.8.8/32"}, set())):
+            changes = fw.sync_blocklist(["8.8.8.8/32"], [], dry_run=True)
+
+        self.assertIn("v4:Blocked", fw.calls)
+        self.assertIn("v6:Blocked", fw.calls)
+        self.assertIn("v4:Init7", fw.calls)
+        self.assertIn("v6:Init7", fw.calls)
+        self.assertEqual(changes["added_ipv4"], ["8.8.8.8/32"])
+        self.assertEqual(changes["removed_ipv4"], ["9.9.9.9/32"])
+
+    def test_sync_evaluates_against_loaded_current_status(self):
+        fw = _FakeFirewall()
+        # type: ignore[assignment]
+        fw.save_state_report = lambda *args, **kwargs: "noop"
+
+        with patch("app.firewall.cumulative_blocklist",
+                   return_value=({"9.9.9.9/32"}, set())):
+            changes = fw.sync_blocklist(["9.9.9.9/32"], [], dry_run=True)
+
+        self.assertEqual(changes["added_ipv4"], [])
+        self.assertEqual(changes["removed_ipv4"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report_backfill.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Tests for legacy report artifact backfill."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from app.db import connect
+from app.db.backfill import backfill_report_artifacts
+from app.db import repositories as repo
+
+
+class ReportBackfillTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp_dir.name)
+        self.conn = connect(str(self.root / "test.db"))
+        self.roots = {
+            "copyparty": str(self.root / "copyparty"),
+            "npm": str(self.root / "npm"),
+            "app": str(self.root / "app"),
+            "mikrotik": str(self.root / "mikrotik"),
+        }
+        for path in self.roots.values():
+            Path(path).mkdir(parents=True)
+
+    def tearDown(self):
+        self.conn.close()
+        self.temp_dir.cleanup()
+
+    def test_dry_run_and_import_are_idempotent(self):
+        Path(self.roots["copyparty"], "2026-11-06-11-30-10.txt").write_text(
+            "copyparty summary\n", encoding="utf-8")
+        Path(self.roots["npm"], "2026-11-06.csv").write_text(
+            "ip,count\n192.0.2.1,3\n", encoding="utf-8")
+        Path(self.roots["app"], "2026-11-06.json").write_text(
+            '{"summary":{"blocked":1}}', encoding="utf-8")
+        Path(self.roots["app"], "data").mkdir()
+        Path(self.roots["app"], "data", "requests.csv").write_text(
+            "ip\n192.0.2.2\n", encoding="utf-8")
+
+        preview = backfill_report_artifacts(
+            self.conn, dry_run=True, roots=self.roots)["reports"]
+        self.assertEqual(preview["would_import"], 3)
+        self.assertEqual(preview["skipped_canonical"], [
+            {"source": "app", "path": "data/requests.csv"}])
+        self.assertEqual(preview["by_source"]["copyparty"]["would_import"], 1)
+        self.assertEqual(preview["by_format"]["csv"]["would_import"], 1)
+        self.assertEqual(repo.list_report_artifacts(self.conn)["items"], [])
+
+        imported = backfill_report_artifacts(
+            self.conn, roots=self.roots)["reports"]
+        self.assertEqual(imported["imported"], 3)
+        self.assertEqual(imported["unmapped"], [])
+
+        repeated = backfill_report_artifacts(
+            self.conn, roots=self.roots)["reports"]
+        self.assertEqual(repeated["imported"], 0)
+        self.assertEqual(repeated["already_imported"], 3)
+
+        items = repo.list_report_artifacts(
+            self.conn, source="copyparty")["items"]
+        self.assertEqual(items[0]["original_path"],
+                         "copyparty/2026-11-06-11-30-10.txt")
+        self.assertEqual(items[0]["generated_at"],
+                         "2026-11-06T11:30:10+00:00")
+
+    def test_malformed_json_is_preserved_with_parse_error(self):
+        Path(self.roots["app"], "2026-11-06.json").write_text(
+            '{"incomplete":', encoding="utf-8")
+
+        result = backfill_report_artifacts(
+            self.conn, roots=self.roots)["reports"]
+        self.assertEqual(result["imported"], 1)
+        artifact = repo.list_report_artifacts(
+            self.conn, source="app")["items"][0]
+        detail = repo.get_report_artifact(self.conn, artifact["id"])
+        self.assertEqual(detail["payload"]["content"], '{"incomplete":')
+        self.assertIn("parse_error", detail["payload"])
+
+    def test_artifacts_survive_sqlite_backup_and_restore(self):
+        artifact_id = repo.insert_report_artifact(
+            self.conn, source="app", report_type="summary",
+            generated_at="2026-11-06T00:00:00+00:00",
+            payload={"summary": {"count": 4}},
+            original_path="app/2026-11-06.json", content_sha256="c" * 64,
+        )
+        restored = connect(str(self.root / "restored.db"))
+        try:
+            self.conn.backup(restored)
+            artifact = repo.get_report_artifact(restored, artifact_id)
+            self.assertEqual(artifact["payload"], {"summary": {"count": 4}})
+        finally:
+            restored.close()
+
+    def test_oversized_file_is_reported_without_deleting_source(self):
+        source = Path(self.roots["npm"], "large.txt")
+        source.write_text("keep", encoding="utf-8")
+        from unittest.mock import patch
+
+        with patch("app.db.backfill.MAX_REPORT_ARTIFACT_BYTES", 1):
+            result = backfill_report_artifacts(
+                self.conn, roots=self.roots)["reports"]
+
+        self.assertEqual(len(result["unmapped"]), 1)
+        self.assertTrue(source.exists())
+        self.assertEqual(result["unmapped"][0]["path"], "large.txt")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report_persistence.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""Tests for dual-writing generated reports to SQLite."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import app.scan
+from app.analyzer import Analyzer
+from app.db import connect
+from app.db import repositories as repo
+from app.utils.blocklist_history import write_blocklist_report
+
+
+class ReportPersistenceTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp_dir.name)
+        self.conn = connect(str(self.root / "test.db"))
+        self.connection_patch = patch("app.db.get_connection",
+                                      return_value=self.conn)
+        self.connection_patch.start()
+
+    def tearDown(self):
+        self.connection_patch.stop()
+        self.conn.close()
+        self.temp_dir.cleanup()
+
+    def test_analyzer_summary_is_stored_with_file_provenance(self):
+        report_dir = self.root / "copyparty"
+        report_dir.mkdir()
+        path = report_dir / "2026-11-06-11-30-10.txt"
+        path.write_text("summary body", encoding="utf-8")
+        analyzer = Analyzer("logs", str(report_dir), quiet=True)
+
+        analyzer.persist_report_artifact(str(path), scan_id=14)
+
+        item = repo.list_report_artifacts(
+            self.conn, source="ids")["items"][0]
+        detail = repo.get_report_artifact(self.conn, item["id"])
+        self.assertEqual(item["original_path"], "ids/2026-11-06-11-30-10.txt")
+        self.assertEqual(detail["scan_id"], 14)
+        self.assertEqual(detail["payload"]["content"], "summary body")
+
+    def test_blocklist_report_is_stored_as_structured_artifact(self):
+        path = write_blocklist_report(
+            str(self.root / "reports"), history_v4=set(), history_v6=set(),
+            prev_v4=set(), prev_v6=set(),
+            complete_v4={"192.0.2.1/32"}, complete_v6=set(),
+            firewall_changes={"added_ipv4": ["192.0.2.1/32"]},
+            dry_run=True, scan_id=22,
+        )
+
+        items = repo.list_report_artifacts(
+            self.conn, source="app", report_type="blocklist_change")["items"]
+        self.assertEqual(len(items), 1)
+        detail = repo.get_report_artifact(self.conn, items[0]["id"])
+        self.assertEqual(detail["scan_id"], 22)
+        self.assertEqual(detail["payload"]["blocked_ipv4"], ["192.0.2.1/32"])
+        self.assertTrue(Path(path).exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_report_routes.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Tests for database-backed report API routes."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from app.db import connect
+from app.db import repositories as repo
+from app.web import create_app
+
+
+class ReportRoutesTest(unittest.TestCase):
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.conn = connect(str(Path(self.temp_dir.name) / "test.db"))
+        self.app = create_app()
+        self.app.config["TESTING"] = True
+        self.client = self.app.test_client()
+        self.connection_patch = patch("app.db.get_connection",
+                                      return_value=self.conn)
+        self.connection_patch.start()
+        self.artifact_id = repo.insert_report_artifact(
+            self.conn, source="copyparty", report_type="text_report",
+            generated_at="2026-11-06T10:00:00+00:00",
+            payload={"format": "text", "content": "summary"},
+            original_path="copyparty/2026-11-06.txt",
+            content_sha256="b" * 64,
+        )
+
+    def tearDown(self):
+        self.connection_patch.stop()
+        self.conn.close()
+        self.temp_dir.cleanup()
+
+    def test_list_detail_and_text_export(self):
+        listed = self.client.get("/api/reports?source=copyparty&limit=1")
+        self.assertEqual(listed.status_code, 200)
+        self.assertEqual(listed.json["items"][0]["id"], self.artifact_id)
+
+        detail = self.client.get(
+            f"/api/report-artifacts/{self.artifact_id}")
+        self.assertEqual(detail.status_code, 200)
+        self.assertEqual(detail.json["payload"]["content"], "summary")
+
+        exported = self.client.get(
+            f"/api/report-artifacts/{self.artifact_id}/export?format=txt")
+        self.assertEqual(exported.status_code, 200)
+        self.assertEqual(exported.data.decode(), "summary")
+        self.assertIn("attachment", exported.headers["Content-Disposition"])
+        self.assertNotIn(".txt.txt", exported.headers["Content-Disposition"])
+
+    def test_invalid_source_cursor_and_missing_id(self):
+        self.assertEqual(
+            self.client.get("/api/reports?source=unknown").status_code, 400)
+        self.assertEqual(
+            self.client.get("/api/reports?cursor=invalid").status_code, 400)
+        self.assertEqual(
+            self.client.get("/api/report-artifacts/9999").status_code, 404)
+
+    def test_legacy_file_routes_remain_registered(self):
+        response = self.client.get("/api/reports/copyparty")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("files", response.json)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_repositories.py

@@ -13,6 +13,7 @@ import os
 import sqlite3
 import tempfile
 import unittest
+from datetime import datetime, timedelta, timezone
 
 from app.db import connect
 from app.db import repositories as repo
@@ -46,7 +47,7 @@ class SchemaTest(unittest.TestCase):
                   "blocklist_reasons", "whitelist_entries",
                   "whitelist_sources", "firewall_snapshots",
                   "firewall_address_list_entries", "firewall_rule_entries",
-                  "whois_info", "log_sources"):
+                  "whois_info", "log_sources", "report_artifacts"):
             self.assertIn(t, tables, f"missing table {t}")
         conn.close()
 
@@ -110,7 +111,8 @@ class BlocklistTest(unittest.TestCase):
                                    added_by="scan:1")
         self.assertEqual(r1["active"], 2)
         # a later scan only sees 5.5.5.5; 6.6.6.6 stays (cumulative).
-        r2 = repo.record_blocklist(self.conn, ["5.5.5.5/32"], added_by="scan:2")
+        r2 = repo.record_blocklist(
+            self.conn, ["5.5.5.5/32"], added_by="scan:2")
         active = [e["cidr"] for e in repo.active_blocklist(self.conn)]
         self.assertIn("6.6.6.6/32", active)
         self.assertEqual(r2["active"], 2)
@@ -122,6 +124,38 @@ class BlocklistTest(unittest.TestCase):
                          [e["cidr"] for e in repo.active_blocklist(self.conn)])
         self.assertFalse(repo.unblock_entry(self.conn, "7.7.7.7/32"))
 
+    def test_prune_old_single_ip_blocks_keeps_ranges(self):
+        old = (datetime.now(timezone.utc) - timedelta(days=31)).isoformat(
+            timespec="seconds")
+        with repo.transaction(self.conn):
+            repo._block_entry_inner(self.conn, "1.2.3.4/32", last_seen=old)
+            repo._block_entry_inner(
+                self.conn, "2001:db8::1/128", last_seen=old)
+            repo._block_entry_inner(self.conn, "1.2.3.0/24", last_seen=old)
+            repo._block_entry_inner(self.conn, "2001:db8::/64", last_seen=old)
+
+        removed = repo.prune_old_single_ip_blocks(
+            self.conn, older_than_days=30)
+        self.assertCountEqual(removed, ["1.2.3.4/32", "2001:db8::1/128"])
+
+        active = {e["cidr"] for e in repo.active_blocklist(self.conn)}
+        self.assertIn("1.2.3.0/24", active)
+        self.assertIn("2001:db8::/64", active)
+        self.assertNotIn("1.2.3.4/32", active)
+        self.assertNotIn("2001:db8::1/128", active)
+
+    def test_prune_old_single_ip_blocks_respects_recency(self):
+        fresh = (datetime.now(timezone.utc) - timedelta(days=5)).isoformat(
+            timespec="seconds")
+        with repo.transaction(self.conn):
+            repo._block_entry_inner(self.conn, "9.9.9.9/32", last_seen=fresh)
+
+        removed = repo.prune_old_single_ip_blocks(
+            self.conn, older_than_days=30)
+        self.assertEqual(removed, [])
+        self.assertIn("9.9.9.9/32",
+                      {e["cidr"] for e in repo.active_blocklist(self.conn)})
+
 
 class WhitelistTest(unittest.TestCase):
     def setUp(self):
@@ -161,5 +195,54 @@ class FirewallTest(unittest.TestCase):
             "SELECT COUNT(*) FROM firewall_rule_entries").fetchone()[0], 1)
 
 
+class ReportArtifactTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = _new_conn()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_artifact_insert_is_idempotent_and_detail_decodes_payload(self):
+        kwargs = {
+            "source": "copyparty", "report_type": "analyzer_summary",
+            "generated_at": "2026-11-01T10:00:00+00:00",
+            "payload": {"summary": {"requests": 12}},
+            "original_path": "reports/copyparty/2026-11-01.txt",
+            "content_sha256": "a" * 64,
+        }
+        first_id = repo.insert_report_artifact(self.conn, **kwargs)
+        duplicate_id = repo.insert_report_artifact(self.conn, **kwargs)
+
+        self.assertEqual(first_id, duplicate_id)
+        self.assertEqual(self.conn.execute(
+            "SELECT COUNT(*) FROM report_artifacts").fetchone()[0], 1)
+        self.assertEqual(repo.get_report_artifact(self.conn, first_id)["payload"],
+                         kwargs["payload"])
+
+    def test_artifact_cursor_orders_stably_and_filters_source(self):
+        for day, source in (("01", "npm"), ("02", "copyparty"),
+                            ("03", "npm")):
+            repo.insert_report_artifact(
+                self.conn, source=source, report_type="summary",
+                generated_at=f"2026-11-{day}T00:00:00+00:00",
+                payload={"day": day}, original_path=f"reports/{day}.txt",
+                content_sha256=day * 64,
+            )
+
+        first = repo.list_report_artifacts(self.conn, source="npm", limit=1)
+        self.assertEqual(len(first["items"]), 1)
+        self.assertIsNotNone(first["next_cursor"])
+        second = repo.list_report_artifacts(
+            self.conn, source="npm", limit=1, cursor=first["next_cursor"])
+        self.assertEqual(len(second["items"]), 1)
+        self.assertIsNone(second["next_cursor"])
+        self.assertGreater(first["items"][0]["generated_at"],
+                           second["items"][0]["generated_at"])
+
+    def test_artifact_cursor_rejects_invalid_token(self):
+        with self.assertRaises(ValueError):
+            repo.list_report_artifacts(self.conn, cursor="invalid")
+
+
 if __name__ == "__main__":
     unittest.main()