feat(pr-generator-core): add environment config parser, command executor, GitHub R… (#28435)

tools/caretaker-agent/cloudrun/pr-generator/pytest.ini

@@ -0,0 +1,13 @@
+[pytest]
+minversion = 8.0
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+asyncio_mode = auto
+addopts =
+    -v
+    --strict-markers
+    --tb=short
+    --cov=workflow
+    --cov-report=term-missing

tools/caretaker-agent/cloudrun/pr-generator/tests/__init__.py

@@ -0,0 +1,4 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Tests package for SSR Code Generator workflow modules."""

tools/caretaker-agent/cloudrun/pr-generator/tests/conftest.py

@@ -0,0 +1,26 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Shared Pytest Fixtures for Workflow Module Tests."""
+
+import os
+import sys
+import pytest
+
+# Ensure workflow directory is in sys.path
+BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+WORKFLOW_DIR = os.path.join(BASE_DIR, "workflow")
+if WORKFLOW_DIR not in sys.path:
+    sys.path.insert(0, WORKFLOW_DIR)
+
+
+@pytest.fixture(autouse=True)
+def reset_env(monkeypatch):
+    """Ensures environment variables are clean and isolated for each test."""
+    monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project-2026")
+    monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "us-central1")
+    monkeypatch.setenv("MODEL_NAME", "gemini-3.5-flash")
+    monkeypatch.setenv("MAX_ATTEMPTS", "5")
+    monkeypatch.setenv("REPO_URL", "https://github.com/test-owner/test-repo.git")
+    monkeypatch.setenv("GIT_TOKEN", "test-github-token-12345")
+    yield

tools/caretaker-agent/cloudrun/pr-generator/tests/test_command_executor.py

@@ -0,0 +1,170 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Unit tests for workflow/command_executor.py."""
+
+import os
+import subprocess
+from unittest.mock import MagicMock, patch
+
+import pytest
+from command_executor import (
+    CommandExecutionError,
+    CommandExecutor,
+    sanitize_identifier,
+    sanitize_relative_path,
+)
+
+
+# --- Sanitization Unit Tests ---
+
+def test_sanitize_relative_path_valid():
+    """Tests that valid relative paths are normalized cleanly."""
+    assert sanitize_relative_path("src/utils/file.ts") == "src/utils/file.ts"
+    assert sanitize_relative_path("a/b/../c/file.ts") == "a/c/file.ts"
+
+
+def test_sanitize_relative_path_traversal():
+    """Tests that path traversal attempts returning '..' are rejected."""
+    assert sanitize_relative_path("../secret/passwords.txt") is None
+    assert sanitize_relative_path("a/../../etc/passwd") is None
+
+
+def test_sanitize_relative_path_absolute():
+    """Tests that absolute paths are rejected."""
+    assert sanitize_relative_path("/etc/passwd") is None
+    assert sanitize_relative_path("/usr/local/bin") is None
+
+
+def test_sanitize_relative_path_null_bytes_and_empty():
+    """Tests stripping of null bytes and handling of empty inputs."""
+    assert sanitize_relative_path("src/utils\x00/file.ts") == "src/utils/file.ts"
+    assert sanitize_relative_path("   \x00   ") is None
+    assert sanitize_relative_path("") is None
+    assert sanitize_relative_path(None) is None
+
+
+def test_sanitize_identifier_valid():
+    """Tests sanitization of alphanumeric identifiers with hyphens/underscores."""
+    assert sanitize_identifier("feature-branch_123") == "feature-branch_123"
+    assert sanitize_identifier("v1.0.0") == "v1.0.0"
+
+
+def test_sanitize_identifier_injection_stripping():
+    """Tests that special command injection characters are removed."""
+    assert sanitize_identifier("branch; rm -rf /") == "branchrm-rf"
+    assert sanitize_identifier("issue#190$(whoami)") == "issue190whoami"
+
+
+def test_sanitize_identifier_empty_fallback():
+    """Tests that empty or invalid inputs fall back to 'default'."""
+    assert sanitize_identifier("") == "default"
+    assert sanitize_identifier(None) == "default"
+    assert sanitize_identifier("!!!") == "default"
+
+
+# --- CommandExecutionError Tests ---
+
+def test_command_execution_error_attributes():
+    """Tests that CommandExecutionError formats exception message and stores attributes."""
+    err = CommandExecutionError(
+        cmd=["git", "status"],
+        returncode=128,
+        stdout="out_data",
+        stderr="err_data",
+    )
+    assert "git status" in str(err)
+    assert "exit code 128" in str(err)
+    assert err.cmd == "git status"
+    assert err.returncode == 128
+    assert err.stdout == "out_data"
+    assert err.stderr == "err_data"
+
+
+# --- CommandExecutor.run Unit Tests ---
+
+@patch("subprocess.run")
+def test_run_list_args_success(mock_subprocess_run):
+    """Tests successful command execution with a list of argument tokens."""
+    mock_subprocess_run.return_value = MagicMock(
+        returncode=0, stdout="hello world\n", stderr=""
+    )
+
+    output = CommandExecutor.run(["echo", "hello", "world"])
+    assert output == "hello world"
+    mock_subprocess_run.assert_called_once()
+    args, kwargs = mock_subprocess_run.call_args
+    assert args[0] == ["echo", "hello", "world"]
+    assert kwargs["check"] is False
+
+
+@patch("subprocess.run")
+def test_run_string_command_shlex_split(mock_subprocess_run):
+    """Tests that string commands are tokenized using shlex without shell=True."""
+    mock_subprocess_run.return_value = MagicMock(
+        returncode=0, stdout="diff output\n", stderr=""
+    )
+
+    output = CommandExecutor.run("git diff --stat origin/main")
+    assert output == "diff output"
+    mock_subprocess_run.assert_called_once()
+    args, kwargs = mock_subprocess_run.call_args
+    assert args[0] == ["git", "diff", "--stat", "origin/main"]
+
+
+@patch("subprocess.run")
+def test_run_inline_env_parsing(mock_subprocess_run):
+    """Tests parsing of inline KEY=VALUE env prefixes in string commands."""
+    mock_subprocess_run.return_value = MagicMock(
+        returncode=0, stdout="installed\n", stderr=""
+    )
+
+    output = CommandExecutor.run('NODE_OPTIONS="--max-old-space-size=4096" npm ci')
+    assert output == "installed"
+    mock_subprocess_run.assert_called_once()
+    args, kwargs = mock_subprocess_run.call_args
+    assert args[0] == ["npm", "ci"]
+    assert kwargs["env"].get("NODE_OPTIONS") == "--max-old-space-size=4096"
+
+
+@patch("subprocess.run")
+def test_run_custom_cwd_and_env(mock_subprocess_run):
+    """Tests custom CWD and environment variable dict propagation."""
+    mock_subprocess_run.return_value = MagicMock(
+        returncode=0, stdout="ok\n", stderr=""
+    )
+
+    custom_env = {"MY_VAR": "custom_val"}
+    output = CommandExecutor.run(["pwd"], cwd="/tmp/pr", env=custom_env)
+    assert output == "ok"
+    mock_subprocess_run.assert_called_once()
+    args, kwargs = mock_subprocess_run.call_args
+    assert kwargs["cwd"] == "/tmp/pr"
+    assert kwargs["env"].get("MY_VAR") == "custom_val"
+
+
+@patch("subprocess.run")
+def test_run_non_zero_exit_code_raises_error(mock_subprocess_run):
+    """Tests that a non-zero exit code raises CommandExecutionError."""
+    mock_subprocess_run.return_value = MagicMock(
+        returncode=1, stdout="some stdout", stderr="fatal error"
+    )
+
+    with pytest.raises(CommandExecutionError) as exc_info:
+        CommandExecutor.run(["git", "checkout", "nonexistent"])
+
+    err = exc_info.value
+    assert err.returncode == 1
+    assert err.stdout == "some stdout"
+    assert err.stderr == "fatal error"
+
+
+@patch("subprocess.run")
+def test_run_timeout_expired(mock_subprocess_run):
+    """Tests that subprocess timeout exceptions propagate cleanly."""
+    mock_subprocess_run.side_effect = subprocess.TimeoutExpired(
+        cmd="long_task", timeout=10.0
+    )
+
+    with pytest.raises(subprocess.TimeoutExpired):
+        CommandExecutor.run(["sleep", "100"], timeout=10.0)

tools/caretaker-agent/cloudrun/pr-generator/tests/test_config.py

@@ -0,0 +1,99 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Unit tests for workflow/config.py."""
+
+import json
+import os
+import pytest
+
+from config import Config, ConfigurationError
+
+
+def test_config_defaults(monkeypatch):
+    """Tests default configuration fallback values when environment variables are unset."""
+    monkeypatch.delenv("REPO_URL", raising=False)
+    monkeypatch.delenv("GOOGLE_CLOUD_PROJECT", raising=False)
+    monkeypatch.delenv("GOOGLE_CLOUD_LOCATION", raising=False)
+    monkeypatch.delenv("MODEL_NAME", raising=False)
+    monkeypatch.delenv("MAX_ATTEMPTS", raising=False)
+
+    cfg = Config()
+    assert cfg.repo_url == "https://github.com/joneba-google/gemini-cli-clone"
+    assert cfg.project_id == "gcli-intern-project-2026"
+    assert cfg.location == "global"
+    assert cfg.model_name == "gemini-3.5-flash"
+    assert cfg.max_attempts == 5
+    assert cfg.repo_name == "gemini-cli-clone"
+
+
+def test_config_max_attempts_valid(monkeypatch):
+    """Tests custom MAX_ATTEMPTS environment variable parsing."""
+    monkeypatch.setenv("MAX_ATTEMPTS", "12")
+    cfg = Config()
+    assert cfg.max_attempts == 12
+
+
+def test_config_max_attempts_invalid_string(monkeypatch):
+    """Tests that invalid MAX_ATTEMPTS strings fall back cleanly to 5."""
+    monkeypatch.setenv("MAX_ATTEMPTS", "invalid_string")
+    cfg = Config()
+    assert cfg.max_attempts == 5
+
+
+def test_config_max_attempts_zero_or_negative(monkeypatch):
+    """Tests lower bound enforcement (max(val, 1)) for zero or negative values."""
+    monkeypatch.setenv("MAX_ATTEMPTS", "0")
+    cfg1 = Config()
+    assert cfg1.max_attempts == 1
+
+    monkeypatch.setenv("MAX_ATTEMPTS", "-5")
+    cfg2 = Config()
+    assert cfg2.max_attempts == 1
+
+
+def test_config_repo_name_derivation(monkeypatch):
+    """Tests repository name parsing from REPO_URL with trailing slashes and .git suffixes."""
+    monkeypatch.setenv("REPO_URL", "https://github.com/my-org/my-custom-repo.git/")
+    cfg = Config()
+    assert cfg.repo_name == "my-custom-repo"
+    assert cfg.pr_repo_path == os.path.join("/tmp/pr", "my-custom-repo")
+    assert cfg.eval_repo_path == os.path.join("/tmp/eval", "my-custom-repo")
+
+
+def test_load_and_validate_firestore_doc_valid(monkeypatch):
+    """Tests loading and validating a valid JSON FIRESTORE_DOC string."""
+    valid_doc = {"workable_spec": {"issue_id": "190"}, "status": "PENDING"}
+    monkeypatch.setenv("FIRESTORE_DOC", json.dumps(valid_doc))
+
+    cfg = Config()
+    parsed = cfg.load_and_validate_firestore_doc()
+    assert parsed["workable_spec"]["issue_id"] == "190"
+    assert parsed["status"] == "PENDING"
+
+
+def test_load_and_validate_firestore_doc_missing(monkeypatch):
+    """Tests error handling when FIRESTORE_DOC environment variable is missing."""
+    monkeypatch.delenv("FIRESTORE_DOC", raising=False)
+    cfg = Config()
+    with pytest.raises(ConfigurationError) as exc_info:
+        cfg.load_and_validate_firestore_doc()
+    assert "Environment variable 'FIRESTORE_DOC' is required" in str(exc_info.value)
+
+
+def test_load_and_validate_firestore_doc_invalid_json(monkeypatch):
+    """Tests error handling when FIRESTORE_DOC is not valid JSON."""
+    monkeypatch.setenv("FIRESTORE_DOC", "{invalid json string")
+    cfg = Config()
+    with pytest.raises(ConfigurationError) as exc_info:
+        cfg.load_and_validate_firestore_doc()
+    assert "Failed to parse 'FIRESTORE_DOC' as JSON" in str(exc_info.value)
+
+
+def test_load_and_validate_firestore_doc_non_dict(monkeypatch):
+    """Tests error handling when FIRESTORE_DOC parses to a non-dict JSON structure."""
+    monkeypatch.setenv("FIRESTORE_DOC", '["array_item1", "array_item2"]')
+    cfg = Config()
+    with pytest.raises(ConfigurationError) as exc_info:
+        cfg.load_and_validate_firestore_doc()
+    assert "Firestore document specification must be a JSON object" in str(exc_info.value)

tools/caretaker-agent/cloudrun/pr-generator/tests/test_github_client.py

@@ -0,0 +1,105 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Unit tests for workflow/github_client.py."""
+
+import io
+import json
+import urllib.error
+from unittest.mock import MagicMock, patch
+
+import pytest
+from github_client import GitHubClient, GitHubClientError
+
+
+def test_github_client_init():
+    """Tests GitHubClient initialization and URL construction."""
+    client = GitHubClient(owner="my-owner", repo="my-repo", token="secret-token")
+    assert client.owner == "my-owner"
+    assert client.repo == "my-repo"
+    assert client._token == "secret-token"
+    assert client._base_url == "https://api.github.com/repos/my-owner/my-repo/pulls"
+
+
+def test_create_pull_request_missing_token():
+    """Tests that create_pull_request raises GitHubClientError when token is missing."""
+    client = GitHubClient(owner="my-owner", repo="my-repo", token=None)
+    with pytest.raises(GitHubClientError) as exc_info:
+        client.create_pull_request("feature-branch", "Fix bug", "PR description")
+    assert "GitHub token is missing" in str(exc_info.value)
+
+
+@patch("urllib.request.urlopen")
+def test_create_pull_request_success(mock_urlopen):
+    """Tests successful pull request creation, verifying headers, payload, and PR number return."""
+    mock_response = MagicMock()
+    mock_response.read.return_value = json.dumps(
+        {"number": 28, "html_url": "https://github.com/my-owner/my-repo/pull/28"}
+    ).encode("utf-8")
+    mock_urlopen.return_value.__enter__.return_value = mock_response
+
+    client = GitHubClient(owner="my-owner", repo="my-repo", token="valid-token")
+    pr_num = client.create_pull_request("feature-branch", "Fix bug", "PR description")
+
+    assert pr_num == "28"
+    mock_urlopen.assert_called_once()
+    req = mock_urlopen.call_args[0][0]
+    assert req.headers["Authorization"] == "Bearer valid-token"
+    assert req.headers["Accept"] == "application/vnd.github+json"
+    assert req.headers["Content-type"] == "application/json"
+
+    data = json.loads(req.data.decode("utf-8"))
+    assert data["title"] == "Fix bug"
+    assert data["body"] == "PR description"
+    assert data["head"] == "feature-branch"
+    assert data["base"] == "main"
+
+
+@patch("urllib.request.urlopen")
+def test_create_pull_request_http_error(mock_urlopen):
+    """Tests HTTPError handling, verifying status code and response body preservation."""
+    error_body = json.dumps({"message": "Validation Failed", "errors": ["Branch already exists"]})
+    mock_fp = io.BytesIO(error_body.encode("utf-8"))
+
+    http_err = urllib.error.HTTPError(
+        url="https://api.github.com/...",
+        code=422,
+        msg="Unprocessable Entity",
+        hdrs={},
+        fp=mock_fp,
+    )
+    mock_urlopen.side_effect = http_err
+
+    client = GitHubClient(owner="my-owner", repo="my-repo", token="valid-token")
+    with pytest.raises(GitHubClientError) as exc_info:
+        client.create_pull_request("feature-branch", "Fix bug", "PR description")
+
+    err_str = str(exc_info.value)
+    assert "HTTP 422" in err_str
+    assert "Validation Failed" in err_str
+
+
+@patch("urllib.request.urlopen")
+def test_create_pull_request_url_error(mock_urlopen):
+    """Tests network URLError handling (e.g. DNS failure or connection refused)."""
+    url_err = urllib.error.URLError(reason="Connection refused")
+    mock_urlopen.side_effect = url_err
+
+    client = GitHubClient(owner="my-owner", repo="my-repo", token="valid-token")
+    with pytest.raises(GitHubClientError) as exc_info:
+        client.create_pull_request("feature-branch", "Fix bug", "PR description")
+
+    err_str = str(exc_info.value)
+    assert "Network Error: Connection refused" in err_str
+
+
+@patch("urllib.request.urlopen")
+def test_create_pull_request_unexpected_exception(mock_urlopen):
+    """Tests unexpected runtime exception handling."""
+    mock_urlopen.side_effect = RuntimeError("System socket crash")
+
+    client = GitHubClient(owner="my-owner", repo="my-repo", token="valid-token")
+    with pytest.raises(GitHubClientError) as exc_info:
+        client.create_pull_request("feature-branch", "Fix bug", "PR description")
+
+    assert "Unexpected API client error" in str(exc_info.value)

tools/caretaker-agent/cloudrun/pr-generator/tests/test_init.py

@@ -0,0 +1,12 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Unit tests for workflow/__init__.py."""
+
+import workflow
+
+
+def test_package_docstring():
+    """Tests that the workflow package contains a valid docstring."""
+    assert workflow.__doc__ is not None
+    assert "GCLI Orchestrator Package" in workflow.__doc__

tools/caretaker-agent/cloudrun/pr-generator/tests/test_preflight_filter.py

@@ -0,0 +1,64 @@
+# Copyright 2026 Google LLC
+# Apache-2.0 License
+
+"""Unit tests for workflow/preflight_filter.py."""
+
+from preflight_filter import (
+    ALLOWED_SANDBOX_FAILURES,
+    PreflightFilter,
+    is_preflight_failure_allowed,
+    strip_ansi,
+)
+
+
+def test_strip_ansi_color_codes():
+    """Tests that strip_ansi cleanly removes ANSI terminal escape codes."""
+    colored_text = "\x1b[31mFAIL\x1b[0m \x1b[1msrc/utils/test.ts\x1b[0m"
+    assert strip_ansi(colored_text) == "FAIL src/utils/test.ts"
+    assert PreflightFilter.strip_ansi(colored_text) == "FAIL src/utils/test.ts"
+
+
+def test_is_preflight_failure_allowed_single_approved_file():
+    """Tests approval of a single known allowed test file failure."""
+    output = "FAIL src/utils/sessionCleanup.test.ts"
+    assert is_preflight_failure_allowed(output) is True
+
+
+def test_is_preflight_failure_allowed_multiple_approved_files():
+    """Tests approval when multiple known allowed test files fail."""
+    output = (
+        "FAIL src/utils/sessionCleanup.test.ts\n"
+        "FAIL src/config/extension-manager-permissions.test.ts"
+    )
+    assert is_preflight_failure_allowed(output) is True
+
+
+def test_is_preflight_failure_allowed_generic_keyword():
+    """Tests approval when failure matches generic container/sandbox exception keywords."""
+    output = "FAILED root-privilege-check in sandbox"
+    assert is_preflight_failure_allowed(output) is True
+
+
+def test_is_preflight_failure_allowed_unapproved_failure():
+    """Tests rejection when an unapproved test file fails."""
+    output = (
+        "FAIL src/utils/sessionCleanup.test.ts\n"
+        "FAIL src/auth/loginService.test.ts"
+    )
+    assert is_preflight_failure_allowed(output) is False
+
+
+def test_is_preflight_failure_allowed_no_failures():
+    """Tests that output with no 'FAIL' or 'FAILED' lines returns False."""
+    output = "PASS src/utils/sessionCleanup.test.ts\nTests: 12 passed, 12 total"
+    assert is_preflight_failure_allowed(output) is False
+
+
+def test_should_ignore_preflight_failure_concatenates_streams():
+    """Tests PreflightFilter.should_ignore_preflight_failure stream concatenation."""
+    stdout = "FAIL src/utils/sessionCleanup.test.ts"
+    stderr = ""
+    assert PreflightFilter.should_ignore_preflight_failure(stdout, stderr) is True
+
+    unapproved_stdout = "FAIL src/core/main.test.ts"
+    assert PreflightFilter.should_ignore_preflight_failure(unapproved_stdout, stderr) is False

tools/caretaker-agent/cloudrun/pr-generator/workflow/__init__.py

@@ -0,0 +1,10 @@
+"""GCLI Orchestrator Package.
+
+This package contains all components of the SSR Agent Orchestrator:
+- config: Configuration loading and validation.
+- command_executor: Subprocess execution utility.
+- github_client: GitHub v3 REST API client.
+- agent_runner: Google Antigravity SDK wrapper.
+- preflight_filter: Preflight test verification filtering.
+- orchestrator: Orchestration state machine coordinating code generation and evaluation.
+"""

tools/caretaker-agent/cloudrun/pr-generator/workflow/command_executor.py

@@ -0,0 +1,154 @@
+"""Command execution and input sanitization module.
+
+Provides safe subprocess execution utilities, path traversal guards,
+and input sanitizers to prevent injection attacks and capture process output cleanly.
+"""
+
+import logging
+import os
+import re
+import shlex
+import subprocess
+
+
+def sanitize_relative_path(path: str | os.PathLike) -> str | None:
+    """Sanitizes an untrusted relative file path to prevent Path Traversal.
+
+    Strips null bytes, normalizes path separators, and ensures the path does not
+    escape the workspace or refer to an absolute root path.
+
+    Args:
+        path: Untrusted file path string or PathLike object.
+
+    Returns:
+        The normalized safe relative path string, or None if malicious/invalid.
+    """
+    if not path:
+        return None
+    raw_str = str(path).replace("\x00", "").strip()
+    if not raw_str:
+        return None
+    clean_path = os.path.normpath(raw_str)
+    if clean_path.startswith("..") or os.path.isabs(clean_path):
+        logging.warning("Path traversal attempt or absolute path detected: %s", path)
+        return None
+    return clean_path
+
+
+def sanitize_identifier(value: str) -> str:
+    """Sanitizes an untrusted string for use in branch names, tags, or CLI identifiers.
+
+    Strips null bytes and removes any character not in [a-zA-Z0-9._-].
+
+    Args:
+        value: Untrusted identifier string.
+
+    Returns:
+        A sanitized alphanumeric identifier string (defaults to 'default' if empty).
+    """
+    if not value:
+        return "default"
+    raw_str = str(value).replace("\x00", "")
+    sanitized = re.sub(r"[^a-zA-Z0-9._-]", "", raw_str)
+    return sanitized or "default"
+
+
+class CommandExecutionError(Exception):
+    """Raised when a subprocess fails to run or returns a non-zero exit code."""
+
+    def __init__(
+        self, cmd: str | list[str], returncode: int, stdout: str, stderr: str
+    ) -> None:
+        """Initializes the error with command results."""
+        cmd_str = " ".join(cmd) if isinstance(cmd, list) else cmd
+        super().__init__(f"Command '{cmd_str}' failed with exit code {returncode}")
+        self.cmd = cmd_str
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class CommandExecutor:
+    """Utility class to execute system-level commands and handle failures."""
+
+    @staticmethod
+    def run(
+        cmd: str | list[str],
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        timeout: float = 3600.0,
+    ) -> str:
+        """Executes a command safely using direct argument lists without shell invocation.
+
+        Args:
+            cmd: The command string or list of argument tokens to execute.
+            cwd: The directory path in which to run the command. Defaults to CWD.
+            env: Custom environment variable dictionary to pass to the process.
+            timeout: Maximum allowed duration in seconds. Defaults to 3600.0s.
+
+        Returns:
+            The trimmed stdout string from the command process.
+
+        Raises:
+            CommandExecutionError: If the process exits with a non-zero status.
+        """
+        active_cwd = cwd or os.getcwd()
+        exec_env = os.environ.copy()
+        if env:
+            exec_env.update(env)
+
+        # Convert string commands into argument tokens, parsing inline KEY=VAL env prefixes
+        if isinstance(cmd, str):
+            tokens = shlex.split(cmd)
+            args: list[str] = []
+            for token in tokens:
+                if "=" in token and not args:
+                    k, v = token.split("=", 1)
+                    exec_env[k] = v
+                else:
+                    args.append(token)
+        else:
+            args = list(cmd)
+
+        cmd_str = " ".join(args)
+        logging.info("Executing command: %s (CWD: %s)", cmd_str, active_cwd)
+
+        try:
+            result = subprocess.run(
+                args,
+                cwd=active_cwd,
+                env=exec_env,
+                capture_output=True,
+                text=True,
+                timeout=timeout,
+                check=False,
+            )
+
+            stdout_str = result.stdout.strip() if result.stdout else ""
+            stderr_str = result.stderr.strip() if result.stderr else ""
+
+            if result.returncode != 0:
+                logging.error(
+                    "Command execution failed: %s (Exit Code: %s)",
+                    cmd_str,
+                    result.returncode,
+                )
+                if stdout_str:
+                    logging.error("Stdout:\n%s", stdout_str)
+                if stderr_str:
+                    logging.error("Stderr:\n%s", stderr_str)
+                raise CommandExecutionError(
+                    cmd=args,
+                    returncode=result.returncode,
+                    stdout=stdout_str,
+                    stderr=stderr_str,
+                )
+
+            return stdout_str
+        except Exception as e:
+            if not isinstance(e, CommandExecutionError):
+                logging.exception(
+                    "An unexpected error occurred during command execution: %s",
+                    cmd_str,
+                )
+            raise

tools/caretaker-agent/cloudrun/pr-generator/workflow/config.py

@@ -0,0 +1,83 @@
+"""Configuration module for the SSR Agent Orchestrator.
+
+This module parses, validates, and holds all configuration parameters and path
+constants needed by the orchestrator. It ensures fast-fail on missing or invalid
+configurations.
+"""
+
+import json
+import os
+from typing import Any
+
+
+class ConfigurationError(Exception):
+    """Raised when configuration loading or validation fails."""
+
+
+class Config:
+    """Manages environmental inputs, paths, and limits for the orchestrator."""
+
+    def __init__(self) -> None:
+        """Initializes the configuration with environment variables and defaults."""
+        # Target repository configuration
+        self.repo_url: str = os.environ.get(
+            "REPO_URL", "https://github.com/joneba-google/gemini-cli-clone"
+        )
+        self.git_token: str | None = os.environ.pop("GIT_TOKEN", None)
+        self.firestore_doc_raw: str | None = os.environ.get("FIRESTORE_DOC")
+        self.firestore_id: str | None = (
+            os.environ.get("FIRESTORE_ID") or os.environ.get("firestore_id")
+        )
+        self.execution_id: str | None = os.environ.get("EXECUTION_ID")
+
+        # Google Cloud Platform configuration
+        self.project_id: str = os.environ.get(
+            "GOOGLE_CLOUD_PROJECT", "gcli-intern-project-2026"
+        )
+        self.location: str = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
+        self.model_name: str = os.environ.get("MODEL_NAME", "gemini-3.5-flash")
+
+        # Global runtime settings
+        try:
+            self.max_attempts: int = max(int(os.environ.get("MAX_ATTEMPTS", "5")), 1)
+        except ValueError:
+            self.max_attempts = 5
+
+        # Workspace directory configuration
+        self.tmp_dir: str = "/tmp"
+        self.pr_dir: str = os.path.join(self.tmp_dir, "pr")
+        self.eval_dir: str = os.path.join(self.tmp_dir, "eval")
+
+        self.repo_name: str = (
+            self.repo_url.rstrip("/").split("/")[-1].replace(".git", "")
+        )
+        self.pr_repo_path: str = os.path.join(self.pr_dir, self.repo_name)
+        self.eval_repo_path: str = os.path.join(self.eval_dir, self.repo_name)
+
+        # Global environment variables to trust the CLI
+        os.environ["GEMINI_CLI_WORKSPACE_TRUSTED"] = "true"
+
+    def load_and_validate_firestore_doc(self) -> dict[str, Any]:
+        """Parses and validates the Firestore JSON input specification.
+
+        Returns:
+            The decoded dictionary of the Firestore document.
+
+        Raises:
+            ConfigurationError: If the document is missing or not valid JSON.
+        """
+        if not self.firestore_doc_raw:
+            raise ConfigurationError(
+                "Environment variable 'FIRESTORE_DOC' is required but was not set."
+            )
+        try:
+            doc_data = json.loads(self.firestore_doc_raw)
+            if not isinstance(doc_data, dict):
+                raise ConfigurationError(
+                    "Firestore document specification must be a JSON object."
+                )
+            return doc_data
+        except json.JSONDecodeError as e:
+            raise ConfigurationError(
+                f"Failed to parse 'FIRESTORE_DOC' as JSON: {e}"
+            ) from e

tools/caretaker-agent/cloudrun/pr-generator/workflow/github_client.py

@@ -0,0 +1,97 @@
+"""GitHub REST API Client module.
+
+Handles GitHub pull request creation and branch push operations cleanly using
+standard urllib to minimize container dependency footprint.
+"""
+
+import json
+import logging
+import urllib.error
+import urllib.request
+
+
+class GitHubClientError(Exception):
+    """Raised when a GitHub API request fails or is rejected."""
+
+
+class GitHubClient:
+    """Lightweight client for communicating with the GitHub v3 REST API."""
+
+    def __init__(self, owner: str, repo: str, token: str | None = None) -> None:
+        """Initializes the GitHub REST Client.
+
+        Args:
+            owner: Owner/organization of the repository.
+            repo: Name of the repository.
+            token: Authentication token. If missing, API calls will fail.
+        """
+        self.owner = owner
+        self.repo = repo
+        self._token = token
+        self._base_url = f"https://api.github.com/repos/{owner}/{repo}/pulls"
+
+    def create_pull_request(
+        self, branch_name: str, title: str, body: str
+    ) -> str:
+        """Submits a POST request to GitHub to create a new Pull Request.
+
+        Args:
+            branch_name: The feature branch to be merged.
+            title: Title of the Pull Request.
+            body: Body description markdown of the Pull Request.
+
+        Returns:
+            The PR number of the successfully created Pull Request as a string.
+
+        Raises:
+            GitHubClientError: If the HTTP request fails or token is missing.
+        """
+        if not self._token:
+            raise GitHubClientError(
+                "GitHub token is missing. Cannot authorize Pull Request creation."
+            )
+
+        data = {
+            "title": title,
+            "body": body,
+            "head": branch_name,
+            "base": "main",
+        }
+
+        req = urllib.request.Request(
+            self._base_url,
+            data=json.dumps(data).encode("utf-8"),
+            headers={
+                "Accept": "application/vnd.github+json",
+                "Authorization": f"Bearer {self._token}",
+                "X-GitHub-Api-Version": "2022-11-28",
+                "Content-Type": "application/json",
+            },
+            method="POST",
+        )
+
+        logging.info(
+            "Sending Pull Request creation request for branch: %s", branch_name
+        )
+        try:
+            with urllib.request.urlopen(req,timeout=60) as response:
+                response_payload = json.loads(response.read().decode("utf-8"))
+                pr_number: str = str(response_payload.get("number", response_payload.get("html_url", "")))
+                logging.info(
+                    "Pull Request created successfully! PR Number: %s", pr_number
+                )
+                return pr_number
+        except urllib.error.URLError as e:
+            if isinstance(e, urllib.error.HTTPError):
+                err_body = e.read().decode("utf-8") if e.fp else "No body content"
+                err_msg = f"HTTP {e.code}: {err_body}"
+            else:
+                err_msg = f"Network Error: {getattr(e, 'reason', e)}"
+
+            logging.error("Failed to create Pull Request: %s", err_msg)
+            raise GitHubClientError(f"GitHub API Error: {err_msg}") from e
+        except Exception as e:
+            logging.exception("Encountered unexpected error during PR creation.")
+            raise GitHubClientError(
+                f"Unexpected API client error: {e}"
+            ) from e

tools/caretaker-agent/cloudrun/pr-generator/workflow/preflight_filter.py

@@ -0,0 +1,66 @@
+"""Preflight test and linting validation filter.
+
+Parses CI tool and unit test terminal outputs to identify and selectively
+bypass known, acceptable test failures (such as specific container/sandbox
+privilege test failures).
+"""
+
+import logging
+import re
+
+_ANSI_ESCAPE_RE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
+
+ALLOWED_SANDBOX_FAILURES: set[str] = {
+    "src/utils/sessionCleanup.test.ts",
+    "src/config/extension-manager-permissions.test.ts",
+    "root-privilege-check",
+    "container-permission-test",
+}
+
+
+def strip_ansi(text: str) -> str:
+    """Removes ANSI terminal styling escape sequences from a string."""
+    return _ANSI_ESCAPE_RE.sub("", text)
+
+
+def is_preflight_failure_allowed(
+    test_output: str,
+    allowed_failures: set[str] = ALLOWED_SANDBOX_FAILURES,
+) -> bool:
+    """Checks if test failures belong strictly to approved container/sandbox exceptions."""
+    clean_output = strip_ansi(test_output)
+    lines = clean_output.splitlines()
+
+    failing_lines = [
+        line for line in lines if "FAIL" in line or "FAILED" in line
+    ]
+    if not failing_lines:
+        return False
+
+    for line in failing_lines:
+        if not any(allowed in line for allowed in allowed_failures):
+            logging.warning("Unapproved preflight test failure detected: %s", line)
+            return False
+
+    logging.info("All detected test failure lines match approved sandbox exceptions.")
+    return True
+
+
+class PreflightFilter:
+    """Utility class to filter ANSI characters and analyze test suite results."""
+
+    @staticmethod
+    def strip_ansi(text: str) -> str:
+        """Removes ANSI terminal styling escape sequences from a string."""
+        return strip_ansi(text)
+
+    @classmethod
+    def should_ignore_preflight_failure(
+        cls,
+        stdout: str | None,
+        stderr: str | None,
+        allowed_failures: set[str] = ALLOWED_SANDBOX_FAILURES,
+    ) -> bool:
+        """Analyzes regression test outputs to see if they can be safely bypassed."""
+        raw_output = (stdout or "") + "\n" + (stderr or "")
+        return is_preflight_failure_allowed(raw_output, allowed_failures)