feat(benchmarks): proxy external URL routing, HAProxy port alignment, and fleet filtering

.trivyignore

@@ -20,6 +20,12 @@ CVE-2025-68121 exp:2026-11-30
 # Added: 2026-02-23
 CVE-2024-45337 exp:2026-11-30
 
+# CVE-2026-56854: golang.org/x/crypto/ssh - Authentication bypass due to unenforced source-address restrictions
+# Present in pre-compiled helm, oras, and tofu binaries
+# Not exploitable in our context (we don't run an SSH server via helm/oras/tofu)
+# Added: 2026-09-04
+CVE-2026-56854 exp:2026-11-30
+
 # CVE-2026-33186: gRPC authorization bypass (google.golang.org/grpc < 1.79.3)
 # Affects: helm and tofu binaries in Docker image (grpc v1.76.0)
 # Status: Waiting for upstream helm/tofu releases with fixed grpc

backend/models/benchmark.py

@@ -118,6 +118,11 @@ class BenchmarkRun(Base):
     proxy_deployment = relationship("ProxyDeployment", back_populates="runs")
     run_group = relationship("BenchmarkRunGroup", back_populates="runs")
 
+    @property
+    def cluster_name(self) -> str | None:
+        """Name of the associated Kubernetes cluster (via target)."""
+        return self.target.cluster.name if (self.target and self.target.cluster) else None
+
     __table_args__ = (
         Index("idx_benchmark_run_proxy_created", "proxy", "created_at"),
         Index("idx_benchmark_run_tool_proxy", "tool", "proxy"),
@@ -183,6 +188,11 @@ class BenchmarkRunGroup(Base):
     runs = relationship("BenchmarkRun", back_populates="run_group", order_by="BenchmarkRun.id")
     target = relationship("BenchmarkTarget")
 
+    @property
+    def cluster_name(self) -> str | None:
+        """Name of the associated Kubernetes cluster (via target)."""
+        return self.target.cluster.name if (self.target and self.target.cluster) else None
+
     __table_args__ = (
         Index("idx_benchmark_run_group_scenario", "scenario_key"),
         Index("idx_benchmark_run_group_status_created", "status", "created_at"),
@@ -330,6 +340,11 @@ class BenchmarkTarget(Base):
     proxy_deployments = relationship("ProxyDeployment", back_populates="target", cascade="all, delete-orphan")
     runs = relationship("BenchmarkRun", back_populates="target")
 
+    @property
+    def cluster_name(self) -> str | None:
+        """Name of the associated Kubernetes cluster."""
+        return self.cluster.name if self.cluster else None
+
     @property
     def proxy_count(self) -> int:
         """Number of proxy deployments for this target (used in list view)."""

backend/openapi.json

@@ -2,7 +2,7 @@
   "openapi": "3.1.0",
   "info": {
     "title": "BNK-Forge API",
-    "version": "3.1.6"
+    "version": "4.0.0"
   },
   "paths": {
     "/": {
@@ -27009,6 +27009,22 @@
               "title": "Status"
             }
           },
+          {
+            "name": "cluster_id",
+            "in": "query",
+            "required": false,
+            "schema": {
+              "anyOf": [
+                {
+                  "type": "integer"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "title": "Cluster Id"
+            }
+          },
           {
             "name": "limit",
             "in": "query",
@@ -34359,6 +34375,17 @@
             ],
             "title": "Target Id"
           },
+          "cluster_name": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Cluster Name"
+          },
           "proxy_deployment_id": {
             "anyOf": [
               {
@@ -34767,6 +34794,17 @@
             ],
             "title": "Target Id"
           },
+          "cluster_name": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Cluster Name"
+          },
           "proxy_deployment_id": {
             "anyOf": [
               {
@@ -35224,6 +35262,17 @@
             "type": "integer",
             "title": "Cluster Id"
           },
+          "cluster_name": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Cluster Name"
+          },
           "llm_base_url": {
             "type": "string",
             "title": "Llm Base Url"
@@ -35375,6 +35424,17 @@
             "type": "integer",
             "title": "Cluster Id"
           },
+          "cluster_name": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Cluster Name"
+          },
           "llm_base_url": {
             "type": "string",
             "title": "Llm Base Url"
@@ -56610,6 +56670,17 @@
             ],
             "title": "Target Id"
           },
+          "cluster_name": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Cluster Name"
+          },
           "proxy": {
             "anyOf": [
               {

backend/requirements.txt

@@ -49,6 +49,7 @@ kr8s==0.20.15
 # HTTP
 httpx==0.28.1
 requests==2.33.0
+urllib3==2.7.0
 
 # Templating (DPU bf.conf renderer — Phase 4)
 # 3.1.6 fixes CVE-2025-27516 (sandbox bypass via |attr filter).

backend/routes/benchmarks.py

@@ -310,13 +310,22 @@ def list_benchmark_runs(
     tool: str | None = Query(None),
     model: str | None = Query(None),
     status: str | None = Query(None),
+    cluster_id: int | None = Query(None),
     limit: int = Query(50, ge=1, le=200),
     offset: int = Query(0, ge=0),
     db: Session = Depends(get_db),
 ):
     """List benchmark runs with optional filters."""
     svc = BenchmarkService(db)
-    runs, total = svc.list_runs(proxy=proxy, tool=tool, model=model, status=status, limit=limit, offset=offset)
+    runs, total = svc.list_runs(
+        proxy=proxy,
+        tool=tool,
+        model=model,
+        status=status,
+        cluster_id=cluster_id,
+        limit=limit,
+        offset=offset,
+    )
     return {"runs": runs, "total": total, "limit": limit, "offset": offset}
 
 
@@ -1107,7 +1116,7 @@ def trigger_benchmark_run(
 
     # 3. Build RunConfig — keys map directly to aiperf CLI flags
     #    See: https://github.com/ai-dynamo/aiperf/blob/main/docs/cli-options.md
-    base_url = deploy.proxy_url or target.llm_base_url
+    base_url = deploy.external_url or deploy.proxy_url or target.llm_base_url
 
     config_json: dict = {
         "url": base_url,
@@ -1272,7 +1281,7 @@ def run_benchmark_scenario(
             code="AGENT_NOT_CONNECTED",
         )
 
-    base_url = deploy.proxy_url or target.llm_base_url
+    base_url = deploy.external_url or deploy.proxy_url or target.llm_base_url
 
     # 3. Expand scenario into a run-group + child runs
     group, runs = bench_svc.create_run_group_from_scenario(
@@ -1470,6 +1479,10 @@ def _agent_ws_authorized(websocket: WebSocket, agent_id: int) -> int | None:
         # claim is mandatory, not merely honoured when present.
         token_agent_id = payload.get("agent_id")
         if token_agent_id is None:
+            # The built-in agent container connects using the bootstrap token minted
+            # before registration (carrying sub=forge-builtin-agent, role=agent, and no agent_id).
+            if payload.get("sub") == "forge-builtin-agent":
+                return None
             logger.warning(
                 "Agent %d WS rejected: token carries no agent_id claim (agent auth required)",
                 agent_id,
@@ -1569,14 +1582,46 @@ async def agent_websocket(websocket: WebSocket, agent_id: int):
     _agent_ws_connections[agent_id] = websocket
     logger.info("Agent %d connected via WebSocket", agent_id)
 
-    # Mark agent as connected
+    # Mark agent as connected and check for pending runs to dispatch
     db = next(get_db())
     try:
         svc = BenchmarkService(db)
         svc.update_agent_status(agent_id, "connected")
         db.commit()
-    except Exception:
-        pass
+
+        pending_run = svc.get_first_pending_run_for_agent(agent_id)
+        if pending_run:
+            group_id = pending_run.run_group_id
+            if group_id:
+                # Route grouped runs through the SAME gated dispatcher the terminal
+                # WS handlers use (MAJOR-2). Its group-guarded atomic claim is the
+                # single serialization point, so a connect-drain racing a
+                # run_completed/run_failed handler can never leave two children of
+                # one group RUNNING — the loser's claim fails the NOT-EXISTS guard.
+                # It claims, sends, and reverts the claim on send failure.
+                await _dispatch_next_group_child(svc, agent_id, group_id)
+                # Reflect the group as RUNNING only once a child is actually running (NIT-D atomic flip).
+                if svc.mark_run_group_running_if_pending(group_id):
+                    db.commit()
+            elif svc.claim_pending_run(pending_run.id):
+                # Standalone (group-less) run: no siblings, single-row atomic claim.
+                db.commit()
+                sent = await send_command_to_agent(
+                    agent_id, {"type": "run", "run_id": pending_run.id, "config": pending_run.config_snapshot}
+                )
+                if sent:
+                    logger.info("Agent %d connect: dispatched pending run #%d", agent_id, pending_run.id)
+                    if pending_run.run_group_id:
+                        group = svc.get_run_group(pending_run.run_group_id)
+                        if group and group.status == BenchmarkRunStatus.PENDING:
+                            group.status = BenchmarkRunStatus.RUNNING
+                            group.started_at = datetime.now(UTC)
+                            db.commit()
+                else:
+                    svc.release_claimed_run(pending_run.id)
+                    db.commit()
+    except Exception as e:
+        logger.warning("Error checking pending runs on agent connect: %s", e)
     finally:
         db.close()
 
@@ -1620,12 +1665,14 @@ async def agent_websocket(websocket: WebSocket, agent_id: int):
                     if run_id and result_data and _agent_owns_run(svc, agent_id, int(run_id)):
                         svc.complete_run_with_aiperf_result(int(run_id), result_data)
                         logger.info("Run #%d completed by agent %d — result ingested", run_id, agent_id)
+                        db.commit()
                         # Gated dispatch: now that this child is done, send the next pending
                         # child of its run-group (one aiperf at a time, strictly sequential).
                         done = svc.get_run(int(run_id))
-                        if done.run_group_id:
+                        if done and done.run_group_id:
                             await _dispatch_next_group_child(svc, agent_id, done.run_group_id)
-                    db.commit()
+                    else:
+                        db.commit()
                 except Exception as e:
                     logger.error("Error ingesting run_completed for run #%s: %s", run_id, e)
                     db.rollback()
@@ -1653,9 +1700,12 @@ async def agent_websocket(websocket: WebSocket, agent_id: int):
                             # Roll up the parent run-group when a child fails.
                             if run.run_group_id:
                                 svc.maybe_finalize_run_group(run.run_group_id)
-                                # Gated dispatch: continue the sweep with the next child.
+                            db.commit()
+                            # Gated dispatch: continue the sweep with the next child.
+                            if run.run_group_id:
                                 await _dispatch_next_group_child(svc, agent_id, run.run_group_id)
-                    db.commit()
+                    else:
+                        db.commit()
                 except Exception as e:
                     logger.error("Error handling run_failed for run #%s: %s", run_id, e)
                     db.rollback()
@@ -1704,9 +1754,17 @@ async def _dispatch_next_group_child(svc: "BenchmarkService", agent_id: int, gro
         return
     nxt_id = nxt.id
     nxt_config = nxt.config_snapshot
-    if not svc.claim_pending_run(nxt_id):
-        # Lost the race — another handler already claimed and dispatched this child.
+    # Group-guarded atomic claim (MAJOR-2 / MAJOR-A): only claim if NO sibling of this group
+    # is already RUNNING. This is the single serialization point shared with the
+    # connect-drain (which routes through here for grouped runs), so two children
+    # of one group can never both be RUNNING — even if the two paths pick different
+    # sibling rows, the second claim fails the NOT-EXISTS guard and skips.
+    if not svc.claim_pending_run(nxt_id, group_id=group_id):
+        # Lost the race — a sibling is already claimed/RUNNING, or another handler
+        # already claimed and dispatched this child.
         return
+    # MINOR-B: Persist the claim BEFORE the awaited network send, freeing the connection/row lock.
+    svc.db.commit()
     sent = await send_command_to_agent(
         agent_id, {"type": "run", "run_id": nxt_id, "config": nxt_config}
     )
@@ -1715,6 +1773,7 @@ async def _dispatch_next_group_child(svc: "BenchmarkService", agent_id: int, gro
     else:
         # Undo the claim so the child can be re-dispatched later.
         svc.release_claimed_run(nxt_id)
+        svc.db.commit()
         logger.warning("Gated dispatch: send failed for run #%d, reverted to pending", nxt_id)
 
 

backend/schemas/benchmarks.py

@@ -157,6 +157,7 @@ class BenchmarkRunResponse(BaseModel):
     config_id: int | None
     agent_id: int | None
     target_id: int | None
+    cluster_name: str | None = None
     proxy_deployment_id: int | None
     scenario_key: str | None = None
     status: str
@@ -389,6 +390,7 @@ class BenchmarkTargetResponse(BaseModel):
     name: str
     description: str | None
     cluster_id: int
+    cluster_name: str | None = None
     llm_base_url: str
     llm_model: str
     llm_namespace: str
@@ -549,6 +551,7 @@ class RunGroupSummary(BaseModel):
     run_label: str | None
     status: str
     target_id: int | None
+    cluster_name: str | None = None
     proxy: str | None
     model: str | None
     total_runs: int