Feat/ambient expense agent runtime

[eliasecchig]

Convert existing ambient sample from Cloud Run to agent runtime

[github-actions]

📋 Security Review Summary

I have performed a security-focused code review of the changes introduced in this PR. The changes primarily focus on migrating the backend of the ambient-expense-agent from Cloud Run to Agent Runtime, modifying FastAPI backend/frontend wiring, and updating the supporting Terraform infrastructure.

🔍 Findings

• 🟢 No security vulnerabilities were found in the changed lines.
• Least-Privilege IAM: Service accounts (e.g. expense-agent-invoker, approval-ui-invoker) are properly restricted and granted target permissions via least-privilege roles like roles/aiplatform.user.
• Identity-Aware Proxy (IAP): The Cloud Run frontend service is correctly configured with iap_enabled = true and restricts ingress specifically to the authorized user email (roles/iap.httpsResourceAccessor).
• Secure Authentication: Backend API communication uses OAuth2 access tokens obtained via standard Application Default Credentials.
• No Secrets Exposure: Review of configuration, environment, and code files confirmed that no hardcoded credentials or secrets were introduced.

[github-actions]

📋 Maintainability Review Summary

We have completed our maintainability review of the pull request. The codebase migration to Agent Runtime is solid, clear, and well-implemented. We identified a couple of minor code duplication issues that could be resolved to improve future maintainability.

🔍 Findings

• Makefile: The inline python command used to extract the Agent Runtime resource ID from deployment_metadata.json is repeated exactly 3 times.
• Terraform: The string representation for the Agent Runtime API base URL is duplicated across several Terraform configuration files.

[github-actions]

📋 Security Review Summary

I have completed a security-focused code review of the pull request changes. Two significant security and architectural authorization issues were identified:

1. 🟠 High: Broken Object Level Authorization (BOLA/IDOR) & Privilege Escalation in the frontend's /approve endpoint, resulting from the frontend acting as a confused deputy with its highly-privileged service account credentials.
2. 🟡 Medium: Authentication Failure on Pub/Sub Push Subscription (pubsub.tf) due to OIDC vs. OAuth2 Access Token mismatch.

Detailed findings and remediation recommendations have been provided as inline comments below.

🔍 Findings

• Broken Object Level Authorization (BOLA/IDOR): Exists in the unauthenticated /approve proxy in frontend/main.py when forwarding user-controlled payloads using the frontend's highly-privileged service account.
• Authentication Mismatch on Pub/Sub Push: The Pub/Sub push configuration in pubsub.tf uses an OIDC ID token which will be rejected by the Vertex AI API.

[github-actions]

📋 Correctness Review Summary

This review focuses on the correctness of changes introduced in the Pull Request. The changes are generally solid, but we have identified a high-severity bug and a medium-severity reliability issue in python/agents/financial-advisor/financial_advisor/__init__.py that can lead to runtime TypeError exceptions or import-time crashes under various environments.

---

🔍 Findings & Recommendations

1. 🔴 High Severity: os.environ.setdefault Type Error in __init__.py

Description:
The code attempts to set the default value of the GOOGLE_CLOUD_PROJECT environment variable using project_id returned by google.auth.default():

_, project_id = google.auth.default()
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", project_id)

If standard Google Cloud credentials can be resolved but do not define a default project ID (which is common for user credentials, local service accounts, or specific credential configurations), project_id will be None.
In Python, calling os.environ.setdefault with None as the second argument raises a runtime exception:
TypeError: str expected, not NoneType
This will immediately crash the entire application upon import.

Recommendation / Corrected Code Suggestion:
Only attempt to set the environment variable if project_id is successfully resolved:

if project_id:
    os.environ.setdefault("GOOGLE_CLOUD_PROJECT", project_id)

---

2. 🟠 High Severity: Import-Time DefaultCredentialsError Crash

Description:
Calling google.auth.default() at the top-level of __init__.py runs immediately on module import. If standard application-default credentials (ADC) are not configured in the execution environment (for instance, during local offline test runs, CI code formatting/linting runs, or automated static analysis), google.auth.default() raises google.auth.exceptions.DefaultCredentialsError.
This completely prevents the financial_advisor package from being imported, causing tests, linters, or setup scripts to crash before they can even run (or before testing frameworks can apply proper mocks).

Recommendation / Corrected Code Suggestion:
Wrap the google.auth.default() invocation in a try-except block to allow the module to be imported and run in environments where credentials are not pre-configured.

---

💡 Corrected Code Block

Here is the complete recommended code for python/agents/financial-advisor/financial_advisor/__init__.py:

import os

import google.auth
from google.auth.exceptions import DefaultCredentialsError

try:
    _, project_id = google.auth.default()
    if project_id:
        os.environ.setdefault("GOOGLE_CLOUD_PROJECT", project_id)
except DefaultCredentialsError:
    # Safely handle environments where default credentials are not pre-configured during import (e.g. CI, offline local dev)
    pass

os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "global")
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "True")

[github-actions]

📋 Maintainability Review Summary

This review focuses on the maintainability of the newly introduced Agent Runtime backend code and configuration for the ambient expense agent. The overall code quality is high, employing modern Python typing and clean FastAPI routing. A few minor cleanups—primarily removing dead code/unused variables and reducing minor string repetition—will ensure long-term readability and compliance with the project's standards.

🔍 Findings

• Unused variables/imports: Cleaned up inline (unused _AGENT_DIR in services.py and unused unpacked project_id in fast_api_app.py).
• Telemetry variable key repetition: Addressed inline (potential duplication of the telemetry environment variable key in telemetry.py).
• No other major maintainability issues (like duplication of business logic, poor naming, or magic numbers) were found.

[github-actions]

📋 Security Review Summary

This security review analyzed the PR changes migrating the ambient-expense-agent from Cloud Run to Vertex AI Agent Runtime (Reasoning Engine). The review focused on identifying concrete security vulnerabilities including hardcoded secrets, SQL/command injection, unsafe deserialization, missing input validation, and insecure access controls.

🔍 Findings

• No security vulnerabilities were found in the changed lines of this pull request.
• The IAM service accounts, least-privilege roles, and OIDC push authentication configurations for Pub/Sub follow standard security best practices.

[github-actions]

📋 Correctness Review Summary

I have performed a correctness review of the changes introduced in this PR.

I identified a minor data propagation gap in fast_api_app.py where the resolved Google Cloud project ID is not set in the environment, causing dynamic session service resolution to fall back unnecessarily. I also identified a potential command setup failure in the local Makefile installation rule. Suggested fixes have been provided inline for both. Other changes are correct and follow robust patterns.

🔍 Findings

• Green / Low (Correctness): Missing environment propagation of default project ID resolved via google.auth.default() in fast_api_app.py. Inline suggestion provided.
• Green / Low (Correctness / Setup): Non-portable and disjoint environment command invocation in the Makefile's install target. Inline suggestion provided.

[github-actions]

📋 Security Review Summary

The security review of the proposed changes in the pull request did not identify any concrete or exploitable security vulnerabilities. The implementation successfully transitions the ambient expense agent to Vertex AI Agent Runtime and properly adheres to GCP security best practices, including least-privilege IAM service accounts, authenticated Pub/Sub push triggers with OIDC, and Google Cloud Identity-Aware Proxy (IAP) for frontend access control.

🔍 Findings

• No security vulnerabilities were found. All security controls are configured correctly.

[github-actions]

📋 Maintainability Review Summary

I have conducted a thorough maintainability review of this pull request, focusing exclusively on identifying potential naming issues, magic numbers, or code duplication on the modified/added lines.

Overall, the codebase migration from Cloud Run to Vertex AI Agent Runtime has been designed and implemented to an exceptionally high standard. The code is modular, beautifully structured, and highly readable.

🔍 Findings

• Makefile (Duplication - Low): There is duplication of the terraform destroy command with its five variables across the initial destroy attempt and the subsequent retry block. Consider extracting this command into a variable to streamline updates.
• Python / Terraform: All other files exhibit high maintainability, consistent naming conventions, clear configurations, and no code duplication.

[github-actions]

🟢 Duplicated terraform destroy command in the retry block of the clean target. To make the Makefile more maintainable, consider extracting this command into a reusable variable or helper block, reducing duplication and ensuring that any future variable updates are consistently applied to both destroy attempts.

[github-actions]

📋 Review Summary

This correctness-focused code review has evaluated the ambient expense agent implementation under python/agents/ambient-expense-agent/. While the integration of the graph-based workflows, Pub/Sub trigger handling, and human-in-the-loop interfaces is conceptually solid, several critical-to-medium correctness and reliability issues were identified in the codebase. This feedback covers workflow error propagation, middleware payload validation, and UI session state parsing.

🔍 General Feedback

🔴 Critical: Unhandled Error Propagation Crash in Agent Routing Workflow

• File: python/agents/ambient-expense-agent/expense_agent/agent.py (Line 101)
• Description: If parse_expense_email encounters an error (such as invalid JSON or base64 decoding issues), it catches the exception and returns an Event with an "error" field: Event(output={"error": ...}). However, route_by_amount does not check for "error" in its node_input and attempts to retrieve node_input.get("amount", 0). Since "amount" is missing from the error dictionary, it defaults to 0. Because 0 is < config.review_threshold, the workflow routes to AUTO_APPROVE. Inside auto_approve, the code attempts to format node_input['amount'] and node_input['submitter']. Since these keys are missing from the error dictionary, this immediately raises a KeyError and crashes the entire workflow.
• Suggestion: Checking for "error" in route_by_amount and raising an exception prevents this silent failure and subsequent crash:

  def route_by_amount(node_input: dict, ctx: Context) -> Event:
      if "error" in node_input:
          raise ValueError(node_input["error"])
      ctx.state["expense_data"] = node_input
      amount = node_input.get("amount", 0)
      ...

🟠 High: Potential TypeError Crash in Subscription Normalization Middleware

• File: python/agents/ambient-expense-agent/expense_agent/fast_api_app.py (Line 98)
• Description: If an incoming Pub/Sub push JSON is missing the "subscription" field (or is set to null/None), data.get("subscription", "") can return None. Calling if "/" in sub on None raises a TypeError: argument of type 'NoneType' is not iterable. Because the middleware only catches json.JSONDecodeError and KeyError, this TypeError is uncaught and will crash the request with an HTTP 500 error.
• Suggestion: Safely verify that sub is a string before performing the check:

  sub = data.get("subscription")
  if isinstance(sub, str) and "/" in sub:
      data["subscription"] = sub.rsplit("/", 1)[-1]
      request._body = json.dumps(data).encode()

🟡 Medium: Stale Content-Length Header on Modified Request Body

• File: python/agents/ambient-expense-agent/expense_agent/fast_api_app.py (Line 101)
• Description: When the subscription normalization middleware modifies request._body to shorten the subscription string, the request's original Content-Length header is left unchanged. Downstream proxies, routers, or parsers that rely on or validate Content-Length will encounter a size mismatch, which can cause requests to be truncated or hang.
• Suggestion: Rebuild the request headers with the updated content length in the ASGI scope:

  if isinstance(sub, str) and "/" in sub:
      data["subscription"] = sub.rsplit("/", 1)[-1]
      new_body = json.dumps(data).encode()
      request._body = new_body
      
      # Rebuild headers with updated content-length
      new_headers = []
      for k, v in request.scope.get("headers", []):
          if k.lower() == b"content-length":
              new_headers.append((k, str(len(new_body)).encode()))
          else:
              new_headers.append((k, v))
      request.scope["headers"] = new_headers

🟡 Medium: Incorrect responded Flag Persistence Across Sequential Events

• File: python/agents/ambient-expense-agent/frontend/main.py (Line 102)
• Description: The loop in _extract_pending_approval processes events sequentially. If it encounters any functionResponse for adk_request_input, it sets responded = True globally for the entire session. If that session later receives a subsequent request for input (e.g., in a multi-step workflow or re-run), responded remains True and the new pending approval is completely ignored and hidden from the frontend.
• Suggestion: Reset responded = False when a new adk_request_input functionCall is processed to ensure the latest request is correctly surfaced:

  elif name == "adk_request_input":
      responded = False
      args = fc.get("args", {})

🟢 Low: Potential AttributeError on Null Session State

• File: python/agents/ambient-expense-agent/frontend/main.py (Line 167)
• Description: In the /pending-approvals endpoint, if a session's "state" field is explicitly present in the JSON but has a null value (None), s.get("state", {}) returns None. Calling .get("expense_data") on None raises an AttributeError and crashes the endpoint.
• Suggestion: Safely unpack the state dictionary:

  review_ids = []
  for s in sessions:
      state = s.get("state") or {}
      expense_data = state.get("expense_data") or {}
      amount = expense_data.get("amount", 0)
      if amount >= REVIEW_THRESHOLD:
          review_ids.append(s["id"])

[github-actions]

[ruff-formatter] _{reported by reviewdog 🐶}

Suggested change

	@functools.cache
	@functools.cache

[github-actions]

[ruff-formatter] _{reported by reviewdog 🐶}

Suggested change

	from google.adk.sessions.in_memory_session_service import InMemorySessionService
	from google.adk.sessions.in_memory_session_service import (
	InMemorySessionService,
	)

[github-actions]

[ruff-formatter] _{reported by reviewdog 🐶}

Suggested change

	from vertexai.agent_engines.templates.adk import _default_instrumentor_builder
	from vertexai.agent_engines.templates.adk import (
	_default_instrumentor_builder,
	)

[github-actions]

[ruff-formatter] _{reported by reviewdog 🐶}

Suggested change

	os.getenv("ALLOW_ORIGINS", "").split(",") if os.getenv("ALLOW_ORIGINS") else None
	os.getenv("ALLOW_ORIGINS", "").split(",")
	if os.getenv("ALLOW_ORIGINS")
	else None

[github-actions]

[ruff-formatter] _{reported by reviewdog 🐶}

Suggested change

	app.description = "Ambient expense agent — processes expense reports via Pub/Sub"
	app.description = (
	"Ambient expense agent — processes expense reports via Pub/Sub"
	)

[github-actions]

[ruff-linter-suggestions] _{reported by reviewdog 🐶}

Suggested change

	from google.adk.sessions.in_memory_session_service import InMemorySessionService
	from google.adk.sessions.in_memory_session_service import (
	InMemorySessionService,
	)

[github-actions]

⚠️ [ruff] <E501> _{reported by reviewdog 🐶}
Line too long (81 > 80)

[github-actions]

⚠️ [ruff] <I001> _{reported by reviewdog 🐶}
Import block is un-sorted or un-formatted

Suggested change

	from google.adk.sessions.in_memory_session_service import InMemorySessionService
	from google.adk.sessions.in_memory_session_service import (
	InMemorySessionService,
	)

[github-actions]

⚠️ [ruff] <E501> _{reported by reviewdog 🐶}
Line too long (84 > 80)

[github-actions]

⚠️ [ruff] <I001> _{reported by reviewdog 🐶}
Import block is un-sorted or un-formatted

Suggested change

	import google.auth
	from vertexai.agent_engines.templates.adk import _default_instrumentor_builder
	import google.auth
	from vertexai.agent_engines.templates.adk import (
	_default_instrumentor_builder,
	)

[github-actions]

⚠️ [ruff] <E501> _{reported by reviewdog 🐶}
Line too long (86 > 80)

[github-actions]

⚠️ [ruff] <E501> _{reported by reviewdog 🐶}
Line too long (85 > 80)

[github-actions]

⚠️ [ruff] <E501> _{reported by reviewdog 🐶}
Line too long (81 > 80)

[github-actions]

📋 Maintainability Review Summary

Overall, the newly introduced code for the Agent Runtime deployment of the Ambient Expense Agent is highly maintainable, modular, and cleanly structured. We identified a few cryptic variable names and minor inconsistencies in naming conventions that could be polished for better long-term readability.

🔍 Findings

All findings have been annotated inline.

[github-actions]

🟡 Cryptic and inconsistent variable naming.

Avoid using short/cryptic variable names like rt. Using a descriptive name like runtime improves readability and matches the variable name in the outer scope.

Suggested change

	rt = get_runtime()
	allowed = streaming_methods if streaming else sync_methods
	if class_method not in allowed:
	raise HTTPException(
	status_code=404,
	detail=f"Unsupported reasoning_engine method: {class_method!r}",
	)
	return getattr(rt, class_method)
	runtime = get_runtime()
	allowed = streaming_methods if streaming else sync_methods
	if class_method not in allowed:
	raise HTTPException(
	status_code=404,
	detail=f"Unsupported reasoning_engine method: {class_method!r}",
	)
	return getattr(runtime, class_method)

[github-actions]

🟡 Cryptic and inconsistent variable naming.

Avoid using cryptic single-letter variable names (like r) outside of list comprehensions/loops. Changing this to response or resp makes it descriptive and consistent with the use of resp in the outer pending_approvals function.

Suggested change

	r = await client.get(url, headers=headers)
	return r.json() if r.status_code == 200 else None
	response = await client.get(url, headers=headers)
	return response.json() if response.status_code == 200 else None

[github-actions]

📋 Correctness Review Summary

This PR migrates the ambient expense agent to Agent Runtime. While reviewing, I identified critical correctness issues in the newly added reasoning_engine_adapter.py that would cause runtime failures (such as ImportError on route dispatch and TypeError when handling synchronous generators during streaming query execution).

[github-actions]

🔴 ImportError on route dispatch: There is no variable named app defined in expense_agent/agent.py. The actual defined workflow/agent is root_agent. Attempting to import app will raise an ImportError and cause route dispatch to fail on the very first request.

Suggested change

	from expense_agent.agent import app as adk_app
	from expense_agent.agent import root_agent as adk_app

[github-actions]

🟠 Uncaught TypeError on synchronous streaming: Using async for event in method(...) unconditionally assumes the registered method is an asynchronous generator. If the method is a synchronous generator function, Python will raise a TypeError: '''generator'' object is not an async iterator at runtime.

Check for asynchronous generator properties before iterating:

Suggested change

	async for event in method(**(body.get("input") or {})):
	async def generator():
	res = method(**(body.get("input") or {}))
	if hasattr(res, "__aiter__"):
	async for event in res:
	yield json.dumps(event) + "\n"
	else:
	for event in res:
	yield json.dumps(event) + "\n"

[github-actions]

🟡 Potential KeyError (Data Validation Gap): Directly indexing body["class_method"] without checking for its presence will raise a KeyError and crash the endpoint with a 500 Internal Server Error if the client sends an unprocessable or malformed JSON payload.

Consider using .get() and validating the input cleanly:

Suggested change

	body = await request.json()
	class_method = body.get("class_method")
	if not class_method:
	raise HTTPException(status_code=400, detail="Missing '''class_method''' in request body")
	method = resolve_method(class_method, streaming=False)

[github-actions]

📋 Security Review Summary

No significant or critical security vulnerabilities were found in the PR's changes. The implementation adheres well to security standards, utilizing Google Cloud Identity-Aware Proxy (IAP) to protect the frontend and IAM controls for backend/Pub/Sub invocations. Structured logging is used safely for metrics and alerts.

🔍 Findings

• Security Architecture: The agent's backend is protected by GCP Vertex AI Agent Runtime IAM permissions, and the frontend is shielded by IAP, ensuring robust access control.
• Input Sanitization & Logging: Inputs from Pub/Sub and requests are safely parsed using standard JSON/base64 parsing. Telemetry and alerts are printed using structured JSON logging, preventing log injection.
• Minor Hygiene: The Dockerfile runs as root by default. While acceptable for a sample, specifying a non-root user is a recommended defense-in-depth practice.

[github-actions]

🟢 Low: Running as Root

The container runs as the root user by default because no USER directive is specified. For production deployments, it is recommended as a defense-in-depth best practice to create and run as a non-privileged user.

Suggested change

	CMD ["uv", "run", "uvicorn", "expense_agent.fast_api_app:app", "--host", "0.0.0.0", "--port", "8080"]
	RUN groupadd -r app && useradd -r -g app app
	USER app
	CMD ["uv", "run", "uvicorn", "expense_agent.fast_api_app:app", "--host", "0.0.0.0", "--port", "8080"]