fix(runners): preserve initial session state keys during rewind

src/google/adk/runners.py

@@ -699,6 +699,17 @@ async def _compute_state_delta_for_rewind(
             state_at_rewind_point[k] = v
 
     current_state = session.state
+# Collect all keys that ever appeared in ANY event's state_delta across
+    # the entire session. Keys present in current_state but absent from all
+    # event state_deltas are "initial state" (set via create_session) and
+    # must be preserved after a rewind.
+    keys_ever_in_event_deltas: set[str] = {
+        k
+        for event in session.events
+        if event.actions.state_delta
+        for k in event.actions.state_delta
+        if not k.startswith('app:') and not k.startswith('user:')
+    }
     rewind_state_delta = {}
 
     # 1. Add/update keys in rewind_state_delta to match state_at_rewind_point.
@@ -712,7 +723,7 @@ async def _compute_state_delta_for_rewind(
     for key in current_state:
       if key.startswith('app:') or key.startswith('user:'):
         continue
-      if key not in state_at_rewind_point:
+      if key not in state_at_rewind_point and key in keys_ever_in_event_deltas:
         rewind_state_delta[key] = None
 
     return rewind_state_delta

src/google/adk/tools/openapi_tool/openapi_spec_parser/rest_api_tool.py

@@ -381,10 +381,15 @@ def _prepare_request_params(
     # Move query params embedded in the path into query_params, since httpx
     # replaces (rather than merges) the URL query string when `params` is set.
     parsed_url = urlparse(url)
-    if parsed_url.query or parsed_url.fragment:
-      for key, values in parse_qs(parsed_url.query).items():
-        query_params.setdefault(key, values[0] if len(values) == 1 else values)
-      url = urlunparse(parsed_url._replace(query="", fragment=""))
+    for part in (parsed_url.query, parsed_url.fragment):
+        if part:
+            for key, values in parse_qs(part).items():
+                query_params.setdefault(
+                    key,
+                    values[0] if len(values) == 1 else values
+                )
+    # URL without query and fragment
+    url = urlunparse(parsed_url._replace(query="", fragment=""))
 
     # Construct body
     body_kwargs: Dict[str, Any] = {}

tests/unittests/tools/openapi_tool/openapi_spec_parser/test_rest_api_tool.py

@@ -1423,6 +1423,47 @@ def test_prepare_request_params_plain_url_unchanged(
     request_params = tool._prepare_request_params([], {})
 
     assert request_params["url"] == "https://example.com/test"
+  def test_prepare_request_params_fragment_params_become_query_params(
+    self, sample_auth_credential, sample_auth_scheme
+  ):
+    # When the ApplicationIntegrationToolset builds an endpoint URL, it sometimes
+    # puts params in the fragment (e.g. #triggerId=my_trigger). Without this fix
+    # those params were silently dropped and the API returned a 400 error.
+    # See: https://github.com/google/adk-python/issues/4598
+    integration_endpoint = OperationEndpoint(
+      base_url="https://integrations.googleapis.com",
+      path=(
+        "/v2/projects/demo/locations/us-central1"
+        "/integrations/MyFlow:execute"
+        "?triggerId=api_trigger/MyFlow"
+        "#httpMethod=POST"
+      ),
+      method="POST",
+    )
+    op = Operation(operationId="run_integration")
+    tool = RestApiTool(
+      name="run_integration",
+      description="Runs a Google Cloud integration flow",
+      endpoint=integration_endpoint,
+      operation=op,
+      auth_credential=sample_auth_credential,
+      auth_scheme=sample_auth_scheme,
+    )
+
+    result = tool._prepare_request_params([], {})
+
+    # Both the query string and fragment params should land in query params
+    assert result["params"]["triggerId"] == "api_trigger/MyFlow"
+    assert result["params"]["httpMethod"] == "POST"
+
+    # The final URL should be clean — no leftover ? or #
+    assert "?" not in result["url"]
+    assert "#" not in result["url"]
+    assert result["url"] == (
+      "https://integrations.googleapis.com"
+      "/v2/projects/demo/locations/us-central1"
+      "/integrations/MyFlow:execute"
+    )
 
 
 def test_snake_to_lower_camel():