mirror of
https://github.com/google-gemini/gemini-cli.git
synced 2026-08-09 00:16:57 -07:00
feat(pr-generator-orchestrator): implement iterative bug-fixing state machine and container worker entrypoint (#28433)
This commit is contained in:
joneba-google
@@ -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()
|
||||
@@ -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
|
||||
@@ -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 <test_file>). 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 #<issue_number>' and including the original issue URL https://github.com/<owner>/<repo>/issues/<issue_number>). "
|
||||
"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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user