harness-tools/skills/document/artifact-presenter/scripts/present.py

#!/usr/bin/env python3
"""Open one review artifact in a reusable right-side Herdr mcat pane."""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Mapping, Sequence

PANE_LABEL = "artifact presenter"

RunCommand = Callable[..., subprocess.CompletedProcess[str]]
FindExecutable = Callable[[str], str | None]


@dataclass(frozen=True)
class PresentationResult:
    status: str
    artifact: Path
    pane_id: str | None = None
    reason: str | None = None

    def message(self) -> str:
        if self.status == "artifact-only":
            return f"artifact-only: {self.reason}; {self.artifact}"
        return f"{self.status}: pane {self.pane_id}; {self.artifact}"


def run_json(
    executable: str,
    args: Sequence[str],
    run: RunCommand,
) -> dict[str, object]:
    completed = run(
        [executable, *args],
        check=True,
        capture_output=True,
        text=True,
    )
    value = json.loads(completed.stdout)
    if not isinstance(value, dict):
        raise ValueError("Herdr returned a non-object response")
    return value


def panes_with_label(response: Mapping[str, object]) -> list[str]:
    result = response.get("result")
    if not isinstance(result, dict):
        return []
    panes = result.get("panes")
    if not isinstance(panes, list):
        return []
    pane_ids: list[str] = []
    for pane in panes:
        if not isinstance(pane, dict) or pane.get("label") != PANE_LABEL:
            continue
        pane_id = pane.get("pane_id")
        if isinstance(pane_id, str) and pane_id:
            pane_ids.append(pane_id)
    return pane_ids


def created_pane_id(response: Mapping[str, object]) -> str:
    result = response.get("result")
    pane = result.get("pane") if isinstance(result, dict) else None
    pane_id = pane.get("pane_id") if isinstance(pane, dict) else None
    if not isinstance(pane_id, str) or not pane_id:
        raise ValueError("Herdr split response is missing result.pane.pane_id")
    return pane_id


def run_mcat(
    herdr: str,
    mcat: str,
    pane_id: str,
    artifact: Path,
    run: RunCommand,
) -> None:
    run(
        [herdr, "pane", "run", pane_id, mcat, str(artifact)],
        check=True,
        capture_output=True,
        text=True,
    )


def present_document(
    document: Path,
    *,
    env: Mapping[str, str] = os.environ,
    which: FindExecutable = shutil.which,
    run: RunCommand = subprocess.run,
) -> PresentationResult:
    artifact = document.expanduser().resolve()
    if not artifact.is_file():
        raise ValueError(f"artifact is not a file: {artifact}")

    herdr = which("herdr")
    mcat = which("mcat")
    workspace_id = env.get("HERDR_WORKSPACE_ID")
    source_pane_id = env.get("HERDR_PANE_ID")
    if not herdr:
        return PresentationResult("artifact-only", artifact, reason="herdr is unavailable")
    if not mcat:
        return PresentationResult("artifact-only", artifact, reason="mcat is unavailable")
    if not workspace_id or not source_pane_id:
        return PresentationResult("artifact-only", artifact, reason="not running in a Herdr pane")

    try:
        pane_list = run_json(herdr, ["pane", "list", "--workspace", workspace_id], run)
        existing_panes = panes_with_label(pane_list)
    except (json.JSONDecodeError, OSError, subprocess.SubprocessError, ValueError) as error:
        return PresentationResult("artifact-only", artifact, reason=f"could not list Herdr panes: {error}")

    for existing_pane in existing_panes:
        try:
            run_mcat(herdr, mcat, existing_pane, artifact, run)
            return PresentationResult("reused", artifact, pane_id=existing_pane)
        except (OSError, subprocess.SubprocessError):
            try:
                run(
                    [herdr, "pane", "rename", existing_pane, f"{PANE_LABEL} stale"],
                    check=False,
                    capture_output=True,
                    text=True,
                )
            except OSError:
                pass

    try:
        split = run_json(
            herdr,
            [
                "pane",
                "split",
                "--pane",
                source_pane_id,
                "--direction",
                "right",
                "--cwd",
                str(artifact.parent),
                "--no-focus",
            ],
            run,
        )
        pane_id = created_pane_id(split)
    except (json.JSONDecodeError, OSError, subprocess.SubprocessError, ValueError) as error:
        return PresentationResult("artifact-only", artifact, reason=f"could not create Herdr viewer: {error}")

    try:
        run(
            [herdr, "pane", "rename", pane_id, PANE_LABEL],
            check=True,
            capture_output=True,
            text=True,
        )
        run_mcat(herdr, mcat, pane_id, artifact, run)
        return PresentationResult("opened", artifact, pane_id=pane_id)
    except (OSError, subprocess.SubprocessError) as error:
        try:
            run(
                [herdr, "pane", "close", pane_id],
                check=False,
                capture_output=True,
                text=True,
            )
        except OSError:
            pass
        return PresentationResult("artifact-only", artifact, reason=f"could not initialize Herdr viewer: {error}")


def create_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("artifact", type=Path, help="Local file to present with mcat")
    return parser


def main() -> int:
    args = create_parser().parse_args()
    try:
        result = present_document(args.artifact)
    except ValueError as error:
        print(error, file=os.sys.stderr)
        return 2
    print(result.message())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())