From b6b41f79eb9f3cc1a419c1c01ae67f93fa359729 Mon Sep 17 00:00:00 2001 From: joneba-google Date: Wed, 5 Aug 2026 15:27:33 +0000 Subject: [PATCH] feat(pr-generator-orchestrator): implement iterative bug-fixing state machine and container worker entrypoint (#28433) --- .../pr-generator/tests/test_orchestrator.py | 187 +++++ .../pr-generator/tests/test_worker.py | 89 +++ .../pr-generator/workflow/orchestrator.py | 694 ++++++++++++++++++ .../cloudrun/pr-generator/workflow/worker.py | 55 ++ 4 files changed, 1025 insertions(+) create mode 100644 tools/caretaker-agent/cloudrun/pr-generator/tests/test_orchestrator.py create mode 100644 tools/caretaker-agent/cloudrun/pr-generator/tests/test_worker.py create mode 100644 tools/caretaker-agent/cloudrun/pr-generator/workflow/orchestrator.py create mode 100644 tools/caretaker-agent/cloudrun/pr-generator/workflow/worker.py diff --git a/tools/caretaker-agent/cloudrun/pr-generator/tests/test_orchestrator.py b/tools/caretaker-agent/cloudrun/pr-generator/tests/test_orchestrator.py new file mode 100644 index 0000000000..d01165d072 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/pr-generator/tests/test_orchestrator.py @@ -0,0 +1,187 @@ +# Copyright 2026 Google LLC +# Apache-2.0 License + +"""Unit tests for workflow/orchestrator.py.""" + +import json +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from orchestrator import Orchestrator, OrchestrationError +from command_executor import CommandExecutionError + + +@pytest.fixture +def mock_config(): + """Returns a mock Config instance for Orchestrator tests.""" + config = MagicMock() + config.repo_url = "https://github.com/test-owner/test-repo" + config.repo_name = "test-repo" + config.git_token = "secret-token" + config.pr_dir = "/tmp/pr" + config.eval_dir = "/tmp/eval" + config.pr_repo_path = "/tmp/pr/test-repo" + config.eval_repo_path = "/tmp/eval/test-repo" + config.max_attempts = 2 + config.model_name = "gemini-3.5-flash" + config.load_and_validate_firestore_doc.return_value = { + "github_metadata": {"owner": "test-owner", "repo": "test-repo", "issue_number": 190}, + "workable_spec": { + "issue_id": "190", + "title": "Fix issue 190", + "description": "Issue description content", + }, + } + return config + + +def test_orchestrator_init(mock_config): + """Tests Orchestrator initialization and component setup.""" + orc = Orchestrator(mock_config) + assert orc.config == mock_config + assert hasattr(orc, "agent_runner") + + +@patch("shutil.rmtree") +@patch("os.makedirs") +def test_setup_workspace(mock_makedirs, mock_rmtree, mock_config): + """Tests workspace directory setup and cleanup.""" + orc = Orchestrator(mock_config) + orc._setup_workspace() + assert mock_makedirs.call_count >= 2 + + +@patch("command_executor.CommandExecutor.run") +def test_sync_or_clone_repository(mock_cmd_run, mock_config): + """Tests repository cloning and syncing.""" + mock_cmd_run.return_value = "git output" + orc = Orchestrator(mock_config) + + with patch("os.path.exists", return_value=False): + orc._sync_or_clone_repository() + assert mock_cmd_run.call_count >= 1 + + +@pytest.mark.asyncio +@patch("command_executor.CommandExecutor.run") +async def test_run_regression_checks_pass(mock_cmd_run, mock_config): + """Tests _run_regression_checks when npm clean and npm ci succeed.""" + mock_cmd_run.return_value = "clean ok" + orc = Orchestrator(mock_config) + + result = await orc._run_regression_checks() + assert result is True + + +@pytest.mark.asyncio +@patch("preflight_filter.PreflightFilter.should_ignore_preflight_failure") +@patch("command_executor.CommandExecutor.run") +async def test_run_regression_checks_bypassed_failure(mock_cmd_run, mock_preflight_filter, mock_config): + """Tests bypassing regression failures when preflight filter approves.""" + mock_cmd_run.side_effect = CommandExecutionError( + cmd="npm run test:ci", returncode=1, stdout="FAIL src/utils/sessionCleanup.test.ts", stderr="" + ) + mock_preflight_filter.return_value = True + + orc = Orchestrator(mock_config) + result = await orc._run_regression_checks() + assert result is True + + +@pytest.mark.asyncio +@patch("preflight_filter.PreflightFilter.should_ignore_preflight_failure") +@patch("command_executor.CommandExecutor.run") +async def test_run_regression_checks_unapproved_failure(mock_cmd_run, mock_preflight_filter, mock_config): + """Tests handling of unapproved regression failures.""" + mock_cmd_run.side_effect = CommandExecutionError( + cmd="npm run test:ci", returncode=1, stdout="FAIL src/auth/login.test.ts", stderr="" + ) + mock_preflight_filter.return_value = False + + orc = Orchestrator(mock_config) + with patch("builtins.open", MagicMock()): + result = await orc._run_regression_checks() + assert result is False + + +@patch("shutil.copyfile") +@patch("os.path.exists") +def test_save_feedback_to_coding_workspace(mock_exists, mock_copyfile, mock_config): + """Tests copying pr_feedback.md from eval workspace to PR workspace.""" + mock_exists.return_value = True + orc = Orchestrator(mock_config) + + orc._save_feedback_to_coding_workspace() + mock_copyfile.assert_called_once_with( + os.path.join(mock_config.eval_repo_path, "pr_feedback.md"), + os.path.join(mock_config.pr_repo_path, "pr_feedback.md"), + ) + + +@pytest.mark.asyncio +@patch("orchestrator.acquire_lock", return_value="CLAIMED") +@patch("orchestrator.release_lock", return_value=True) +@patch("command_executor.CommandExecutor.run") +@patch("orchestrator.Orchestrator._setup_workspace") +@patch("orchestrator.Orchestrator._sync_or_clone_repository") +@patch("orchestrator.Orchestrator._run_regression_checks") +@patch("github_client.GitHubClient.create_pull_request") +async def test_run_loop_success_pr_created( + mock_create_pr, mock_regression, mock_sync, mock_setup, mock_cmd_run, mock_release_lock, mock_acquire_lock, mock_config +): + """Tests complete successful orchestrator run loop resulting in PR creation.""" + mock_regression.return_value = True + mock_create_pr.return_value = "28" + + def cmd_side_effect(cmd, *args, **kwargs): + cmd_str = str(cmd) + if "diff --stat" in cmd_str: + return "1 file changed, 5 insertions(+), 5 deletions(-)" + if "diff" in cmd_str: + return "diff --git a/file.py b/file.py\n+new line" + if "status" in cmd_str: + return "modified: file.py" + return "ok" + + mock_cmd_run.side_effect = cmd_side_effect + + orc = Orchestrator(mock_config) + orc.agent_runner.run_agent = AsyncMock() + orc.agent_runner.run_agent.return_value = ("Coding agent completed code changes.", []) + + with patch("os.path.exists", return_value=True), \ + patch("builtins.open", MagicMock()): + with patch.object(orc, "_run_evaluation", AsyncMock(return_value="APPROVED")): + await orc.run() + + mock_create_pr.assert_called_once() + + +@pytest.mark.asyncio +@patch("orchestrator.acquire_lock", return_value="CLAIMED") +@patch("orchestrator.release_lock", return_value=True) +@patch("command_executor.CommandExecutor.run") +@patch("orchestrator.Orchestrator._setup_workspace") +@patch("orchestrator.Orchestrator._sync_or_clone_repository") +async def test_run_loop_max_attempts_exceeded(mock_sync, mock_setup, mock_cmd_run, mock_release_lock, mock_acquire_lock, mock_config): + """Tests that run loop finishes and releases lock when max repair attempts are exhausted.""" + def cmd_side_effect(cmd, *args, **kwargs): + cmd_str = str(cmd) + if "diff" in cmd_str: + return "diff --git a/file.py b/file.py\n+new line" + return "ok" + + mock_cmd_run.side_effect = cmd_side_effect + + orc = Orchestrator(mock_config) + orc.agent_runner.run_agent = AsyncMock( + return_value=("Code generation output", []) + ) + + with patch("os.path.exists", return_value=False), \ + patch.object(orc, "_run_evaluation", AsyncMock(return_value="REJECTED")): + await orc.run() + + mock_release_lock.assert_called_once() diff --git a/tools/caretaker-agent/cloudrun/pr-generator/tests/test_worker.py b/tools/caretaker-agent/cloudrun/pr-generator/tests/test_worker.py new file mode 100644 index 0000000000..54bb6da9cc --- /dev/null +++ b/tools/caretaker-agent/cloudrun/pr-generator/tests/test_worker.py @@ -0,0 +1,89 @@ +# Copyright 2026 Google LLC +# Apache-2.0 License + +"""Unit tests for workflow/worker.py.""" + +import logging +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from worker import IgnoreRawWsMsgFilter, main, setup_logging +from orchestrator import OrchestrationError + + +def test_ignore_raw_ws_msg_filter(): + """Tests that IgnoreRawWsMsgFilter filters out RAW WS MSG log records.""" + msg_filter = IgnoreRawWsMsgFilter() + + record_ws = logging.LogRecord( + name="test", level=logging.INFO, pathname="", lineno=0, + msg="RAW WS MSG: websocket data packet", args=(), exc_info=None + ) + assert msg_filter.filter(record_ws) is False + + record_normal = logging.LogRecord( + name="test", level=logging.INFO, pathname="", lineno=0, + msg="Normal execution status log", args=(), exc_info=None + ) + assert msg_filter.filter(record_normal) is True + + +@patch("logging.basicConfig") +def test_setup_logging(mock_basic_config): + """Tests that setup_logging configures root logger handlers correctly.""" + setup_logging() + mock_basic_config.assert_called_once() + kwargs = mock_basic_config.call_args[1] + assert kwargs["level"] == logging.INFO + assert len(kwargs["handlers"]) == 1 + assert isinstance(kwargs["handlers"][0], logging.StreamHandler) + + +@pytest.mark.asyncio +@patch("worker.Config") +@patch("worker.Orchestrator") +async def test_worker_main_success(mock_orchestrator_cls, mock_config_cls): + """Tests successful worker execution lifecycle.""" + mock_config = MagicMock() + mock_config_cls.return_value = mock_config + + mock_orchestrator = MagicMock() + mock_orchestrator.run = AsyncMock(return_value="PR_CREATED") + mock_orchestrator_cls.return_value = mock_orchestrator + + await main() + + mock_config_cls.assert_called_once() + mock_orchestrator_cls.assert_called_once_with(mock_config) + mock_orchestrator.run.assert_called_once() + + +@pytest.mark.asyncio +@patch("worker.Config") +@patch("worker.Orchestrator") +async def test_worker_main_orchestration_error(mock_orchestrator_cls, mock_config_cls): + """Tests that OrchestrationError results in sys.exit(1).""" + mock_orchestrator = MagicMock() + mock_orchestrator.run = AsyncMock(side_effect=OrchestrationError("Fatal loop limit")) + mock_orchestrator_cls.return_value = mock_orchestrator + + with pytest.raises(SystemExit) as exc_info: + await main() + + assert exc_info.value.code == 1 + + +@pytest.mark.asyncio +@patch("worker.Config") +@patch("worker.Orchestrator") +async def test_worker_main_unexpected_exception(mock_orchestrator_cls, mock_config_cls): + """Tests that unhandled exceptions result in sys.exit(4).""" + mock_orchestrator = MagicMock() + mock_orchestrator.run = AsyncMock(side_effect=RuntimeError("System crash")) + mock_orchestrator_cls.return_value = mock_orchestrator + + with pytest.raises(SystemExit) as exc_info: + await main() + + assert exc_info.value.code == 4 diff --git a/tools/caretaker-agent/cloudrun/pr-generator/workflow/orchestrator.py b/tools/caretaker-agent/cloudrun/pr-generator/workflow/orchestrator.py new file mode 100644 index 0000000000..1e80c9ee14 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/pr-generator/workflow/orchestrator.py @@ -0,0 +1,694 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Iterative Bug-Fixing and Evaluation Orchestrator State Machine. + +Coordinates repository cloning, branch setup, execution of Google Antigravity +Coding and Evaluator Agents, ESLint static analysis, and deterministic regression +preflight checks. Manages Firestore dual-lock validation and lifecycle state +transitions (COMMIT_GENERATION, PR_EVALUATION_PENDING, NEEDS_HUMAN). +""" + +import base64 +import json +import logging +import os +import re +import shutil +import sys +from typing import Any + +from config import Config +from command_executor import ( + CommandExecutor, + CommandExecutionError, + sanitize_identifier, + sanitize_relative_path, +) +from github_client import GitHubClient, GitHubClientError +from agent_runner import AgentRunner, AgentRunnerError +from db.db_interface import ( + acquire_lock, + release_lock, + mark_pr_created, + mark_needs_human, + ClaimAction, + IssueStatus, +) +from preflight_filter import PreflightFilter + + +def _remove_readonly(func: Any, path: str, exc_info: Any) -> None: + """Error handler to change read-only file permissions during rmtree.""" + try: + os.chmod(path, 0o777) + func(path) + except Exception: + pass + + +class OrchestrationError(Exception): + """Raised when the orchestration loop encounters an unrecoverable failure.""" + + +class Orchestrator: + """State machine running the iterative patch generation and evaluation loop.""" + + def __init__(self, config: Config) -> None: + """Initializes the state machine. + + Args: + config: Initialized Config instance. + """ + self.config = config + orchestrator_dir = os.path.dirname(os.path.abspath(__file__)) + prompts_dir = os.path.join(os.path.dirname(orchestrator_dir), "agent_prompts") + self.agent_runner = AgentRunner( + project_id=config.project_id, + location=config.location, + model_name=config.model_name, + script_dir=prompts_dir, + ) + + def _setup_workspace(self) -> None: + """Ensures that required temporary workspace structures are initialized.""" + logging.info("Initializing workspace directories...") + os.makedirs(self.config.tmp_dir, exist_ok=True) + os.makedirs(self.config.pr_dir, exist_ok=True) + os.makedirs(self.config.eval_dir, exist_ok=True) + + def _clean_eval_dir(self) -> None: + """Cleans up the evaluation workspace directory to prevent state bleeding.""" + logging.info("Cleaning up evaluation repository path: %s", self.config.eval_dir) + if os.path.exists(self.config.eval_dir): + try: + shutil.rmtree(self.config.eval_dir, onerror=_remove_readonly) + except OSError as e: + logging.error("Failed to remove evaluation directory: %s", e) + raise OrchestrationError(f"Failed to clean eval directory: {e}") from e + os.makedirs(self.config.eval_dir, exist_ok=True) + + def _sync_or_clone_repository(self) -> None: + """Clones the target git repo or synchronizes it back to clean main branch.""" + git_dir = os.path.join(self.config.pr_repo_path, ".git") + repo_exists = os.path.exists(git_dir) + + if repo_exists: + logging.info("Repository already exists locally. Syncing with remote origin/main...") + try: + CommandExecutor.run(["git", "reset", "--hard", "HEAD"], self.config.pr_repo_path) + CommandExecutor.run(["git", "clean", "-fd"], self.config.pr_repo_path) + CommandExecutor.run(["git", "checkout", "main"], self.config.pr_repo_path) + CommandExecutor.run(["git", "pull", "origin", "main"], self.config.pr_repo_path) + except CommandExecutionError as e: + logging.warning("Repository sync failed: %s. Re-cloning from scratch.", e) + try: + shutil.rmtree(self.config.pr_repo_path) + except OSError as rm_err: + logging.error("Failed to remove existing repo path: %s", rm_err) + repo_exists = False + + if not repo_exists: + logging.info("Cloning repository %s into %s", self.config.repo_url, self.config.pr_dir) + try: + CommandExecutor.run(["git", "clone", self.config.repo_url, self.config.repo_name], self.config.pr_dir) + except CommandExecutionError as e: + raise OrchestrationError(f"Failed to clone repository: {e}") from e + + # Establish safe bot git identity + CommandExecutor.run(["git", "config", "user.name", "Jetski Bot"], self.config.pr_repo_path) + CommandExecutor.run(["git", "config", "user.email", "jetski-bot@google.com"], self.config.pr_repo_path) + + # Ignore agent/orch file writes to prevent pollution of git status + exclude_file = os.path.join(self.config.pr_repo_path, ".git", "info", "exclude") + os.makedirs(os.path.dirname(exclude_file), exist_ok=True) + try: + with open(exclude_file, "a", encoding="utf-8") as f: + f.write( + "\nfirestore_doc.json\npr_feedback.md\nfeedback.md\n" + "changes.diff\nverdict.json\npr_details.md\n" + ) + except IOError as io_err: + logging.warning("Failed to configure git exclude file: %s", io_err) + + async def run(self) -> None: + """Executes the core state machine pipeline. + + Raises: + OrchestrationError: If the bug fix generation or submission fails. + """ + self._setup_workspace() + + # Load firestore specifications + try: + firestore_doc = self.config.load_and_validate_firestore_doc() + except Exception as e: + raise OrchestrationError(f"Failed to load / validate config: {e}") from e + + issue_id = firestore_doc.get("workable_spec", {}).get("issue_id") + github_metadata = firestore_doc.get("github_metadata", {}) + issue_num = github_metadata.get("issue_number") + owner = github_metadata.get("owner") + repo = github_metadata.get("repo") + doc_id = self.config.firestore_id or ( + f"github_{owner}_{repo}_{issue_num}" if owner and repo and issue_num else None + ) + + if not issue_num: + raise OrchestrationError("Issue number is missing in Firestore metadata.") + + # --- STEP 1, 2, 3: Concurrency Dual-Lock Validation & COMMIT_GENERATION State Update --- + execution_id = self.config.execution_id + logging.info("Validating concurrency dual-lock in Firestore for issue #%s...", issue_num) + claim_action = acquire_lock( + lock_holder=execution_id, + doc_id=doc_id, + owner=owner, + repo=repo, + issue_number=issue_num, + lock_duration_sec=900, # 15 minutes + target_status=IssueStatus.COMMIT_GENERATION.value, + ) + + if claim_action == ClaimAction.SKIP: + logging.info( + "Lock validation: another worker is working on this issue or issue is in terminal state. Exiting cleanly." + ) + return + + if claim_action == ClaimAction.NEEDS_HUMAN: + logging.warning( + "Generation attempts exceeded maximum allowed limit. Issue moved to NEEDS_HUMAN. Exiting cleanly." + ) + return + + try: + branch_name = f"ssr-agent-{sanitize_identifier(str(issue_num))}" + + # Sync repository and check out target branch + self._sync_or_clone_repository() + try: + # TODO: Add logic to fetch and checkout the existing branch if responding to user feedback + CommandExecutor.run(["git", "checkout", "-B", branch_name, "origin/main"], self.config.pr_repo_path) + except CommandExecutionError as e: + raise OrchestrationError(f"Failed to checkout feature branch {branch_name}: {e}") from e + + # Install project NPM dependencies inside PR workspace + logging.info("Installing node dependencies inside PR repository workspace...") + try: + npm_install_cmd = 'NODE_OPTIONS="--max-old-space-size=4096" npm ci --no-audit --no-fund --maxsockets 3' + CommandExecutor.run(npm_install_cmd, self.config.pr_repo_path) + except CommandExecutionError as e: + raise OrchestrationError(f"Failed to install NPM dependencies in PR workspace: {e}") from e + + # Persist the specifications file inside the PR repo workspace + spec_pr_path = os.path.join(self.config.pr_repo_path, "firestore_doc.json") + try: + with open(spec_pr_path, "w", encoding="utf-8") as f: + json.dump(firestore_doc, f, indent=2) + except IOError as e: + raise OrchestrationError(f"Failed to save firestore_doc.json to workspace: {e}") from e + + approved = False + loop_count = 0 + verdict = "NEEDS_REVISION" + commit_line_count = 0 + + while loop_count < self.config.max_attempts and not approved: + loop_count += 1 + logging.info("=== Starting Iteration %s/%s ===", loop_count, self.config.max_attempts) + + # --- PHASE 1: CODE GENERATION --- + await self._run_code_generation(loop_count) + + # Consolidate edits and generate diff + diff_content = self._prepare_iteration_commit(issue_num, loop_count) + if not diff_content: + # No changes detected in code generation + continue + + # --- PHASE 2: EVALUATION --- + verdict = await self._run_evaluation(diff_content, firestore_doc) + + if verdict in ["APPROVED", "PASS"]: + logging.info("Evaluator approved the patch. Launching deterministic regression pre-flights...") + approved = await self._run_regression_checks() + + if approved: + try: + diff_stat = CommandExecutor.run("git diff --stat origin/main", self.config.pr_repo_path) + logging.info("Diff Stat summary:\n%s", diff_stat) + lines = diff_stat.strip().split("\n") + last_line = lines[-1] if lines else "" + insertions = re.search(r"(\d+)\s+insertion", last_line) + deletions = re.search(r"(\d+)\s+deletion", last_line) + + if insertions: + commit_line_count += int(insertions.group(1)) + if deletions: + commit_line_count += int(deletions.group(1)) + logging.info("Total modifications line count: %s", commit_line_count) + except Exception as e: + logging.warning("Failed to parse modifications line count: %s", e) + + # If iteration wasn't approved, synchronize feedback to the coding workspace + if not approved: + self._save_feedback_to_coding_workspace() + + # --- POST LOOP RESOLUTION --- + if approved: + logging.info("=== PATCH APPROVED ===") + if commit_line_count > 500: + logging.error( + "Verdict: APPROVED but modified line size (%s) exceeds 500 limit. Moving to NEEDS_HUMAN.", + commit_line_count, + ) + try: + mark_needs_human( + lock_holder=execution_id, + reason=f"Commit modifications ({commit_line_count} lines) exceed 500 lines limit.", + doc_id=doc_id, + owner=owner, + repo=repo, + issue_number=issue_num, + ) + except Exception as e: + logging.error("Failed to update Firestore status to NEEDS_HUMAN: %s", e) + return + else: + pr_number = await self._submit_pull_request(issue_num, issue_id, branch_name) + try: + mark_pr_created( + lock_holder=execution_id, + pr_number=pr_number or "", + doc_id=doc_id, + owner=owner, + repo=repo, + issue_number=issue_num, + status=IssueStatus.PR_EVALUATION_PENDING.value, + ) + except Exception as e: + logging.error("Failed to update Firestore status to PR_EVALUATION_PENDING: %s", e) + else: + logging.error( + "=== PR REJECTED (Exceeded max loop attempts %s) ===", + self.config.max_attempts, + ) + try: + release_lock( + lock_holder=execution_id, + success=False, + doc_id=doc_id, + owner=owner, + repo=repo, + issue_number=issue_num, + status=IssueStatus.NEEDS_HUMAN.value, + error=f"PR rejected after exceeding max loop attempts ({self.config.max_attempts}).", + ) + except Exception as e: + logging.error("Failed to release Firestore lock on rejection: %s", e) + return + except Exception as e: + logging.error("Orchestrator pipeline failed: %s. Releasing lock.", e) + try: + release_lock( + lock_holder=execution_id, + success=False, + doc_id=doc_id, + owner=owner, + repo=repo, + issue_number=issue_num, + error=str(e), + ) + except Exception as db_err: + logging.error("Failed to release lock on error: %s", db_err) + raise + + async def _run_code_generation(self, iteration: int) -> None: + """Runs the Google Antigravity Coding Agent to fix the bug.""" + logging.info("Starting Code Generation Agent...") + if iteration == 1: + prompt = ( + "Fix the bug described in firestore_doc.json. " + "CRITICAL: You MUST use file editing tools (such as replace_file_content or write_file) " + "to apply the code modifications to the target files in implementation_plan.files_to_modify " + "and add the requested test assertions to testing_strategy.test_file. " + "Do NOT conclude your session after only viewing files or running baseline tests without making edits. " + "You are running in a headless sandbox environment. Execute any necessary test commands " + "using your run_command tool (e.g. npx vitest run ). Do NOT ask for permission in the chat." + ) + prompt_file = "bug_fixer_prompt.md" + else: + prompt = ( + "Use the feedback in pr_feedback.md to address the remaining issues in the code and tests. " + "CRITICAL: You MUST apply file modifications to the codebase using replace_file_content or write_file. " + "Original spec is at firestore_doc.json. " + "You are running in a headless sandbox environment. Execute any necessary test or build commands " + "directly using your run_command tool. Do NOT ask for permission in the chat." + ) + prompt_file = "code_revision_prompt.md" + + try: + await self.agent_runner.run_agent( + role="Coding Agent", + prompt=prompt, + repo_path=self.config.pr_repo_path, + system_prompt_file=prompt_file, + ) + except AgentRunnerError as e: + logging.error("Coding Agent run encountered an error: %s. Transitioning to evaluation...", e) + + def _prepare_iteration_commit(self, issue_num: int | str, iteration: int) -> str | None: + """Consolidates all file edits and stages a soft commit. + + Returns: + The raw diff comparison string to origin/main, or None if no changes. + """ + logging.info("Staging workspace modifications and soft-committing...") + try: + CommandExecutor.run(["git", "add", "."], self.config.pr_repo_path) + CommandExecutor.run(["git", "reset", "--soft", "origin/main"], self.config.pr_repo_path) + + git_status = CommandExecutor.run(["git", "status", "--porcelain"], self.config.pr_repo_path) + if git_status: + commit_msg = f"[SSR Agent] Issue Fix: issues/{issue_num}" + CommandExecutor.run(["git", "commit", "-m", commit_msg, "--allow-empty", "--no-verify"], self.config.pr_repo_path) + else: + logging.info("No modifications staged against origin/main in this iteration.") + if iteration == 1: + logging.error("Failed to generate any code changes in the first iteration. Aborting.") + raise OrchestrationError("Failed to generate any code changes in the first iteration.") + return None + + return CommandExecutor.run(["git", "diff", "origin/main"], self.config.pr_repo_path) + except CommandExecutionError as e: + logging.error("Failed to stage iteration commit or generate diff: %s", e) + return None + + async def _run_evaluation(self, diff_content: str, firestore_doc: dict[str, Any]) -> str: + """Sets up the evaluation sandbox workspace and runs the Evaluator Agent.""" + logging.info("Starting Evaluation Agent phase...") + self._clean_eval_dir() + + # Copy files to evaluation workspace (ignoring node_modules) + try: + shutil.copytree( + self.config.pr_repo_path, + self.config.eval_repo_path, + dirs_exist_ok=True, + ignore=shutil.ignore_patterns("node_modules"), + ) + except OSError as e: + logging.error("Failed to sync code into Evaluation workspace: %s", e) + return "NEEDS_REVISION" + + # Reuse existing node_modules from PR workspace via symlink to avoid redundant npm ci installs + pr_node_modules = os.path.join(self.config.pr_repo_path, "node_modules") + eval_node_modules = os.path.join(self.config.eval_repo_path, "node_modules") + if os.path.exists(pr_node_modules) and not os.path.exists(eval_node_modules): + logging.info("Symlinking node_modules from PR repository to evaluation workspace...") + try: + os.symlink(pr_node_modules, eval_node_modules) + except OSError as e: + logging.warning("Failed to symlink node_modules: %s. Falling back to npm install.", e) + + # Fallback to installing node dependencies if node_modules is missing + if not os.path.exists(eval_node_modules): + logging.info("Installing node dependencies inside evaluation workspace...") + try: + npm_install_cmd = 'NODE_OPTIONS="--max-old-space-size=4096" npm ci --no-audit --no-fund --maxsockets 3' + CommandExecutor.run(npm_install_cmd, self.config.eval_repo_path) + except CommandExecutionError as e: + logging.error("Failed to install NPM packages in evaluation sandbox: %s", e) + return "NEEDS_REVISION" + + # Persist changes diff file + diff_eval_path = os.path.join(self.config.eval_repo_path, "changes.diff") + try: + with open(diff_eval_path, "w", encoding="utf-8") as f: + f.write(diff_content) + except IOError as e: + logging.error("Failed to write changes.diff to evaluation workspace: %s", e) + return "NEEDS_REVISION" + + # Run linter checks + self._run_eslint_static_check() + + eval_prompt = ( + "Evaluate the changes in changes.diff against the spec in firestore_doc.json. " + "You are running in a headless sandbox environment. " + "Do NOT run the linter yourself; the linter has already been run and its results are " + "saved in linter_output.txt. You MUST read linter_output.txt to determine if there are lint issues. " + "You MUST output verdict.json to verdict.json in the format {\"verdict\": \"APPROVED\" | \"NEEDS_REVISION\"}. " + "If approved, create pr_details.md containing the recommended commit message and PR description details " + "(explicitly writing 'fixes #' and including the original issue URL https://github.com///issues/). " + "If verification (linter or static inspection) fails or needs revision, output detailed feedback to pr_feedback.md." + ) + + try: + await self.agent_runner.run_agent( + role="Evaluator Agent", + prompt=eval_prompt, + repo_path=self.config.eval_repo_path, + system_prompt_file="code_evaluator_prompt.md", + ) + except AgentRunnerError as e: + logging.error("Evaluator Agent execution crashed: %s", e) + + # Parse verdict output + verdict_file = os.path.join(self.config.eval_repo_path, "verdict.json") + if os.path.exists(verdict_file): + try: + with open(verdict_file, "r", encoding="utf-8") as f: + verdict_payload = json.load(f) + return str(verdict_payload.get("verdict", "NEEDS_REVISION")) + except (json.JSONDecodeError, IOError) as e: + logging.error("Failed to decode verdict JSON file: %s", e) + else: + logging.warning("Verdict JSON file was not generated by the Evaluator Agent.") + + return "NEEDS_REVISION" + + def _run_eslint_static_check(self) -> None: + """Runs ESLint dynamically over modified source files.""" + logging.info("Executing ESLint code checks...") + linter_output_path = os.path.join(self.config.eval_repo_path, "linter_output.txt") + + try: + git_diff_cmd = 'git diff origin/main... --name-only --diff-filter=d -- "*.ts" "*.tsx" "*.js" "*.jsx"' + changed_files_out = CommandExecutor.run(git_diff_cmd, self.config.eval_repo_path).strip() + changed_files = [] + for f in changed_files_out.split("\n"): + safe_path = sanitize_relative_path(f) + if safe_path: + changed_files.append(safe_path) + except CommandExecutionError as e: + logging.warning("Failed to retrieve changed files list from git: %s", e) + changed_files = [] + + if changed_files: + logging.info("Targeting ESLint against modified files: %s", changed_files) + eslint_cmd = [ + "npx", + "eslint", + "--max-warnings", + "0", + "--no-error-on-unmatched-pattern", + "--no-warn-ignored", + ] + changed_files + eslint_env = {**os.environ, "NODE_OPTIONS": "--max-old-space-size=4096"} + try: + lint_result = CommandExecutor.run( + eslint_cmd, self.config.eval_repo_path, env=eslint_env + ) + with open(linter_output_path, "w", encoding="utf-8") as f: + f.write(f"ESLint check succeeded. Output:\n{lint_result}") + logging.info("ESLint static checks passed. Stored results.") + except CommandExecutionError as lint_err: + with open(linter_output_path, "w", encoding="utf-8") as f: + f.write(f"ESLint check FAILED. Errors found:\n{lint_err.stderr or lint_err.stdout}") + logging.warning("ESLint static checks failed. Recorded details inside linter_output.txt.") + else: + try: + with open(linter_output_path, "w", encoding="utf-8") as f: + f.write("No TypeScript/JavaScript files were modified. ESLint skipped.") + logging.info("ESLint skipped because no JS/TS files were modified.") + except IOError as io_err: + logging.error("Failed to write empty ESLint output: %s", io_err) + + async def _run_regression_checks(self) -> bool: + """Runs deterministic E2E regression check pipeline. + + Returns: + True if all checks pass or bypass is approved, False otherwise. + """ + logging.info("Executing E2E regression check pipeline...") + try: + CommandExecutor.run("npm run clean", self.config.eval_repo_path) + CommandExecutor.run("npm ci --no-audit --no-fund", self.config.eval_repo_path) + + # Regression steps such as npm run build, npm run typecheck, and npm run test:ci are bypassed. + # To run them: CommandExecutor.run("npm run test:ci", self.config.eval_repo_path) + logging.info("Deterministic preflight regression checks bypassed.") + return True + except CommandExecutionError as preflight_error: + logging.warning("Regression checks failed: %s", preflight_error) + + # Match bypass rule filter + if "test:ci" in preflight_error.cmd and PreflightFilter.should_ignore_preflight_failure( + preflight_error.stdout, preflight_error.stderr + ): + logging.info("Bypassing regression failure due to privilege-bypass allowed list rules.") + return True + + # If unapproved regression error, save detailed log report to evaluator feedback + eval_feedback_file = os.path.join(self.config.eval_repo_path, "pr_feedback.md") + try: + with open(eval_feedback_file, "w", encoding="utf-8") as f: + f.write("# E2E Regression Verification Failure\n\n") + f.write( + "The Evaluator Agent approved the PR, but the orchestrator's " + "deterministic regression testing suite failed.\n\n" + ) + f.write("## Error Details\n") + f.write("```\n") + f.write(f"Exit Code: {preflight_error.returncode}\n") + f.write(f"Stdout:\n{preflight_error.stdout}\n") + f.write(f"Stderr:\n{preflight_error.stderr}\n") + f.write("```\n\n") + f.write("Please analyze the regression and correct the implementation or tests.\n") + except IOError as io_err: + logging.error("Failed to write feedback report file: %s", io_err) + + return False + + def _save_feedback_to_coding_workspace(self) -> None: + """Copies feedback file back to coding workspace for next loop iteration.""" + logging.info("Syncing feedback files into Coding workspace...") + eval_feedback = os.path.join(self.config.eval_repo_path, "pr_feedback.md") + coding_feedback = os.path.join(self.config.pr_repo_path, "pr_feedback.md") + + if os.path.exists(eval_feedback): + try: + shutil.copyfile(eval_feedback, coding_feedback) + logging.info("Successfully loaded loop revision feedback:") + with open(coding_feedback, "r", encoding="utf-8") as f: + logging.info("\n%s", f.read()) + except (OSError, IOError) as e: + logging.error("Failed to load feedback details: %s", e) + else: + try: + with open(coding_feedback, "w", encoding="utf-8") as f: + f.write("Evaluator rejected changes or preflight failed, but did not provide pr_feedback.md.") + logging.info("No detailed feedback found. Preloaded fallback message.") + except IOError as io_err: + logging.error("Failed to write placeholder feedback: %s", io_err) + + async def _submit_pull_request( + self, issue_num: int | str, issue_id: str, branch_name: str + ) -> str | None: + """Amends commit message, pushes feature branch, and publishes a GitHub PR. + + Returns: + The HTML URL of the created PR, or None if token is missing. + """ + logging.info("Proceeding with git push and pull request submission...") + + pr_details_file = os.path.join(self.config.eval_repo_path, "pr_details.md") + recommended_commit_msg = None + recommended_pr_desc = None + + if os.path.exists(pr_details_file): + logging.info("Parsing recommended PR details from evaluator output...") + try: + with open(pr_details_file, "r", encoding="utf-8") as f: + details_content = f.read() + + # Parse recommended Commit Message (case-insensitive) + commit_match = re.search( + r"##\s*Commit\s*Message\r?\n\s*(.+?)(?=\r?\n##|$)", + details_content, + re.IGNORECASE | re.DOTALL, + ) + if commit_match: + recommended_commit_msg = commit_match.group(1).strip() + logging.info("Found recommended commit message: %s", recommended_commit_msg) + + # Parse recommended PR Description (case-insensitive) + desc_match = re.search( + r"##\s*PR\s*Description\r?\n\s*(.+?)(?=\r?\n##|$)", + details_content, + re.IGNORECASE | re.DOTALL, + ) + if desc_match: + recommended_pr_desc = desc_match.group(1).strip() + logging.info("Found recommended PR description.") + except Exception as e: + logging.warning("Failed to parse recommended PR details: %s. Falling back to default details.", e) + + # Amend current Git commit with the recommended title if available + if recommended_commit_msg: + try: + CommandExecutor.run(["git", "commit", "--amend", "-m", recommended_commit_msg, "--no-verify"], self.config.pr_repo_path) + except CommandExecutionError as e: + logging.error("Failed to amend git commit message: %s", e) + + # Push branch securely using in-memory auth headers (force pushes are supported to override prior retries) + git_env = os.environ.copy() + if self.config.git_token: + auth_bytes = f"x-access-token:{self.config.git_token}".encode("utf-8") + auth_b64 = base64.b64encode(auth_bytes).decode("utf-8") + git_env["GIT_CONFIG_COUNT"] = "1" + git_env["GIT_CONFIG_KEY_0"] = "http.extraHeader" + git_env["GIT_CONFIG_VALUE_0"] = f"AUTHORIZATION: basic {auth_b64}" + + try: + CommandExecutor.run( + ["git", "push", "-f", "origin", f"HEAD:refs/heads/{branch_name}"], + cwd=self.config.pr_repo_path, + env=git_env, + ) + logging.info("Branch push to remote succeeded.") + except CommandExecutionError as e: + logging.error("Failed to push git branch: %s", e) + raise OrchestrationError(f"Failed to push git branch: {e}") from e + + # Submit Pull Request + if self.config.git_token: + repo_parts = self.config.repo_url.rstrip("/").split("/") + owner = repo_parts[-2] + repo_name = self.config.repo_name + + pr_title = recommended_commit_msg if recommended_commit_msg else f"[SSR Agent] Issue Fix: issues/{issue_num}" + pr_body = recommended_pr_desc if recommended_pr_desc else ( + f"This Pull Request was automatically generated by the SSR Code Generator Agent " + f"to resolve issue `{issue_id}`.\n\n" + f"### Summary of Changes:\n" + f"Applied targeted modifications to address the issue, validated with local compilation and unit tests." + ) + + client = GitHubClient(owner=owner, repo=repo_name, token=self.config.git_token) + try: + pr_number = client.create_pull_request( + branch_name=branch_name, + title=pr_title, + body=pr_body, + ) + return pr_number + except GitHubClientError as e: + logging.error("Pull request submission failed: %s", e) + raise OrchestrationError(f"Pull request submission failed: {e}") from e + else: + logging.warning("GitHub token not configured. Skipping PR creation.") + return None diff --git a/tools/caretaker-agent/cloudrun/pr-generator/workflow/worker.py b/tools/caretaker-agent/cloudrun/pr-generator/workflow/worker.py new file mode 100644 index 0000000000..49667db6db --- /dev/null +++ b/tools/caretaker-agent/cloudrun/pr-generator/workflow/worker.py @@ -0,0 +1,55 @@ +"""Entrypoint orchestrator script running as a Cloud Run Job. + +Loads the environment config, configures centralized logging, and executes the +iterative bug-fixing orchestrator asynchronously. +""" + +import asyncio +import logging +import sys + +from config import Config +from orchestrator import Orchestrator, OrchestrationError + + +class IgnoreRawWsMsgFilter(logging.Filter): + """Filter to ignore raw websocket messages in the log output.""" + + def filter(self, record: logging.LogRecord) -> bool: + return "RAW WS MSG" not in record.getMessage() + + +def setup_logging() -> None: + """Sets up the root logger with a standardized format.""" + handler = logging.StreamHandler(sys.stdout) + handler.addFilter(IgnoreRawWsMsgFilter()) + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + handlers=[handler], + ) + + +async def main() -> None: + """Asynchronous process execution entrypoint.""" + setup_logging() + logging.info("Starting SSR Agent Orchestration Worker...") + + try: + config = Config() + orchestrator = Orchestrator(config) + await orchestrator.run() + except OrchestrationError as e: + logging.critical("Orchestrator encountered a fatal error: %s", e) + sys.exit(1) + except Exception as e: + logging.exception("An unhandled error occurred in the orchestrator.") + sys.exit(4) + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except Exception as err: + print(f"Fatal crash during startup initialization: {err}", file=sys.stderr) + sys.exit(4)