feat(pem): add authenticated direct-query gRPC endpoint

src/carnot/BUILD.bazel

@@ -44,6 +44,11 @@ pl_cc_library(
             "carnot_executable.cc",
         ],
     ),
+    visibility = [
+        "//src/carnot:__subpackages__",
+        "//src/experimental:__subpackages__",
+        "//src/vizier/services/agent:__subpackages__",
+    ],
     deps = [
         "//src/carnot/exec:cc_library",
         "//src/carnot/exec/ml:cc_library",

src/carnot/exec/BUILD.bazel

@@ -16,7 +16,10 @@
 
 load("//bazel:pl_build_system.bzl", "pl_cc_binary", "pl_cc_library", "pl_cc_test", "pl_cc_test_library")
 
-package(default_visibility = ["//src/carnot:__subpackages__"])
+package(default_visibility = [
+    "//src/carnot:__subpackages__",
+    "//src/vizier/services/agent:__subpackages__",
+])
 
 pl_cc_library(
     name = "cc_library",

src/carnot/udf/BUILD.bazel

@@ -19,6 +19,7 @@ load("//bazel:pl_build_system.bzl", "pl_cc_binary", "pl_cc_library", "pl_cc_test
 package(default_visibility = [
     "//src/carnot:__subpackages__",
     "//src/vizier/funcs:__subpackages__",
+    "//src/vizier/services/agent:__subpackages__",
 ])
 
 pl_cc_library(

src/vizier/services/agent/pem/BUILD.bazel

@@ -20,6 +20,18 @@ load("//bazel:pl_build_system.bzl", "pl_cc_binary", "pl_cc_library", "pl_cc_test
 
 package(default_visibility = ["//src/vizier:__subpackages__"])
 
+# Compile-time kill switch for the direct-query endpoint.
+# Operators who do not want the feature available even as a disabled-by-default
+# option build with `bazel build … --//src/vizier/services/agent/pem:direct_query=disabled`
+# (or `--define=PX_PEM_DIRECT_QUERY=disabled`). This propagates
+# `-DPX_PEM_DIRECT_QUERY_DISABLED` into cc_library; the entire feature body
+# in direct_query_server.cc and pem_manager.cc is `#ifndef`'d out and
+# stub methods return UNIMPLEMENTED. See DIRECT_QUERY_SECURITY.md.
+config_setting(
+    name = "direct_query_disabled",
+    define_values = {"PX_PEM_DIRECT_QUERY": "disabled"},
+)
+
 pl_cc_library(
     name = "cc_library",
     srcs = glob(
@@ -30,7 +42,27 @@ pl_cc_library(
         ],
     ),
     hdrs = glob(["*.h"]),
+    defines = select({
+        ":direct_query_disabled": ["PX_PEM_DIRECT_QUERY_DISABLED"],
+        "//conditions:default": [],
+    }),
     deps = [
+        # direct-query server deps (direct_query_server.{h,cc}):
+        "//src/api/proto/vizierpb:vizier_pl_cc_proto",
+        "//src/carnot",
+        "//src/carnot:cc_library",
+        "//src/carnot/carnotpb:carnot_pl_cc_proto",
+        "//src/carnot/exec:cc_library",
+        "//src/carnot/funcs:cc_library",
+        "//src/carnot/planner/compiler:cc_library",
+        "//src/carnot/planpb:plan_pl_cc_proto",
+        "//src/carnot/udf:cc_library",
+        # HS256 verify uses BoringSSL HMAC directly and parses claims via rapidjson.
+        "@boringssl//:crypto",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_github_rlyeh_sole//:sole",
+        "@com_github_tencent_rapidjson//:rapidjson",
+        # existing PEM deps:
         "//src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_cc_proto",
         "//src/integrations/grpc_clocksync:cc_library",
         "//src/shared/tracepoint_translation:cc_library",
@@ -51,6 +83,26 @@ pl_cc_test(
     ],
 )
 
+# TDD contract for the PEM direct-query endpoint.
+# See DIRECT_QUERY_CONTRACT.md for the behavioral spec.
+pl_cc_test(
+    name = "direct_query_server_test",
+    srcs = ["direct_query_server_test.cc"],
+    deps = [
+        ":cc_library",
+        "//src/api/proto/vizierpb:vizier_pl_cc_proto",
+        "//src/carnot",
+        "//src/carnot:cc_library",
+        "//src/carnot/exec:cc_library",
+        "//src/carnot/funcs:cc_library",
+        "//src/carnot/udf:cc_library",
+        "//src/table_store:cc_library",
+        "@com_github_arun11299_cpp_jwt//:cpp_jwt",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_github_rlyeh_sole//:sole",
+    ],
+)
+
 pl_cc_binary(
     name = "pem",
     srcs = ["pem_main.cc"],

src/vizier/services/agent/pem/DIRECT_QUERY_CONTRACT.md

@@ -0,0 +1,89 @@
+# PEM direct-query gRPC endpoint — contract
+
+## Why
+
+Today in-cluster services read PEM data by querying the **vizier-query-broker**
+(the standard `ExecuteScript` path). That works and is the primary path. This
+feature adds the *node-local* alternative: make the normal `vizier-pem` itself
+serve `ExecuteScript` directly over gRPC, so an on-node client can query its
+node-local PEM with no broker hop and no cloud dependency.
+
+This capability was already proved in the experimental `standalone_pem`
+(`src/experimental/standalone_pem/vizier_server.h` — `px::vizier::agent::VizierServer`
+implementing `api::vizierpb::VizierService::ExecuteScript` against a local Carnot).
+The two differences for the real PEM:
+
+1. **Metadata-connected.** The normal PEM has the metadata service, so per-pod PxL
+   filters (`df[df.ctx['pod'] == ...]`) resolve — the gap that made standalone_pem
+   return empty per-pod results. Reuse the PEM's existing Carnot +
+   table_store + metadata state; do **not** stand up a second Carnot.
+2. **Authenticated.** standalone_pem was insecure (`WithDirectCredsInsecure`). The
+   real PEM is in `pl` and must require a **valid cluster service JWT** (the same
+   `jwt-signing-key` used by kelvin, query-broker, and metadata-server).
+
+## The endpoint
+
+- Service: `px.api.vizierpb.VizierService` (the generated gRPC service).
+- Method implemented: **`ExecuteScript`** (server-streaming). Mutations/tracepoints
+  are **out of scope** — return `UNIMPLEMENTED` for `req.mutation()==true`.
+  (standalone_pem handles mutations; the read path never mutates.)
+- Transport: gRPC over TLS using the in-cluster self-signed CA
+  (`SSL::DefaultGRPCServerCreds()`). Insecure fallback only when `PL_DISABLE_SSL=1`
+  is explicitly set for a dev/soak cluster.
+
+## Config (flags / env) — gated OFF by default
+
+| flag / env                          | default | meaning                                            |
+|-------------------------------------|---------|----------------------------------------------------|
+| `--direct_query_enabled` / `PL_PEM_DIRECT_QUERY_ENABLED` | `false` | master switch; when false the port is never opened |
+| `--direct_query_port` / `PL_PEM_DIRECT_QUERY_PORT`       | `50305` | gRPC listen port for the direct-query service      |
+| `--direct_query_jwt_signing_key` / `PL_JWT_SIGNING_KEY`  | `""`    | HMAC key the bearer JWT must verify against         |
+
+Default-off so existing PEM deployments are byte-for-byte unchanged until opted in.
+Opt out at runtime with `PL_PEM_DIRECT_QUERY_ENABLED=false`, or compile it out with
+`--//src/vizier/services/agent/pem:direct_query=disabled`.
+
+## Auth contract
+
+Every `ExecuteScript` call MUST present `authorization: Bearer <jwt>` metadata.
+The JWT is verified with `PL_JWT_SIGNING_KEY` and must:
+- have a valid signature (HS256) against the signing key,
+- be unexpired (`exp` in the future),
+- carry a service/audience claim acceptable to vizier
+  (minted via `GenerateJWTForService` in `src/shared/services/utils`).
+
+Missing/invalid/expired token → `grpc::StatusCode::UNAUTHENTICATED`. No token must
+ever fall through to query execution.
+
+## Behavioral contract (the executable spec → `direct_query_server_test.cc`)
+
+1. **flag-off → no listener.** With `direct_query_enabled=false`, nothing listens on
+   the port; the PEM starts exactly as today.
+2. **flag-on → serves ExecuteScript.** With it enabled + a signing key set, a gRPC
+   client with a valid bearer JWT gets a streamed response (status OK) for a trivial
+   PxL (e.g. `import px; px.display(px.DataFrame('http_events'))`).
+3. **auth required.** Same call with (a) no token, (b) a token signed by the wrong
+   key, (c) an expired token → each `UNAUTHENTICATED`, no rows.
+4. **metadata-connected filter.** A PxL with a per-pod filter returns only that pod's
+   rows (proves the metadata gap from standalone_pem is closed on the real PEM). May
+   be an integration test tagged `requires_metadata` if a unit Carnot fixture can't
+   supply pod context.
+5. **mutation rejected.** `req.mutation()==true` → `UNIMPLEMENTED` (scope guard).
+6. **no regression.** The existing PEM agent registration / Carnot / Stirling path is
+   unchanged when the flag is off (assert via the existing PEM smoke/unit tests).
+
+## Client integration (informational)
+
+The existing in-cluster clients already have the pieces:
+- **JWT mint**: `GenerateJWTForService` in `src/shared/services/utils/jwt.go`; mount
+  the `pl-cluster-secrets/jwt-signing-key` via `secretKeyRef`.
+- **gRPC metadata**: attach `authorization: Bearer <jwt>` to each call.
+- **Address**: `<HOST_IP>:50305` (HOST_IP via downward API, or the pod's node IP).
+- **TLS**: use `WithDisableTLSVerification` against the cluster's self-signed CA,
+  matching the broker path.
+
+## Done =
+
+`direct_query_server_test.cc` green under `bazel test`, PEM image builds via the
+vizier-release workflow, and a live cluster shows an on-node client ruling in an
+e2e scenario off the node-local PEM with the same verdict it gets via the broker.