Authenticate A2A requests and fix jwt-validation pipeline reload

README.md

@@ -1047,6 +1047,64 @@ After all benchmarks finish, the script prints and writes `e2e-results.md`:
 - **Parallel jobs mode** (`--parallel-jobs`): all benchmarks run concurrently; each gets unique OTEL-collector and Prometheus ports so local port-forwards don't collide.
 - **In-cluster mode** (`--in-cluster`): one Kubernetes Job is created per benchmark from `k8s/job.yaml`; `e2e-test.sh` streams the logs and checks the job's success status.
 
+## IBAC Comparison Script
+
+`run-ibac-comparison.sh` runs the **same benchmark twice for an
+apples-to-apples comparison** — once with an AuthBridge plugin preset and
+once as a baseline with no plugins — then compares the two runs with
+[`analyze-run.sh --compare`](#analyzing-traces-with-analyze-runsh). Use it
+to measure the cost and effect of a plugin pipeline (e.g. IBAC) against an
+otherwise identical run.
+
+For each invocation it:
+
+1. Deletes existing deployments, then runs `deploy-and-evaluate.sh` with the
+   selected `--plugin-preset` (the *plugin* run).
+2. Deletes deployments again, then runs `deploy-and-evaluate.sh` with no
+   preset (the *baseline* run).
+3. Calls `analyze-run.sh --compare <plugin>,<baseline>` to report the delta.
+
+Both runs share a short random experiment id so their names are unique
+across repeated invocations yet paired to each other. Experiment names are
+derived from the parameters, e.g.
+`gsm8k-10-parallel-1-<id>-ibac` (plugin) and `gsm8k-10-parallel-1-<id>`
+(baseline).
+
+The judge is configured from `OPENAI_API_BASE` / `OPENAI_API_KEY` (both
+required); these are wired through to the runner and the analysis step.
+
+### Usage
+
+```bash
+# Defaults: gsm8k, tool_calling, 10 tasks, 1 session, ibac-only preset
+./run-ibac-comparison.sh
+
+# Compare the full pipeline (auth + parsers + IBAC) against baseline
+./run-ibac-comparison.sh --plugin-preset full
+
+# Larger run with explicit model and concurrency
+./run-ibac-comparison.sh --model gcp/gemini-3-flash-preview \
+    --benchmark gsm8k --max-tasks 50 --max-parallel-sessions 4
+
+# Show help
+./run-ibac-comparison.sh --help
+```
+
+### Options
+
+| Option | Description | Default |
+|--------|-------------|---------|
+| `--model MODEL` | Model name | `gcp/gemini-3-flash-preview` |
+| `--benchmark NAME` | Benchmark name | `gsm8k` |
+| `--agent NAME` | Agent name | `tool_calling` |
+| `--max-tasks N` | Maximum number of tasks to evaluate | `10` |
+| `--max-parallel-sessions N` | Concurrent evaluation sessions | `1` |
+| `--plugin-preset PRESET` | Preset for the plugin run: `auth-only`, `ibac-only`, `full` | `ibac-only` |
+| `-h, --help` | Show help and exit | - |
+
+For preset contents and pipeline mechanics, see
+[AuthBridge Plugin Pipeline](#authbridge-plugin-pipeline).
+
 ## Current Limitations
 
 - No retry mechanism for failed operations

exgentic_a2a_runner/analyze_traces.py

@@ -33,6 +33,7 @@ class TraceRecord:
     experiment_name: str = "default"
     start_time: str = ""
     evaluation_result: bool | None = None
+    status_message: str = ""
 
     # Timing from child spans (seconds)
     session_creation_s: float = 0.0
@@ -109,6 +110,7 @@ def parse_traces(data: dict) -> list[TraceRecord]:
         num_parallel = int(meta.get("num_parallel_tasks", 0))
         session_id = meta.get("session_id", "unknown")
         status = root.get("statusCode", "UNSET")
+        status_message = root.get("statusMessage", "")
         evaluation_result = meta.get("evaluation_result")
 
         # Model from the invoke_agent child span or root metadata
@@ -134,6 +136,7 @@ def parse_traces(data: dict) -> list[TraceRecord]:
             experiment_name=experiment_name,
             start_time=root.get("startTime", ""),
             evaluation_result=evaluation_result,
+            status_message=status_message,
         )
 
         # Extract timing from child spans — collect chat spans separately
@@ -190,9 +193,11 @@ def parse_traces(data: dict) -> list[TraceRecord]:
             record.llm_total_s += chat_latency
             record.llm_count += 1
             child_attrs = parse_attrs(chat_span)
-            token_count = child_attrs.get("llm", {}).get("token_count", {})
-            record.llm_input_tokens += int(token_count.get("prompt", 0) or 0)
-            record.llm_output_tokens += int(token_count.get("completion", 0) or 0)
+            # Token usage from the OpenTelemetry gen_ai semantic-convention keys
+            # (gen_ai.usage.input_tokens/output_tokens) that current tracers emit.
+            gen_ai_usage = child_attrs.get("gen_ai", {}).get("usage", {})
+            record.llm_input_tokens += int(gen_ai_usage.get("input_tokens", 0) or 0)
+            record.llm_output_tokens += int(gen_ai_usage.get("output_tokens", 0) or 0)
 
             # Classify as before or after initial observation
             is_after_obs = False
@@ -630,6 +635,14 @@ def print_metric_row(label: str, metric_fn, fmt: str = ".2f") -> None:
         count_row += f" | {len(exp_groups[exp]):>{col_w}d}"
     print(count_row)
 
+    err_row = f"{'Error Rate (%)':<35s}"
+    for exp in experiments:
+        traces = exp_groups[exp]
+        n = len(traces)
+        rate = sum(1 for t in traces if t.status == "ERROR") / n * 100 if n else 0
+        err_row += f" | {rate:>{col_w}.1f}"
+    print(err_row)
+
     eval_row = f"{'Eval Success Rate (%)':<35s}"
     for exp in experiments:
         traces = exp_groups[exp]
@@ -686,6 +699,51 @@ def print_metric_row(label: str, metric_fn, fmt: str = ".2f") -> None:
 
     print()
 
+    # --- Error breakdown by type (verbatim root statusMessage) ---
+    # Count each ERROR-status root span under its statusMessage, per experiment.
+    # ERROR traces with no message are grouped under "(no message)".
+    A2A_PREFIX = "A2A task ended in state 'failed':"
+
+    def error_label(t: TraceRecord) -> str:
+        msg = (t.status_message or "").splitlines()[0].strip() if t.status_message else ""
+        if msg.startswith(A2A_PREFIX):
+            msg = msg[len(A2A_PREFIX):].strip()
+        if not msg:
+            return "(no message)"
+        return msg if len(msg) <= 100 else msg[:97] + "..."
+
+    # error type -> {experiment -> count}
+    err_counts: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
+    for r in records:
+        if r.status == "ERROR":
+            err_counts[error_label(r)][r.experiment_name] += 1
+
+    print("=" * 140)
+    print("Error Breakdown by Type")
+    print("=" * 140)
+    print()
+
+    if not err_counts:
+        print("No errors in any experiment.")
+        print()
+    else:
+        err_header = f"{'Error Type':<100s}"
+        for exp in experiments:
+            err_header += f" | {exp:>{col_w}s}"
+        print(err_header)
+        print("-" * len(err_header))
+
+        # Sort error types by total count across experiments (most frequent first)
+        for err_type in sorted(err_counts, key=lambda e: -sum(err_counts[e].values())):
+            row = f"{err_type:<100s}"
+            for exp in experiments:
+                count = err_counts[err_type].get(exp, 0)
+                total = len(exp_groups[exp])
+                pct = (count / total * 100) if total else 0
+                row += f" | {f'{count} ({pct:.0f}%)':>{col_w}s}"
+            print(row)
+        print()
+
 
 def main() -> int:
     import argparse

exgentic_a2a_runner/authbridge/apply-pipeline.sh

@@ -60,6 +60,25 @@ TOKEN_BROKER_URL="${TOKEN_BROKER_URL:-}"
 TOKEN_BROKER_AUDIENCE="${TOKEN_BROKER_AUDIENCE:-}"
 export TOKEN_BROKER_URL TOKEN_BROKER_AUDIENCE
 
+# --- jwt-validation config (consumed only when jwt-validation is selected,
+# which every preset does). The plugin needs issuer + keycloak_url/realm that
+# the operator ships as flat keys in the generic `authbridge-config` ConfigMap —
+# NOT in the per-agent config.yaml the merge reads. Source them from there so
+# the operator stays the source of truth; an explicit JWT_VALIDATION_* env
+# override wins. Keys match authbridge's jwtValidationConfig struct
+# (issuer + keycloak_url + keycloak_realm); Configure() DisallowUnknownFields,
+# so do NOT introduce provider* keys here — those belong to token-exchange.
+AUTHBRIDGE_BASE_CM="${AUTHBRIDGE_BASE_CM:-authbridge-config}"
+_ab_base_key() {
+    # Echo a single flat key from the operator base ConfigMap, or empty.
+    kubectl -n "$NAMESPACE" get configmap "$AUTHBRIDGE_BASE_CM" \
+        -o jsonpath="{.data.$1}" 2>/dev/null || true
+}
+JWT_VALIDATION_ISSUER="${JWT_VALIDATION_ISSUER:-$(_ab_base_key ISSUER)}"
+JWT_VALIDATION_KEYCLOAK_URL="${JWT_VALIDATION_KEYCLOAK_URL:-$(_ab_base_key KEYCLOAK_URL)}"
+JWT_VALIDATION_KEYCLOAK_REALM="${JWT_VALIDATION_KEYCLOAK_REALM:-$(_ab_base_key KEYCLOAK_REALM)}"
+export JWT_VALIDATION_ISSUER JWT_VALIDATION_KEYCLOAK_URL JWT_VALIDATION_KEYCLOAK_REALM
+
 echo "Applying AuthBridge pipeline to $NAMESPACE/$AGENT_NAME"
 echo "  Plugins: $PIPELINE_PLUGINS"
 if [ -n "${PIPELINE_OVERLAY_FILE:-}" ]; then
@@ -129,6 +148,26 @@ if $ibac_active && [ ! -s "$PROMPT_FILE" ]; then
     exit 1
 fi
 
+# --- Pre-flight: confirm the issuer is resolved when jwt-validation is
+# active. Without it the fragment renders an empty issuer and the sidecar
+# rejects the reload ("jwt-validation config: issuer is required"), leaving
+# the pod serving stale config — the exact failure this guard prevents.
+jwt_validation_active=false
+for tok in $PIPELINE_PLUGINS; do
+    name=${tok%%:*}
+    policy=${tok#*:}; [[ "$policy" == "$tok" ]] && policy=enforce
+    if [[ "$name" == "jwt-validation" && "$policy" != "off" ]]; then
+        jwt_validation_active=true
+        break
+    fi
+done
+if $jwt_validation_active && [ -z "${JWT_VALIDATION_ISSUER:-}" ]; then
+    echo "ERROR: jwt-validation is active but no issuer could be resolved." >&2
+    echo "       Expected key ISSUER in ConfigMap $NAMESPACE/$AUTHBRIDGE_BASE_CM," >&2
+    echo "       or set JWT_VALIDATION_ISSUER explicitly." >&2
+    exit 1
+fi
+
 # --- Render every plugin fragment with envsubst into a temp dir. The
 # merge script reads these as the rendered defaults; --plugin-config-file
 # overrides are deep-merged on top inside pipeline-merge.py.

exgentic_a2a_runner/authbridge/pipeline-merge.py

@@ -29,6 +29,7 @@
 """
 
 import argparse
+import copy
 import sys
 from pathlib import Path
 from typing import Any
@@ -118,22 +119,63 @@ def build_entry(
     policy: str,
     fragment: dict[str, Any],
     overrides: dict[str, Any] | None,
+    operator_config: dict[str, Any] | None,
 ) -> dict[str, Any]:
-    """Build the final plugin entry: fragment defaults + overrides + on_error."""
+    """Build the final plugin entry.
+
+    Config precedence (later wins on leaf conflicts):
+        operator base config  <  fragment defaults  <  --config-file overrides
+
+    The operator base is the config block the operator already rendered for
+    this plugin in the ConfigMap we read from stdin. Preserving it is the
+    whole point: several plugins (jwt-validation's issuer, token-exchange's
+    token_url/identity/routes) are configured ONLY by the operator, and a
+    prior version of this script replaced the plugin list wholesale — which
+    dropped that config and made the sidecar reject the reload
+    ("issuer is required", "token_url is required"). We now start from the
+    operator's block and layer our fragment + overrides on top, matching how
+    authbridge's own `abctl edit` splices rather than rebuilds.
+    """
     entry: dict[str, Any] = {"name": name}
-    cfg = fragment.get("config")
-    if isinstance(cfg, dict) and cfg:
-        entry["config"] = dict(cfg)
+    config: dict[str, Any] = {}
+    if isinstance(operator_config, dict) and operator_config:
+        config = copy.deepcopy(operator_config)
+    frag_cfg = fragment.get("config")
+    if isinstance(frag_cfg, dict) and frag_cfg:
+        deep_merge(config, copy.deepcopy(frag_cfg))
     if overrides:
-        entry.setdefault("config", {})
-        deep_merge(entry["config"], overrides)
+        deep_merge(config, overrides)
+    if config:
+        entry["config"] = config
     # `enforce` is the framework default — omit on_error to keep diffs
     # minimal. observe/off are explicit.
     if policy != "enforce":
         entry["on_error"] = policy
     return entry
 
 
+def index_operator_config(operator: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Map plugin name -> its `config` block as rendered by the operator in
+    the ConfigMap read from stdin. Scans both chains; missing/blank config
+    yields no entry."""
+    by_name: dict[str, dict[str, Any]] = {}
+    pipeline = operator.get("pipeline")
+    if not isinstance(pipeline, dict):
+        return by_name
+    for chain in ("inbound", "outbound"):
+        section = pipeline.get(chain)
+        if not isinstance(section, dict):
+            continue
+        for entry in section.get("plugins", []) or []:
+            if not isinstance(entry, dict):
+                continue
+            n = entry.get("name")
+            cfg = entry.get("config")
+            if isinstance(n, str) and isinstance(cfg, dict) and cfg:
+                by_name[n] = cfg
+    return by_name
+
+
 def validate_mutex(resolved: dict[str, str]) -> None:
     """Reject mutually-exclusive plugin pairs both active (not off) in the
     same chain."""
@@ -207,31 +249,35 @@ def main() -> int:
         if text.strip():
             ibac_prompt = text
 
-    # Build inbound and outbound chains in canonical order.
+    # Index the config blocks the operator already rendered, so build_entry
+    # can preserve them (issuer, token_url/identity, …) instead of dropping
+    # them. Done before we overwrite the plugin lists below.
+    operator_config = index_operator_config(operator)
+
+    # Build inbound and outbound chains in canonical order. We're
+    # authoritative for the chain composition (which plugins, in what order,
+    # with what on_error) by design (see §4.3), but each plugin's config is
+    # seeded from the operator's rendered block so operator-supplied settings
+    # survive.
     inbound_entries: list[dict[str, Any]] = []
     for name in INBOUND_ORDER:
-        # Skip plugins not in the operator base when our policy is `off`
-        # — no need to emit a no-op entry for something that doesn't
-        # exist downstream. The operator base today enables every
-        # supported plugin, so this branch effectively never fires; it's
-        # here for the case where the operator drops one in the future.
-        # (Conservative behavior: always emit, since we don't read the
-        # operator base here. We *always* emit.)
         fragment = load_fragment(args.plugins_dir, name)
-        entry = build_entry(name, resolved[name], fragment, overrides.get(name))
+        entry = build_entry(name, resolved[name], fragment,
+                            overrides.get(name), operator_config.get(name))
         inbound_entries.append(entry)
 
     outbound_entries: list[dict[str, Any]] = []
     for name in OUTBOUND_ORDER:
         fragment = load_fragment(args.plugins_dir, name)
-        entry = build_entry(name, resolved[name], fragment, overrides.get(name))
+        entry = build_entry(name, resolved[name], fragment,
+                            overrides.get(name), operator_config.get(name))
         if name == "ibac" and ibac_prompt is not None and resolved[name] != "off":
             cfg = entry.setdefault("config", {})
             cfg["system_prompt"] = ibac_prompt
         outbound_entries.append(entry)
 
-    # Replace operator's plugin lists with our resolved ones. We're
-    # authoritative for the chain composition by design (see §4.3).
+    # Replace operator's plugin lists with our resolved ones (composition is
+    # ours; per-plugin config was preserved via operator_config above).
     pipeline = operator.setdefault("pipeline", {})
     pipeline.setdefault("inbound", {})["plugins"] = inbound_entries
     pipeline.setdefault("outbound", {})["plugins"] = outbound_entries

exgentic_a2a_runner/authbridge/plugins/jwt-validation.yaml

@@ -1,8 +1,30 @@
 # jwt-validation plugin fragment.
 #
-# The operator base config supplies issuer/keycloak_url/keycloak_realm.
-# This fragment exists so the merge can keep the plugin in the resolved
-# selection (or flip its on_error policy) without re-rendering the
-# operator's heavy config block. Per-plugin overrides may be supplied
-# via --plugin-config-file (see AUTHBRIDGE_PIPELINE_SPEC.md §4.5).
+# Placeholders below are expanded by apply-pipeline.sh (envsubst) before
+# the merge runs. apply-pipeline.sh sources the issuer / keycloak values
+# from the operator's `authbridge-config` ConfigMap (flat keys ISSUER,
+# KEYCLOAK_URL, KEYCLOAK_REALM) and exports them as the JWT_VALIDATION_*
+# vars below; a JWT_VALIDATION_* override in the environment wins over the
+# operator value.
+#
+# Historically this fragment was config-less on the assumption that the
+# operator base config already carried a rendered jwt-validation config
+# block. It does not — the operator ships flat env keys in a separate
+# ConfigMap — so pipeline-merge.py (which replaces the inbound plugin
+# list wholesale) emitted a bare `name: jwt-validation`, and the sidecar
+# rejected the reload with "issuer is required". Rendering the config
+# here, like plugins/ibac.yaml, keeps the operator as the source of truth
+# while giving the plugin the config it needs.
+#
+# Schema matches the authbridge jwtValidationConfig struct
+# (authbridge/authlib/plugins/jwtvalidation/plugin.go @ v0.6.0-alpha.12):
+# issuer (required), keycloak_url + keycloak_realm (derive jwks_url).
+# Configure() calls DisallowUnknownFields(), so ONLY these keys are
+# accepted — note `provider`/`provider_url`/`provider_realm` belong to the
+# token-exchange plugin, NOT jwt-validation. Per-plugin overrides may still
+# be supplied via --plugin-config-file (see AUTHBRIDGE_PIPELINE_SPEC.md §4.5).
 name: jwt-validation
+config:
+  issuer: "${JWT_VALIDATION_ISSUER}"
+  keycloak_url: "${JWT_VALIDATION_KEYCLOAK_URL}"
+  keycloak_realm: "${JWT_VALIDATION_KEYCLOAK_REALM}"