226 lines
7.1 KiB
Python
226 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import Counter
|
|
from datetime import datetime, timedelta
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
def _read_safe(root: Path, relative: str) -> str | None:
|
|
"""Read a file if it exists, otherwise return None."""
|
|
path = root / relative
|
|
if path.exists():
|
|
return path.read_text(encoding="utf-8")
|
|
return None
|
|
|
|
|
|
def _last_modified(root: Path, relative: str) -> datetime | None:
|
|
"""Return the last-modified time of a file, or None."""
|
|
path = root / relative
|
|
if path.exists():
|
|
ts = path.stat().st_mtime
|
|
return datetime.fromtimestamp(ts)
|
|
return None
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Individual guardrail checks
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def check_repeated_reads(root: Path) -> dict:
|
|
"""Detect files that appear to be read repeatedly in AGENT_HANDOFF.md or project context.
|
|
|
|
Returns a dict with keys: status, details.
|
|
- status: 'ok', 'warning', 'error'
|
|
- details: human-readable explanation
|
|
"""
|
|
handoff_text = _read_safe(root, "AGENT_HANDOFF.md")
|
|
prompt_log = _read_safe(root, ".rdb/prompt-history.md") # optional
|
|
|
|
text_parts = []
|
|
if handoff_text is not None:
|
|
text_parts.append(handoff_text)
|
|
if prompt_log is not None:
|
|
text_parts.append(prompt_log)
|
|
|
|
if not text_parts:
|
|
return {"status": "ok", "details": "No agent history files to analyse."}
|
|
|
|
combined = "\n".join(text_parts)
|
|
file_mentions = re.findall(r"[A-Za-z0-9_/.-]+\.md", combined)
|
|
counts = Counter(file_mentions)
|
|
|
|
repeated = {path: count for path, count in counts.items() if count > 2}
|
|
|
|
if not repeated:
|
|
return {"status": "ok", "details": "No files appear to be read excessively."}
|
|
|
|
top_files = ", ".join(f"{p} ({c}x)" for p, c in list(repeated.items())[:5])
|
|
return {
|
|
"status": "warning",
|
|
"details": f"Files mentioned more than twice: {top_files}",
|
|
}
|
|
|
|
|
|
def check_repeated_commands(root: Path) -> dict:
|
|
"""Detect repeated command entries in RUN_LOG.md.
|
|
|
|
Returns a dict with keys: status, details.
|
|
"""
|
|
run_log = _read_safe(root, "RUN_LOG.md")
|
|
if not run_log:
|
|
return {"status": "ok", "details": "No RUN_LOG.md found."}
|
|
|
|
lines = [l.strip() for l in run_log.splitlines() if l.startswith("|")]
|
|
events = []
|
|
for line in lines:
|
|
parts = [p.strip() for p in line.split("|")][1:-1]
|
|
if len(parts) >= 2:
|
|
events.append(f"{parts[0]}|{parts[1]}")
|
|
|
|
counts = Counter(events)
|
|
repeated = {k: v for k, v in counts.items() if v > 2}
|
|
|
|
if not repeated:
|
|
return {"status": "ok", "details": "No significantly repeated commands detected."}
|
|
|
|
items = ", ".join(f"'{k}' ({v}x)" for k, v in list(repeated.items())[:5])
|
|
return {
|
|
"status": "warning",
|
|
"details": f"Repeated entries: {items}",
|
|
}
|
|
|
|
|
|
def check_no_recent_file_changes(root: Path) -> dict:
|
|
"""Flag if project files have not been modified in a long time (> 48 h).
|
|
|
|
Returns a dict with keys: status, details.
|
|
"""
|
|
monitored = [
|
|
"TASKS.md",
|
|
"PROJECT_STATE.md",
|
|
"AGENT_HANDOFF.md",
|
|
"context/product-brief.md",
|
|
"context/architecture.md",
|
|
]
|
|
now = _now()
|
|
stale: list[str] = []
|
|
|
|
for rel in monitored:
|
|
ts = _last_modified(root, rel)
|
|
if ts is None:
|
|
continue
|
|
age = now - ts
|
|
if age > timedelta(hours=48):
|
|
stale.append(rel)
|
|
|
|
if not stale:
|
|
return {"status": "ok", "details": "Project files have recent updates."}
|
|
|
|
return {
|
|
"status": "warning",
|
|
"details": f"No changes in the last 48 hours: {', '.join(stale)}",
|
|
}
|
|
|
|
|
|
def check_no_test_run_recorded(root: Path) -> dict:
|
|
"""Flag if RUN_LOG.md does not mention tests or pytest.
|
|
|
|
Returns a dict with keys: status, details.
|
|
"""
|
|
run_log = _read_safe(root, "RUN_LOG.md")
|
|
if not run_log:
|
|
return {"status": "ok", "details": "No RUN_LOG.md found — can't check."}
|
|
|
|
lower = run_log.lower()
|
|
keywords = ["pytest", "test passed", "tests passed", "all tests"]
|
|
if any(kw in lower for kw in keywords):
|
|
return {"status": "ok", "details": "Test activity recorded."}
|
|
|
|
return {
|
|
"status": "warning",
|
|
"details": "No test execution recorded in RUN_LOG.md.",
|
|
}
|
|
|
|
|
|
def check_run_log_updated(root: Path) -> dict:
|
|
"""Check whether TASKS.md status changes are reflected in RUN_LOG.md.
|
|
|
|
Returns a dict with keys: status, details.
|
|
"""
|
|
tasks_text = _read_safe(root, "TASKS.md")
|
|
run_log = _read_safe(root, "RUN_LOG.md")
|
|
|
|
if not tasks_text or not run_log:
|
|
return {"status": "ok", "details": "Cannot compare — missing files."}
|
|
|
|
log_lines = [l.strip() for l in run_log.splitlines() if l.startswith("|")]
|
|
if not log_lines:
|
|
return {
|
|
"status": "warning",
|
|
"details": "RUN_LOG.md exists but contains no log entries.",
|
|
}
|
|
|
|
# Check done tasks have corresponding RUN_LOG entries
|
|
done_tasks = re.findall(r"(TASK-\d+).*?Status:\s*Done", tasks_text, re.DOTALL)
|
|
for task_id in done_tasks:
|
|
found = any(task_id in line for line in log_lines)
|
|
if not found:
|
|
return {
|
|
"status": "warning",
|
|
"details": f"Task {task_id} marked Done but no corresponding RUN_LOG entry found.",
|
|
}
|
|
|
|
return {"status": "ok", "details": "TASKS.md and RUN_LOG.md appear consistent."}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run_all_guardrails(root: Path) -> dict:
|
|
"""Run all guardrail checks and return combined results."""
|
|
checks = {
|
|
"Repeated reads": check_repeated_reads(root),
|
|
"Repeated commands": check_repeated_commands(root),
|
|
"No recent file changes": check_no_recent_file_changes(root),
|
|
"No test run recorded": check_no_test_run_recorded(root),
|
|
"Run log updated": check_run_log_updated(root),
|
|
}
|
|
|
|
statuses = [v["status"] for v in checks.values()]
|
|
if "error" in statuses:
|
|
overall = "action required"
|
|
elif "warning" in statuses:
|
|
overall = "review recommended"
|
|
else:
|
|
overall = "all clear"
|
|
|
|
return {"checks": checks, "overall_status": overall}
|
|
|
|
|
|
def format_report(results: dict) -> str:
|
|
"""Format guardrail results as a human-readable report."""
|
|
lines = ["# Guardrail Report", ""]
|
|
|
|
for name, result in results["checks"].items():
|
|
icon_map = {"ok": "[green]✓[/green]", "warning": "[yellow]⚠[/yellow]", "error": "[red]✗[/red]"}
|
|
icon = icon_map.get(result["status"], "?")
|
|
lines.append(f"- {icon} **{name}:** {result['details']}")
|
|
|
|
status_icon_map = {
|
|
"all clear": "[green]✓[/green]",
|
|
"review recommended": "[yellow]⚠[/yellow]",
|
|
"action required": "[red]✗[/red]",
|
|
}
|
|
icon = status_icon_map.get(results["overall_status"], "?")
|
|
|
|
lines.append("")
|
|
lines.append(f"**Overall status:** {icon} {results['overall_status']}")
|
|
|
|
return "\n".join(lines)
|