fix(a2a): address PR #3704 review findings — security, observability, type design

Security-critical:
- Resolve A2A task lookup ambiguity: Rust runtime pre-resolves agent from
  URL path and injects agent_id into task-method proxy params; Python
  service refuses to guess when (task_id, no agent_id) matches multiple
  rows. Prevents cross-agent task read/cancel via shared task_id.
- Encrypt push webhook auth_token at rest via services_auth.encode_auth;
  new list_push_configs_for_dispatch returns decrypted tokens for the
  trusted Rust sidecar only. Fixes secret-at-rest leak.
- Upsert semantics on push-config re-registration: mutable fields
  (auth_token, events, enabled) update in place instead of silently
  keeping stale config. Key rotations now take effect.
- Fail-closed on query-param auth decrypt failure (matches header path).
- Refuse empty/missing A2A_RUST_AUTH_SECRET at startup; add cross-field
  RuntimeConfig validation (retry budget, TTL ordering, session TTL).
- Feature-flag gate on internal A2A endpoints: reject trusted requests
  when MCPGATEWAY_A2A_ENABLED=false (defense-in-depth).
- events/flush rejects events for unknown task_ids (400) to prevent
  visibility bypass.

Observability:
- logger.exception on _authorize_internal_a2a_method errors.
- logger.warning on agent-visibility denial across 7 internal endpoints.
- warn! on JSON parse failures in 4 proxy sites (server.rs) and
  malformed Last-Event-ID headers.
- warn! on session fingerprint mismatch (security-relevant signal).
- MetricsCollector.webhook_retry_exhausted counter + record method.
- Log Rust-cache invalidation scheduling failures instead of silent pass.
- Rollback-after-failure in task persistence now logs and calls invalidate.

Type design / schema:
- New A2ATaskState enum with terminal-state validator on A2ATaskUpdate.
- Unknown task-state protocol values now warn (passed through for
  forward-compat).
- Alembic: tenant=String(255), icon_url=String(767) match the ORM
  (prevents Postgres VARCHAR-length drift between fresh and migrated DBs).
- Rename two A2A migration files to hash-first convention matching the
  rest of the project.
- trust.rs doc/comment clarifies the header is SHA256, not HMAC.
- BTreeMap for query-param merge ensures deterministic ordering.
- encode_auth_context uses .expect() — empty header would look anonymous.

Test coverage:
- Deny-path regressions for internal A2A endpoints: RBAC denied,
  wrong-team, feature-disabled (20 new cases).
- Rust↔Python bridge: ConnectError/ConnectTimeout/ReadTimeout/PoolTimeout
  now wrap as RustA2ARuntimeError with correct is_timeout flag.
- Push config: encryption round-trip, token rotation, upsert semantics,
  undecryptable-legacy heal path, dispatch-listing SQL visibility scope.
- Rust concurrency: session concurrent lookup/invalidate leaves storage
  consistent; concurrent create produces unique IDs; queue overflow
  under concurrent producers rejects with Full.
- RuntimeConfig cross-field validation (5 scenarios).
- Rust push: webhook retry-exhaustion metric increments.
- Binary startup refuses without auth_secret.
- select_downstream_agent: ORDER BY name + enabled filter assertions.

Deferred to follow-up issues:
- Typestate wrappers for trust-boundary types (#4233)
- SSE client-disconnect integration test (#4234)

Signed-off-by: Jonathan Springer <jps@s390x.com>

Author: jonpspri

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "^.secrets.baseline|package-lock.json|Cargo.lock|scripts/sign_image.sh|scripts/zap|sonar-project.properties|uv.lock|go.sum|mcpgateway/sri_hashes.json|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-04-15T16:31:17Z",
+  "generated_at": "2026-04-15T18:51:42Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -5136,31 +5136,31 @@
         "hashed_secret": "f2b14f68eb995facb3a1c35287b778d5bd785511",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 5497,
+        "line_number": 5687,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "f42a3fabe1e9bed059d727f47eb752e3aa61b977",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 5554,
+        "line_number": 5744,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "b85788b459aa4d67e1070930dae6d0827756aadb",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 5592,
+        "line_number": 5782,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "52dcc83ec1e54426ad58a64854d1eb8d5f5d9685",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 5593,
+        "line_number": 5783,
         "type": "Secret Keyword",
         "verified_result": null
       }
@@ -7502,119 +7502,111 @@
         "hashed_secret": "41db7c65f665e40b44178f95f5f86473ca17160e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 72,
+        "line_number": 73,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "4ea8d2335b430796cf3f500368c5b0f5b1dc90f5",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 190,
+        "line_number": 191,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "1a91d62f7ca67399625a4368a6ab5d4a3baa6073",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 191,
+        "line_number": 192,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "5ef3af6cd9b5aa907fa618fac829baaec6763b42",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 385,
+        "line_number": 387,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "94c9325c8ade4f6327d87e240713c2fd7fdbfc63",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 386,
+        "line_number": 388,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "7bb0c6efd2edb1ae20dab6dd260895ad8895e07e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 422,
+        "line_number": 424,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "ddbeee31dc29b74fcede0bea4d3fc5276d9148c8",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 448,
+        "line_number": 451,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "122758dd535bc6d246f36a686b29c7c40fab1315",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 799,
+        "line_number": 803,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "fa727f587e9f6115da6d4eb600dd8b6fe06cf69a",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 801,
+        "line_number": 805,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "250cdef3ce09000fa9a939a13e5f73433d96a852",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 2354,
+        "line_number": 2399,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "f2b14f68eb995facb3a1c35287b778d5bd785511",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 2363,
-        "type": "Secret Keyword",
-        "verified_result": null
-      },
-      {
-        "hashed_secret": "1902e3d6fc4e78a0bcc50ba12b882769afbf4a8c",
-        "is_secret": false,
-        "is_verified": false,
-        "line_number": 2395,
+        "line_number": 2408,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "d92342976d720ff38cf5dcb329be41959ab1ba6c",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 2816,
+        "line_number": 3241,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "a343c9b5b1abfcf385d5c6f6dc0282f4d88a416e",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 2977,
+        "line_number": 3402,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "c00dbbc9dadfbe1e232e93a729dd4752fade0abf",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 3009,
+        "line_number": 3434,
         "type": "Secret Keyword",
         "verified_result": null
       }

crates/a2a_runtime/src/auth.rs

@@ -131,9 +131,13 @@ pub fn apply_invoke_auth(
     }
 
     // Merge auth query params (override existing keys).
+    //
+    // Use BTreeMap so the resulting query string is lexicographically
+    // ordered.  HashMap iteration order is randomized per-process, which
+    // would (a) break HMAC/signed-URL flows that require a canonical
+    // serialization and (b) make log/diff comparison non-deterministic.
     if !query_params.is_empty() {
-        // Collect existing pairs, then layer auth on top.
-        let mut merged: HashMap<String, String> = url
+        let mut merged: std::collections::BTreeMap<String, String> = url
             .query_pairs()
             .map(|(k, v)| (k.into_owned(), v.into_owned()))
             .collect();

crates/a2a_runtime/src/config.rs

@@ -202,6 +202,61 @@ impl RuntimeConfig {
                 )
             })
     }
+
+    /// Cross-field validation invoked by [`crate::run`] at startup.
+    ///
+    /// Clap handles per-field parsing, but several invariants cross multiple
+    /// fields.  Catching these at startup beats surfacing them as strange
+    /// runtime behaviour (timeouts that can never trigger, retry budgets
+    /// that exceed the client timeout, etc.).
+    pub fn validate_cross_field(&self) -> Result<(), String> {
+        if self.max_concurrent == 0 {
+            return Err("A2A_RUST_MAX_CONCURRENT must be >= 1".to_string());
+        }
+        if self.agent_cache_max_entries == 0 {
+            return Err("A2A_RUST_AGENT_CACHE_MAX_ENTRIES must be >= 1".to_string());
+        }
+        if self.circuit_failure_threshold == 0 {
+            return Err("A2A_RUST_CIRCUIT_FAILURE_THRESHOLD must be >= 1".to_string());
+        }
+
+        // Retry budget: total backoff (sum of geometric series up to
+        // max_retries * retry_backoff_ms) should fit within the per-request
+        // timeout, or the circuit breaker will trip before retries even run.
+        // Use a conservative upper bound: max_retries * retry_backoff_ms.
+        let retry_budget_ms = self
+            .retry_backoff_ms
+            .saturating_mul(u64::from(self.max_retries));
+        if retry_budget_ms > self.request_timeout_ms {
+            return Err(format!(
+                "retry budget ({}ms = {} retries × {}ms backoff) exceeds request timeout ({}ms); \
+                 retries can never complete",
+                retry_budget_ms, self.max_retries, self.retry_backoff_ms, self.request_timeout_ms
+            ));
+        }
+
+        // Cache TTL sanity: L2 (Redis, shared) should outlive L1 (in-process)
+        // so a node restart does not reload a stale-by-comparison entry from
+        // L2.  Allow equality (both set to the same value) for simple
+        // deployments.
+        if self.l2_cache_ttl_secs > 0 && self.l2_cache_ttl_secs < self.agent_cache_ttl_secs {
+            return Err(format!(
+                "A2A_RUST_L2_CACHE_TTL_SECS ({}) must be >= A2A_RUST_AGENT_CACHE_TTL_SECS ({})",
+                self.l2_cache_ttl_secs, self.agent_cache_ttl_secs
+            ));
+        }
+
+        // Session TTL: when sessions are enabled, the session must outlive
+        // a typical request round-trip; otherwise fast-path cache hits will
+        // never trigger.
+        if self.session_enabled && self.session_ttl_secs == 0 {
+            return Err(
+                "A2A_RUST_SESSION_TTL_SECS must be > 0 when sessions are enabled".to_string(),
+            );
+        }
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -287,4 +342,58 @@ mod tests {
         assert!(config.session_enabled);
         assert_eq!(config.event_store_max_events, 1000);
     }
+
+    #[test]
+    fn validate_cross_field_accepts_defaults() {
+        let config = test_config();
+        assert!(config.validate_cross_field().is_ok());
+    }
+
+    #[test]
+    fn validate_cross_field_rejects_zero_max_concurrent() {
+        let mut config = test_config();
+        config.max_concurrent = 0;
+        let err = config.validate_cross_field().expect_err("should reject");
+        assert!(err.contains("MAX_CONCURRENT"));
+    }
+
+    #[test]
+    fn validate_cross_field_rejects_retry_budget_exceeding_timeout() {
+        // 5 retries × 1000ms backoff = 5000ms retry budget,
+        // but request timeout is only 2000ms — retries cannot complete.
+        let mut config = test_config();
+        config.request_timeout_ms = 2_000;
+        config.retry_backoff_ms = 1_000;
+        config.max_retries = 5;
+        let err = config.validate_cross_field().expect_err("should reject");
+        assert!(err.contains("retry budget"));
+        assert!(err.contains("exceeds request timeout"));
+    }
+
+    #[test]
+    fn validate_cross_field_rejects_l2_ttl_shorter_than_l1() {
+        // L2 (Redis, shared) should outlive L1 or node restarts reload stale data.
+        let mut config = test_config();
+        config.agent_cache_ttl_secs = 600;
+        config.l2_cache_ttl_secs = 60;
+        let err = config.validate_cross_field().expect_err("should reject");
+        assert!(err.contains("L2_CACHE_TTL_SECS"));
+    }
+
+    #[test]
+    fn validate_cross_field_rejects_zero_session_ttl_when_sessions_enabled() {
+        let mut config = test_config();
+        config.session_enabled = true;
+        config.session_ttl_secs = 0;
+        let err = config.validate_cross_field().expect_err("should reject");
+        assert!(err.contains("SESSION_TTL_SECS"));
+    }
+
+    #[test]
+    fn validate_cross_field_allows_zero_session_ttl_when_sessions_disabled() {
+        let mut config = test_config();
+        config.session_enabled = false;
+        config.session_ttl_secs = 0;
+        assert!(config.validate_cross_field().is_ok());
+    }
 }

crates/a2a_runtime/src/lib.rs

@@ -53,6 +53,26 @@ fn build_http_client(config: &RuntimeConfig) -> Result<Client, reqwest::Error> {
 }
 
 pub async fn run(config: RuntimeConfig) -> Result<(), RuntimeError> {
+    if config
+        .auth_secret
+        .as_deref()
+        .map(str::is_empty)
+        .unwrap_or(true)
+    {
+        return Err(RuntimeError::Config(
+            "A2A_RUST_AUTH_SECRET is required: the runtime cannot build trust headers or decrypt \
+             encrypted auth blobs without a shared secret. Refusing to start."
+                .to_string(),
+        ));
+    }
+
+    // Surface cross-field config errors at startup rather than as silent
+    // runtime misbehaviour (timeouts that never trigger, retries that
+    // exceed their deadline, cache tiers that churn each other, etc.).
+    config
+        .validate_cross_field()
+        .map_err(RuntimeError::Config)?;
+
     let client = build_http_client(&config)?;
     let config_arc = Arc::new(config.clone());
 
@@ -131,7 +151,10 @@ pub async fn run(config: RuntimeConfig) -> Result<(), RuntimeError> {
             rx,
             client.clone(),
             config.backend_base_url.clone(),
-            config.auth_secret.clone().unwrap_or_default(),
+            config
+                .auth_secret
+                .clone()
+                .expect("auth_secret validated non-empty in run()"),
             Duration::from_millis(config.event_flush_interval_ms),
             config.event_flush_batch_size,
         );
@@ -288,7 +311,7 @@ mod tests {
             max_response_body_bytes: 1024,
             max_retries: 0,
             retry_backoff_ms: 1,
-            auth_secret: None,
+            auth_secret: Some("test-shared-secret".to_string()),
             backend_base_url: "http://127.0.0.1:4444".to_string(),
             max_concurrent: 1,
             max_queued: Some(4),
@@ -353,6 +376,30 @@ mod tests {
             .expect("runtime should start and shut down cleanly");
     }
 
+    #[tokio::test]
+    async fn run_rejects_missing_auth_secret() {
+        let mut config = test_config();
+        config.auth_secret = None; // pragma: allowlist secret
+        let err = run(config)
+            .await
+            .expect_err("missing auth_secret must fail");
+        assert!(
+            matches!(err, RuntimeError::Config(_)),
+            "expected Config error, got {err:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn run_rejects_empty_auth_secret() {
+        let mut config = test_config();
+        config.auth_secret = Some(String::new());
+        let err = run(config).await.expect_err("empty auth_secret must fail");
+        assert!(
+            matches!(err, RuntimeError::Config(_)),
+            "expected Config error, got {err:?}"
+        );
+    }
+
     #[tokio::test]
     async fn run_serves_uds_until_exit_after_startup() {
         let path = temp_socket_path("a2a-run");