From 0bfaa6ec1aac64a0c5529bc3e37b25de12b87d22 Mon Sep 17 00:00:00 2001 From: Yoav Katz Date: Sun, 26 Jul 2026 21:12:46 +0300 Subject: [PATCH 1/5] Authenticate A2A requests and fix jwt-validation pipeline reload Wire a Keycloak bearer token through the A2A evaluation path so requests are accepted by the agent's authbridge sidecar, and fix the pipeline merge so jwt-validation reloads instead of failing with "issuer is required". - a2a_client: attach Authorization: Bearer from config.auth_token (A2A_AUTH_TOKEN) to the agent-card fetch and every send_message. - evaluate-benchmark.sh: obtain a rossoctl-realm token via the direct-access-grant flow when A2A_AUTH_TOKEN is unset. - pipeline-merge.py: seed each plugin entry from the operator's rendered config block (deep-merge fragment + overrides on top) instead of replacing the plugin list wholesale, preserving operator-only config such as jwt-validation's issuer and token-exchange's token_url. - jwt-validation.yaml: render issuer/keycloak_url/keycloak_realm from the operator authbridge-config ConfigMap; document the DisallowUnknownFields schema constraint. - apply-pipeline.sh: source JWT_VALIDATION_* from the base ConfigMap and pre-flight fail if the issuer can't be resolved while jwt-validation is active. - deploy-agent.sh: fail fast with a clear message when python3/PyYAML is missing and a plugin selector was supplied. - run-ibac-comparison.sh: new wrapper to run a benchmark with a plugin preset vs a baseline and compare. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yoav Katz --- .../authbridge/apply-pipeline.sh | 39 +++++ .../authbridge/pipeline-merge.py | 82 +++++++--- .../authbridge/plugins/jwt-validation.yaml | 32 +++- exgentic_a2a_runner/deploy-agent.sh | 24 +++ exgentic_a2a_runner/evaluate-benchmark.sh | 40 +++++ .../exgentic_a2a_runner/a2a_client.py | 9 +- exgentic_a2a_runner/run-ibac-comparison.sh | 140 ++++++++++++++++++ 7 files changed, 342 insertions(+), 24 deletions(-) create mode 100755 exgentic_a2a_runner/run-ibac-comparison.sh diff --git a/exgentic_a2a_runner/authbridge/apply-pipeline.sh b/exgentic_a2a_runner/authbridge/apply-pipeline.sh index 3847229..e3fa5b9 100755 --- a/exgentic_a2a_runner/authbridge/apply-pipeline.sh +++ b/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. diff --git a/exgentic_a2a_runner/authbridge/pipeline-merge.py b/exgentic_a2a_runner/authbridge/pipeline-merge.py index a0e7d3b..3cfe749 100755 --- a/exgentic_a2a_runner/authbridge/pipeline-merge.py +++ b/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,15 +119,34 @@ 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": @@ -134,6 +154,28 @@ def build_entry( 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 diff --git a/exgentic_a2a_runner/authbridge/plugins/jwt-validation.yaml b/exgentic_a2a_runner/authbridge/plugins/jwt-validation.yaml index 3791cc5..1ecc207 100644 --- a/exgentic_a2a_runner/authbridge/plugins/jwt-validation.yaml +++ b/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}" diff --git a/exgentic_a2a_runner/deploy-agent.sh b/exgentic_a2a_runner/deploy-agent.sh index 8846d47..abab018 100755 --- a/exgentic_a2a_runner/deploy-agent.sh +++ b/exgentic_a2a_runner/deploy-agent.sh @@ -180,6 +180,30 @@ if [ -z "$BENCHMARK_NAME" ] || [ -z "$AGENT_NAME_INPUT" ]; then exit 1 fi +# Prerequisite check: AuthBridge plugin resolution (Step 8) and the skip_hosts +# merge (Step 7.5) both shell out to `python3` with PyYAML. That happens deep +# into the deploy after minutes of network work, so a missing venv/module fails +# late (ModuleNotFoundError: No module named 'yaml'). Detect it here, up front, +# and only when a plugin selector was actually supplied (the sole trigger for +# those Python steps). +if [ -n "$PIPELINE_PRESET" ] || [ ${#PIPELINE_SELECTORS[@]} -gt 0 ] || [ -n "$PIPELINE_OVERLAY_FILE" ]; then + if ! command -v python3 >/dev/null 2>&1; then + echo "Error: python3 is required for AuthBridge plugin resolution but was not found on PATH." >&2 + echo " You passed a plugin selector (--plugin/--no-plugin/--plugin-preset/--plugin-config-file)," >&2 + echo " which needs python3 + PyYAML. Install python3 and re-run." >&2 + exit 1 + fi + if ! python3 -c 'import yaml' 2>/dev/null; then + echo "Error: python3 is on PATH but the PyYAML module is missing." >&2 + echo " $(command -v python3) cannot 'import yaml'. This usually means the wrong" >&2 + echo " Python/venv is active. Activate the venv with PyYAML, or install it:" >&2 + echo " pip3 install --user pyyaml" >&2 + echo " brew install libyaml && pip3 install pyyaml # macOS" >&2 + echo " sudo apt install python3-yaml # Debian/Ubuntu" >&2 + exit 1 + fi +fi + # Determine agent name and image # Automatically add exgentic-a2a- prefix if not already present if [[ "$AGENT_NAME_INPUT" == exgentic-a2a-* ]]; then diff --git a/exgentic_a2a_runner/evaluate-benchmark.sh b/exgentic_a2a_runner/evaluate-benchmark.sh index 1725659..02d550b 100755 --- a/exgentic_a2a_runner/evaluate-benchmark.sh +++ b/exgentic_a2a_runner/evaluate-benchmark.sh @@ -378,6 +378,46 @@ source .venv/bin/activate export EXGENTIC_MCP_SERVER_URL="${MCP_BASE_URL}/mcp" export A2A_BASE_URL="$AGENT_BASE_URL" +# Obtain a Keycloak bearer token for the A2A agent endpoint. The agent runs +# behind an authbridge sidecar that validates a JWT from the "rossoctl" realm, +# so unauthenticated requests get HTTP 401. This mirrors the direct-access-grant +# flow used by deploy-benchmark.sh / deploy-agent.sh (client_id=rossoctl). +# Skip if A2A_AUTH_TOKEN is already provided in the environment/.env. +if [ -z "${A2A_AUTH_TOKEN:-}" ]; then + echo "Obtaining Keycloak token for A2A agent..." + KEYCLOAK_API="$(keycloak_api_url)" + KEYCLOAK_USERNAME="${KEYCLOAK_USERNAME:-admin}" + + # Resolve the password the same way deploy-agent.sh does: prefer an explicit + # KEYCLOAK_PASSWORD, else the rossoctl-test-user cluster secret, else "admin". + if [ -z "${KEYCLOAK_PASSWORD:-}" ] || [ "${KEYCLOAK_PASSWORD}" = "unknown" ]; then + KC_SECRET_PASSWORD=$("$KUBECTL_BIN" get secret rossoctl-test-user -n keycloak \ + -o jsonpath='{.data.password}' 2>/dev/null | base64 -d 2>/dev/null || echo "") + KEYCLOAK_PASSWORD="${KC_SECRET_PASSWORD:-admin}" + fi + + A2A_TOKEN_RESPONSE=$(curl -s -X POST "$KEYCLOAK_API/realms/rossoctl/protocol/openid-connect/token" \ + -H "Content-Type: application/x-www-form-urlencoded" \ + -d "username=${KEYCLOAK_USERNAME}" \ + -d "password=${KEYCLOAK_PASSWORD}" \ + -d "grant_type=password" \ + -d "client_id=rossoctl" 2>/dev/null || echo "") + + A2A_AUTH_TOKEN=$(echo "$A2A_TOKEN_RESPONSE" | grep -o '"access_token":"[^"]*"' | sed 's/"access_token":"\([^"]*\)"/\1/') + if [ -z "$A2A_AUTH_TOKEN" ]; then + echo "Error: Could not obtain Keycloak token for the A2A agent." + echo " Keycloak: $KEYCLOAK_API (realm rossoctl, client_id rossoctl, user $KEYCLOAK_USERNAME)" + echo " Response: $A2A_TOKEN_RESPONSE" + echo " Ensure Direct Access Grants are enabled for the rossoctl client" + echo " (deploy-agent.sh enables them) or set A2A_AUTH_TOKEN explicitly." + exit 1 + fi + export A2A_AUTH_TOKEN + echo "✓ Obtained A2A bearer token" +else + echo "Using A2A_AUTH_TOKEN from environment" +fi + # Set tool prefix when using MCP gateway (gateway namespaces tools with a prefix) if [ "$USE_MCP_GATEWAY" = "true" ]; then export EXGENTIC_MCP_TOOL_PREFIX="${EXGENTIC_MCP_TOOL_PREFIX:-exgentic_${BENCHMARK_NAME}_}" diff --git a/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py b/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py index 3c80427..84afa1e 100644 --- a/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py +++ b/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py @@ -96,9 +96,16 @@ async def _async_send_prompt(self, prompt: str, timeout_s: float, otel_context=N # keys intent off authbridge's own session store (populated by a2a-parser), # not this header — but emitting it is harmless and useful for tracing. x_session_id = uuid.uuid4().hex + headers = {"x-session-id": x_session_id} + # Attach the Keycloak-issued bearer token so authbridge (the agent's + # sidecar) accepts the request; without it the endpoint returns 401. + # The same header is applied to the agent-card fetch and every + # send_message call because the client is reused for both. + if self.config.auth_token: + headers["Authorization"] = f"Bearer {self.config.auth_token}" httpx_client = httpx.AsyncClient( timeout=timeout_s, - headers={"x-session-id": x_session_id}, + headers=headers, ) if self.otel_enabled: try: diff --git a/exgentic_a2a_runner/run-ibac-comparison.sh b/exgentic_a2a_runner/run-ibac-comparison.sh new file mode 100755 index 0000000..fc9a33c --- /dev/null +++ b/exgentic_a2a_runner/run-ibac-comparison.sh @@ -0,0 +1,140 @@ +#!/usr/bin/env bash +# +# Run a benchmark twice for comparison: once with a plugin preset and once +# without any plugins (baseline). Experiment names are derived from the +# parameters. +# +# Usage: +# ./run-ibac-comparison.sh [OPTIONS] +# +# Options: +# --model MODEL Model name (default: gcp/gemini-3-flash-preview) +# --benchmark NAME Benchmark name (default: gsm8k) +# --agent NAME Agent name (default: tool_calling) +# --max-tasks N Maximum number of tasks to evaluate (default: 10) +# --max-parallel-sessions N Number of concurrent evaluation sessions (default: 1) +# --plugin-preset PRESET Plugin preset for the first run: +# auth-only | ibac-only | full (default: ibac-only) +# -h, --help Show this help and exit +# +# The judge is configured from the OPENAI_API_BASE / OPENAI_API_KEY environment +# variables (as in the original template). +# +# Examples: +# ./run-ibac-comparison.sh +# ./run-ibac-comparison.sh --model gcp/gemini-3-flash-preview --benchmark gsm8k --max-tasks 10 --max-parallel-sessions 1 +# ./run-ibac-comparison.sh --plugin-preset full + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# Defaults (from the template) +MODEL="gcp/gemini-3-flash-preview" +BENCHMARK="gsm8k" +AGENT="tool_calling" +MAX_TASKS=10 +MAX_PARALLEL_SESSIONS=1 +PLUGIN_PRESET="ibac-only" + +while [[ $# -gt 0 ]]; do + case "$1" in + --model) + MODEL="$2" + shift 2 + ;; + --benchmark) + BENCHMARK="$2" + shift 2 + ;; + --agent) + AGENT="$2" + shift 2 + ;; + --max-tasks) + MAX_TASKS="$2" + shift 2 + ;; + --max-parallel-sessions) + MAX_PARALLEL_SESSIONS="$2" + shift 2 + ;; + --plugin-preset) + PLUGIN_PRESET="$2" + shift 2 + ;; + -h|--help) + sed -n '2,26p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//' + exit 0 + ;; + *) + echo "Error: unknown option: $1" >&2 + echo "Run '$0 --help' for usage." >&2 + exit 1 + ;; + esac +done + +# Validate the plugin preset. deploy-and-evaluate.sh accepts only these values; +# passing anything else (e.g. "auth_only" with an underscore) produces a broken +# auth pipeline rather than a clear error, so reject it up front. +case "$PLUGIN_PRESET" in + auth-only|ibac-only|full) + ;; + *) + echo "Error: invalid --plugin-preset: '$PLUGIN_PRESET'" >&2 + echo "Valid values: auth-only | ibac-only | full" >&2 + exit 1 + ;; +esac + +# Experiment names derived from the params. The plugin run is suffixed with the +# preset name (e.g. "-ibac" for ibac-only) to keep names meaningful per preset. +PRESET_SUFFIX="${PLUGIN_PRESET%-only}" +EXPERIMENT_BASE="${BENCHMARK}-${MAX_TASKS}-parallel-${MAX_PARALLEL_SESSIONS}" +EXPERIMENT_PLUGIN="${EXPERIMENT_BASE}-${PRESET_SUFFIX}" + +# Judge configuration (from the template environment). +: "${OPENAI_API_BASE:?OPENAI_API_BASE must be set}" +: "${OPENAI_API_KEY:?OPENAI_API_KEY must be set}" + +echo "==========================================" +echo "Model: $MODEL" +echo "Benchmark: $BENCHMARK" +echo "Agent: $AGENT" +echo "Max tasks: $MAX_TASKS" +echo "Max parallel sessions: $MAX_PARALLEL_SESSIONS" +echo "Plugin preset: $PLUGIN_PRESET" +echo "Plugin experiment: $EXPERIMENT_PLUGIN" +echo "Baseline experiment: $EXPERIMENT_BASE" +echo "==========================================" + +# Run 1: with the selected plugin preset. +"$SCRIPT_DIR/delete-all-deployments.sh" +env IBAC_JUDGE_ENDPOINT="$OPENAI_API_BASE" \ + IBAC_JUDGE_MODEL="$MODEL" \ + JUDGE_BEARER="$OPENAI_API_KEY" \ + "$SCRIPT_DIR/deploy-and-evaluate.sh" \ + --benchmark "$BENCHMARK" \ + --agent "$AGENT" \ + --model "openai/$MODEL" \ + --max-tasks "$MAX_TASKS" \ + --max-parallel-sessions "$MAX_PARALLEL_SESSIONS" \ + --plugin-preset "$PLUGIN_PRESET" \ + --experiment "$EXPERIMENT_PLUGIN" + +# Run 2: baseline (no plugin preset). +"$SCRIPT_DIR/delete-all-deployments.sh" +env IBAC_JUDGE_ENDPOINT="$OPENAI_API_BASE" \ + IBAC_JUDGE_MODEL="$MODEL" \ + JUDGE_BEARER="$OPENAI_API_KEY" \ + "$SCRIPT_DIR/deploy-and-evaluate.sh" \ + --benchmark "$BENCHMARK" \ + --agent "$AGENT" \ + --model "openai/$MODEL" \ + --max-tasks "$MAX_TASKS" \ + --max-parallel-sessions "$MAX_PARALLEL_SESSIONS" \ + --experiment "$EXPERIMENT_BASE" + +# Compare the two runs. +"$SCRIPT_DIR/analyze-run.sh" -c "${EXPERIMENT_PLUGIN},${EXPERIMENT_BASE}" From fecf3b57b823cbc119c7a77cc51b49ad4e7d8f89 Mon Sep 17 00:00:00 2001 From: Yoav Katz Date: Mon, 27 Jul 2026 12:16:01 +0300 Subject: [PATCH 2/5] Track A2A task failure state and gen_ai token usage Surface A2A execution failures as errors, fix trace token accounting to the current OpenTelemetry gen_ai keys, and make comparison experiment names unique per run. - a2a_client: add A2ATaskError and inspect the task's terminal state; a task ending in failed/rejected/canceled is raised as an error even when it returned descriptive text, so the runner marks Agent.Call / Agent.Session spans ERROR instead of counting it as a success. - analyze_traces: read token usage from gen_ai.usage.input_tokens/ output_tokens (the keys current tracers emit) rather than the stale llm.token_count.prompt/completion path. - run-ibac-comparison.sh: append a short random id to experiment names so repeated runs with the same params don't collide; read /dev/urandom via a bounded head to avoid an intermittent SIGPIPE-induced exit under pipefail + set -e. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yoav Katz --- exgentic_a2a_runner/analyze_traces.py | 8 ++-- .../exgentic_a2a_runner/a2a_client.py | 41 ++++++++++++++++++- exgentic_a2a_runner/run-ibac-comparison.sh | 14 ++++++- 3 files changed, 57 insertions(+), 6 deletions(-) diff --git a/exgentic_a2a_runner/analyze_traces.py b/exgentic_a2a_runner/analyze_traces.py index 68d5cd3..47e067c 100644 --- a/exgentic_a2a_runner/analyze_traces.py +++ b/exgentic_a2a_runner/analyze_traces.py @@ -190,9 +190,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 diff --git a/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py b/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py index 84afa1e..719a8a0 100644 --- a/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py +++ b/exgentic_a2a_runner/exgentic_a2a_runner/a2a_client.py @@ -16,6 +16,24 @@ logger = logging.getLogger(__name__) +class A2ATaskError(RuntimeError): + """Raised when the A2A task terminates in a failure state. + + The A2A agent can report an execution failure (e.g. "LLM produced + unparseable output") by ending the task in a ``failed``/``rejected``/ + ``canceled`` state while still returning descriptive text. That text is a + normal response, not an exception, so without this the runner would treat + the session as successful. Raising here lets the runner's failure path mark + the Agent.Call / Agent.Session spans as ERROR. + """ + + def __init__(self, state: str, message: str): + self.state = state + self.agent_message = message + detail = message.strip() or "" + super().__init__(f"A2A task ended in state '{state}': {detail}") + + class A2AProxyClient: """Client for A2A protocol communication using the standard a2a-sdk.""" @@ -78,7 +96,7 @@ async def _async_send_prompt(self, prompt: str, timeout_s: float, otel_context=N import httpx from a2a.client import ClientConfig, ClientFactory, create_text_message_object from a2a.client.card_resolver import A2ACardResolver - from a2a.types import Role, TextPart + from a2a.types import Role, TaskState, TextPart # Restore the OTEL context from the calling thread so that # httpx instrumentation creates spans under the correct parent. @@ -145,6 +163,11 @@ async def _async_send_prompt(self, prompt: str, timeout_s: float, otel_context=N result_text = "" task_id = None event_count = 0 + # Track the task's terminal state. The agent signals an execution + # failure by ending the task in a non-completed state while still + # emitting descriptive text, so we must inspect the state rather + # than assume any returned text means success. + final_state: Optional[TaskState] = None request_metadata = {"session_id": session_id} if session_id else None async for response in client.send_message(message, request_metadata=request_metadata): @@ -156,6 +179,11 @@ async def _async_send_prompt(self, prompt: str, timeout_s: float, otel_context=N task_id = task.id logger.debug(f"Task ID: {task_id}") + # Latest known task state wins; the terminal event carries + # the final state (completed / failed / rejected / ...). + if task.status and task.status.state: + final_state = task.status.state + # Extract text from artifact events if event and hasattr(event, "artifact") and event.artifact: for part in event.artifact.parts: @@ -169,8 +197,17 @@ async def _async_send_prompt(self, prompt: str, timeout_s: float, otel_context=N result_text += part.root.text logger.debug( - f"Completed: {event_count} events, {len(result_text)} chars" + f"Completed: {event_count} events, {len(result_text)} chars, " + f"final state: {final_state}" ) + + # A task that ended in a failure state is an Agent.Call error even + # though it returned text. Raise so the runner marks the spans ERROR. + # completed/working/submitted/input-required/unknown are not treated + # as errors here (a wrong-but-valid answer stays a successful call). + if final_state in (TaskState.failed, TaskState.rejected, TaskState.canceled): + raise A2ATaskError(final_state.value, result_text) + return result_text finally: diff --git a/exgentic_a2a_runner/run-ibac-comparison.sh b/exgentic_a2a_runner/run-ibac-comparison.sh index fc9a33c..7ff71f5 100755 --- a/exgentic_a2a_runner/run-ibac-comparison.sh +++ b/exgentic_a2a_runner/run-ibac-comparison.sh @@ -88,10 +88,21 @@ case "$PLUGIN_PRESET" in ;; esac +# A short random id keeps experiment names unique across repeated invocations +# with the same parameters (5 lowercase-alphanumeric chars). +# +# Read a fixed, bounded chunk from /dev/urandom BEFORE filtering: piping the +# unbounded stream straight into `head -c 5` makes `head` close the pipe early, +# which sends SIGPIPE to `tr` and — under `set -o pipefail` + `set -e` — kills +# the whole script intermittently. `head` on the source bounds the read so no +# writer is left with a closed pipe. +EXPERIMENT_ID="$(LC_ALL=C tr -dc 'a-z0-9' < <(head -c 256 /dev/urandom) | cut -c1-5)" + # Experiment names derived from the params. The plugin run is suffixed with the # preset name (e.g. "-ibac" for ibac-only) to keep names meaningful per preset. +# The random id is appended so both runs in a comparison share the same id. PRESET_SUFFIX="${PLUGIN_PRESET%-only}" -EXPERIMENT_BASE="${BENCHMARK}-${MAX_TASKS}-parallel-${MAX_PARALLEL_SESSIONS}" +EXPERIMENT_BASE="${BENCHMARK}-${MAX_TASKS}-parallel-${MAX_PARALLEL_SESSIONS}-${EXPERIMENT_ID}" EXPERIMENT_PLUGIN="${EXPERIMENT_BASE}-${PRESET_SUFFIX}" # Judge configuration (from the template environment). @@ -105,6 +116,7 @@ echo "Agent: $AGENT" echo "Max tasks: $MAX_TASKS" echo "Max parallel sessions: $MAX_PARALLEL_SESSIONS" echo "Plugin preset: $PLUGIN_PRESET" +echo "Experiment id: $EXPERIMENT_ID" echo "Plugin experiment: $EXPERIMENT_PLUGIN" echo "Baseline experiment: $EXPERIMENT_BASE" echo "==========================================" From 9d8ca3fb11a2639d210d22e74f0ed23aafac9cf2 Mon Sep 17 00:00:00 2001 From: Yoav Katz Date: Mon, 27 Jul 2026 12:22:32 +0300 Subject: [PATCH 3/5] Document run-ibac-comparison.sh in the README Add an "IBAC Comparison Script" section describing the plugin-vs-baseline comparison wrapper: what it does, the shared random experiment id, the judge env requirements, usage examples, and an options table matching the script's --help defaults. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yoav Katz --- README.md | 58 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/README.md b/README.md index cb8d29f..f96f267 100644 --- a/README.md +++ b/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 ,` 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--ibac` (plugin) and `gsm8k-10-parallel-1-` +(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 From a8651311e8d39c0a90f697e3d7cc50da56a5a8ac Mon Sep 17 00:00:00 2001 From: Yoav Katz Date: Mon, 27 Jul 2026 18:16:07 +0300 Subject: [PATCH 4/5] Show error rate in experiment comparison report Add an Error Rate (%) row to the --compare/-c report, printed just before Eval Success Rate. Uses the same ERROR-status definition as the per-parallel and per-group tables. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yoav Katz --- exgentic_a2a_runner/analyze_traces.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/exgentic_a2a_runner/analyze_traces.py b/exgentic_a2a_runner/analyze_traces.py index 47e067c..bbc61ec 100644 --- a/exgentic_a2a_runner/analyze_traces.py +++ b/exgentic_a2a_runner/analyze_traces.py @@ -632,6 +632,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] From 6149a02fde0c151492d2fc8d2906a409376164a3 Mon Sep 17 00:00:00 2001 From: Yoav Katz Date: Mon, 27 Jul 2026 18:36:11 +0300 Subject: [PATCH 5/5] Add per-experiment error-type breakdown to comparison report The --compare/-c report now includes an "Error Breakdown by Type" table below the main comparison, counting ERROR-status root spans by their verbatim statusMessage per experiment, shown as count (pct%). - Thread statusMessage from the root Agent.Session span into TraceRecord. - Group ERROR traces by first line of statusMessage; blank -> "(no message)". Strip the "A2A task ended in state 'failed':" prefix and truncate labels to 100 chars. - Rows are the union of error types across experiments, sorted by total frequency; the section always prints, showing "No errors" when empty. Co-Authored-By: Claude Opus 4.8 (1M context) Signed-off-by: Yoav Katz --- exgentic_a2a_runner/analyze_traces.py | 48 +++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/exgentic_a2a_runner/analyze_traces.py b/exgentic_a2a_runner/analyze_traces.py index bbc61ec..df31cc9 100644 --- a/exgentic_a2a_runner/analyze_traces.py +++ b/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 @@ -696,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