diff --git a/tools/caretaker-agent/cloudrun/triage-worker/main.py b/tools/caretaker-agent/cloudrun/triage-worker/main.py index 48108202ff..6ce6e5a137 100644 --- a/tools/caretaker-agent/cloudrun/triage-worker/main.py +++ b/tools/caretaker-agent/cloudrun/triage-worker/main.py @@ -7,6 +7,7 @@ from google.cloud import firestore from triage_orchestrator import process_issue_triage from utils.validator import validate_triage_result from utils.egress import send_label_action, send_comment_action +from utils.events import publish_issue_ready_for_code from db.issues_store import IssuesStore, ClaimAction, ReleaseAction FEATURE_CLOSED_COMMENT = ( @@ -146,6 +147,9 @@ def main() -> None: send_label_action( owner, repo, issue_number, [f"effort/{effort.lower()}"] ) + publish_issue_ready_for_code( + owner, repo, issue_number, workable_spec + ) store.release_lock( owner, repo, 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 index c267323f76..fc8ef822da 100644 --- a/tools/caretaker-agent/cloudrun/triage-worker/tests/test_integration_main.py +++ b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_integration_main.py @@ -82,7 +82,8 @@ class TestIntegrationMain(unittest.TestCase): }).encode("utf-8")).decode("utf-8"), "WORKFLOW_EXECUTION_ID": "test-workflow-exec-101", "PROJECT_ID": "test-gcp-project", - "EGRESS_TOPIC_ID": "test-egress-actions" + "EGRESS_TOPIC_ID": "test-egress-actions", + "READY_FOR_CODE_TOPIC_ID": "test-ready-topic" }) self.env_patcher.start() @@ -140,7 +141,10 @@ class TestIntegrationMain(unittest.TestCase): @patch("main.process_issue_triage") @patch("main.send_label_action") - def test_ok_quality_flow(self, mock_send_label, mock_triage): + @patch("main.publish_issue_ready_for_code") + def test_ok_quality_flow( + self, mock_publish_event, mock_send_label, mock_triage + ): """Verifies end-to-end flow for OK quality issues.""" self.stored_data = { "status": "UNTRIAGED", @@ -168,6 +172,9 @@ class TestIntegrationMain(unittest.TestCase): mock_send_label.assert_called_once_with( "owner", "repo", 42, ["effort/small"] ) + mock_publish_event.assert_called_once_with( + "owner", "repo", 42, INTEGRATION_OK_PAYLOAD["workable_spec"] + ) # Verify state transition in store data self.assertEqual(self.stored_data["status"], "TRIAGED") diff --git a/tools/caretaker-agent/cloudrun/triage-worker/tests/test_main.py b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_main.py index 1f3ba513f7..6592e1bbcc 100644 --- a/tools/caretaker-agent/cloudrun/triage-worker/tests/test_main.py +++ b/tools/caretaker-agent/cloudrun/triage-worker/tests/test_main.py @@ -44,7 +44,8 @@ class TestMainExecutionLoop(unittest.TestCase): "ISSUE_DETAILS": encoded, "WORKFLOW_EXECUTION_ID": "exec-123", "PROJECT_ID": "test-project", - "EGRESS_TOPIC_ID": "test-topic" + "EGRESS_TOPIC_ID": "test-topic", + "READY_FOR_CODE_TOPIC_ID": "test-ready-topic" }) self.env_patcher.start() @@ -137,8 +138,14 @@ class TestMainExecutionLoop(unittest.TestCase): @patch("main.process_issue_triage") @patch("main.send_label_action") - def test_main_ok_quality_flow(self, mock_send_label, mock_triage): - """OK quality issues dispatch effort label and release TRIAGED spec.""" + @patch("main.publish_issue_ready_for_code") + def test_main_ok_quality_flow( + self, mock_publish_event, mock_send_label, mock_triage + ): + """ + OK quality issues dispatch effort label, release TRIAGED spec, + and publish ready-for-code event. + """ self.mock_store.acquire_lock.return_value = ClaimAction.PROCEED output = json.dumps({ "triage_metadata": {"quality": "OK", "effort_estimate": "SMALL"}, @@ -162,6 +169,9 @@ class TestMainExecutionLoop(unittest.TestCase): status="TRIAGED", workable_spec=VALID_SPEC, ) + mock_publish_event.assert_called_once_with( + "owner", "repo", 42, VALID_SPEC + ) @patch("main.process_issue_triage") def test_main_failure_triggers_retry_release(self, mock_triage): diff --git a/tools/caretaker-agent/cloudrun/triage-worker/utils/events.py b/tools/caretaker-agent/cloudrun/triage-worker/utils/events.py new file mode 100644 index 0000000000..2822238fa6 --- /dev/null +++ b/tools/caretaker-agent/cloudrun/triage-worker/utils/events.py @@ -0,0 +1,54 @@ +import os +import json +from google.cloud import pubsub_v1 + + +def publish_issue_ready_for_code( + owner: str, repo: str, issue_number: int, workable_spec: dict +) -> None: + """ + Publishes an issue-ready-for-code event to Pub/Sub to trigger the + downstream Code Generation Workflow. + + Args: + owner: GitHub repository owner name. + repo: GitHub repository name. + issue_number: GitHub issue number. + workable_spec: Structured Workable Spec dictionary generated by triage. + """ + project_id = os.environ.get("PROJECT_ID") + topic_id = os.environ.get("READY_FOR_CODE_TOPIC_ID") + + if not project_id: + print("[WORKER] Warning: Missing PROJECT_ID, skipping ready-for-code event.") + return + + if not topic_id: + print( + "[WORKER] Warning: Missing READY_FOR_CODE_TOPIC_ID, " + "skipping ready-for-code event." + ) + return + + payload = { + "github_metadata": { + "owner": owner, + "repo": repo, + "issue_number": issue_number, + }, + "workable_spec": workable_spec, + } + + try: + publisher = pubsub_v1.PublisherClient() + topic_path = publisher.topic_path(project_id, topic_id) + data = json.dumps(payload).encode("utf-8") + future = publisher.publish(topic_path, data) + message_id = future.result() + print( + f"[WORKER] Published ready-for-code event to Pub/Sub ({topic_id}). " + f"Message ID: {message_id}" + ) + except Exception as e: + print(f"[WORKER] Error publishing ready-for-code event to Pub/Sub: {e}") + raise