Nexus/scripts/generate_bug_scan_registry.py

#!/usr/bin/env python3
"""Generate docs/audit/2026-06-04-bug-scan-registry.json — hand-written source SSOT."""

from __future__ import annotations

import json
import re
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
OUT = ROOT / "docs/audit/2026-06-04-bug-scan-registry.json"

# Batch 1–6 coverage (path relative to ROOT)
BATCH_1_6 = {
    "server/api/install.py": {"batch_id": 1, "status": "done", "needs_full_rescan": False},
    "web/app/install.html": {"batch_id": 1, "status": "done", "needs_full_rescan": False},
    "server/application/services/script_job_callback.py": {
        "batch_id": 1,
        "status": "done",
        "needs_full_rescan": False,
    },
    "server/api/agent.py": {"batch_id": 1, "status": "partial", "needs_full_rescan": True},
    "server/api/servers.py": {"batch_id": 1, "status": "partial", "needs_full_rescan": True},
    "server/background/schedule_runner.py": {"batch_id": 2, "status": "done", "needs_full_rescan": False},
    "server/application/services/auth_service.py": {
        "batch_id": 2,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "server/application/services/sync_engine_v2.py": {
        "batch_id": 2,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "server/infrastructure/redis/script_execution_store.py": {
        "batch_id": 2,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "frontend/src/composables/terminal/useTerminalSessions.ts": {
        "batch_id": 2,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "server/background/retry_runner.py": {"batch_id": 3, "status": "done", "needs_full_rescan": False},
    "server/application/services/files_browse_service.py": {
        "batch_id": 3,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "frontend/src/pages/DashboardPage.vue": {
        "batch_id": 3,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "server/api/sync_v2.py": {"batch_id": 4, "status": "partial", "needs_full_rescan": True},
    "server/api/files.py": {"batch_id": 4, "status": "done", "needs_full_rescan": False},
    "server/background/heartbeat_flush.py": {
        "batch_id": 4,
        "status": "partial",
        "needs_full_rescan": True,
    },
    "frontend/src/composables/push/usePushProgress.ts": {
        "batch_id": 4,
        "status": "done",
        "needs_full_rescan": False,
    },
    "server/api/schemas.py": {"batch_id": 4, "status": "partial", "needs_full_rescan": True},
    "server/background/self_monitor.py": {"batch_id": 5, "status": "done", "needs_full_rescan": False},
    "server/api/auth_jwt.py": {"batch_id": 6, "status": "done", "needs_full_rescan": False},
    "server/background/script_execution_flush.py": {
        "batch_id": 6,
        "status": "done",
        "needs_full_rescan": False,
    },
    "frontend/src/composables/files/useFilesBrowse.ts": {
        "batch_id": 6,
        "status": "done",
        "needs_full_rescan": False,
    },
    "frontend/src/stores/files.ts": {"batch_id": 6, "status": "done", "needs_full_rescan": False},
}

EXCLUDED_JUNK = {"scripts/sedul3hFb"}
# Local secrets — not in repo SSOT; example only in docker/.env.example
EXCLUDED_LOCAL = {"docker/.env", ".env"}

LARGE_FILE_THRESHOLD = 800

# Planned batch assignments (relative path -> batch_id)
PLANNED_BATCHES: dict[str, int] = {}

def _add_batch(batch_id: int, paths: list[str]) -> None:
    for p in paths:
        PLANNED_BATCHES[p] = batch_id


# Phase A B7-B16
_add_batch(7, ["server/api/servers.py"])
_add_batch(8, ["server/api/sync_v2.py"])
_add_batch(
    9,
    [
        "server/api/webssh.py",
        "server/api/auth.py",
        "server/api/websocket.py",
        "server/api/terminal.py",
        "server/api/settings.py",
    ],
)
_add_batch(
    10,
    [
        "server/api/scripts.py",
        "server/api/search.py",
        "server/api/assets.py",
        "server/api/files.py",
        "server/api/health.py",
    ],
)
_add_batch(
    11,
    [
        "server/api/dependencies.py",
        "server/api/remote_path_validation.py",
        "server/api/schema_path_validators.py",
        "server/api/schemas.py",
        "server/api/__init__.py",
    ],
)
_add_batch(12, ["server/api/install.py"])
_add_batch(13, ["server/api/agent.py"])
_add_batch(14, ["server/api/auth_jwt.py"])

# Phase B B17-B24
_add_batch(17, ["server/application/services/script_service.py"])
_add_batch(18, ["server/application/services/sync_engine_v2.py"])
_add_batch(19, ["server/application/services/auth_service.py"])
_add_batch(
    20,
    [
        "server/application/services/script_job_callback.py",
        "server/application/services/files_browse_service.py",
        "server/application/services/server_service.py",
        "server/application/services/sync_service.py",
        "server/application/services/script_jobs.py",
    ],
)
_add_batch(
    21,
    [
        "server/application/__init__.py",
        "server/application/services/__init__.py",
    ],
)
_add_batch(
    22,
    [
        "server/background/schedule_runner.py",
        "server/background/retry_runner.py",
        "server/background/heartbeat_flush.py",
        "server/background/self_monitor.py",
        "server/background/script_execution_flush.py",
    ],
)
_add_batch(
    23,
    [
        "server/background/ip_allowlist_refresh.py",
        "server/background/upload_staging_cleanup.py",
        "server/background/__init__.py",
    ],
)
_add_batch(24, ["server/main.py", "server/config.py", "server/__init__.py"])

# Phase C B25-B34 — filled dynamically below after collecting infra paths

INFRA_DB = sorted(
    [
        "server/infrastructure/database/admin_repo.py",
        "server/infrastructure/database/audit_log_repo.py",
        "server/infrastructure/database/crypto.py",
        "server/infrastructure/database/db_credential_repo.py",
        "server/infrastructure/database/engine_compat.py",
        "server/infrastructure/database/migrations.py",
        "server/infrastructure/database/password_preset_repo.py",
        "server/infrastructure/database/platform_node_repo.py",
        "server/infrastructure/database/push_schedule_repo.py",
        "server/infrastructure/database/script_repo.py",
        "server/infrastructure/database/server_repo.py",
        "server/infrastructure/database/session.py",
        "server/infrastructure/database/setting_repo.py",
        "server/infrastructure/database/ssh_key_preset_repo.py",
        "server/infrastructure/database/ssh_session_repo.py",
        "server/infrastructure/database/sync_log_repo.py",
        "server/infrastructure/database/__init__.py",
    ]
)
INFRA_SSH = sorted(
    [
        "server/infrastructure/ssh/asyncssh_pool.py",
        "server/infrastructure/ssh/files_capability.py",
        "server/infrastructure/ssh/remote_archive.py",
        "server/infrastructure/ssh/remote_probe.py",
        "server/infrastructure/ssh/remote_shell.py",
        "server/infrastructure/ssh/__init__.py",
    ]
)
INFRA_OTHER = sorted(
    [
        "server/infrastructure/redis/client.py",
        "server/infrastructure/redis/script_callback_rate.py",
        "server/infrastructure/redis/script_execution_store.py",
        "server/infrastructure/redis/__init__.py",
        "server/infrastructure/telegram/__init__.py",
        "server/infrastructure/agent_url.py",
        "server/infrastructure/subscription_parser.py",
        "server/infrastructure/__init__.py",
    ]
)
DOMAIN_UTILS = sorted(
    [
        "server/domain/__init__.py",
        "server/domain/models/__init__.py",
        "server/domain/repositories/__init__.py",
        "server/utils/files_chmod_policy.py",
        "server/utils/files_elevation.py",
        "server/utils/posix_paths.py",
        "server/utils/sync_error_message.py",
        "server/utils/text_io.py",
        "server/utils/unix_ls.py",
    ]
)


def _chunk(paths: list[str], size: int) -> list[list[str]]:
    return [paths[i : i + size] for i in range(0, len(paths), size)]


bid = 25
for chunk in _chunk(INFRA_DB, 5):
    _add_batch(bid, chunk)
    bid += 1
for chunk in _chunk(INFRA_SSH, 5):
    _add_batch(bid, chunk)
    bid += 1
_add_batch(bid, INFRA_OTHER)
bid += 1
for chunk in _chunk(DOMAIN_UTILS, 5):
    _add_batch(bid, chunk)
    bid += 1

# Phase D frontend — pages B35+
FRONTEND_PAGES = sorted((ROOT / "frontend/src/pages").glob("*.vue"))
FRONTEND_PAGES = [str(p.relative_to(ROOT)).replace("\\", "/") for p in FRONTEND_PAGES]
bid = 35
for chunk in _chunk(FRONTEND_PAGES, 5):
    _add_batch(bid, chunk)
    bid += 1

# composables — useTerminalSessions alone first (large)
_add_batch(bid, ["frontend/src/composables/terminal/useTerminalSessions.ts"])
bid += 1
COMPOSABLES = sorted(
    (ROOT / "frontend/src/composables").rglob("*"),
    key=lambda p: str(p),
)
COMPOSABLES = [
    str(p.relative_to(ROOT)).replace("\\", "/")
    for p in COMPOSABLES
    if p.is_file() and p.suffix in {".ts", ".vue"}
    and "useTerminalSessions.ts" not in p.name
]
for chunk in _chunk(COMPOSABLES, 5):
    _add_batch(bid, chunk)
    bid += 1

COMPONENTS = sorted(
    (ROOT / "frontend/src/components").rglob("*.vue"),
    key=lambda p: str(p),
)
COMPONENTS = [str(p.relative_to(ROOT)).replace("\\", "/") for p in COMPONENTS]
for chunk in _chunk(COMPONENTS, 5):
    _add_batch(bid, chunk)
    bid += 1

_add_batch(
    bid,
    [
        "frontend/src/api/index.ts",
        "frontend/src/stores/auth.ts",
        "frontend/src/stores/files.ts",
        "frontend/src/router/index.ts",
        "frontend/src/App.vue",
    ],
)
bid += 1

FRONTEND_REST = sorted(
    p
    for p in (ROOT / "frontend/src").rglob("*")
    if p.is_file()
    and p.suffix in {".ts", ".scss"}
    and "assets" not in p.parts
    and p not in {ROOT / f for f in [
        "frontend/src/api/index.ts",
        "frontend/src/stores/auth.ts",
        "frontend/src/stores/files.ts",
        "frontend/src/router/index.ts",
        "frontend/src/App.vue",
        "frontend/src/main.ts",
    ]}
)
FRONTEND_REST_PATHS = []
for p in FRONTEND_REST:
    rel = str(p.relative_to(ROOT)).replace("\\", "/")
    if rel.startswith("frontend/src/pages/"):
        continue
    if rel.startswith("frontend/src/composables/"):
        continue
    if rel.startswith("frontend/src/components/"):
        continue
    if rel.endswith((".png", ".svg", ".jpg", ".ico")):
        continue
    FRONTEND_REST_PATHS.append(rel)
for chunk in _chunk(sorted(FRONTEND_REST_PATHS), 5):
    _add_batch(bid, chunk)
    bid += 1
if (ROOT / "frontend/src/main.ts").exists():
    PLANNED_BATCHES.setdefault("frontend/src/main.ts", bid)

# Phase E
_add_batch(
    bid,
    [
        "web/app/index.html",
        "web/app/install.html",
        "web/app/api.js",
        "web/app/layout.js",
        "web/app/preview.html",
    ],
)
bid += 1
_add_batch(
    bid,
    [
        "web/app/preview-v3.html",
        "web/app/preview-v4.html",
        "web/app/preview-v5.html",
        "web/agent/agent.py",
        "web/agent/agent.sh",
    ],
)
bid += 1
_add_batch(
    bid,
    [
        "web/agent/install.sh",
        "web/agent/uninstall.sh",
        "web/agent/config.example.json",
    ],
)

# Phase F tests
TESTS = sorted((ROOT / "tests").rglob("*.py"))
TESTS = [str(p.relative_to(ROOT)).replace("\\", "/") for p in TESTS]
for chunk in _chunk(TESTS, 5):
    _add_batch(bid, chunk)
    bid += 1

DEPLOY = sorted(
    p
    for p in (ROOT / "deploy").rglob("*")
    if p.is_file() and p.name != "gate_log.jsonl"
)
DEPLOY = [str(p.relative_to(ROOT)).replace("\\", "/") for p in DEPLOY]
for chunk in _chunk(DEPLOY, 5):
    _add_batch(bid, chunk)
    bid += 1

SCRIPTS = sorted(
    p
    for p in (ROOT / "scripts").rglob("*")
    if p.is_file() and p.name not in EXCLUDED_JUNK
)
SCRIPTS = [str(p.relative_to(ROOT)).replace("\\", "/") for p in SCRIPTS]
for chunk in _chunk(SCRIPTS, 5):
    _add_batch(bid, chunk)
    bid += 1

DOCKER_DEPLOY_ROOT = sorted(
    [
        "docker/generate_env.py",
        "docker/sync_root_env.py",
        "docker/entrypoint.sh",
        "docker/.env.example",
        "docker/README.md",
        "Dockerfile",
        ".dockerignore",
    ]
)
_add_batch(bid, [p for p in DOCKER_DEPLOY_ROOT if (ROOT / p).exists()])
bid += 1
_add_batch(
    bid,
    [
        "frontend/vite.config.mts",
        "frontend/package.json",
        "requirements.txt",
        "requirements-dev.txt",
        "ruff.toml",
    ],
)


def collect_must_audit() -> list[Path]:
    paths: list[Path] = []
    # server py
    paths.extend(ROOT.glob("server/**/*.py"))
    # frontend src
    for p in (ROOT / "frontend/src").rglob("*"):
        if not p.is_file():
            continue
        if p.suffix not in {".ts", ".vue", ".scss"}:
            continue
        if "assets" in p.parts and p.suffix in {".png", ".svg", ".jpg", ".ico"}:
            continue
        paths.append(p)
    # web app root html/js
    for name in ["*.html", "*.js"]:
        paths.extend((ROOT / "web/app").glob(name))
    # agent
    paths.extend((ROOT / "web/agent").rglob("*"))
    paths = [p for p in paths if p.is_file()]
    # tests
    paths.extend((ROOT / "tests").rglob("*.py"))
    # deploy scripts docker (selected)
    for p in (ROOT / "deploy").rglob("*"):
        if p.is_file() and p.name != "gate_log.jsonl":
            paths.append(p)
    for p in (ROOT / "scripts").rglob("*"):
        if p.is_file() and p.name not in EXCLUDED_JUNK:
            paths.append(p)
    for p in (ROOT / "docker").rglob("*"):
        if p.is_file():
            paths.append(p)
    for extra in [
        "Dockerfile",
        ".dockerignore",
        "frontend/vite.config.mts",
        "frontend/package.json",
        "requirements.txt",
        "requirements-dev.txt",
        "ruff.toml",
    ]:
        ep = ROOT / extra
        if ep.exists():
            paths.append(ep)
    # dedupe
    seen: set[str] = set()
    out: list[Path] = []
    for p in sorted(set(paths), key=lambda x: str(x)):
        rel = str(p.relative_to(ROOT)).replace("\\", "/")
        if rel in seen:
            continue
        seen.add(rel)
        out.append(p)
    return out


def line_count(path: Path) -> int:
    try:
        return sum(1 for _ in path.open("rb"))
    except OSError:
        return 0


def main() -> None:
    must = collect_must_audit()
    entries: list[dict] = []
    covered_assets = 0

    for path in sorted((ROOT / "web/app/assets").rglob("*"), key=lambda x: str(x)):
        if not path.is_file():
            continue
        rel = str(path.relative_to(ROOT)).replace("\\", "/")
        entries.append(
            {
                "path": rel,
                "category": "generated",
                "lines": line_count(path),
                "batch_id": None,
                "status": "covered-by-frontend-src",
                "needs_full_rescan": False,
                "planned_batch": "B-Assets",
            }
        )
        covered_assets += 1

    for path in must:
        rel = str(path.relative_to(ROOT)).replace("\\", "/")
        if rel in EXCLUDED_JUNK:
            entries.append(
                {
                    "path": rel,
                    "category": "excluded-junk",
                    "lines": line_count(path),
                    "batch_id": None,
                    "status": "excluded",
                    "needs_full_rescan": False,
                }
            )
            continue
        if rel in EXCLUDED_LOCAL:
            entries.append(
                {
                    "path": rel,
                    "category": "excluded-local",
                    "lines": line_count(path),
                    "batch_id": None,
                    "status": "excluded",
                    "needs_full_rescan": False,
                    "note": "local secrets; audit docker/.env.example instead",
                }
            )
            continue
        prev = BATCH_1_6.get(rel, {})
        planned = PLANNED_BATCHES.get(rel)
        n = line_count(path)
        large = n > LARGE_FILE_THRESHOLD
        entry = {
            "path": rel,
            "category": "must_audit",
            "lines": n,
            "large_file": large,
            "batch_id": prev.get("batch_id") or planned,
            "planned_batch": planned,
            "status": prev.get("status", "pending"),
            "needs_full_rescan": prev.get("needs_full_rescan", False),
            "read_range": None,
            "findings_count": 0,
        }
        entries.append(entry)

    must_audit = [e for e in entries if e["category"] == "must_audit"]
    pending = sum(1 for e in must_audit if e["status"] == "pending")
    partial_rescan = sum(1 for e in must_audit if e.get("needs_full_rescan"))
    done = sum(1 for e in must_audit if e["status"] == "done")

    doc = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "summary": {
            "total_must_audit": len(must_audit),
            "pending": pending,
            "done": done,
            "partial_rescan": partial_rescan,
            "covered_by_frontend_src_assets": covered_assets,
            "excluded_junk": len(EXCLUDED_JUNK),
            "max_batch_id": max(PLANNED_BATCHES.values()) if PLANNED_BATCHES else 0,
        },
        "entries": entries,
    }
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    print(json.dumps(doc["summary"], indent=2))


if __name__ == "__main__":
    main()