# Ambient Expense Agent — Event-Driven Pub/Sub Workflow with HITL

## Intent

A clone-and-study **ambient agent**: there is no interactive chat loop. Expense
reports arrive as **Pub/Sub** events, get pushed to an ADK trigger endpoint, and
flow through an **ADK 2.0 graph-based `Workflow`**. Business rules stay in code
(a `$100` threshold routes low-value expenses straight to auto-approval); only
high-value expenses reach an LLM `review_agent`, which then **pauses for
human-in-the-loop approval** via `RequestInput` before logging a decision. The
graph itself is compact — the interesting parts are the **ambient plumbing**:
authenticated Pub/Sub push, session-based HITL discovery, an IAP-protected
approval UI, and Cloud Monitoring email alerts, all provisioned by **Terraform**.

## When To Use

- The user wants an **event-driven / ambient** agent (triggered by Pub/Sub, no
  user typing at it) rather than a request/response chat agent.
- The user wants **business rules in code + the LLM only for judgment calls**
  (deterministic threshold routing, LLM used just for risk analysis).
- The user needs **human-in-the-loop approval** that pauses a workflow mid-run
  and resumes it later from a separate UI.
- The user wants a reproducible **two-service Cloud Run + Pub/Sub + IAP +
  Monitoring** deployment as a starting point.

## Eval

- **Scenarios Path**: none — this recipe ships **no eval datasets** and no
  `tests/eval/` directory; nothing is wired into `make test`.
- What exists instead lives under **`tests/`**:
  - `tests/test_runnability.py` — imports `expense_agent.agent`, patches
    `google.auth.default()`, and asserts `root_agent` is defined (a smoke test).
  - `tests/test_integration.py` — drives the **full flow in-process** over
    `httpx.ASGITransport` (no real servers): Pub/Sub trigger → auto-approve,
    trigger → review → HITL approve, → HITL reject, and subscription-name
    normalization. This is the closest thing to behavioral coverage.
- **To add eval:** create `tests/eval/` with an `eval_config.yaml` + `datasets/`
  (mirroring the RAG recipes' `agents-cli` LLM-judge format), grading the
  `review_agent`'s risk assessment against fixed expense payloads.

## End-to-end flow

```
expense published to Pub/Sub topic "expense-reports"
   -> authenticated OIDC push -> POST /apps/expense_agent/trigger/pubsub
   -> parse_expense_email (base64/plain JSON -> ExpenseData fields)
   -> route_by_amount ($100 threshold, stashes expense_data in ctx.state)
        |                                   |
     < $100                             >= $100
        |                                   |
   auto_approve                        review_agent (LLM)
   (logs INFO, done)                     -> emit_expense_alert (JSON stdout)
                                          -> request_approval (RequestInput -> PAUSE)
```

While paused, a structured `WARNING` log fires a **log-based metric → alert
policy → email** to the manager. The manager opens the **IAP-protected approval
UI**, which finds the pending `adk_request_input` by scanning ADK session events
and resumes the workflow via `POST /run`:

```
   ... PAUSED ... -> manager approves/rejects in approval UI
   -> frontend POST /approve -> backend POST /run (functionResponse)
   -> process_decision (logs approved/rejected, emits final summary) -> done
```

The Pub/Sub push completes as soon as the workflow **pauses** — the ack deadline
is not held for the human; the HITL resume is entirely out-of-band.

## Most interesting files to study (in order)

### Trigger server (the ambient entry point)
1. **`expense_agent/fast_api_app.py`** — the whole "ambient" seam. Builds the ADK
   app with `get_fast_api_app(..., trigger_sources=["pubsub"])` so Pub/Sub can
   POST expenses to `/apps/{app}/trigger/pubsub`. Adds middleware that
   **normalizes** `projects/.../subscriptions/NAME` → `NAME`, because the ADK
   trigger handler uses the subscription as the session `user_id` — and the
   frontend must query with that same short name.

### Deployment plumbing (Terraform — what makes it production-ambient)
2. **`terraform/pubsub.tf`** — the trigger wiring: an `expense-reports` topic and
   an **authenticated push** subscription whose `push_config` targets the backend
   trigger URL with an `oidc_token`. `600s` ack deadline, exponential retry, and
   a **dead-letter topic after 5 attempts**.
3. **`terraform/cloud_run.tf`** — the two services. Backend runs the ADK agent;
   frontend is `iap_enabled = true` and gets `BACKEND_URL`, `APP_NAME`,
   `PUBSUB_SUBSCRIPTION`, and `USE_SERVICE_AUTH` wired as env. `min_instance_count
   = 1` keeps one instance warm (relevant to in-memory sessions — see Gotchas).
4. **`terraform/monitoring.tf`** — the alert path. A `google_logging_metric`
   filters `jsonPayload.alert_type="expense_review"` from Cloud Run logs; an alert
   policy fires an **email notification channel** with a markdown link straight to
   the approval UI. This is what turns a stdout log into a manager's email.
5. **`terraform/iam.tf`** — least-privilege service accounts: `pubsub_invoker`
   (`run.invoker` on backend + `serviceAccountTokenCreator` for the Pub/Sub
   service agent's OIDC), `frontend_invoker`, and the **IAP** bindings
   (`iap.httpsResourceAccessor` for the notification email + `run.invoker` for the
   IAP service agent).

### Approval UI (HITL frontend — separate uv sub-project)
6. **`frontend/main.py`** — a thin proxy, not an agent. `GET /pending-approvals`
   queries the **backend's built-in ADK session APIs**, then `_extract_pending_
   approval` scans each session's events for an `adk_request_input` function call
   **without a matching response** (and grabs the `emit_expense_alert` risk
   summary). `POST /approve` forwards a `functionResponse` to the backend's
   `POST /run` to resume the paused workflow. On Cloud Run it mints ID tokens for
   service-to-service auth; locally it calls the backend unauthenticated.

### Agent graph & config (read last)
7. **`expense_agent/config.py`** — the auth bootstrap and the two knobs. Picks
   AI Studio if `GOOGLE_API_KEY` is set, else Vertex via `google.auth.default()`
   **at import time**. Exposes `model` (from `MODEL_NAME` env) and
   `review_threshold = 100.0`.
8. **`expense_agent/agent.py`** — the payoff, once you understand how events
   arrive and approvals flow back. Defines the `ExpenseData` Pydantic schema, the
   function nodes (`parse_expense_email`, `route_by_amount`, `auto_approve`,
   `request_approval`, `process_decision`), the LLM `review_agent`
   (`emit_expense_alert` tool), and the **`Workflow` graph** that stitches them
   together — a mixed function/LLM graph with conditional routing and a
   `RequestInput` HITL pause. `expense_agent/__init__.py` runs `load_dotenv()`
   before the submodule imports (hence the `# noqa: E402`).

## Data handling

- **No documents, no ingestion.** "Data" here is the **expense event payload**
  moving between graph nodes, shaped by the `ExpenseData` Pydantic model
  (`amount`, `submitter`, `category`, `description`, `date`).
- **Decoding:** `parse_expense_email` accepts the Pub/Sub message `data` as
  base64 (real Pub/Sub) or plain JSON (local testing) and coerces fields with
  safe defaults.
- **State handoff:** `route_by_amount` stashes the parsed dict into
  `ctx.state["expense_data"]` so the HITL `request_approval` node and the
  frontend can read it after the pause.
- **Sessions are the durable substrate.** Pending approvals are discovered by
  scanning **ADK session events** for an unanswered `adk_request_input`. The
  session `user_id` is the **Pub/Sub subscription name** (normalized by the
  middleware). No session service URI is configured, so the ADK default
  **in-memory** session service is used.
- **Structured logs are the "output".** `auto_approve`, `emit_expense_alert`,
  and `process_decision` `print(json.dumps(...))` to stdout; Cloud Run captures
  these as structured Cloud Logging entries that drive the alert metric.

## Gotchas / things to know

- **Sessions are in-memory.** With no persistent session service configured,
  pending approvals live only in the backend instance's memory —
  `min_instance_count = 1` keeps one instance warm, but a restart/redeploy
  **drops in-flight approvals**. Wire a persistent session service for real use.
- **The `$100` threshold lives in two places.** `config.review_threshold` (backend
  routing) **and** `REVIEW_THRESHOLD = 100` in `frontend/main.py` (pre-filters
  which sessions to show). Keep them in sync or the UI and the agent disagree.
- **`user_id` == subscription name.** The frontend queries pending approvals
  using `PUBSUB_SUBSCRIPTION` (default `test-sub` locally). If it doesn't match
  the `subscription` field in trigger requests, the UI shows nothing — which is
  exactly why `fast_api_app.py` normalizes the fully-qualified subscription path.
- **Model comes from `MODEL_NAME`.** `config.py` reads `os.getenv("MODEL_NAME")`
  with **no default** (so it is `None` if unset). `.env.example` sets
  `gemini-3.5-flash`; tests set it explicitly. Don't hardcode model names
  and don't use deprecated ones (`gemini-2.0-flash`, `gemini-2.5-flash`).
- **Import-time credentials.** Without `GOOGLE_API_KEY`, importing `config`
  calls `google.auth.default()` immediately — a credential-less import fails
  unless ADC is present. `test_runnability.py` patches it.
- **Alerts only fire when deployed.** `emit_expense_alert` just prints locally;
  the email path (log-based metric → alert policy → channel) exists only in the
  deployed Cloud Monitoring setup.
- **IAP propagation lag.** After `make deploy`, IAP can take **5–10 minutes** to
  propagate; a `403 Forbidden` on the approval UI early on is expected.

## Where to run things

`Makefile` targets:

- `make install` / `make install-frontend` — `uv sync` (backend / frontend
  sub-project).
- `make dev` — run the backend trigger server (`fast_api_app.py`, port 8080).
- `make dev-frontend` — run the approval UI against `BACKEND_URL=localhost:8080`
  (port 8081).
- `make playground` — local ADK web UI (`adk web`, port 8501).
- `make test` — `pytest tests/ -xvs` (runnability + in-process integration).
- `make lint` — `codespell` + `ruff check --fix` + `ruff format` + `mypy`.
- `make deploy NOTIFICATION_EMAIL=...` — build both images via Cloud Build, then
  `terraform apply` the whole stack.
- `make remote-test` — publish a `$250` test expense to the deployed topic.
- `make clean NOTIFICATION_EMAIL=...` — `terraform destroy` (retries once after
  60s for alert-policy propagation).

## Reuse (copy as-is)

- **`terraform/`** is self-contained — two Cloud Run services, Pub/Sub (+
  dead-letter), IAM, IAP, and Cloud Monitoring. Copy the directory and set
  `project_id`, `region`, `notification_email`, and the `backend_image` /
  `frontend_image` vars. It expects **prebuilt images**, so build them first (the
  `Makefile` uses Cloud Build) before `terraform apply`.
- **`frontend/`** is an independent uv sub-project (its own `pyproject.toml`,
  `uv.lock`, and `Dockerfile`) — a generic **ADK HITL approval proxy**. It has no
  code coupling to the backend beyond the ADK session-API contract; point it at
  any ADK agent via `BACKEND_URL`, `APP_NAME`, and `PUBSUB_SUBSCRIPTION`.
- There is **no code coupling into `expense_agent/`**: the agent is configured
  purely through env (`MODEL_NAME`, `GOOGLE_API_KEY` / `GOOGLE_CLOUD_*`), and the
  frontend reaches it only over HTTP.
