fix(samples): separate skipped from failed issues in the monitoring agent

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 963583295

Author: GWeale

contributing/samples/adk_team/adk_issue_monitoring_agent/main.py

@@ -15,7 +15,9 @@
 import asyncio
 import logging
 import re
+import sys
 import time
+from typing import Literal
 
 from adk_issue_monitoring_agent.agent import root_agent
 from adk_issue_monitoring_agent.settings import BOT_ALERT_SIGNATURE
@@ -40,12 +42,17 @@
 APP_NAME = "issue_monitoring_app"
 USER_ID = "issue_monitoring_user"
 
+# An issue is skipped when there is deliberately nothing to review, which is a
+# normal outcome and not a failure.
+_Outcome = Literal["audited", "skipped", "failed"]
+
 
 async def process_single_issue(
     runner: InMemoryRunner, issue_number: int, maintainers: list[str]
-) -> tuple[float, int]:
+) -> tuple[float, int, _Outcome]:
   start_time = time.perf_counter()
   start_api_calls = get_api_call_count()
+  outcome: _Outcome = "audited"
 
   try:
     # 1. Fetch the main issue AND the comments
@@ -87,6 +94,7 @@ async def process_single_issue(
         return (
             time.perf_counter() - start_time,
             get_api_call_count() - start_api_calls,
+            "skipped",
         )
 
       if (
@@ -111,6 +119,7 @@ async def process_single_issue(
       return (
           time.perf_counter() - start_time,
           get_api_call_count() - start_api_calls,
+          "skipped",
       )
 
     logger.info(
@@ -147,14 +156,15 @@ async def process_single_issue(
 
   except Exception as e:
     logger.error(f"Error processing issue #{issue_number}: {e}", exc_info=True)
+    outcome = "failed"
 
   # Calculate duration and API calls regardless of success or failure
   duration = time.perf_counter() - start_time
   issue_api_calls = get_api_call_count() - start_api_calls
-  return duration, issue_api_calls
+  return duration, issue_api_calls, outcome
 
 
-async def main():
+async def main() -> int:
   logger.info(f"--- Starting Issue Monitoring Agent for {OWNER}/{REPO} ---")
   reset_api_call_count()
 
@@ -164,25 +174,29 @@ async def main():
     logger.info(f"Found {len(maintainers)} maintainers.")
   except Exception as e:
     logger.critical(f"Failed to fetch maintainers: {e}")
-    return
+    return 1
 
   # Step 2: Fetch target issues
   try:
     all_issues = get_target_issues(OWNER, REPO)
   except Exception as e:
     logger.critical(f"Failed to fetch issue list: {e}")
-    return
+    return 1
 
   total_count = len(all_issues)
   if total_count == 0:
     logger.info("No issues matched criteria. Run finished.")
-    return
+    return 0
 
   logger.info(f"Found {total_count} issues to process.")
 
   # Initialize the runner ONCE for the entire run
   runner = InMemoryRunner(agent=root_agent, app_name=APP_NAME)
 
+  audited_count = 0
+  skipped_count = 0
+  failed_count = 0
+
   # Step 3: Iterate through issues async 'CONCURRENCY_LIMIT' at a time
   for i in range(0, total_count, CONCURRENCY_LIMIT):
     chunk = all_issues[i : i + CONCURRENCY_LIMIT]
@@ -192,13 +206,28 @@ async def main():
         process_single_issue(runner, issue_num, maintainers)
         for issue_num in chunk
     ]
-    await asyncio.gather(*tasks)
+    results = await asyncio.gather(*tasks)
+
+    for _, _, outcome in results:
+      if outcome == "audited":
+        audited_count += 1
+      elif outcome == "skipped":
+        skipped_count += 1
+      else:
+        failed_count += 1
 
     if (i + CONCURRENCY_LIMIT) < total_count:
       await asyncio.sleep(SLEEP_BETWEEN_CHUNKS)
 
   logger.info(f"--- Run Finished. Total API calls: {get_api_call_count()} ---")
+  logger.info(f"Successfully processed {audited_count} issues.")
+  if skipped_count:
+    logger.info(f"Skipped {skipped_count} issues.")
+  if failed_count:
+    logger.error(f"Failed to process {failed_count} issues.")
+
+  return 1 if failed_count else 0
 
 
 if __name__ == "__main__":
-  asyncio.run(main())
+  sys.exit(asyncio.run(main()))

tests/unittests/test_samples.py

@@ -354,3 +354,102 @@ async def run_async(
   assert ("Failed to process 1 issues." in caplog.text) == (
       failing_issue is not None
   )
+
+
+@pytest.mark.parametrize(
+    "failing_issue, expected_exit_code", [(None, 0), (4, 1)]
+)
+async def test_issue_monitoring_agent_separates_skips_from_failures(
+    failing_issue: int | None,
+    expected_exit_code: int,
+    monkeypatch,
+    caplog,
+):
+  """A skipped issue is not a failure, and a failed one is not a success."""
+  for key, value in _DUMMY_ENV.items():
+    monkeypatch.setenv(key, value)
+
+  reviewed: list[int] = []
+
+  class _FakeSession:
+    id = "fake-session"
+
+  class _FakeSessionService:
+
+    async def create_session(
+        self, *, user_id: str, app_name: str
+    ) -> _FakeSession:
+      return _FakeSession()
+
+  class _FakeRunner:
+    """Stands in for InMemoryRunner, failing the audit of one issue."""
+
+    def __init__(self, *, agent: Any, app_name: str) -> None:
+      self.session_service = _FakeSessionService()
+
+    async def run_async(
+        self, *, user_id: str, session_id: str, new_message: types.Content
+    ) -> AsyncIterator[Event]:
+      text = new_message.parts[0].text
+      issue_number = int(text.split("#")[1].split(":")[0])
+      reviewed.append(issue_number)
+      if issue_number == failing_issue:
+        raise RuntimeError("model backend unavailable")
+      yield Event(
+          author="agent",
+          content=types.Content(
+              role="model", parts=[types.Part(text="Not spam.")]
+          ),
+      )
+
+  with _sample_module(
+      SAMPLES_DIR / "adk_team" / "adk_issue_monitoring_agent", "main"
+  ) as main_module:
+    # 1 and 4 are audited, 2 is skipped because the bot already alerted on it,
+    # 3 is skipped because only a maintainer has written on it.
+    details = {
+        n: {"user": {"login": "maintainer"}, "body": "tracking"} for n in (2, 3)
+    }
+    details[1] = {"user": {"login": "outsider"}, "body": "buy things"}
+    details[4] = {"user": {"login": "outsider"}, "body": "buy more things"}
+    comments = {
+        2: [{
+            "user": {"login": main_module.BOT_NAME},
+            "body": main_module.BOT_ALERT_SIGNATURE,
+        }],
+        3: [{"user": {"login": "maintainer"}, "body": "still looking"}],
+    }
+
+    monkeypatch.setattr(main_module, "InMemoryRunner", _FakeRunner)
+    monkeypatch.setattr(main_module, "SLEEP_BETWEEN_CHUNKS", 0)
+    monkeypatch.setattr(
+        main_module,
+        "get_repository_maintainers",
+        lambda owner, repo: ["maintainer"],
+    )
+    monkeypatch.setattr(
+        main_module, "get_target_issues", lambda owner, repo: [1, 2, 3, 4]
+    )
+    monkeypatch.setattr(
+        main_module,
+        "get_issue_details",
+        lambda owner, repo, issue_number: details[issue_number],
+    )
+    monkeypatch.setattr(
+        main_module,
+        "get_issue_comments",
+        lambda owner, repo, issue_number: comments.get(issue_number, []),
+    )
+    with caplog.at_level(logging.INFO, logger="google_adk"):
+      exit_code = await main_module.main()
+
+  assert exit_code == expected_exit_code
+  # Every reviewable issue still reaches the agent: one failure must not abort
+  # the batch.
+  assert sorted(reviewed) == [1, 4]
+  expected_successes = 2 if failing_issue is None else 1
+  assert f"Successfully processed {expected_successes} issues." in caplog.text
+  assert "Skipped 2 issues." in caplog.text
+  assert ("Failed to process 1 issues." in caplog.text) == (
+      failing_issue is not None
+  )