diff --git a/tools/caretaker-agent/cloudrun/triage-worker/.dockerignore b/tools/caretaker-agent/cloudrun/triage-worker/.dockerignore new file mode 100644 index 0000000000..705caee9b2 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/triage-worker/.dockerignore @@ -0,0 +1,9 @@ +__pycache__/ +*.pyc +*.pyo +*.pyd +.pytest_cache/ +venv/ +experimental/ +tests/ +.env diff --git a/tools/caretaker-agent/cloudrun/triage-worker/.gemini/triage_orchestrator.md b/tools/caretaker-agent/cloudrun/triage-worker/.gemini/triage_orchestrator.md index edccc745bc..4307a2bbca 100644 --- a/tools/caretaker-agent/cloudrun/triage-worker/.gemini/triage_orchestrator.md +++ b/tools/caretaker-agent/cloudrun/triage-worker/.gemini/triage_orchestrator.md @@ -9,7 +9,7 @@ You are a triage coordinator agent. When presented with a GitHub issue: ### Triage Workflow: 1. **Invoke the `quality` skill** to analyze the issue's quality. 2. If the quality is **"OK"**: - - **Codebase Exploration:** Explore the repository codebase using your search and navigation tools (such as `list_dir`, `find_by_name`, and `grep_search`) to locate the actual files, functions, and test files related to the issue. Do not guess or assume file paths. + - **Codebase Exploration:** Explore the repository codebase using your search and navigation tools (such as `list_directory`, `find_file`, and `search_directory`) to locate the actual files, functions, and test files related to the issue. Do not guess or assume file paths. - **Invoke the `effort` skill** to estimate the work required. - **Invoke the `spec_generator` skill** to create the technical implementation plan that follows the strict template. 3. If the quality is **not "OK"** (e.g., SPAM, EMPTY, FEATURE, or NEEDS_INFO), populate empty/default values for the effort and spec fields as specified below. diff --git a/tools/caretaker-agent/cloudrun/triage-worker/Dockerfile b/tools/caretaker-agent/cloudrun/triage-worker/Dockerfile new file mode 100644 index 0000000000..35341c4bd4 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/triage-worker/Dockerfile @@ -0,0 +1,11 @@ +FROM python:3.13-slim +WORKDIR /app +ENV PYTHONUNBUFFERED=1 +RUN apt-get update && apt-get install -y \ + git \ + curl \ + && rm -rf /var/lib/apt/lists/* +RUN git clone https://github.com/google-gemini/gemini-cli.git /opt/gemini-cli +COPY . . +RUN pip install --no-cache-dir -r requirements.txt +CMD ["python", "main.py"] \ No newline at end of file diff --git a/tools/caretaker-agent/cloudrun/triage-worker/requirements.txt b/tools/caretaker-agent/cloudrun/triage-worker/requirements.txt index 09d97bd1f1..fa7b37ec26 100644 --- a/tools/caretaker-agent/cloudrun/triage-worker/requirements.txt +++ b/tools/caretaker-agent/cloudrun/triage-worker/requirements.txt @@ -1,2 +1,5 @@ google-cloud-firestore>=2.15.0, <3.0.0 -google-cloud-pubsub \ No newline at end of file +google-cloud-storage +google-cloud-pubsub +google-antigravity>=0.1.0 +python-dotenv \ No newline at end of file diff --git a/tools/caretaker-agent/cloudrun/triage-worker/tests/test_agent_logger.py b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_agent_logger.py new file mode 100644 index 0000000000..f28275c0f0 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_agent_logger.py @@ -0,0 +1,52 @@ +import unittest +from unittest.mock import patch +from google.antigravity.types import Text +from utils.agent_logger import extract_final_output, log_agent_run +from triage_orchestrator import process_issue_triage + + +class TestAgentLogger(unittest.TestCase): + + def test_extract_final_output(self): + """Verifies final step output extraction and step filtering.""" + self.assertEqual(extract_final_output(None), "") + self.assertEqual(extract_final_output([]), "") + chunks = [ + Text(text="Thought 1", step_index=0), + Text(text="Result: SUCCESS", step_index=1), + Text(text=" Additional", step_index=1), + ] + self.assertEqual( + extract_final_output(chunks), "Result: SUCCESS Additional" + ) + + @patch("utils.agent_logger.upload_to_bucket") + def test_log_agent_run(self, mock_upload): + """Verifies log routing to GCS and serialization behavior.""" + chunks = [Text(text="Thought", step_index=0)] + + # Test OFF mode (should not call upload) + log_agent_run("repo", 42, chunks, mode="OFF") + mock_upload.assert_not_called() + + # Test GCS mode (should call upload and serialize the chunks) + log_agent_run("repo", 42, chunks, mode="GCS") + mock_upload.assert_called_once() + + # Verify GCS upload payload contains our serialized text + args, _ = mock_upload.call_args + self.assertIn('"text": "Thought"', args[2]) + + @patch("triage_orchestrator.upload_to_bucket") + @patch("triage_orchestrator.Agent") + def test_process_issue_triage_error(self, mock_agent, mock_upload): + """Verifies error handling and GCS upload on SDK failures.""" + mock_agent.return_value.__aenter__.side_effect = Exception("API Error") + success, raw_output = process_issue_triage({"issue_number": 42}) + self.assertFalse(success) + self.assertIn("API Error", raw_output) + mock_upload.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/caretaker-agent/cloudrun/triage-worker/tests/test_integration_main.py b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_integration_main.py new file mode 100644 index 0000000000..dc38e3af07 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_integration_main.py @@ -0,0 +1,299 @@ +""" +Integration tests for main.py. + +Verifies workflow execution across main.py and issues_store.py. +External network boundaries (Firestore database client and LLM inference) +are mocked. +""" + +import unittest +from unittest.mock import patch, MagicMock +import os +import json +import base64 +from db.issues_store import IssuesStore, ClaimAction, ReleaseAction +import main as main_module +from main import main + +VALID_WORKABLE_SPEC = { + "issue_id": "owner/repo#42", + "summary": {"problem": "p", "root_cause": "r", "context": "c"}, + "implementation_plan": { + "files_to_modify": ["src/app.ts"], "steps": ["Fix bug"] + }, + "testing_strategy": { + "test_file": "tests/app.test.ts", + "expected_behavior": "Pass", + "verification_steps": ["Check"], + "framework": "Vitest" + } +} + +INTEGRATION_OK_PAYLOAD = { + "triage_metadata": { + "quality": "OK", + "reasoning": "Actionable bug report.", + "comment": "", + "effort_estimate": "SMALL", + "effort_reasoning": "Easy fix." + }, + "workable_spec": VALID_WORKABLE_SPEC +} + +INTEGRATION_NEEDS_INFO_PAYLOAD = { + "triage_metadata": { + "quality": "NEEDS_INFO", + "reasoning": ( + "The issue reports a crash on startup, but lacks any actual " + "details." + ), + "comment": ( + "Hi! Thanks for commenting on this issue, we need more " + "information to triage the bug." + ), + "effort_estimate": "", + "effort_reasoning": "" + }, + "workable_spec": {} +} + +INTEGRATION_INVALID_EFFORT_PAYLOAD = { + "triage_metadata": { + "quality": "OK", + "reasoning": "Some reasoning.", + "comment": "", + "effort_estimate": "HUGE", + "effort_reasoning": "This will take a while to fix." + }, + "workable_spec": VALID_WORKABLE_SPEC +} + + +class TestIntegrationMain(unittest.TestCase): + + def setUp(self): + # Mock environment variables + self.env_patcher = patch.dict(os.environ, { + "ISSUE_DETAILS": base64.b64encode(json.dumps({ + "issue_number": 42, + "repository": "owner/repo", + "title": "Fix crash", + "body": "App crashes on start" + }).encode("utf-8")).decode("utf-8"), + "WORKFLOW_EXECUTION_ID": "test-workflow-exec-101", + "PROJECT_ID": "test-gcp-project", + "EGRESS_TOPIC_ID": "test-egress-actions" + }) + self.env_patcher.start() + + # Mock the Firestore database client at the network boundary + self.mock_db = MagicMock() + self.db_patcher = patch( + "main.firestore.Client", return_value=self.mock_db + ) + self.db_patcher.start() + + self.mock_doc_ref = MagicMock() + self.mock_snapshot = MagicMock() + self.mock_transaction = MagicMock() + + self.mock_db.collection.return_value.document.return_value = ( + self.mock_doc_ref + ) + self.mock_db.transaction.return_value = self.mock_transaction + self.mock_doc_ref.get.return_value = self.mock_snapshot + self.mock_snapshot.exists = True + + # In-memory document state simulation + self.stored_data = {} + self.mock_snapshot.to_dict.side_effect = lambda: self.stored_data + + def mock_update(doc_ref, updates): + if "status" in updates: + self.stored_data["status"] = updates["status"] + if "workable_spec" in updates: + self.stored_data["workable_spec"] = updates["workable_spec"] + if "lock.holder" in updates: + if "lock" not in self.stored_data: + self.stored_data["lock"] = {} + self.stored_data["lock"]["holder"] = updates["lock.holder"] + + # Bind mock_update to execute whenever transaction.update is invoked + self.mock_transaction.update.side_effect = mock_update + + # Mock IssuesStore instance + self.mock_store = MagicMock() + self.store_patcher = patch( + "main.IssuesStore", return_value=self.mock_store + ) + self.store_patcher.start() + + # Wire mock_store methods to execute real store logic against mock_db + real_store = IssuesStore(self.mock_db, "issues") + self.mock_store.acquire_lock.side_effect = real_store.acquire_lock + self.mock_store.release_lock.side_effect = real_store.release_lock + + def tearDown(self): + self.store_patcher.stop() + self.db_patcher.stop() + self.env_patcher.stop() + + @patch("main.process_issue_triage") + @patch("main.send_label_action") + def test_ok_quality_flow(self, mock_send_label, mock_triage): + """Verifies end-to-end flow for OK quality issues.""" + self.stored_data = { + "status": "UNTRIAGED", + "triage_attempts": 0, + "lock": {"holder": None, "expires_at": None} + } + mock_triage.return_value = (True, json.dumps(INTEGRATION_OK_PAYLOAD)) + + with self.assertRaises(SystemExit) as ctx: + main() + + self.assertEqual(ctx.exception.code, 0) + self.mock_store.acquire_lock.assert_called_once_with( + "owner", "repo", 42, "test-workflow-exec-101" + ) + self.mock_store.release_lock.assert_called_once_with( + "owner", + "repo", + 42, + "test-workflow-exec-101", + success=True, + status="TRIAGED", + workable_spec=INTEGRATION_OK_PAYLOAD["workable_spec"], + ) + mock_send_label.assert_called_once_with( + "owner", "repo", 42, ["effort/small"] + ) + + # Verify state transition in store data + self.assertEqual(self.stored_data["status"], "TRIAGED") + self.assertEqual( + self.stored_data["workable_spec"], VALID_WORKABLE_SPEC + ) + self.assertIsNone(self.stored_data["lock"]["holder"]) + + @patch("main.process_issue_triage") + @patch("main.send_comment_action") + def test_needs_info_flow(self, mock_send_comment, mock_triage): + """Verifies end-to-end flow for NEEDS_INFO issues.""" + self.stored_data = { + "status": "UNTRIAGED", + "triage_attempts": 0, + "lock": {"holder": None, "expires_at": None} + } + payload_data = json.dumps(INTEGRATION_NEEDS_INFO_PAYLOAD) + mock_triage.return_value = (True, payload_data) + + with self.assertRaises(SystemExit) as ctx: + main() + + self.assertEqual(ctx.exception.code, 0) + self.mock_store.acquire_lock.assert_called_once_with( + "owner", "repo", 42, "test-workflow-exec-101" + ) + self.mock_store.release_lock.assert_called_once_with( + "owner", + "repo", + 42, + "test-workflow-exec-101", + success=True, + status="NEEDS_INFO", + ) + expected_comment = ( + INTEGRATION_NEEDS_INFO_PAYLOAD["triage_metadata"]["comment"] + ) + mock_send_comment.assert_called_once_with( + "owner", "repo", 42, expected_comment + ) + + self.assertEqual(self.stored_data["status"], "NEEDS_INFO") + self.assertIsNone(self.stored_data["lock"]["holder"]) + + @patch("main.process_issue_triage") + @patch("main.send_label_action") + def test_auto_close_flows(self, mock_send_label, mock_triage): + """Verifies end-to-end flow for auto-closed issues.""" + for quality in ["SPAM", "EMPTY", "FEATURE"]: + self.mock_store.acquire_lock.reset_mock() + self.mock_store.release_lock.reset_mock() + mock_send_label.reset_mock() + mock_triage.reset_mock() + + self.stored_data = { + "status": "UNTRIAGED", + "triage_attempts": 0, + "lock": {"holder": None, "expires_at": None} + } + payload = {"triage_metadata": {"quality": quality}} + mock_triage.return_value = (True, json.dumps(payload)) + + with self.assertRaises(SystemExit) as ctx: + main() + + self.assertEqual(ctx.exception.code, 0) + self.mock_store.acquire_lock.assert_called_once_with( + "owner", "repo", 42, "test-workflow-exec-101" + ) + self.mock_store.release_lock.assert_called_once_with( + "owner", + "repo", + 42, + "test-workflow-exec-101", + success=True, + status="AUTO_CLOSE", + ) + mock_send_label.assert_called_once_with( + "owner", "repo", 42, ["auto-close"] + ) + + self.assertEqual(self.stored_data["status"], "AUTO_CLOSE") + self.assertIsNone(self.stored_data["lock"]["holder"]) + + @patch("main.process_issue_triage") + def test_validation_failure_triggers_retry(self, mock_triage): + """Verifies retry state transition when validation fails.""" + self.stored_data = { + "status": "UNTRIAGED", + "triage_attempts": 0, + "lock": {"holder": None, "expires_at": None} + } + payload_data = json.dumps(INTEGRATION_INVALID_EFFORT_PAYLOAD) + mock_triage.return_value = (True, payload_data) + + with self.assertRaises(SystemExit) as ctx: + main() + + self.assertEqual(ctx.exception.code, 1) + self.mock_store.acquire_lock.assert_called_once_with( + "owner", "repo", 42, "test-workflow-exec-101" + ) + self.mock_store.release_lock.assert_called_once_with( + "owner", "repo", 42, "test-workflow-exec-101", success=False + ) + self.assertEqual(self.stored_data["status"], "UNTRIAGED") + self.assertIsNone(self.stored_data["lock"]["holder"]) + + def test_max_attempts_escalates_to_needs_human(self): + """Verifies escalation to NEEDS_HUMAN when triage_attempts >= 2.""" + self.stored_data = { + "status": "UNTRIAGED", + "triage_attempts": 2, + "lock": {"holder": None, "expires_at": None}, + } + + with self.assertRaises(SystemExit) as ctx: + main() + + self.assertEqual(ctx.exception.code, 0) + self.mock_store.acquire_lock.assert_called_once_with( + "owner", "repo", 42, "test-workflow-exec-101" + ) + self.assertEqual(self.stored_data["status"], "NEEDS_HUMAN") + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/caretaker-agent/cloudrun/triage-worker/triage_orchestrator.py b/tools/caretaker-agent/cloudrun/triage-worker/triage_orchestrator.py index 53d53f7cd6..1371740e99 100644 --- a/tools/caretaker-agent/cloudrun/triage-worker/triage_orchestrator.py +++ b/tools/caretaker-agent/cloudrun/triage-worker/triage_orchestrator.py @@ -1,16 +1,95 @@ -""" -Handles LLM inference for issue triage. -(Stubbed implementation for execution loop integration). -""" +import os +import asyncio +from utils.agent_logger import ( + upload_to_bucket, + log_agent_run, + extract_final_output, +) +from google.antigravity import Agent, LocalAgentConfig +from google.antigravity.hooks.policy import allow, deny def process_issue_triage(payload: dict) -> tuple[bool, str]: """ - Stubbed entrypoint for issue triage processing. - - Args: - payload: Dictionary containing issue details (issue_number, repository). - - Returns: - tuple[bool, str]: (success, raw_output) + LLM inference via Antigravity SDK. """ - return False, "Not implemented: LLM triage orchestrator" + issue_num = payload.get("issue_number") + title = payload.get("title", "") + body = payload.get("body", "") + repo_name = payload.get("repository", "") + + current_dir = os.path.dirname(os.path.abspath(__file__)) + system_prompt_path = os.path.join( + current_dir, ".gemini", "triage_orchestrator.md" + ) + target_cwd = os.environ.get("TARGET_CWD", "/opt/gemini-cli") + gcs_logging = os.environ.get("GCS_LOGGING", "GCS").upper() + + policies = [ + # Deny all tools by default + deny("*"), + + # Whitelist specific read-only and skill tools + allow("view_file"), + allow("list_directory"), + allow("find_file"), + allow("search_directory"), + allow("activate_skill"), + allow("finish") + ] + + with open(system_prompt_path, "r", encoding="utf-8") as f: + system_instructions = f.read() + + skills_dir = os.path.join(current_dir, ".gemini", "skills") + prompt = ( + f"Repository: {repo_name}\n" + f"Issue Number: {issue_num}\n" + f"Title: {title}\n" + f"Description: {body}" + ) + + async def run_triage(): + config = LocalAgentConfig( + system_instructions=system_instructions, + skills_paths=[skills_dir], + api_key=os.environ.get("GEMINI_API_KEY"), + workspaces=[target_cwd, skills_dir], + policies=policies, + ) + + print( + f"[LOGIC] [Issue #{issue_num}] Initializing Antigravity Agent..." + ) + async with Agent(config) as agent: + print( + f"[LOGIC] [Issue #{issue_num}] Sending triage request..." + ) + response = await agent.chat(prompt) + + # Resolve all execution chunks (thoughts, tool calls, and results) + resolved_chunks = await response.resolve() + + # Extract the final step's output + text_output = extract_final_output(resolved_chunks) + + log_agent_run( + repo_name, + issue_num, + resolved_chunks, + mode=gcs_logging, + ) + + print(f"[LOGIC] Agent Response:\n{text_output}") + + return True, text_output + + try: + success, raw_output = asyncio.run(run_triage()) + return success, raw_output + except Exception as e: + error_msg = f"Error during Antigravity Agent run: {e}" + print(f"[LOGIC] {error_msg}") + if gcs_logging == "GCS": + # If agent failed/crashed before chunks resolved, upload traceback string directly + upload_to_bucket(repo_name, issue_num, error_msg) + return False, error_msg diff --git a/tools/caretaker-agent/cloudrun/triage-worker/utils/agent_logger.py b/tools/caretaker-agent/cloudrun/triage-worker/utils/agent_logger.py new file mode 100644 index 0000000000..9623067052 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/triage-worker/utils/agent_logger.py @@ -0,0 +1,119 @@ +import json +import datetime +import os +import uuid +from google.cloud import storage +from google.antigravity.types import Text + +BUCKET_NAME = os.environ.get("TRIAGE_DEBUG_LOGS_BUCKET") +_storage_client = None + + +def _get_storage_client() -> storage.Client: + global _storage_client + if _storage_client is None: + _storage_client = storage.Client() + return _storage_client + + +def upload_to_bucket(repository: str, issue_number: str | int, payload: str) -> None: + """ + Uploads a string payload directly to the triage debug logs GCS bucket. + """ + if not payload or not BUCKET_NAME: + if not BUCKET_NAME: + print("[LOGIC] Warning: Missing TRIAGE_DEBUG_LOGS_BUCKET, skipping GCS upload.") + return + try: + storage_client = _get_storage_client() + bucket = storage_client.bucket(BUCKET_NAME) + + timestamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d_%H%M%S") + safe_repo = str(repository).replace("/", "_") if repository else "unknown" + unique_id = uuid.uuid4().hex[:8] + blob_name = f"{safe_repo}/issue_{issue_number}_{timestamp}_{unique_id}_debug.log" + + blob = bucket.blob(blob_name) + blob.upload_from_string(payload, content_type="text/plain") + print(f"[LOGIC] Uploaded debug logs to gs://{BUCKET_NAME}/{blob_name}") + except Exception as e: + print(f"[LOGIC] Error uploading debug logs to GCS: {e}") + +def _format_debug_trajectory(resolved_chunks: list) -> str: + """ + Formats Antigravity Agent resolved stream chunks (thoughts, tool calls, + tool outputs) into a structured, human-readable JSON string. + """ + if not resolved_chunks: + return "[]" + serializable = [] + for chunk in resolved_chunks: + try: + dumped = chunk.model_dump() + except AttributeError: + dumped = chunk.dict() + dumped["chunk_type"] = chunk.__class__.__name__ + serializable.append(dumped) + return json.dumps(serializable, indent=2, default=str) + + +def log_agent_run( + repository: str, + issue_number: str | int, + resolved_chunks: list, + mode: str = "GCS", +) -> None: + """ + Logs the agent execution trajectory based on the mode parameter. + + Modes: + - "LOCAL": Saves log file to local disk (requires LOCAL_LOG_DIR env var). + - "GCS": Uploads log file to the GCS bucket. + - Any other mode (e.g. "OFF"): Skips logging. + """ + mode_upper = str(mode).upper() + if not resolved_chunks or mode_upper not in ("LOCAL", "GCS"): + return + + try: + debug_log_str = _format_debug_trajectory(resolved_chunks) + + if mode_upper == "LOCAL": + local_dir = os.environ.get("LOCAL_LOG_DIR") + if not local_dir: + print( + "[LOGIC] Error: LOCAL_LOG_DIR env var is not configured." + ) + return + os.makedirs(local_dir, exist_ok=True) + log_path = os.path.join( + local_dir, f"gemini_cli_{issue_number}_debug.json" + ) + with open(log_path, "w", encoding="utf-8") as f: + f.write(debug_log_str) + print(f"[LOGIC] 📄 Saved local agent trajectory log to: {log_path}") + + elif mode_upper == "GCS": + upload_to_bucket(repository, issue_number, debug_log_str) + + except Exception as e: + print(f"[LOGIC] Error logging agent run: {e}") + +def extract_final_output(resolved_chunks: list) -> str: + """ + Extracts the final response text generated by the agent during the last execution step. + + Since the agent outputs conversational planning text during its tool-calling turns, + this function isolates the final turn (highest step_index) and filters for text chunks, + stripping away any intermediate planning text, thoughts, or markdown backticks around the JSON. + """ + if not resolved_chunks: + return "" + # Filter for Text chunks (skip Thought and ToolCall chunks) + text_chunks = [c for c in resolved_chunks if isinstance(c, Text)] + if not text_chunks: + return "" + + # Final output is in the last step + last_step = text_chunks[-1].step_index + return "".join(c.text for c in text_chunks if c.step_index == last_step).strip() \ No newline at end of file