NotesAgentic/scripts/measure-sync-memory.py

"""Windows component measurements, not the S-09 service/release benchmark.

Build the current Rust library test executable, then run each fixed workload in
its own process. Reports observed OS cumulative peaks, with sampling coverage.
No third-party Python packages required. Usage: python scripts/measure-sync-memory.py
"""
from __future__ import annotations
import argparse
import ctypes as c
from ctypes import wintypes as w
import hashlib
import json
import os
from pathlib import Path
import platform
import re
import statistics
import subprocess
import sys
import time
from datetime import datetime, timezone

ROOT = Path(__file__).resolve().parents[1]
WORKLOADS = {
    "stream_verify": "payloads::tests::hundred_mib_verification_is_bounded_and_rejects_growth_or_truncation",
    "write_recovery": "workspace::tests::hundred_mib_spooled_writes_and_journal_recovery_keep_receipts_exactly_once",
    "rename_delete": "workspace::tests::hundred_mib_file_operations_recover_each_copy_stage_without_duplicate_events",
    "conflicts": "sync_resolution::tests::hundred_mib_conflicts_resolve_all_choices_and_reopen_without_duplicate_jobs",
    "discovery_rebind": "sync_initial::tests::hundred_mib_discovery_preview_and_rebinding_preserve_all_current_files",
}

class Counters(c.Structure):
    _fields_ = [("cb", w.DWORD), ("PageFaultCount", w.DWORD)] + [
        (name, c.c_size_t) for name in (
            "PeakWorkingSetSize", "WorkingSetSize", "QuotaPeakPagedPoolUsage",
            "QuotaPagedPoolUsage", "QuotaPeakNonPagedPoolUsage", "QuotaNonPagedPoolUsage",
            "PagefileUsage", "PeakPagefileUsage", "PrivateUsage")]

class Monitor:
    def __init__(self):
        self.kernel = c.WinDLL("kernel32", use_last_error=True)
        self.psapi = c.WinDLL("psapi", use_last_error=True)
        self.kernel.OpenProcess.argtypes = [w.DWORD, w.BOOL, w.DWORD]
        self.kernel.OpenProcess.restype = w.HANDLE
        self.kernel.CloseHandle.argtypes = [w.HANDLE]
        self.kernel.CloseHandle.restype = w.BOOL
        self.psapi.GetProcessMemoryInfo.argtypes = [w.HANDLE, c.POINTER(Counters), w.DWORD]
        self.psapi.GetProcessMemoryInfo.restype = w.BOOL

    def run(self, command, log: Path, timeout=120):
        env = {key: value for key, value in os.environ.items() if key.upper() in {
            "SYSTEMROOT", "WINDIR", "PATH", "TEMP", "TMP", "USERPROFILE", "LOCALAPPDATA"}}
        env["PYTHONUTF8"] = "1"
        start = time.perf_counter()
        samples = errors = 0
        first = last = None
        peaks = {"working_set_bytes": 0, "commit_bytes": 0, "sampled_private_bytes": 0}
        with log.open("wb") as output:
            process = subprocess.Popen(command, cwd=ROOT, stdin=subprocess.DEVNULL,
                stdout=output, stderr=subprocess.STDOUT, env=env,
                creationflags=subprocess.CREATE_NO_WINDOW)
            handle = self.kernel.OpenProcess(0x1000 | 0x0010, False, process.pid)
            try:
                if not handle:
                    raise OSError("MEMORY_PROCESS_OPEN_FAILED")
                while process.poll() is None:
                    if time.perf_counter() - start > timeout:
                        raise TimeoutError("MEMORY_WORKLOAD_TIMEOUT")
                    counters = Counters()
                    counters.cb = c.sizeof(counters)
                    if self.psapi.GetProcessMemoryInfo(handle, c.byref(counters), counters.cb):
                        samples += 1
                        last = time.perf_counter() - start
                        if first is None:
                            first = last
                        peaks["working_set_bytes"] = max(peaks["working_set_bytes"], counters.PeakWorkingSetSize)
                        peaks["commit_bytes"] = max(peaks["commit_bytes"], counters.PeakPagefileUsage)
                        peaks["sampled_private_bytes"] = max(peaks["sampled_private_bytes"], counters.PrivateUsage)
                    else:
                        errors += 1
                    time.sleep(0.01)
                elapsed = time.perf_counter() - start
                return dict(exit_code=process.returncode, elapsed_seconds=elapsed,
                    samples=samples, read_errors=errors, first_sample_seconds=first,
                    last_sample_seconds=last, unobserved_tail_seconds=None if last is None else elapsed-last,
                    peaks=peaks, log=log.name)
            finally:
                if process.poll() is None:
                    process.kill()
                    process.wait(timeout=10)
                if handle:
                    self.kernel.CloseHandle(handle)

def digest(path):
    with path.open("rb") as file:
        return hashlib.file_digest(file, "sha256").hexdigest()

def build(directory):
    command = ["cargo", "test", "--manifest-path", "frontend/src-tauri/Cargo.toml",
        "--features", "desktop", "--lib", "--no-run", "--message-format=json"]
    with (directory / "build.log").open("wb") as error:
        result = subprocess.run(command, cwd=ROOT, stdout=subprocess.PIPE, stderr=error, check=True)
    executables = []
    for line in result.stdout.decode("utf-8").splitlines():
        try:
            message = json.loads(line)
        except ValueError:
            continue
        if message.get("reason") == "compiler-artifact" and message.get("profile", {}).get("test") and message.get("target", {}).get("name") == "notesagent_host" and message.get("executable"):
            executables.append(Path(message["executable"]))
    if len(executables) != 1:
        raise RuntimeError("MEMORY_TEST_EXECUTABLE_AMBIGUOUS")
    return executables[0]

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--runs", type=int, default=3)
    parser.add_argument("--output", type=Path, default=ROOT / ".build/sync-memory")
    args = parser.parse_args()
    if os.name != "nt" or not 1 <= args.runs <= 10:
        parser.error("Windows required; runs must be 1..10")
    directory = args.output.resolve()
    directory.mkdir(parents=True, exist_ok=True)
    report = dict(schema=1, created_at=datetime.now(timezone.utc).isoformat(),
        scope="Single Rust component test process; excludes WebView, Core, service and system file cache",
        acceptance="NOT_ASSESSED", polling_interval_seconds=0.01, runs=[],
        limitations=["OS cumulative peaks observed before exit; final sampling gap is reported",
            "Patterns and local hardware differ from the complete release benchmark",
            "No network/load/RSS acceptance threshold is inferred from these measurements"])
    try:
        exe = build(directory)
        source = hashlib.sha256()
        paths = sorted((ROOT / "frontend/src-tauri/src").rglob("*.rs")) + sorted((ROOT / "frontend/src-tauri/tests").rglob("*.rs"))
        for path in paths:
            source.update(str(path.relative_to(ROOT)).replace(chr(92), "/").encode())
            source.update(bytes.fromhex(digest(path)))
        report["build"] = dict(executable=exe.name, executable_sha256=digest(exe),
            sources_sha256=source.hexdigest(), cargo_lock_sha256=digest(ROOT / "frontend/src-tauri/Cargo.lock"),
            git_head=subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip(),
            rustc=subprocess.check_output(["rustc", "--version"], text=True).strip())
        import winreg
        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"HARDWARE\DESCRIPTION\System\CentralProcessor\0") as key:
            cpu = winreg.QueryValueEx(key, "ProcessorNameString")[0].strip()
        report["machine"] = dict(os=platform.platform(), cpu=cpu, logical_processors=os.cpu_count(),
            python=platform.python_version(), disk="not measured", network="not exercised")
        monitor = Monitor()
        calibration = monitor.run([sys.executable, "-c",
            "import time; data=bytearray(64*1024*1024); time.sleep(0.3)"], directory / "calibration.log")
        report["calibration"] = calibration
        if calibration["exit_code"] != 0 or calibration["peaks"]["working_set_bytes"] < 64*1024*1024 or calibration["peaks"]["commit_bytes"] < 64*1024*1024:
            raise RuntimeError("MEMORY_CALIBRATION_FAILED")
        for name, test in WORKLOADS.items():
            for iteration in range(1, args.runs + 1):
                log = directory / f"{name}-{iteration}.log"
                value = monitor.run([str(exe), "--exact", test, "--nocapture", "--test-threads=1"], log)
                value.update(workload=name, test=test, iteration=iteration)
                output = log.read_text(encoding="utf-8", errors="replace")
                value["passed"] = value["exit_code"] == 0 and value["samples"] > 0 and bool(re.search(r"test result: ok\. 1 passed; 0 failed; 0 ignored;", output))
                report["runs"].append(value)
                print(json.dumps(dict(workload=name, iteration=iteration, passed=value["passed"], peaks=value["peaks"])), flush=True)
                if not value["passed"]:
                    raise RuntimeError("MEMORY_WORKLOAD_FAILED")
        report["summary"] = {name: {
            metric: dict(max=max(values), median=statistics.median(values), min=min(values))
            for metric in ("working_set_bytes", "commit_bytes")
            for values in [[item["peaks"][metric] for item in report["runs"] if item["workload"] == name]]
        } for name in WORKLOADS}
        report["measurement_completed"] = True
    finally:
        (directory / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2)+"\n", encoding="utf-8")

if __name__ == "__main__":
    main()