minimal-run-and-audit

어떤 스킬인가요?

README 우선 딥러닝 저장소 재현을 위한 Rigor Run 스킬입니다. 선택한 스모크 테스트나 문서화된 추론 또는 평가 명령어에서 증거를 구체적으로 포착하거나 정규화하고, 저장소 파일이 변경된 경우 패치 노트를 포함하여 표준화된 `repro_outputs/` 파일을 작성하는 것이 주된 작업일 때 사용하십시오. 이 스킬은 훈련 실행, 초기 저장소 인수, 일반적인 환경 설정, 논문 검색, 대상 선택, 과학적 의미가 내포된 변경 사항, 또는 이 스킬만으로는 엔드투엔드 오케스트레이션에 사용해서는 안 됩니다.

설치 방법

프로젝트 폴더에서 아래 명령어를 실행하고, 설치할 에이전트를 선택하세요.

설치 명령을 직접 실행해야 적용됩니다. 지원 에이전트와 필요한 권한·라이선스는 제작자의 안내를 확인하세요.

스킬 원문 파일

display_name: Rigor Run
short_description: Rigor Run mode for selected inference, evaluation, smoke, or sanity execution evidence.
default_prompt: Run the selected smoke, inference, evaluation, or sanity command conservatively, capture execution evidence, and write SUMMARY.md COMMANDS.md LOG.md and status.json.

# Reporting Policy

## Tone

Keep reports short, factual, and easy to audit.

## Requirements

- separate facts from inferences
- mention the documented command explicitly
- mention whether the non-training run was full, partial, smoke-only, sanity-only, or blocked
- explain the main blocker without burying it
- when patches were applied, mention patch state briefly in `SUMMARY.md` and keep the full audit in `PATCHES.md`

## Output priorities

1. clear overall result
2. copyable commands
3. concise process trace
4. stable machine-readable state
5. patch evidence when relevant

## Avoid

- long narrative journals
- vague "it should work" language
- hiding unsupported assumptions
- treating training startup or resume as part of this skill

#!/usr/bin/env python3
"""Execute a short non-training command and normalize the evidence."""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple


SHARED_SCRIPTS = Path(__file__).resolve().parents[3] / "shared" / "scripts"
if not all((SHARED_SCRIPTS / name).is_file() for name in (
    "runtime_runner.py", "model_adapter.py", "command_utils.py", "resource_monitor.py"
)):
    SHARED_SCRIPTS = (Path(__file__).resolve().parents[2] / "ai-research-reproduction"
                      / "_bundled" / "shared" / "scripts")
if not (SHARED_SCRIPTS / "model_adapter.py").is_file():
    raise RuntimeError("Shared runtime missing: install all RigorPilot skills, including ai-research-reproduction.")
if str(SHARED_SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SHARED_SCRIPTS))

from runtime_runner import run_persistent_command
from model_adapter import ModelAdapterError, load_model_profile, missing_capabilities


METRIC_RE = re.compile(
    r"\b([A-Za-z][A-Za-z0-9_.-]{1,31})\s*[:=]\s*(-?\d+(?:\.\d+)?(?:[eE][+-]?\d+)?)"
)


def combine_logs(parts: Iterable[str]) -> str:
    return "\n".join(part for part in parts if part).strip()


def decode_stream(value: Any) -> str:
    # On POSIX, subprocess.TimeoutExpired carries captured output as bytes
    # even when the run was started with text=True.
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return value or ""


def parse_metrics(text: str) -> Dict[str, Any]:
    observed_metrics: Dict[str, float] = {}
    best_metric: Optional[Dict[str, Any]] = None

    for match in METRIC_RE.finditer(text):
        name = match.group(1)
        value = float(match.group(2))
        observed_metrics[name] = value

    priority_names = [
        name for name in observed_metrics
        if not any(token in name.lower() for token in {"loss", "lr", "time", "mem", "epoch", "step", "iter", "iteration"})
    ]
    if priority_names:
        chosen = priority_names[-1]
        best_metric = {"name": chosen, "value": observed_metrics[chosen]}
    elif observed_metrics:
        chosen = list(observed_metrics)[-1]
        best_metric = {"name": chosen, "value": observed_metrics[chosen]}

    return {
        "observed_metrics": observed_metrics,
        "best_metric": best_metric,
    }


def run_git(repo: Path, args: List[str]) -> subprocess.CompletedProcess[str]:
    try:
        return subprocess.run(
            ["git", *args],
            cwd=repo,
            capture_output=True,
            text=True,
            timeout=15,
            check=False,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired) as exc:
        # A missing or hanging git binary must degrade to the documented
        # "git-unavailable" evidence path, not crash the runner.
        return subprocess.CompletedProcess(["git", *args], returncode=127, stdout="", stderr=str(exc))


def git_status_snapshot(repo: Path) -> Tuple[Optional[Dict[str, str]], Dict[str, Any]]:
    probe = run_git(repo, ["rev-parse", "--is-inside-work-tree"])
    if probe.returncode != 0 or probe.stdout.strip() != "true":
        return None, {
            "collection_method": "git-status-diff",
            "available": False,
            "reason": "git-unavailable-or-not-a-worktree",
        }

    result = run_git(repo, ["status", "--porcelain=v1", "--untracked-files=all"])
    if result.returncode != 0:
        return None, {
            "collection_method": "git-status-diff",
            "available": False,
            "reason": "git-status-failed",
            "stderr": result.stderr.strip(),
        }

    snapshot: Dict[str, str] = {}
    for raw_line in result.stdout.splitlines():
        line = raw_line.rstrip()
        if len(line) < 4:
            continue
        status = line[:2]
        path = line[3:]
        if " -> " in path:
            _old, _arrow, path = path.partition(" -> ")
        normalized = path.replace("\\", "/").strip()
        if normalized:
            snapshot[normalized] = status
    return snapshot, {
        "collection_method": "git-status-diff",
        "available": True,
        "status_entries": len(snapshot),
    }


def diff_status_snapshots(
    before: Optional[Dict[str, str]],
    after: Optional[Dict[str, str]],
) -> Dict[str, List[str]]:
    if before is None or after is None:
        return {
            "changed_files": [],
            "new_files": [],
            "deleted_files": [],
            "touched_paths": [],
            "touched_symbols": [],
        }

    changed_files: List[str] = []
    new_files: List[str] = []
    deleted_files: List[str] = []

    for path, status in after.items():
        previous_status = before.get(path)
        if previous_status == status:
            continue
        normalized_status = status.replace(" ", "")
        if "D" in normalized_status:
            deleted_files.append(path)
            continue
        if "?" in normalized_status or "A" in normalized_status:
            new_files.append(path)
            continue
        changed_files.append(path)

    touched_paths = []
    for path in [*changed_files, *new_files, *deleted_files]:
        if path not in touched_paths:
            touched_paths.append(path)
    return {
        "changed_files": changed_files,
        "new_files": new_files,
        "deleted_files": deleted_files,
        "touched_paths": touched_paths,
        "touched_symbols": [],
    }


def exclude_runtime_snapshot(
    repo: Path,
    runtime_dir: Path,
    snapshot: Optional[Dict[str, str]],
) -> Optional[Dict[str, str]]:
    if snapshot is None:
        return None
    try:
        prefix = runtime_dir.resolve().relative_to(repo.resolve()).as_posix().rstrip("/") + "/"
    except ValueError:
        return snapshot
    return {path: status for path, status in snapshot.items() if not path.startswith(prefix)}


def execute_command(
    repo: Path,
    command: str,
    timeout: int,
    shell_mode: str = "direct",
    runtime_root: Optional[Path] = None,
    model_adapter: Optional[Dict[str, Any]] = None,
    monitor_gpu: bool = False,
) -> Dict[str, Any]:
    before_status, before_capture = git_status_snapshot(repo)
    selected_runtime_root = (runtime_root or (repo / "repro_outputs" / "_runtime")).resolve()
    execution = run_persistent_command(
        repo=repo,
        command=command,
        timeout=timeout,
        runtime_root=selected_runtime_root,
        shell_mode=shell_mode,
        model_adapter=model_adapter,
        monitor_gpu=monitor_gpu,
    )
    after_status, after_capture = git_status_snapshot(repo)
    after_status = exclude_runtime_snapshot(repo, Path(execution["runtime_dir"]), after_status)
    if after_status is not None:
        after_capture["status_entries"] = len(after_status)
        after_capture["runtime_artifacts_excluded"] = True
    execution.update(diff_status_snapshots(before_status, after_status))
    execution["evidence_capture"] = {
        **after_capture,
        "before_status_entries": before_capture.get("status_entries"),
    }
    return execution


def decide_outcome(command: str, timeout: int, execution: Dict[str, Any], metric_data: Dict[str, Any]) -> Dict[str, Any]:
    combined_text = combine_logs(
        [
            f"STDOUT:\n{execution['stdout'].strip()}" if execution.get("stdout", "").strip() else "",
            f"STDERR:\n{execution['stderr'].strip()}" if execution.get("stderr", "").strip() else "",
        ]
    )

    if execution.get("launch_error"):
        return {
            "status": "blocked",
            "documented_command_status": "blocked",
            "main_blocker": f"Executable not found for command: {execution['launch_error']}",
            "execution_log": [f"Command failed before launch: {execution['launch_error']}"],
            "monitoring_scope": "no_run",
        }

    if execution.get("cancelled"):
        return {
            "status": "partial",
            "documented_command_status": "partial",
            "main_blocker": "The selected command was cancelled through the runtime control file.",
            "execution_log": [combined_text] if combined_text else ["Command cancelled."],
            "monitoring_scope": "runtime_cancel",
        }

    if execution.get("timed_out"):
        return {
            "status": "partial",
            "documented_command_status": "partial",
            "main_blocker": f"Selected command did not finish within {timeout} seconds.",
            "execution_log": [combined_text or f"Command timed out after {timeout} seconds."],
            "monitoring_scope": f"timeout:{timeout}s",
        }

    if execution.get("returncode") == 0:
        return {
            "status": "success",
            "documented_command_status": "success",
            "main_blocker": "None.",
            "execution_log": [combined_text] if combined_text else [],
            "monitoring_scope": "process_completion",
        }

    return {
        "status": "partial",
        "documented_command_status": "partial",
        "main_blocker": f"Selected command exited with code {execution.get('returncode')}.",
        "execution_log": [combined_text] if combined_text else [f"Command `{command}` exited non-zero."],
        "monitoring_scope": "process_completion",
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Run a short non-training command and summarize the evidence.")
    parser.add_argument("--repo", required=True, help="Path to the target repository.")
    parser.add_argument("--command", required=True, help="Command to execute.")
    parser.add_argument("--timeout", type=int, default=60, help="Execution timeout in seconds.")
    parser.add_argument(
        "--shell-mode",
        choices=["direct", "native"],
        default="direct",
        help="Use direct argv execution by default; native shell execution requires explicit opt-in.",
    )
    parser.add_argument(
        "--runtime-root",
        default="",
        help="Directory for persistent runtime state and streamed logs (default: <repo>/repro_outputs/_runtime).",
    )
    parser.add_argument("--model-profile-json", default="", help="Optional provider-neutral model identity/capability profile.")
    parser.add_argument(
        "--require-model-capability",
        action="append",
        default=[],
        help="Required model capability; repeat as needed.",
    )
    parser.add_argument("--monitor-gpu", action="store_true", help="Sample NVIDIA device-level telemetry when available.")
    args = parser.parse_args()
    if args.timeout <= 0:
        parser.error("--timeout must be greater than zero")

    repo = Path(args.repo).resolve()
    runtime_root = Path(args.runtime_root).resolve() if args.runtime_root else None
    try:
        model_adapter = load_model_profile(Path(args.model_profile_json) if args.model_profile_json else None)
        missing = missing_capabilities(model_adapter, args.require_model_capability)
    except ModelAdapterError as exc:
        parser.error(str(exc))
    if missing:
        parser.error(f"model profile is missing required capabilities: {', '.join(missing)}")
    execution = execute_command(
        repo,
        args.command,
        args.timeout,
        args.shell_mode,
        runtime_root,
        model_adapter,
        args.monitor_gpu,
    )
    metric_data = parse_metrics(combine_logs([execution.get("stdout", ""), execution.get("stderr", "")]))
    outcome = decide_outcome(args.command, args.timeout, execution, metric_data)

    payload = {
        "status": outcome["status"],
        "documented_command_status": outcome["documented_command_status"],
        "main_blocker": outcome["main_blocker"],
        "execution_log": outcome["execution_log"],
        "monitoring_scope": outcome["monitoring_scope"],
        "execution_mode": execution.get("execution_mode", args.shell_mode),
        "runtime_run_id": execution.get("runtime_run_id"),
        "runtime_dir": execution.get("runtime_dir"),
        "runtime_status": execution.get("runtime_status"),
        "runtime_state_path": execution.get("runtime_state_path"),
        "runtime_events_path": execution.get("runtime_events_path"),
        "stdout_log_path": execution.get("stdout_log_path"),
        "stderr_log_path": execution.get("stderr_log_path"),
        "stdout_truncated": execution.get("stdout_truncated", False),
        "stderr_truncated": execution.get("stderr_truncated", False),
        "cancelled": execution.get("cancelled", False),
        "duration_seconds": execution.get("duration_seconds"),
        "runtime_attempt": execution.get("runtime_attempt", 1),
        "runtime_retry_of": execution.get("runtime_retry_of"),
        "resources_log_path": execution.get("resources_log_path"),
        "resource_summary": execution.get("resource_summary", {}),
        "model_adapter": execution.get("model_adapter"),
        "best_metric": metric_data["best_metric"],
        "observed_metrics": metric_data["observed_metrics"],
        "changed_files": execution.get("changed_files", []),
        "new_files": execution.get("new_files", []),
        "deleted_files": execution.get("deleted_files", []),
        "touched_paths": execution.get("touched_paths", []),
        "touched_symbols": execution.get("touched_symbols", []),
        "evidence_capture": execution.get("evidence_capture", {}),
    }
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    return 0


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

#!/usr/bin/env python3
"""Compatibility wrapper for trusted verify output bundles."""

from __future__ import annotations

import importlib.util
from pathlib import Path


def load_shared_module():
    module_path = Path(__file__).resolve().parents[3] / "shared" / "scripts" / "write_run_bundle.py"
    if not module_path.is_file():
        module_path = (Path(__file__).resolve().parents[2] / "ai-research-reproduction"
                       / "_bundled" / "shared" / "scripts" / "write_run_bundle.py")
    if not module_path.is_file():
        raise RuntimeError("Shared writer missing: install all RigorPilot skills, including ai-research-reproduction.")
    spec = importlib.util.spec_from_file_location("write_run_bundle", module_path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"Unable to load shared writer module from {module_path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def main() -> int:
    module = load_shared_module()
    return module.main(default_mode="repro", default_output_dir="repro_outputs")


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

---
name: minimal-run-and-audit
description: Rigor Run skill for README-first deep learning repo reproduction. Use when the task is specifically to capture or normalize evidence from the selected smoke test or documented inference or evaluation command and write standardized `repro_outputs/` files, including patch notes when repository files changed. Do not use for training execution, initial repo intake, generic environment setup, paper lookup, target selection, hidden scientific-meaning changes, or end-to-end orchestration by itself.
---

# minimal-run-and-audit

Use this as the Rigor Run skill. The installed slug remains
`minimal-run-and-audit` for compatibility.

Use the shared operating principles in
`../ai-research-reproduction/references/agent-operating-principles.md`; this skill should make run
evidence auditable without turning every command into a rigid protocol.

## When to apply

- After a reproduction target and setup plan exist.
- When the main skill needs execution evidence and normalized outputs.
- When a smoke test, documented inference run, documented evaluation run, or other short non-training verification is appropriate.
- When the user already knows what command should be attempted and wants execution plus reporting only.

## When not to apply

- During initial repo scanning.
- When environment or assets are still undefined enough to make execution meaningless.
- When the task is a literature lookup rather than repository execution.
- When the user is still deciding which reproduction target should count as the main run.

## Clear boundaries

- This skill owns normalized reporting for an attempted command.
- It may receive execution evidence from the main skill or a thin helper.
- It does not choose the overall target on its own.
- It does not perform broad paper analysis.
- It does not own training startup, resume, or long-running training state.
- It should not normalize risky code edits into acceptable practice.
- It must not hide changes that alter evaluation, preprocessing, checkpoints,
  metrics, or other scientific meaning.

## Input expectations

- selected reproduction goal
- runnable commands or smoke commands
- environment and asset assumptions
- optional patch metadata

## Output expectations

- execution result summary
- standardized `repro_outputs/` files
- `SCIENTIFIC_CHANGELOG.md` for changed scientific meaning and evidence status
- `COMPARABILITY_REPORT.md` for README/paper/baseline comparability
- clear distinction between verified, partial, and blocked states
- `PATCHES.md` when repo files changed

## Notes

Use `references/reporting-policy.md`, `../ai-research-reproduction/references/research-rigor-principles.md`, `scripts/run_command.py`, and `scripts/write_outputs.py`.