Feat/agent strict code reviewer

examples/templates/strict_pr_reviewer/__init__.py

@@ -0,0 +1,38 @@
+"""
+Strict PR Code Reviewer — read-only GitHub PR review (no code changes, no GitHub posts).
+
+Fetches PR metadata and file patches via the GitHub API, then delivers a strict written review.
+"""
+
+from __future__ import annotations
+
+from .agent import (
+    StrictPrReviewerAgent,
+    default_agent,
+    edges,
+    entry_node,
+    entry_points,
+    goal,
+    nodes,
+    pause_nodes,
+    terminal_nodes,
+)
+from .config import AgentMetadata, RuntimeConfig, default_config, metadata
+
+__version__ = "1.0.0"
+
+__all__ = [
+    "AgentMetadata",
+    "RuntimeConfig",
+    "StrictPrReviewerAgent",
+    "default_agent",
+    "default_config",
+    "edges",
+    "entry_node",
+    "entry_points",
+    "goal",
+    "metadata",
+    "nodes",
+    "pause_nodes",
+    "terminal_nodes",
+]

examples/templates/strict_pr_reviewer/__main__.py

@@ -0,0 +1,224 @@
+"""
+CLI entry point for Strict PR Code Reviewer.
+
+Uses AgentRuntime-style setup consistent with other example templates.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import sys
+
+import click
+
+from .agent import StrictPrReviewerAgent, default_agent
+
+
+def setup_logging(verbose: bool = False, debug: bool = False) -> None:
+    """Configure logging for execution visibility."""
+    if debug:
+        level, fmt = logging.DEBUG, "%(asctime)s %(name)s: %(message)s"
+    elif verbose:
+        level, fmt = logging.INFO, "%(message)s"
+    else:
+        level, fmt = logging.WARNING, "%(levelname)s: %(message)s"
+    logging.basicConfig(level=level, format=fmt, stream=sys.stderr)
+    logging.getLogger("framework").setLevel(level)
+
+
+@click.group()
+@click.version_option(version="1.0.0")
+def cli() -> None:
+    """Strict PR Code Reviewer — read-only GitHub pull request review."""
+    pass
+
+
+@cli.command()
+@click.option("--quiet", "-q", is_flag=True, help="Only output result JSON")
+@click.option("--verbose", "-v", is_flag=True, help="Show execution details")
+@click.option("--debug", is_flag=True, help="Show debug logging")
+def run(quiet: bool, verbose: bool, debug: bool) -> None:
+    """Execute the PR reviewer agent (requires interactive intake for PR details)."""
+    if not quiet:
+        setup_logging(verbose=verbose, debug=debug)
+
+    result = asyncio.run(default_agent.run({}))
+
+    output_data = {
+        "success": result.success,
+        "steps_executed": result.steps_executed,
+        "output": result.output,
+    }
+    if result.error:
+        output_data["error"] = result.error
+
+    click.echo(json.dumps(output_data, indent=2, default=str))
+    sys.exit(0 if result.success else 1)
+
+
+@cli.command()
+@click.option("--verbose", "-v", is_flag=True, help="Show execution details")
+@click.option("--debug", is_flag=True, help="Show debug logging")
+def tui(verbose: bool, debug: bool) -> None:
+    """Launch the TUI dashboard for interactive PR review."""
+    setup_logging(verbose=verbose, debug=debug)
+
+    try:
+        from framework.tui.app import AdenTUI
+    except ImportError:
+        click.echo(
+            "TUI requires the 'textual' package. Install with: uv pip install textual"
+        )
+        sys.exit(1)
+
+    from pathlib import Path
+
+    from framework.llm import LiteLLMProvider
+    from framework.runner.tool_registry import ToolRegistry
+    from framework.runtime.agent_runtime import create_agent_runtime
+    from framework.runtime.event_bus import EventBus
+    from framework.runtime.execution_stream import EntryPointSpec
+
+    async def run_with_tui() -> None:
+        agent = StrictPrReviewerAgent()
+
+        agent._event_bus = EventBus()
+        agent._tool_registry = ToolRegistry()
+
+        storage_path = Path.home() / ".hive" / "agents" / "strict_pr_reviewer"
+        storage_path.mkdir(parents=True, exist_ok=True)
+
+        mcp_config_path = Path(__file__).parent / "mcp_servers.json"
+        if mcp_config_path.exists():
+            agent._tool_registry.load_mcp_config(mcp_config_path)
+
+        llm = LiteLLMProvider(
+            model=agent.config.model,
+            api_key=agent.config.api_key,
+            api_base=agent.config.api_base,
+        )
+
+        tools = list(agent._tool_registry.get_tools().values())
+        tool_executor = agent._tool_registry.get_executor()
+        graph = agent._build_graph()
+
+        runtime = create_agent_runtime(
+            graph=graph,
+            goal=agent.goal,
+            storage_path=storage_path,
+            entry_points=[
+                EntryPointSpec(
+                    id="start",
+                    name="Start PR review",
+                    entry_node="intake",
+                    trigger_type="manual",
+                    isolation_level="isolated",
+                ),
+            ],
+            llm=llm,
+            tools=tools,
+            tool_executor=tool_executor,
+        )
+
+        await runtime.start()
+
+        try:
+            app = AdenTUI(runtime)
+            await app.run_async()
+        finally:
+            await runtime.stop()
+
+    asyncio.run(run_with_tui())
+
+
+@cli.command()
+@click.option("--json", "output_json", is_flag=True)
+def info(output_json: bool) -> None:
+    """Show agent information."""
+    info_data = default_agent.info()
+    if output_json:
+        click.echo(json.dumps(info_data, indent=2))
+    else:
+        click.echo(f"Agent: {info_data['name']}")
+        click.echo(f"Version: {info_data['version']}")
+        click.echo(f"Description: {info_data['description']}")
+        click.echo(f"\nNodes: {', '.join(info_data['nodes'])}")
+        click.echo(f"Client-facing: {', '.join(info_data['client_facing_nodes'])}")
+        click.echo(f"Entry: {info_data['entry_node']}")
+        click.echo(f"Terminal: {', '.join(info_data['terminal_nodes'])}")
+
+
+@cli.command()
+def validate() -> None:
+    """Validate agent structure."""
+    validation = default_agent.validate()
+    if validation["valid"]:
+        click.echo("Agent is valid")
+        for warning in validation["warnings"]:
+            click.echo(f"  WARNING: {warning}")
+    else:
+        click.echo("Agent has errors:")
+        for error in validation["errors"]:
+            click.echo(f"  ERROR: {error}")
+    sys.exit(0 if validation["valid"] else 1)
+
+
+@cli.command()
+@click.option("--verbose", "-v", is_flag=True)
+def shell(verbose: bool) -> None:
+    """Interactive PR review session (CLI, no TUI)."""
+    asyncio.run(_interactive_shell(verbose))
+
+
+async def _interactive_shell(verbose: bool = False) -> None:
+    """Async interactive shell."""
+    setup_logging(verbose=verbose)
+
+    click.echo("=== Strict PR Code Reviewer ===")
+    click.echo("Provide a PR URL or owner/repo/# when prompted. Type 'quit' to exit.\n")
+
+    agent = StrictPrReviewerAgent()
+    await agent.start()
+
+    try:
+        while True:
+            try:
+                user_input = await asyncio.get_event_loop().run_in_executor(
+                    None, input, "PR> "
+                )
+                if user_input.lower() in ["quit", "exit", "q"]:
+                    click.echo("Goodbye!")
+                    break
+
+                click.echo("\nRunning review workflow...\n")
+
+                result = await agent.trigger_and_wait("start", {})
+
+                if result is None:
+                    click.echo("\n[Execution timed out]\n")
+                    continue
+
+                if result.success:
+                    out = result.output
+                    if isinstance(out, dict) and "final_report" in out:
+                        click.echo(out["final_report"])
+                    click.echo()
+                else:
+                    click.echo(f"\nFailed: {result.error}\n")
+
+            except KeyboardInterrupt:
+                click.echo("\nGoodbye!")
+                break
+            except Exception as e:
+                click.echo(f"Error: {e}", err=True)
+                import traceback
+
+                traceback.print_exc()
+    finally:
+        await agent.stop()
+
+
+if __name__ == "__main__":
+    cli()