fix(cli): normalize trigger user ids for sessions

Merge https://github.com/google/adk-python/pull/5402

## Summary
- normalize Pub/Sub subscription and Eventarc source metadata before reusing them as session user ids
- replace slash-separated resource paths with path-safe -- delimiters while preserving the full resource identity
- add trigger endpoint regression tests that verify the created sessions are stored under the normalized user ids

## Testing
- python3 -m py_compile src/google/adk/cli/trigger_routes.py tests/unittests/cli/test_trigger_routes.py
- python3 -m pytest tests/unittests/cli/test_trigger_routes.py -k "path_safe or with_subscription_metadata or source_from_ce_header" (fails during collection in this environment: ModuleNotFoundError: No module named 'fastapi')

Co-authored-by: George Weale <gweale@google.com>
COPYBARA_INTEGRATE_REVIEW=https://github.com/google/adk-python/pull/5402 from MukundaKatta:codex/trigger-user-id-path-safe a97467903bcadc92fc7a916ccef95638c58b3a65
PiperOrigin-RevId: 963505905

src/google/adk/cli/trigger_routes.py

@@ -208,6 +208,22 @@ class TriggerResponse(BaseModel):
   )
 
 
+def _make_trigger_user_id(
+    raw_value: Optional[str],
+    *,
+    default: str,
+) -> str:
+  """Normalize trigger metadata into a session-safe user_id."""
+  if not raw_value:
+    return default
+
+  normalized = raw_value.strip().strip("/")
+  if not normalized:
+    return default
+
+  return normalized.replace("/", "--")
+
+
 # ---------------------------------------------------------------------------
 # Trigger Router
 # ---------------------------------------------------------------------------
@@ -411,8 +427,9 @@ class TriggerRouter:
       async def trigger_pubsub(
           app_name: str, req: PubSubTriggerRequest, request: Request
       ) -> TriggerResponse:
-        subscription = req.subscription or "pubsub-caller"
-        user_id = subscription.replace("/", "--")
+        user_id = _make_trigger_user_id(
+            req.subscription, default="pubsub-caller"
+        )
 
         decoded_data = None
         data_payload = None
@@ -478,10 +495,10 @@ class TriggerRouter:
           app_name: str, req: EventarcTriggerRequest, request: Request
       ) -> TriggerResponse:
 
-        source = (
-            req.source or request.headers.get("ce-source") or "eventarc-caller"
+        user_id = _make_trigger_user_id(
+            req.source or request.headers.get("ce-source"),
+            default="eventarc-caller",
         )
-        user_id = source.strip("/").replace("/", "--")
 
         logger.info(
             "Eventarc trigger: source=%s, type=%s, id=%s",

tests/unittests/cli/test_trigger_routes.py

@@ -460,6 +460,24 @@ class TestTriggerPubSub:
     assert len(captured_user_ids) == 1
     assert captured_user_ids[0] == "pubsub-caller"
 
+  def test_subscription_user_id_is_path_safe(
+      self, client, mock_session_service
+  ):
+    """Pub/Sub subscription-derived user_id is stored without slashes."""
+    message_data = base64.b64encode(b"test").decode("utf-8")
+    payload = {
+        "message": {"data": message_data},
+        "subscription": "projects/p/subscriptions/orders-sub",
+    }
+
+    resp = client.post("/apps/test_app/trigger/pubsub", json=payload)
+
+    assert resp.status_code == 200
+    assert (
+        "projects--p--subscriptions--orders-sub"
+        in mock_session_service.sessions["test_app"]
+    )
+
   def test_unknown_app_fails_early(
       self, client, mock_agent_loader, mock_session_service
   ):
@@ -610,6 +628,25 @@ class TestTriggerEventarc:
     assert len(captured_user_ids) == 1
     assert captured_user_ids[0] == "eventarc-caller"
 
+  def test_eventarc_source_user_id_is_path_safe(
+      self, client, mock_session_service
+  ):
+    """Eventarc ce-source-derived user_id is stored without slashes."""
+    payload = {
+        "data": {"key": "value"},
+    }
+    resp = client.post(
+        "/apps/test_app/trigger/eventarc",
+        json=payload,
+        headers={"ce-source": "//pubsub.googleapis.com/projects/p/topics/t"},
+    )
+
+    assert resp.status_code == 200
+    assert (
+        "pubsub.googleapis.com--projects--p--topics--t"
+        in mock_session_service.sessions["test_app"]
+    )
+
   def test_complex_event_data(self, client, monkeypatch):
     """Complex nested event data is serialized as JSON for the agent."""
     captured_messages = []