[Not ready for review]

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/engine_state.h

@@ -128,14 +128,18 @@ class EngineState : public NotCopyable {
     return std::make_unique<plan::PlanState>(func_registry_.get());
   }
 
+  // max_output_rows_per_table of 0 means unlimited. That is the right default
+  // for the single-node executable and for tests, where the caller controls the
+  // input, but a server accepting queries from a peer should pass a real cap:
+  // results are buffered in memory before they are streamed out.
   std::unique_ptr<planner::CompilerState> CreateLocalExecutionCompilerState(
-      types::Time64NSValue time_now) {
+      types::Time64NSValue time_now, int64_t max_output_rows_per_table = 0) {
     auto rel_map = table_store_->GetRelationMap();
     // Use an empty string for query result address, because the local execution mode should use
     // the Local GRPC result server to send results to.
     return std::make_unique<planner::CompilerState>(
         std::move(rel_map), planner::SensitiveColumnMap{}, registry_info_.get(), time_now,
-        /* max_output_rows_per_table */ 0,
+        max_output_rows_per_table,
         /* result address */ "",
         /* ssl target name override*/ "", planner::RedactionOptions{}, nullptr, nullptr,
         planner::DebugInfo{});

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/exec/local_grpc_result_server.h

@@ -64,7 +64,12 @@ class LocalResultSinkServer final : public carnotpb::ResultSinkService::Service
     return ::grpc::Status::OK;
   }
 
-  void ResetQueryResults() { query_results_.clear(); }
+  void ResetQueryResults() {
+    // Takes the same lock TransferResultChunk does: a caller resetting between
+    // queries races an in-flight chunk write otherwise.
+    const std::lock_guard<std::mutex> lock(result_mutex_);
+    query_results_.clear();
+  }
 
  private:
   // List of the query results received.
@@ -75,12 +80,23 @@ class LocalResultSinkServer final : public carnotpb::ResultSinkService::Service
 
 class LocalGRPCResultSinkServer {
  public:
-  LocalGRPCResultSinkServer() {
-    grpc::ServerBuilder builder;
+  // Create reports a bind failure instead of aborting.
+  //
+  // The default constructor CHECK-fails, which is fine for tests and the
+  // single-node carnot executable but not for a long-lived server: a failure to
+  // bind an ephemeral loopback port would take the whole process down. Callers
+  // that have something to lose should use this.
+  static StatusOr<std::unique_ptr<LocalGRPCResultSinkServer>> Create() {
+    auto server =
+        std::unique_ptr<LocalGRPCResultSinkServer>(new LocalGRPCResultSinkServer(NoAbortTag{}));
+    if (server->grpc_server_ == nullptr) {
+      return error::ResourceUnavailable(
+          "failed to start the local result sink server on 127.0.0.1:0");
+    }
+    return server;
+  }
 
-    builder.AddListeningPort("127.0.0.1:0", grpc::InsecureServerCredentials());
-    builder.RegisterService(&result_sink_server_);
-    grpc_server_ = builder.BuildAndStart();
+  LocalGRPCResultSinkServer() : LocalGRPCResultSinkServer(NoAbortTag{}) {
     CHECK(grpc_server_ != nullptr);
   }
 
@@ -151,6 +167,17 @@ class LocalGRPCResultSinkServer {
   }
 
  private:
+  // Tag type selecting the constructor that leaves grpc_server_ null on a bind
+  // failure instead of aborting.
+  struct NoAbortTag {};
+  explicit LocalGRPCResultSinkServer(NoAbortTag) {
+    grpc::ServerBuilder builder;
+
+    builder.AddListeningPort("127.0.0.1:0", grpc::InsecureServerCredentials());
+    builder.RegisterService(&result_sink_server_);
+    grpc_server_ = builder.BuildAndStart();
+  }
+
   std::unique_ptr<grpc::Server> grpc_server_;
   LocalResultSinkServer result_sink_server_;
 };

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

@@ -14,12 +14,36 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+load("@bazel_skylib//rules:common_settings.bzl", "bool_flag")
 load("@io_bazel_rules_docker//cc:image.bzl", "cc_image")
 load("@io_bazel_rules_docker//container:container.bzl", "container_image")
 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. Defaults to True, so
+# the feature is compiled in and gated at runtime by --direct_query_enabled
+# (itself false by default). Operators who do not want the feature present in
+# the binary at all build with
+# `bazel build … --//src/vizier/services/agent/pem:direct_query=false`, which
+# puts `-DPX_PEM_DIRECT_QUERY_DISABLED` on this package's compilations only:
+# the 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.
+#
+# A build setting rather than `--define`: `--define` lives in the global
+# configuration key, so it applies to every target in the build and shows up in
+# every dependency's configuration; this flag is scoped to the targets that
+# read it.
+bool_flag(
+    name = "direct_query",
+    build_setting_default = True,
+)
+
+config_setting(
+    name = "direct_query_disabled",
+    flag_values = {":direct_query": "False"},
+)
+
 pl_cc_library(
     name = "cc_library",
     srcs = glob(
@@ -30,7 +54,30 @@ pl_cc_library(
         ],
     ),
     hdrs = glob(["*.h"]),
+    # local_defines, not defines: the macro only guards this package's sources,
+    # so there is no reason to push it onto every transitive dependent's
+    # compilation. Targets that need to compile against the disabled build
+    # (direct_query_server_test below) select on the same config_setting.
+    local_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",
+        "//src/shared/services/jwt:cc_library",
+        "//src/vizier/services/agent/shared/vizier_results:cc_library",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_github_rlyeh_sole//:sole",
+        # 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 +98,34 @@ 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"],
+    # The kill-switch assertions are `#ifdef PX_PEM_DIRECT_QUERY_DISABLED`, and
+    # cc_library's local_defines deliberately do not reach here — so select on
+    # the flag again, otherwise a `--//…:direct_query=false` run would silently
+    # compile the enabled branch and skip the coverage entirely.
+    defines = select({
+        ":direct_query_disabled": ["PX_PEM_DIRECT_QUERY_DISABLED"],
+        "//conditions:default": [],
+    }),
+    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.