AirCoding/AirPlan/docs/spec/AirPlan-ParaV2/plugins/airndb/scripts/airndb_mode.py

#!/usr/bin/env python3
"""Bootstrap AirNDB network debugging context files."""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Tuple

from airndb_capture import ensure_capture_tool

MARKER_BEGIN = "<!-- AIRNDB:BEGIN -->"
MARKER_END = "<!-- AIRNDB:END -->"


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def airndb_agents_block() -> str:
    return f"""{MARKER_BEGIN}
## AirNDB Network Debug Workflow

1. Use AirNDB for `/airndb` sessions that need tcpdump/WinDump packet capture, pcap reading, BPF filters, or network evidence.
2. On first entry, detect tcpdump/WinDump; on Windows, if none is available, download official `WinDump.exe` from https://www.winpcap.org/windump/install/, verify SHA1, and write `AirPlan/state/airndb/tool.env`.
3. Only capture authorized traffic; prefer short bounded captures with `-nn`, `-s 0`, `-c <count>`, and narrow BPF filters.
4. Before capture, load:
   - `AirPlan/AGENTS.md`
   - `AirPlan/docs/architecture/adr/`
   - `AirPlan/docs/architecture/c4/module.md`
   - `AirPlan/docs/network/airndb-log.md`
5. Store pcap, text summaries, and JSON reports under `AirPlan/docs/network/airndb-captures/`.
6. Record exact command, interface, filter, capture window, artifact paths, key observations, and residual risk in `AirPlan/docs/network/airndb-log.md`.
7. AirDbg may call AirNDB when debugging needs DNS/TCP/UDP/TLS/HTTP, ports, proxy, firewall, packet-loss, retransmit, reset, or pcap evidence.
8. Update ADR/C4 when network boundaries, capture tooling, observability, ports, protocols, DNS, proxy, TLS, or runtime topology become durable architecture context.
{MARKER_END}
"""


def c4_module_template() -> str:
    return """# C4 Module

## System Context
- TODO: Describe the system, users, and important external systems.

## Containers
- TODO: Describe runtime/deployable units and network boundaries.

## Modules

| Module | Responsibility | Network Interfaces | Dependencies | Data Ownership | Network Debug Notes |
| --- | --- | --- | --- | --- | --- |
| TODO | TODO | TODO | TODO | TODO | TODO |

## Network / Observability Boundaries
- tcpdump/WinDump capture points, interfaces, container/WSL/VM/host boundaries: TODO
- Ports, protocols, DNS, proxy, TLS, firewall, NAT, or gateway notes: TODO

## Change Log
- TODO: Record network-boundary or observability changes discovered by AirNDB.
"""


def adr_template() -> str:
    return """# ADR-0001: AirNDB Packet Capture Governance

- Status: Accepted
- Date: TODO

## Context
Network debugging needs durable, AI-readable packet capture context and bounded evidence collection.

## Decision
Use AirNDB to build safe tcpdump/WinDump commands, capture or read pcap artifacts, and maintain `AirPlan/docs/network/airndb-log.md`, C4 module docs, and ADR records when network boundaries or diagnostics change.

## Consequences
- Packet evidence can be reused by future AirDbg or AirNDB sessions.
- Captures must stay bounded and authorized.
- pcap artifacts may contain sensitive data and need careful handling.

## Alternatives
- Chat-only packet notes: rejected because commands, filters, and pcap paths are easy to lose.
"""


def network_log_template() -> str:
    return """# AirNDB Network Debug Log

Append entries for AirNDB packet capture sessions.

## Entry Template

### YYYY-MM-DD: short network issue title

- Scope / authorization: TODO
- Symptom: TODO
- Interface: TODO
- Filter: TODO
- Capture window: TODO
- Command: TODO
- Artifacts: TODO
- Key observations: TODO
- AirDbg handoff: TODO
- ADR/C4 updates: TODO
- Residual risk: TODO
"""


def captures_gitignore_template() -> str:
    return """*.pcap
*.pcapng
*.cap
*.txt
*.json
!.gitignore
"""


def airndb_gitignore_template() -> str:
    return """tool.env
"""


def write_if_missing(path: Path, content: str) -> bool:
    if path.exists():
        return False
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8", newline="\n")
    return True


def upsert_agents_md(path: Path) -> str:
    block = airndb_agents_block().rstrip() + "\n"
    if path.exists():
        original = path.read_text(encoding="utf-8")
        existed = True
    else:
        original = "# AGENTS.md\n\n"
        existed = False

    begin = original.find(MARKER_BEGIN)
    end = original.find(MARKER_END)
    if begin >= 0 and end > begin:
        end += len(MARKER_END)
        updated = original[:begin].rstrip() + "\n\n" + block + original[end:].lstrip()
        status = "updated"
    else:
        updated = original.rstrip() + "\n\n" + block
        status = "updated" if existed else "created"

    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(updated, encoding="utf-8", newline="\n")
    return status


def artifact_map(project_root: Path) -> Dict[str, Path]:
    return {
        "AGENTS.md": project_root / "AirPlan" / "AGENTS.md",
        "c4_module": project_root / "AirPlan" / "docs" / "architecture" / "c4" / "module.md",
        "adr_dir": project_root / "AirPlan" / "docs" / "architecture" / "adr",
        "adr_0001": project_root / "AirPlan" / "docs" / "architecture" / "adr" / "ADR-0001-airndb-packet-capture-governance.md",
        "network_log": project_root / "AirPlan" / "docs" / "network" / "airndb-log.md",
        "captures_dir": project_root / "AirPlan" / "docs" / "network" / "airndb-captures",
        "captures_gitignore": project_root / "AirPlan" / "docs" / "network" / "airndb-captures" / ".gitignore",
        "tool_env": project_root / "AirPlan" / "state" / "airndb" / "tool.env",
        "airndb_gitignore": project_root / "AirPlan" / "state" / "airndb" / ".gitignore",
        "state": project_root / "AirPlan" / "state" / "airndb" / "state.json",
    }


def write_state(path: Path, enabled: bool, project_root: Path) -> None:
    artifacts = artifact_map(project_root)
    health = {
        name: artifacts[name].exists()
        for name in [
            "AGENTS.md",
            "c4_module",
            "adr_dir",
            "adr_0001",
            "network_log",
            "captures_dir",
            "captures_gitignore",
            "tool_env",
            "airndb_gitignore",
        ]
    }
    payload = {
        "enabled": enabled,
        "updatedAt": now_iso(),
        "projectRoot": str(project_root),
        "artifactHealth": health,
        "captureTool": ensure_capture_tool(project_root, auto_install=False),
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")


def enter_mode(project_root: Path) -> Tuple[str, Dict[str, str]]:
    artifacts = artifact_map(project_root)
    results: Dict[str, str] = {}
    results["AGENTS.md"] = upsert_agents_md(artifacts["AGENTS.md"])
    results["c4_module"] = "created" if write_if_missing(artifacts["c4_module"], c4_module_template()) else "exists"
    artifacts["adr_dir"].mkdir(parents=True, exist_ok=True)
    results["adr_dir"] = "exists"
    results["adr_0001"] = "created" if write_if_missing(artifacts["adr_0001"], adr_template()) else "exists"
    results["network_log"] = "created" if write_if_missing(artifacts["network_log"], network_log_template()) else "exists"
    artifacts["captures_dir"].mkdir(parents=True, exist_ok=True)
    results["captures_dir"] = "exists"
    results["captures_gitignore"] = "created" if write_if_missing(artifacts["captures_gitignore"], captures_gitignore_template()) else "exists"
    results["airndb_gitignore"] = "created" if write_if_missing(artifacts["airndb_gitignore"], airndb_gitignore_template()) else "exists"
    tool_status = ensure_capture_tool(project_root, auto_install=True)
    results["capture_tool"] = tool_status.get("status", "unknown")
    results["capture_tool_path"] = tool_status.get("tool", "")
    results["capture_tool_source"] = tool_status.get("source", "")
    results["capture_tool_env"] = tool_status.get("projectEnv", "")
    if tool_status.get("driverHint"):
        results["capture_driver_hint"] = tool_status["driverHint"]
    write_state(artifacts["state"], True, project_root)
    return "enabled", results


def exit_mode(project_root: Path) -> Tuple[str, Dict[str, str]]:
    artifacts = artifact_map(project_root)
    write_state(artifacts["state"], False, project_root)
    return "disabled", {}


def status_mode(project_root: Path) -> Tuple[str, Dict[str, str]]:
    artifacts = artifact_map(project_root)
    state_file = artifacts["state"]
    enabled = False
    if state_file.exists():
        try:
            payload = json.loads(state_file.read_text(encoding="utf-8"))
            enabled = bool(payload.get("enabled"))
        except json.JSONDecodeError:
            enabled = False
    results = {
        name: ("ok" if path.exists() else "missing")
        for name, path in artifacts.items()
        if name != "state"
    }
    return ("enabled" if enabled else "disabled"), results


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Manage AirNDB network debug artifacts.")
    parser.add_argument("--mode", choices=["enter", "setup", "exit", "status"], default="enter")
    parser.add_argument("--project", default=".")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    project_root = Path(args.project).expanduser().resolve()
    if args.mode in {"enter", "setup"}:
        mode_state, result = enter_mode(project_root)
    elif args.mode == "exit":
        mode_state, result = exit_mode(project_root)
    else:
        mode_state, result = status_mode(project_root)
    print(f"airndb_mode={mode_state}")
    print(f"project_root={project_root}")
    for key, value in result.items():
        print(f"{key}={value}")


if __name__ == "__main__":
    main()