govoplan/tools/devkit/govoplan_devkit/monitoring.py

"""Read-only run discovery and bounded provisional output, never passing evidence."""

from __future__ import annotations

from datetime import datetime
import os
from pathlib import Path
import re
import time

from .common import (
    atomic_json,
    identifier,
    now,
    read_json,
    redact,
    reject_symlinks,
    state_root,
)

MAX_LIVE_BYTES = 65536
MAX_HISTORY_ENTRIES = 10000
MAX_HISTORY_SCAN = 500
ANSI = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\))")
CONTROLS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f]")


def display_text(value: str) -> str:
    return redact(CONTROLS.sub("", ANSI.sub("", value)))


def _closed_lines(value: bytes) -> bytes:
    boundary = value.rfind(b"\n")
    return value[: boundary + 1] if boundary >= 0 else b""


def capture_text(snapshot, *, provisional: bool = False) -> str:
    """Do not expose partial secret/context fragments at live or retention boundaries."""
    data = snapshot.stdout
    omitted = snapshot.omitted_stdout_bytes
    if omitted:
        head = _closed_lines(data[: snapshot.stdout_head_bytes])
        tail = data[snapshot.stdout_head_bytes :]
        # A rolling tail can start inside Authorization or a secret: drop that fragment.
        tail = tail.partition(b"\n")[2]
        if provisional:
            tail = _closed_lines(tail)
        return (
            display_text(head.decode("utf-8", errors="replace"))
            + f"\n[Output truncated: {omitted} bytes omitted between retained head and tail; cut boundary lines are withheld.]\n"
            + display_text(tail.decode("utf-8", errors="replace"))
        )
    if provisional:
        data = _closed_lines(data)
    return display_text(data.decode("utf-8", errors="replace"))


def bounded_display(value: str, maximum: int, *, tail_only: bool = False) -> str:
    if type(maximum) is not int or maximum < 1:
        raise ValueError("Display bound must be a positive integer")
    encoded = value.encode("utf-8")
    if len(encoded) <= maximum:
        return value
    if tail_only:
        return encoded[-maximum:].decode("utf-8", errors="ignore")
    marker = b"\n[Display bound: middle omitted; retained beginning and final output follow.]\n"
    if maximum <= len(marker):
        return encoded[-maximum:].decode("utf-8", errors="ignore")
    available = max(0, maximum - len(marker))
    head = available // 2
    return (
        encoded[:head].decode("utf-8", errors="ignore")
        + marker.decode()
        + encoded[-(available - head) :].decode("utf-8", errors="ignore")
    )


def write_live(log_path: Path, stage_id: str, snapshot, started: float) -> None:
    # Complete lines only, even at final callback: the separately finalized log
    # may include an unterminated final line after normal redaction.
    excerpt = bounded_display(
        capture_text(snapshot, provisional=True), MAX_LIVE_BYTES, tail_only=True
    )
    atomic_json(
        log_path.with_suffix(".live.json"),
        {
            "schema_version": 1,
            "run_id": log_path.parent.name,
            "stage_id": stage_id,
            "provisional": True,
            "updated_at": now(),
            "elapsed_seconds": round(time.monotonic() - started, 3),
            "output_truncated": snapshot.truncated,
            "omitted_bytes": snapshot.omitted_stdout_bytes,
            "excerpt": excerpt,
        },
    )


def read_live(
    workspace_root: Path, state_dir: Path | None, run_id: str, stage_id: str
) -> dict:
    identifier(run_id)
    identifier(stage_id)
    path = (
        state_root(workspace_root, state_dir)
        / "runs"
        / run_id
        / (stage_id + ".live.json")
    )
    if not path.exists() and not path.is_symlink():
        return {"provisional": True, "excerpt": "", "live_available": False}
    payload = read_json(path, max_bytes=MAX_LIVE_BYTES * 6 + 8192)
    if (
        not isinstance(payload, dict)
        or payload.get("schema_version") != 1
        or payload.get("run_id") != run_id
        or payload.get("stage_id") != stage_id
        or payload.get("provisional") is not True
        or not isinstance(payload.get("excerpt"), str)
        or len(payload["excerpt"].encode("utf-8")) > MAX_LIVE_BYTES
        or not isinstance(payload.get("updated_at"), str)
        or type(payload.get("elapsed_seconds")) not in {int, float}
        or payload["elapsed_seconds"] < 0
        or type(payload.get("output_truncated")) is not bool
    ):
        raise ValueError("Invalid provisional log snapshot")
    return {
        "provisional": True,
        "live_available": True,
        "excerpt": display_text(payload["excerpt"]),
        "updated_at": payload.get("updated_at"),
        "elapsed_seconds": payload.get("elapsed_seconds"),
        "output_truncated": payload.get("output_truncated"),
    }


def elapsed_seconds(receipt: dict) -> float | None:
    try:
        started = datetime.fromisoformat(receipt["generated_at"])
        finished = datetime.fromisoformat(receipt.get("finished_at") or now())
        return max(0, round((finished - started).total_seconds(), 3))
    except (KeyError, TypeError, ValueError):
        return None


def list_runs(args) -> dict:
    from .runner import read_receipt

    limit = getattr(args, "limit", 10)
    before = getattr(args, "before", None)
    if type(limit) is not int or not 1 <= limit <= 100:
        raise ValueError("Run history limit must be between 1 and 100")
    if before:
        identifier(before)
    base = state_root(args.workspace_root, args.state_dir) / "runs"
    reject_symlinks(base)
    names = []
    if base.exists():
        with os.scandir(base) as entries:
            for index, entry in enumerate(entries):
                if index >= MAX_HISTORY_ENTRIES:
                    raise ValueError(
                        "Run history exceeds its directory bound; archive old evidence explicitly before listing"
                    )
                if not entry.name.startswith("."):
                    identifier(entry.name)
                    names.append(entry.name)
    names = sorted(
        (name for name in names if before is None or name < before), reverse=True
    )
    rows, examined, cursor = [], 0, None
    wanted_project = (
        str(args.project.resolve()) if getattr(args, "project", None) else None
    )
    for identity in names:
        examined += 1
        cursor = identity
        try:
            record = read_receipt(args.workspace_root, args.state_dir, identity)
            if wanted_project and record.get("project_file") != wanted_project:
                if examined >= MAX_HISTORY_SCAN:
                    break
                continue
            rows.append(
                {
                    "run_id": identity,
                    "status": record["status"],
                    "phase": record.get("phase"),
                    "profile": record.get("profile"),
                    "generated_at": record.get("generated_at"),
                    "elapsed_seconds": elapsed_seconds(record),
                    "snapshot_verified": record["snapshot_verified"],
                    "passed_stages": sum(
                        item["status"] == "passed" for item in record["stages"]
                    ),
                    "total_stages": len(record["stages"]),
                }
            )
        except (OSError, ValueError) as exc:
            # Corrupt/newest evidence is visible, never silently replaced with an older pass.
            rows.append(
                {
                    "run_id": display_text(identity),
                    "status": "invalid",
                    "error": display_text(str(exc)),
                }
            )
        if len(rows) >= limit or examined >= MAX_HISTORY_SCAN:
            break
    next_cursor = cursor if examined < len(names) else None
    lines = [
        f"{item['run_id']}: {item['status']}"
        + (
            f" ({item.get('profile')}; {item.get('elapsed_seconds')}s)"
            if item["status"] != "invalid"
            else " — " + item["error"]
        )
        for item in rows
    ]
    if not lines:
        lines = ["No matching check runs found; no verification is implied."]
    if next_cursor:
        lines.append("More history: use --before " + display_text(next_cursor))
    return {
        "runs": rows,
        "next_cursor": next_cursor,
        "examined": examined,
        "summary": lines,
        "_exit_code": 1 if any(item["status"] == "invalid" for item in rows) else 0,
    }


def latest_run(args) -> dict:
    from argparse import Namespace
    from .runner import read_receipt, summarize

    selected = list_runs(Namespace(**{**vars(args), "limit": 1}))
    if not selected["runs"]:
        return {
            "status": "not_found",
            "summary": selected["summary"],
            "next_cursor": selected["next_cursor"],
            "_exit_code": 1,
        }
    row = selected["runs"][0]
    if row["status"] == "invalid":
        raise ValueError(
            "Latest run is invalid; inspect run history instead of assuming an older pass"
        )
    return summarize(read_receipt(args.workspace_root, args.state_dir, row["run_id"]))