Add agent support to Process class

agent.py

@@ -0,0 +1,45 @@
+import subprocess
+import sys
+
+
+def agent(prompt: str, model: str = "sonnet") -> str:
+    """
+    Call an agent with the given prompt.
+
+    Args:
+        prompt: The prompt to send to the agent
+        model: The model to use (default: "sonnet")
+
+    Returns:
+        The response from the agent
+    """
+    cmd = [
+        "bunx",
+        "--bun",
+        "claude",
+        "--dangerously-skip-permissions",
+        "--model",
+        model,
+        "-p",
+        prompt
+    ]
+
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            check=True
+        )
+        # Mirror stdout to our process's stdout
+        if result.stdout:
+            print(result.stdout.strip())
+        # Mirror stderr to our process's stderr
+        if result.stderr:
+            print(result.stderr, file=sys.stderr)
+        return result.stdout.strip()
+    except subprocess.CalledProcessError as e:
+        # Mirror error to stderr before raising
+        if e.stderr:
+            print(e.stderr, file=sys.stderr)
+        raise RuntimeError(f"Agent call failed: {e.stderr}") from e

fix_lint_issues.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""Process to fix lint issues one at a time using an agent."""
+
+import os
+import subprocess
+import tempfile
+from human_do_task import Process
+
+
+def count_lint_issues(lint_output: str) -> int:
+    """Count the number of lint issues in the output."""
+    return len([line for line in lint_output.strip().split('\n') if line])
+
+
+def run_linter() -> tuple[str, int]:
+    """Run the linter and return output and count of issues."""
+    try:
+        # Run flake8 (Python linter) - adjust command for your project
+        result = subprocess.run(
+            ['flake8', '.'],
+            capture_output=True,
+            text=True,
+            check=False
+        )
+        output = result.stdout + result.stderr
+        count = count_lint_issues(output)
+        return output, count
+    except FileNotFoundError:
+        print("Linter not found. Please install flake8: pip install flake8")
+        return "", 0
+
+
+def save_to_file(content: str, filename: str) -> None:
+    """Save content to a file."""
+    with open(filename, 'w') as f:
+        f.write(content)
+
+
+def revert_changes() -> None:
+    """Revert the last git changes."""
+    subprocess.run(['git', 'checkout', '.'], check=False)
+    if os.path.exists('commit_msg.txt'):
+        os.remove('commit_msg.txt')
+
+
+def commit_changes() -> None:
+    """Commit the current changes using the saved commit message."""
+    if os.path.exists('commit_msg.txt'):
+        with open('commit_msg.txt', 'r') as f:
+            commit_msg = f.read().strip()
+        subprocess.run(['git', 'add', '-A'], check=True)
+        subprocess.run(['git', 'commit', '-m', commit_msg], check=True)
+        os.remove('commit_msg.txt')
+        print(f"Committed: {commit_msg}")
+    else:
+        print("No commit message found")
+
+
+def perform(go: Process) -> None:
+    """Main process to fix lint issues one by one."""
+
+    # Initial setup
+    go.tell("Please ensure you have a clean git working directory before starting.")
+
+    iteration = 0
+    max_iterations = 50  # Prevent infinite loops
+
+    while iteration < max_iterations:
+        iteration += 1
+        print(f"\n=== Iteration {iteration} ===\n")
+
+        # Run linter and save results
+        go.do(lambda: print("Running linter..."))
+        lint_output, issue_count = run_linter()
+
+        if issue_count == 0:
+            print("No lint issues found! Process complete.")
+            break
+
+        go.do(lambda: save_to_file(lint_output, f'lint_results_{iteration}.txt'))
+        print(f"Found {issue_count} lint issue(s)")
+
+        # Ask agent to fix ONE issue
+        prompt = f"""
+Look at the lint issues in this output and fix EXACTLY ONE RANDOM issue:
+
+{lint_output}
+
+Instructions:
+1. Pick only ONE RANDOM lint issue from the output
+2. Fix that single issue in the code
+3. Do NOT fix any other issues
+4. Do NOT create a commit
+5. Write a brief commit message (one line) describing what you fixed and save it to 'commit_msg.txt'
+6. Return the exact lint issue you fixed
+
+Example commit message: "Fix E501 line too long in example.py:42"
+"""
+
+        go.agent(prompt)
+
+        # Check if lint issues decreased
+        go.do(lambda: print("\nChecking if lint issues decreased..."))
+        new_lint_output, new_issue_count = run_linter()
+
+        if new_issue_count < issue_count:
+            print(f"Success! Lint issues reduced from {issue_count} to {new_issue_count}")
+
+            # Ask human for approval
+            if go.ask_yes_no(f"The agent fixed 1 issue (from {issue_count} to {new_issue_count} issues). Review the changes and approve?"):
+                go.do(commit_changes)
+                print("Changes committed successfully")
+            else:
+                go.do(revert_changes)
+                print("Changes reverted by user")
+                if not go.ask_yes_no("Do you want to continue fixing other issues?"):
+                    break
+        else:
+            print(f"Failed to reduce issues (still {new_issue_count} issues). Reverting...")
+            go.do(revert_changes)
+
+            if not go.ask_yes_no("The fix didn't work. Try again?"):
+                break
+
+    if iteration >= max_iterations:
+        print(f"Reached maximum iterations ({max_iterations}). Stopping.")
+
+    print("\nProcess complete!")
+
+
+def verify(go: Process) -> None:
+    """Verify the lint fixing process."""
+
+    # Check git is available
+    go.verify(lambda: subprocess.run(['git', '--version'], capture_output=True).returncode == 0)
+
+    # Check linter is available
+    go.verify(lambda: subprocess.run(['which', 'flake8'], capture_output=True).returncode == 0 or
+              subprocess.run(['which', 'ruff'], capture_output=True).returncode == 0)
+
+    # Check we're in a git repository
+    go.verify(lambda: os.path.exists('.git'))
+
+    # Check git working directory is clean
+    result = subprocess.run(['git', 'status', '--porcelain'], capture_output=True, text=True)
+    go.verify(lambda: len(result.stdout.strip()) == 0)
+
+    go.verify(go.that("the agent command (bunx claude) is available"))
+
+
+if __name__ == "__main__":
+    Process.run(perform, verify)

human_do_task.py

@@ -4,6 +4,8 @@
 import traceback
 import typing
 
+from agent import agent as call_agent
+
 
 class Clock(object):
     started: typing.Optional[datetime.datetime]
@@ -51,16 +53,19 @@ def __init__(self) -> None:
         self._automated = Clock()
         self._automated.start()
         self._manual = Clock()
+        self._agent = Clock()
         self._test_result: list[str] = []
         self._automated_steps = 0
         self._manual_steps = 0
+        self._agent_steps = 0
 
     class ManualSection:
         def __init__(self, go: Process) -> None:
             self.go = go
 
         def __enter__(self) -> None:
             self.go._automated.stop()
+            self.go._agent.stop()
             self.go._manual.start()
 
         def __exit__(
@@ -138,6 +143,17 @@ def impl() -> bool:
 
         return impl
 
+    def agent(self, prompt: str, model: str = "sonnet") -> str:
+        self._agent_steps += 1
+        self._automated.stop()
+        self._agent.start()
+        try:
+            result = call_agent(prompt, model)
+        finally:
+            self._agent.stop()
+            self._automated.start()
+        return result
+
     def print_test_results(self) -> None:
         if self._test_result:
             print("Verification failed. Please fix the process and try again.")
@@ -146,9 +162,10 @@ def print_test_results(self) -> None:
 
     def _print_stats(self) -> None:
         self._automated.stop()
-        total_time = self._automated.elapsed + self._manual.elapsed
+        total_time = self._automated.elapsed + self._manual.elapsed + self._agent.elapsed
         print(f"Process complete in {total_time}.")
         print(
             f"   Automated: {self._automated_steps} steps in {self._automated.elapsed}."
         )
         print(f"   Manual: {self._manual_steps} steps in {self._manual.elapsed}.")
+        print(f"   Agent: {self._agent_steps} steps in {self._agent.elapsed}.")

sample_with_lint_issues.py

@@ -0,0 +1,22 @@
+# Sample file with intentional lint issues for testing
+
+def badly_formatted_function(x,y,z):
+    result=x+y+z   # Missing spaces around operators
+    return  result  # Extra spaces
+
+class poorlyNamedClass:  # Class name should be CamelCase
+    def __init__(self):
+        unused_variable = 42  # F841: local variable assigned but never used
+        pass
+
+def function_with_long_line():
+    # E501: Line too long
+    really_long_string = "This is a really really really really really really really really really really long line that exceeds the recommended length"
+    return really_long_string
+
+import os  # E402: Module level import not at top of file
+
+def unused_function():  # This function is never called
+    pass
+
+# Missing newline at end of file

test_agent.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python3
+
+from human_do_task import Process
+
+
+def perform(go: Process) -> None:
+    go.tell("Starting process with agent interaction")
+
+    response = go.agent("What is 2+2?")
+    print(f"Agent response: {response}")
+
+    go.tell("Process complete")
+
+
+def verify(go: Process) -> None:
+    go.tell("Verification not implemented")
+
+
+if __name__ == "__main__":
+    Process.run(perform, verify)