fix(cli): expose trigger delivery id so tools can detect redeliveries - #7325

Link to Issue or Description of Change

1. Link to an existing issue (if applicable):

• Closes: Pub/Sub trigger: a redelivered message re-runs the agent in a new session, repeating tool side effects that already happened #7322

Problem:

When a Pub/Sub trigger run fails after a tool has already caused a side effect (a payment, an email, a ticket), the endpoint returns 500 and Pub/Sub redelivers the message. Every delivery runs in a new session (uuid4()), and the Pub/Sub messageId is only logged, never passed to the agent. So a tool has nothing stable to deduplicate on, and the side effect happens again. The same applies to Eventarc: Pub/Sub-wrapped events don't carry ce-id into the agent's input.

Solution:

This implements option 1 from the issue: pass the delivery identity into the run.

When a trigger creates its session, it now seeds the session state with a trigger_delivery entry (exposed as TRIGGER_DELIVERY_STATE_KEY):

Session state is new on every delivery, so the deduplication itself has to happen at the provider or in an external store, not in state. A tool derives an idempotency key from this value and passes it on. A Pub/Sub messageId is only unique per topic, so the key includes the subscription (for Eventarc, event_source + id):

def pay_invoice(invoice: str, amount: str, tool_context: ToolContext) -> dict:
  delivery = tool_context.state["trigger_delivery"]
  key = f"{delivery['subscription']}:{delivery['id']}"
  return payments.charge(invoice, amount, idempotency_key=key)

Why this approach:

• Additive and backwards compatible. The session ID, the agent's input text, and the HTTP status codes are unchanged, so existing agents behave exactly as before.
• Works with any session service. The ID comes from the request itself, so it works with the default in-memory service and across multiple Cloud Run instances, where deriving the session ID from the messageId (option 2) would not help on its own.
• In-process 429 retries already reuse the session, so they see the same value.

Option 2 (a deterministic session_id) changes redelivery behaviour and would need a guard against concurrent redeliveries, so I left it out of this PR. I'm happy to follow up on it, or on the docs (option 3) in adk-docs, if maintainers want either.

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Added TestTriggerDeliveryIdentity to tests/unittests/cli/test_trigger_routes.py (7 tests):

• Pub/Sub messageId, subscription and publishTime are stored in session state.
• A Pub/Sub request without messageId still records source, with id set to None.
• A redelivered Pub/Sub message lets a tool skip a side effect it already performed, using a subscription + messageId key checked outside the session (the scenario from the issue: first delivery returns 500, the redelivery returns 200, one payment).
• A structured CloudEvent's id, source and type are stored.
• Binary content mode reads the id from the ce-* headers.
• A Pub/Sub-wrapped Eventarc event without ce-id falls back to the messageId.
• An in-process retry after a 429 still sees the delivery identity.

All 7 new tests fail on main and pass with this change.

$ pytest tests/unittests/cli/test_trigger_routes.py -q
78 passed in 9.48s

Same result on Python 3.10, 3.11 and 3.14. The rest of tests/unittests/cli also passes; the only failures on my Windows machine are 32 platform-specific tests (symlinks, path separators, gcloud deploy), and they fail identically on main.

pyink, isort, ruff, addlicense, codespell and the compliance checks pass. mypy reports no new errors in trigger_routes.py.

Manual End-to-End (E2E) Tests:

I used the reproduction script from #7322 unchanged: a stock LlmAgent on the Gemini class pointed at a local fake Gemini endpoint that returns 503 once after the pay_invoice call, with a loop that redelivers the same push envelope like Pub/Sub. No network access or API key is needed.

I ran it twice: once with the original tool, and once with the tool changed to pass a subscription + messageId idempotency key to the (fake) provider, whose key store lives outside the ADK session. The second run then sends a new message, to check that a new messageId is still charged:

def pay_invoice(invoice: str, amount: str, tool_context: ToolContext) -> dict:
    """Pay an invoice. Irreversible."""
    delivery = tool_context.state.get("trigger_delivery") or {}
    print(f"  pay_invoice sees trigger_delivery={delivery or None}")
    if delivery.get("id"):
        # Session state is new on every delivery, so dedupe at the provider,
        # keyed on subscription + messageId (messageId is unique per topic).
        key = f"{delivery['subscription']}:{delivery['id']}"
        if key in __main__.DONE:
            return {"status": "PAID", "note": "provider: idempotency key already used"}
        __main__.DONE.add(key)
    __main__.PAYMENTS.append((invoice, amount))
    return {"status": "PAID", "payment_number": len(__main__.PAYMENTS)}

Note: DONE is an in-memory check-and-add ledger that stands in for the provider's idempotency store, to keep the harness self-contained. It is not the recommended pattern: check → act → record pays twice after an ambiguous failure or an overlapping redelivery. The TRIGGER_DELIVERY_STATE_KEY docstring recommends the provider's idempotency key, or an atomic reservation before the effect when using your own store.

Before (main @ 044a1ec): the tool sees no delivery identity and pays twice for one message.

google-adk 2.10.0
  pay_invoice sees trigger_delivery=None
delivery 1 of messageId 1234567890: HTTP 500
  pay_invoice sees trigger_delivery=None
delivery 2 of messageId 1234567890: HTTP 200
payments made for messageId 1234567890: 2
  pay_invoice sees trigger_delivery=None
delivery 1 of messageId 1234567891: HTTP 200
payments made for messageId 1234567891: 1

After (this PR): both deliveries carry the same key, so the provider charges once; the next message has a new messageId and is charged normally.

google-adk 2.10.0
  pay_invoice sees trigger_delivery={'source': 'pubsub', 'id': '1234567890', 'subscription': 'projects/p/subscriptions/invoices', 'publish_time': None}
delivery 1 of messageId 1234567890: HTTP 500
  pay_invoice sees trigger_delivery={'source': 'pubsub', 'id': '1234567890', 'subscription': 'projects/p/subscriptions/invoices', 'publish_time': None}
delivery 2 of messageId 1234567890: HTTP 200
payments made for messageId 1234567890: 1
  pay_invoice sees trigger_delivery={'source': 'pubsub', 'id': '1234567891', 'subscription': 'projects/p/subscriptions/invoices', 'publish_time': None}
delivery 1 of messageId 1234567891: HTTP 200
payments made for messageId 1234567891: 1

The unmodified script from the issue still reports 2 payments, as expected: this change gives tools what they need to deduplicate, and doesn't change what happens to a tool that ignores it.

Checklist

• I have read the CONTRIBUTING.md document.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end.
• Any dependent changes have been merged and published in downstream modules.

Additional context

The key is documented on TRIGGER_DELIVERY_STATE_KEY (including where to dedupe and how to build the key) and in both endpoints' OpenAPI descriptions. The ack-deadline and dead-letter guidance from the issue will follow as a separate PR in adk-docs.