feat(caretaker): publish workable spec event to ready-for-code Pub/Sub topic (#28588)

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,

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")

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):

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