From aee13b045e0b35f561779d79846eb17f8e107076 Mon Sep 17 00:00:00 2001 From: robbond Date: Tue, 2 Jun 2026 11:41:27 +0100 Subject: [PATCH] chore: mark TASK-006 complete --- TASKS.md | 36 +++++ TEST_PLAN.md | 14 ++ src/rdb_discovery/cli.py | 10 ++ src/rdb_discovery/guardrails.py | 225 +++++++++++++++++++++++++++++ tests/test_guardrails.py | 249 ++++++++++++++++++++++++++++++++ 5 files changed, 534 insertions(+) create mode 100644 src/rdb_discovery/guardrails.py create mode 100644 tests/test_guardrails.py diff --git a/TASKS.md b/TASKS.md index 4e4d72e..b211409 100644 --- a/TASKS.md +++ b/TASKS.md @@ -95,6 +95,25 @@ Acceptance Criteria: - Add/update tests - Run python -m pytest +## TASK-006 — Agent execution guardrails + +Status: Done + +Goal: +Detect agent stalls, repeated reads, long reasoning loops, and non-progressing execution. + +Acceptance Criteria: + +- `rdb guardrails` command exists ✓ +- Detects repeated file reads via AGENT_HANDOFF.md analysis ✓ +- Detects repeated command entries in RUN_LOG.md ✓ +- Flags project files not modified in >48 hours ✓ +- Flags missing test run records in RUN_LOG.md ✓ +- Checks TASKS.md ↔ RUN_LOG.md consistency ✓ +- Produces clear human-readable report with overall status ✓ +- All existing tests still pass (33 → 51) ✓ +- Guardrail-specific tests added (18 new tests) ✓ + ## TASK-007 — Improve generated agent prompts Status: Todo @@ -110,3 +129,20 @@ Acceptance Criteria: - Prompt tells agent to inspect first, then edit - Prompt limits scope to one small implementation step - Add/update tests + +## TASK-012 — Agent telemetry + +Goal: +Capture actual agent actions rather than inferring behaviour +from documentation and run logs. + +Examples: + +- file reads +- file writes +- command execution +- test execution +- git activity + +Output: +.rdb/session-log.jsonl diff --git a/TEST_PLAN.md b/TEST_PLAN.md index fbb813f..0ae7e1d 100644 --- a/TEST_PLAN.md +++ b/TEST_PLAN.md @@ -15,3 +15,17 @@ Expected result: Tests pass CLI help displays + +## Guardrails Validation + +Run: + +```bash +rdb guardrails +``` + +Expected result: + +A report listing at least 5 checks (repeated reads, repeated commands, +recent file changes, test run record, RUN_LOG consistency) with an +overall status summary. diff --git a/src/rdb_discovery/cli.py b/src/rdb_discovery/cli.py index 842f6d3..a04737f 100644 --- a/src/rdb_discovery/cli.py +++ b/src/rdb_discovery/cli.py @@ -12,6 +12,7 @@ from .handoff import build_handoff from .status import project_stage, task_counts, update_project_state, update_agent_handoff from .tasks import generate_agent_prompt, get_next_task, update_task_status from .templates import CONTEXT_FILES, write_file_if_missing +from .guardrails import run_all_guardrails, format_report app = typer.Typer(help="RDB discovery and delivery workflow CLI.") console = Console() @@ -186,3 +187,12 @@ def prompt() -> None: root = root_path() result = generate_agent_prompt(root) console.print(result) + + +@app.command() +def guardrails() -> None: + """Review agent runs for signs of non-progress (stalls, repeats, loops).""" + root = root_path() + results = run_all_guardrails(root) + report = format_report(results) + console.print(report) diff --git a/src/rdb_discovery/guardrails.py b/src/rdb_discovery/guardrails.py new file mode 100644 index 0000000..1eb3f42 --- /dev/null +++ b/src/rdb_discovery/guardrails.py @@ -0,0 +1,225 @@ +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) diff --git a/tests/test_guardrails.py b/tests/test_guardrails.py new file mode 100644 index 0000000..46a13b6 --- /dev/null +++ b/tests/test_guardrails.py @@ -0,0 +1,249 @@ +from pathlib import Path +from datetime import datetime, timedelta + +import os +import pytest + +from rdb_discovery.guardrails import ( + check_repeated_reads, + check_repeated_commands, + check_no_recent_file_changes, + check_no_test_run_recorded, + check_run_log_updated, + run_all_guardrails, + format_report, +) + + +# -- check_repeated_reads -- + +def test_check_repeated_reads_ok(tmp_path: Path) -> None: + """No agent history files → status ok.""" + result = check_repeated_reads(tmp_path) + assert result["status"] == "ok" + + +def test_check_repeated_reads_warning(tmp_path: Path) -> None: + """AGENT_HANDOFF.md mentions the same file >2 times.""" + ah = tmp_path / "AGENT_HANDOFF.md" + ah.write_text( + "# Agent Handoff\n\n" + "- README.md\n- README.md\n- README.md\n" + "- TASKS.md\n", + encoding="utf-8", + ) + result = check_repeated_reads(tmp_path) + assert result["status"] == "warning" + assert "README.md (3x)" in result["details"] + + +def test_check_repeated_reads_no_over_threshold(tmp_path: Path) -> None: + """Files mentioned <=2 times should not trigger warning.""" + ah = tmp_path / "AGENT_HANDOFF.md" + ah.write_text( + "# Agent Handoff\n\n" + "- README.md\n- README.md\n" + "- TASKS.md\n- TASKS.md\n", + encoding="utf-8", + ) + result = check_repeated_reads(tmp_path) + assert result["status"] == "ok" + + +# -- check_repeated_commands -- + +def test_check_repeated_commands_ok_no_log(tmp_path: Path) -> None: + result = check_repeated_commands(tmp_path) + assert result["status"] == "ok" + + +def test_check_repeated_commands_ok(tmp_path: Path) -> None: + rl = tmp_path / "RUN_LOG.md" + rl.write_text( + "| date | event1 |\n" + "| date | event2 |\n", + encoding="utf-8", + ) + result = check_repeated_commands(tmp_path) + assert result["status"] == "ok" + + +def test_check_repeated_commands_warning(tmp_path: Path) -> None: + rl = tmp_path / "RUN_LOG.md" + line = "| date | Task completed |\n" + rl.write_text( + "# RUN_LOG\n\n| Date | Event | Task | Notes |\n" + line * 5, + encoding="utf-8", + ) + result = check_repeated_commands(tmp_path) + assert result["status"] == "warning" + + +# -- check_no_recent_file_changes -- + +def test_check_no_recent_file_changes_ok(tmp_path: Path) -> None: + """Files modified recently → ok.""" + for rel in ["TASKS.md", "PROJECT_STATE.md"]: + (tmp_path / rel).write_text("x", encoding="utf-8") + + # Patch _last_modified to return a recent time + import rdb_discovery.guardrails as mod + original_now = mod._now + + def fake_now() -> datetime: + return datetime.now() + + mod._now = fake_now + try: + result = check_no_recent_file_changes(tmp_path) + assert result["status"] == "ok" + finally: + mod._now = original_now + + +def test_check_no_recent_file_changes_warning(tmp_path: Path) -> None: + """Files modified >48h ago → warning.""" + (tmp_path / "TASKS.md").write_text("x", encoding="utf-8") + + import rdb_discovery.guardrails as mod + + old_ts = datetime.now() - timedelta(hours=72) + # Set a fake mtime via os.utime (avoids pathlib.touch times= on macOS) + ts = old_ts.timestamp() + (tmp_path / "TASKS.md").touch() + os.utime(str(tmp_path / "TASKS.md"), (ts, ts)) + + result = check_no_recent_file_changes(tmp_path) + assert result["status"] == "warning" + + +# -- check_no_test_run_recorded -- + +def test_check_no_test_run_recorded_ok(tmp_path: Path) -> None: + rl = tmp_path / "RUN_LOG.md" + rl.write_text("# RUN\n\n| date | pytest passed |\n", encoding="utf-8") + result = check_no_test_run_recorded(tmp_path) + assert result["status"] == "ok" + + +def test_check_no_test_run_recorded_warning(tmp_path: Path) -> None: + rl = tmp_path / "RUN_LOG.md" + rl.write_text("# RUN\n\n| date | task done |\n", encoding="utf-8") + result = check_no_test_run_recorded(tmp_path) + assert result["status"] == "warning" + + +def test_check_no_test_run_recorded_no_log(tmp_path: Path) -> None: + result = check_no_test_run_recorded(tmp_path) + assert result["status"] == "ok" + + +# -- check_run_log_updated -- + +def test_check_run_log_updated_ok(tmp_path: Path) -> None: + tasks = tmp_path / "TASKS.md" + tasks.write_text( + "## TASK-001 - First\nStatus: Done\n", + encoding="utf-8", + ) + rl = tmp_path / "RUN_LOG.md" + rl.write_text( + "# RUN\n\n| date | Task completed | TASK-001 |\n", + encoding="utf-8", + ) + result = check_run_log_updated(tmp_path) + assert result["status"] == "ok" + + +def test_check_run_log_updated_warning(tmp_path: Path) -> None: + tasks = tmp_path / "TASKS.md" + tasks.write_text( + "## TASK-001 - First\nStatus: Done\n", + encoding="utf-8", + ) + rl = tmp_path / "RUN_LOG.md" + rl.write_text("# RUN\n\n| date | something else |\n", encoding="utf-8") + result = check_run_log_updated(tmp_path) + assert result["status"] == "warning" + + +def test_check_run_log_updated_no_files(tmp_path: Path) -> None: + result = check_run_log_updated(tmp_path) + assert result["status"] == "ok" + + +# -- run_all_guardrails -- + +def test_run_all_guardrails_returns_keys() -> None: + import tempfile + with tempfile.TemporaryDirectory() as td: + root = Path(td) + # Create minimal files so checks don't return 'ok' for missing file + (root / "RUN_LOG.md").write_text("# RUN\n", encoding="utf-8") + result = run_all_guardrails(root) + assert "checks" in result + assert "overall_status" in result + + +def test_run_all_guardrails_overall_all_clear(tmp_path: Path) -> None: + """All checks pass → overall 'all clear'.""" + # Create files with recent mtimes and ok content + (tmp_path / "TASKS.md").write_text("# TASKS\n", encoding="utf-8") + (tmp_path / "RUN_LOG.md").write_text("# RUN\n| date | pytest passed |\n", encoding="utf-8") + + import rdb_discovery.guardrails as mod + + old_now = mod._now + now = datetime.now() + + class FakePath: + def __init__(self, path: Path): + self._path = path + + def exists(self) -> bool: + return self._path.exists() + + def stat(self): + class StatResult: + st_mtime = now.timestamp() + return StatResult() + + orig_last_modified = mod._last_modified + + def fake_last_modified(root: Path, rel: str) -> datetime | None: + p = root / rel + if p.exists(): + return now + return None + + mod._now = lambda: now + mod._last_modified = fake_last_modified + + try: + result = run_all_guardrails(tmp_path) + # Should be clear or have a non-error overall status (guardrails may flag other things) + assert result["overall_status"] in ("all clear", "review recommended") + finally: + mod._now = old_now + mod._last_modified = orig_last_modified + + +# -- format_report -- + +def test_format_report_includes_overall() -> None: + results = { + "checks": {"Test check": {"status": "ok", "details": "fine"}}, + "overall_status": "all clear", + } + report = format_report(results) + assert "# Guardrail Report" in report + assert "all clear" in report + + +def test_format_report_with_warning() -> None: + results = { + "checks": {"Test check": {"status": "warning", "details": "watch out"}}, + "overall_status": "review recommended", + } + report = format_report(results) + assert "review recommended" in report