397 lines
13 KiB
Python
397 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
Ralph Loop - SubagentStop Hook for Check Agent Loop Control
|
|
|
|
Based on the Ralph Wiggum technique for autonomous agent loops.
|
|
Uses completion promises to control when the check agent can stop.
|
|
|
|
Mechanism:
|
|
- Intercepts when check subagent tries to stop (SubagentStop event)
|
|
- If verify commands configured in worktree.yaml, runs them to verify
|
|
- Otherwise, reads check.jsonl to get dynamic completion markers ({reason}_FINISH)
|
|
- Blocks stopping until verification passes or all markers found
|
|
- Has max iterations as safety limit
|
|
|
|
State file: .trellis/.ralph-state.json
|
|
- Tracks current iteration count per session
|
|
- Resets when task changes
|
|
"""
|
|
|
|
# IMPORTANT: Suppress all warnings FIRST
|
|
import warnings
|
|
warnings.filterwarnings("ignore")
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# IMPORTANT: Force stdout to use UTF-8 on Windows
|
|
# This fixes UnicodeEncodeError when outputting non-ASCII characters
|
|
if sys.platform == "win32":
|
|
import io as _io
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
|
elif hasattr(sys.stdout, "detach"):
|
|
sys.stdout = _io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8", errors="replace") # type: ignore[union-attr]
|
|
|
|
# =============================================================================
|
|
# Configuration
|
|
# =============================================================================
|
|
|
|
MAX_ITERATIONS = 5 # Safety limit to prevent infinite loops
|
|
STATE_TIMEOUT_MINUTES = 30 # Reset state if older than this
|
|
STATE_FILE = ".trellis/.ralph-state.json"
|
|
WORKTREE_YAML = ".trellis/worktree.yaml"
|
|
DIR_WORKFLOW = ".trellis"
|
|
FILE_CURRENT_TASK = ".current-task"
|
|
|
|
# Only control loop for check agent
|
|
TARGET_AGENT = "check"
|
|
|
|
|
|
def find_repo_root(start_path: str) -> str | None:
|
|
"""Find git repo root from start_path upwards"""
|
|
current = Path(start_path).resolve()
|
|
while current != current.parent:
|
|
if (current / ".git").exists():
|
|
return str(current)
|
|
current = current.parent
|
|
return None
|
|
|
|
|
|
def get_current_task(repo_root: str) -> str | None:
|
|
"""Read current task directory path"""
|
|
current_task_file = os.path.join(repo_root, DIR_WORKFLOW, FILE_CURRENT_TASK)
|
|
if not os.path.exists(current_task_file):
|
|
return None
|
|
|
|
try:
|
|
with open(current_task_file, "r", encoding="utf-8") as f:
|
|
content = f.read().strip()
|
|
if not content:
|
|
return None
|
|
normalized = content.replace("\\", "/")
|
|
while normalized.startswith("./"):
|
|
normalized = normalized[2:]
|
|
if normalized.startswith("tasks/"):
|
|
normalized = f".trellis/{normalized}"
|
|
return normalized
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_verify_commands(repo_root: str) -> list[str]:
|
|
"""
|
|
Read verify commands from worktree.yaml.
|
|
|
|
Returns list of commands to run, or empty list if not configured.
|
|
Uses simple YAML parsing without external dependencies.
|
|
"""
|
|
yaml_path = os.path.join(repo_root, WORKTREE_YAML)
|
|
if not os.path.exists(yaml_path):
|
|
return []
|
|
|
|
try:
|
|
with open(yaml_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
# Simple YAML parsing for verify section
|
|
# Look for "verify:" followed by list items
|
|
lines = content.split("\n")
|
|
in_verify_section = False
|
|
commands = []
|
|
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
|
|
# Check for section start
|
|
if stripped.startswith("verify:"):
|
|
in_verify_section = True
|
|
continue
|
|
|
|
# Check for new section (not indented, ends with :)
|
|
if (
|
|
not line.startswith(" ")
|
|
and not line.startswith("\t")
|
|
and stripped.endswith(":")
|
|
and stripped != ""
|
|
):
|
|
in_verify_section = False
|
|
continue
|
|
|
|
# If in verify section, look for list items
|
|
if in_verify_section:
|
|
# Skip comments and empty lines
|
|
if stripped.startswith("#") or stripped == "":
|
|
continue
|
|
# Parse list item (- command)
|
|
if stripped.startswith("- "):
|
|
cmd = stripped[2:].strip()
|
|
if cmd:
|
|
commands.append(cmd)
|
|
|
|
return commands
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def run_verify_commands(repo_root: str, commands: list[str]) -> tuple[bool, str]:
|
|
"""
|
|
Run verify commands and return (success, message).
|
|
|
|
All commands must pass for success.
|
|
"""
|
|
for cmd in commands:
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
cwd=repo_root,
|
|
capture_output=True,
|
|
timeout=120, # 2 minute timeout per command
|
|
)
|
|
if result.returncode != 0:
|
|
stderr = result.stderr.decode("utf-8", errors="replace")
|
|
stdout = result.stdout.decode("utf-8", errors="replace")
|
|
error_output = stderr or stdout
|
|
# Truncate long output
|
|
if len(error_output) > 500:
|
|
error_output = error_output[:500] + "..."
|
|
return False, f"Command failed: {cmd}\n{error_output}"
|
|
except subprocess.TimeoutExpired:
|
|
return False, f"Command timed out: {cmd}"
|
|
except Exception as e:
|
|
return False, f"Command error: {cmd} - {str(e)}"
|
|
|
|
return True, "All verify commands passed"
|
|
|
|
|
|
def get_completion_markers(repo_root: str, task_dir: str) -> list[str]:
|
|
"""
|
|
Read check.jsonl and generate completion markers from reasons.
|
|
|
|
Each entry's "reason" field becomes {REASON}_FINISH marker.
|
|
Example: {"file": "...", "reason": "TypeCheck"} -> "TYPECHECK_FINISH"
|
|
"""
|
|
check_jsonl_path = os.path.join(repo_root, task_dir, "check.jsonl")
|
|
markers = []
|
|
|
|
if not os.path.exists(check_jsonl_path):
|
|
# Fallback: if no check.jsonl, use default marker
|
|
return ["ALL_CHECKS_FINISH"]
|
|
|
|
try:
|
|
with open(check_jsonl_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
item = json.loads(line)
|
|
reason = item.get("reason", "")
|
|
if reason:
|
|
# Convert to uppercase and add _FINISH suffix
|
|
marker = f"{reason.upper().replace(' ', '_')}_FINISH"
|
|
if marker not in markers:
|
|
markers.append(marker)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# If no markers found, use default
|
|
if not markers:
|
|
markers = ["ALL_CHECKS_FINISH"]
|
|
|
|
return markers
|
|
|
|
|
|
def load_state(repo_root: str) -> dict:
|
|
"""Load Ralph Loop state from file"""
|
|
state_path = os.path.join(repo_root, STATE_FILE)
|
|
if not os.path.exists(state_path):
|
|
return {"task": None, "iteration": 0, "started_at": None}
|
|
|
|
try:
|
|
with open(state_path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {"task": None, "iteration": 0, "started_at": None}
|
|
|
|
|
|
def save_state(repo_root: str, state: dict) -> None:
|
|
"""Save Ralph Loop state to file"""
|
|
state_path = os.path.join(repo_root, STATE_FILE)
|
|
try:
|
|
# Ensure directory exists
|
|
os.makedirs(os.path.dirname(state_path), exist_ok=True)
|
|
with open(state_path, "w", encoding="utf-8") as f:
|
|
json.dump(state, f, indent=2, ensure_ascii=False)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def check_completion(agent_output: str, markers: list[str]) -> tuple[bool, list[str]]:
|
|
"""
|
|
Check if all completion markers are present in agent output.
|
|
|
|
Returns:
|
|
(all_complete, missing_markers)
|
|
"""
|
|
missing = []
|
|
for marker in markers:
|
|
if marker not in agent_output:
|
|
missing.append(marker)
|
|
|
|
return len(missing) == 0, missing
|
|
|
|
|
|
def main():
|
|
try:
|
|
input_data = json.load(sys.stdin)
|
|
except json.JSONDecodeError:
|
|
# If can't parse input, allow stop
|
|
sys.exit(0)
|
|
|
|
# Get event info
|
|
hook_event = input_data.get("hook_event_name", "")
|
|
|
|
# Only handle SubagentStop event
|
|
if hook_event != "SubagentStop":
|
|
sys.exit(0)
|
|
|
|
# Get subagent info
|
|
# Field names per Claude Code SubagentStop event schema:
|
|
# agent_type, last_assistant_message, agent_id, agent_transcript_path, cwd
|
|
# The event does NOT carry a `prompt` field, so finish-phase detection
|
|
# based on a `[finish]` marker in the user prompt is no longer possible
|
|
# here; finish-phase skip logic should be reintroduced via task.json
|
|
# state (e.g. current_phase) in a follow-up.
|
|
agent_type = input_data.get("agent_type", "")
|
|
last_assistant_message = input_data.get("last_assistant_message", "")
|
|
cwd = input_data.get("cwd", os.getcwd())
|
|
|
|
# Only control check agent
|
|
if agent_type != TARGET_AGENT:
|
|
sys.exit(0)
|
|
|
|
# Find repo root
|
|
repo_root = find_repo_root(cwd)
|
|
if not repo_root:
|
|
sys.exit(0)
|
|
|
|
# Get current task
|
|
task_dir = get_current_task(repo_root)
|
|
if not task_dir:
|
|
sys.exit(0)
|
|
|
|
# Load state
|
|
state = load_state(repo_root)
|
|
|
|
# Reset state if task changed or state is too old
|
|
should_reset = False
|
|
if state.get("task") != task_dir:
|
|
should_reset = True
|
|
elif state.get("started_at"):
|
|
try:
|
|
started = datetime.fromisoformat(state["started_at"])
|
|
if (datetime.now() - started).total_seconds() > STATE_TIMEOUT_MINUTES * 60:
|
|
should_reset = True
|
|
except (ValueError, TypeError):
|
|
should_reset = True
|
|
|
|
if should_reset:
|
|
state = {
|
|
"task": task_dir,
|
|
"iteration": 0,
|
|
"started_at": datetime.now().isoformat(),
|
|
}
|
|
|
|
# Increment iteration
|
|
state["iteration"] = state.get("iteration", 0) + 1
|
|
current_iteration = state["iteration"]
|
|
|
|
# Save state
|
|
save_state(repo_root, state)
|
|
|
|
# Safety check: max iterations
|
|
if current_iteration >= MAX_ITERATIONS:
|
|
# Allow stop, reset state for next run
|
|
state["iteration"] = 0
|
|
save_state(repo_root, state)
|
|
output = {
|
|
"decision": "allow",
|
|
"reason": f"Max iterations ({MAX_ITERATIONS}) reached. Stopping to prevent infinite loop.",
|
|
}
|
|
print(json.dumps(output, ensure_ascii=False))
|
|
sys.exit(0)
|
|
|
|
# Check if verify commands are configured
|
|
verify_commands = get_verify_commands(repo_root)
|
|
|
|
if verify_commands:
|
|
# Use programmatic verification
|
|
passed, message = run_verify_commands(repo_root, verify_commands)
|
|
|
|
if passed:
|
|
# All verify commands passed, allow stop
|
|
state["iteration"] = 0
|
|
save_state(repo_root, state)
|
|
output = {
|
|
"decision": "allow",
|
|
"reason": "All verify commands passed. Check phase complete.",
|
|
}
|
|
print(json.dumps(output, ensure_ascii=False))
|
|
sys.exit(0)
|
|
else:
|
|
# Verification failed, block stop
|
|
output = {
|
|
"decision": "block",
|
|
"reason": f"Iteration {current_iteration}/{MAX_ITERATIONS}. Verification failed:\n{message}\n\nPlease fix the issues and try again.",
|
|
}
|
|
print(json.dumps(output, ensure_ascii=False))
|
|
sys.exit(0)
|
|
else:
|
|
# No verify commands, fall back to completion markers
|
|
markers = get_completion_markers(repo_root, task_dir)
|
|
all_complete, missing = check_completion(last_assistant_message, markers)
|
|
|
|
if all_complete:
|
|
# All checks complete, allow stop
|
|
state["iteration"] = 0
|
|
save_state(repo_root, state)
|
|
output = {
|
|
"decision": "allow",
|
|
"reason": "All completion markers found. Check phase complete.",
|
|
}
|
|
print(json.dumps(output, ensure_ascii=False))
|
|
sys.exit(0)
|
|
else:
|
|
# Missing markers, block stop and continue
|
|
output = {
|
|
"decision": "block",
|
|
"reason": f"""Iteration {current_iteration}/{MAX_ITERATIONS}. Missing completion markers: {", ".join(missing)}.
|
|
|
|
IMPORTANT: You must ACTUALLY run the checks, not just output the markers.
|
|
- Did you run lint? What was the output?
|
|
- Did you run typecheck? What was the output?
|
|
- Did they actually pass with zero errors?
|
|
|
|
Only output a marker (e.g., LINT_FINISH) AFTER:
|
|
1. You have executed the corresponding command
|
|
2. The command completed with zero errors
|
|
3. You have shown the command output in your response
|
|
|
|
Do NOT output markers just to escape the loop. The loop exists to ensure quality.""",
|
|
}
|
|
print(json.dumps(output, ensure_ascii=False))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|