@@ -55,4 +55,5 @@ terraform.tfvars
 *.html
 
 examples/**/node_modules/
-examples/**/package-lock.json
+examples/**/package-lock.json
+examples/public-mesh-openai-demo/certs/
@@ -5,3 +5,18 @@ NATS_URL=nats.meshprotocol.dev:4222
 NATS_TOKEN=your_token_here
 REGISTRY_URL=https://api.meshprotocol.dev
 OPENAI_API_KEY=sk-your-openai-key
+SESSION_TOKEN_SECRET=replace-with-strong-shared-secret
+DATAPLANE_BIND=0.0.0.0:50051
+DATAPLANE_PUBLIC_ENDPOINT=grpc://127.0.0.1:50051
+PROVIDER_ED25519_PRIVATE_KEY_BASE64=base64_pkcs8_private_key_for_provider
+CONSUMER_ED25519_PRIVATE_KEY_BASE64=base64_pkcs8_private_key_for_consumer
+ALLOW_LEGACY_DATAPLANE=0
+DATAPLANE_ALLOW_INSECURE=0
+DATAPLANE_TLS_CA_CERT_PATH=../certs/dev/ca.crt
+DATAPLANE_TLS_SERVER_CERT_PATH=../certs/dev/server.crt
+DATAPLANE_TLS_SERVER_KEY_PATH=../certs/dev/server.key
+DATAPLANE_TLS_REQUIRE_CLIENT_CERT=1
+DATAPLANE_TLS_CLIENT_CERT_PATH=../certs/dev/client.crt
+DATAPLANE_TLS_CLIENT_KEY_PATH=../certs/dev/client.key
+DATAPLANE_TLS_SERVER_NAME=localhost
+METRICS_PORT=9095
@@ -0,0 +1,88 @@
+# Security Rollout Guide (Data Plane)
+
+## Current Mode
+
+- Payload path migrated from `mesh.tasks.*` / `mesh.results.*` to DataPlane gRPC.
+- Handshake now enforces:
+  - signed `session_token` (HS256),
+  - Ed25519 consumer/provider signatures,
+  - ephemeral X25519 key exchange.
+- DID Document now also publishes `key_agreement` (X25519) for provider/consumer.
+- Transfer/Result now use `AES-256-GCM` with per-message nonce and `sequence`.
+- Provider now exposes Prometheus metrics on `:METRICS_PORT/metrics`.
+- gRPC now supports TLS and optional mTLS (client cert validation).
+
+## Rollout Strategy
+
+1. Deploy provider with `dataplane_capability=v1-e2e` annotation (automatic in current provider script).
+2. Enable consumer strict mode (`ALLOW_LEGACY_DATAPLANE=0`) only after provider(s) are updated.
+3. Monitor provider metrics logs:
+   - `handshakeFailed`
+   - `decryptFailed`
+   - `replayDetected`
+4. If handshake failures spike, temporarily set `ALLOW_LEGACY_DATAPLANE=1` on consumers during incident triage.
+5. After stabilization window, remove any legacy fallback and block providers that do not advertise `v1-e2e`.
+
+## Local TLS/mTLS Quickstart
+
+1. Generate dev certs:
+   - `npm run certs:dev`
+2. Generate Ed25519 keys:
+   - `npm run keys:provider`
+   - `npm run keys:consumer`
+3. Copy `.env.example` to `.env` and fill:
+   - `NATS_TOKEN`, `OPENAI_API_KEY`, `SESSION_TOKEN_SECRET`
+   - `PROVIDER_ED25519_PRIVATE_KEY_BASE64`, `CONSUMER_ED25519_PRIVATE_KEY_BASE64`
+4. Start provider:
+   - `npm run run:provider`
+5. Start consumer:
+   - `npm run run:consumer -- "What is 17 * 23?"`
+6. Verify metrics:
+   - `curl -s http://localhost:9095/metrics | rg "mesh_provider_dataplane_"`
+
+## Operational Hardening
+
+- Keep `SESSION_TOKEN_SECRET` rotated and out of source control.
+- Keep Ed25519 private keys outside repository and inject only via environment.
+- Enforce TLS transport for gRPC endpoints in non-local environments:
+  - Set `DATAPLANE_ALLOW_INSECURE=0`
+  - Set `DATAPLANE_TLS_SERVER_CERT_PATH` + `DATAPLANE_TLS_SERVER_KEY_PATH`
+  - Set `DATAPLANE_TLS_CA_CERT_PATH`
+- Enable mTLS when both sides are ready:
+  - Provider: `DATAPLANE_TLS_REQUIRE_CLIENT_CERT=1`
+  - Consumer: `DATAPLANE_TLS_CLIENT_CERT_PATH` + `DATAPLANE_TLS_CLIENT_KEY_PATH`
+- Use `DATAPLANE_TLS_SERVER_NAME` on consumer when certificate CN/SAN does not match endpoint host.
+- Alert on any `replayDetected` event (treat as suspicious traffic).
+- Reject providers without `did_document` in registry metadata.
+
+## Key Generation
+
+- Generate consumer keypair:
+  - `npm run keys:consumer`
+- Generate provider keypair:
+  - `npm run keys:provider`
+- Copy `<ROLE>_ED25519_PRIVATE_KEY_BASE64` into `.env`.
+- Use `<ROLE>_ED25519_PUBLIC_KEY_BASE64` only for diagnostics/audits (public data).
+
+## Prometheus Metrics
+
+Main metrics:
+- `mesh_provider_dataplane_handshake_total{outcome,reason}`
+- `mesh_provider_dataplane_transfer_total{outcome,reason}`
+- `mesh_provider_dataplane_decrypt_failures_total{reason}`
+- `mesh_provider_dataplane_replay_total{reason}`
+- `mesh_provider_dataplane_phase_duration_seconds{phase}`
+
+## Deploy Checklist (Production)
+
+- `session_token` signing secret rotated and sourced from secret manager.
+- gRPC with `DATAPLANE_ALLOW_INSECURE=0`.
+- Valid server cert/key configured and signed by trusted CA.
+- mTLS enabled (`DATAPLANE_TLS_REQUIRE_CLIENT_CERT=1`) where client identity is required.
+- `DATAPLANE_TLS_SERVER_NAME` validated in consumer environment.
+- Ed25519 private keys injected by runtime secret manager (never in repo).
+- Alerts configured for:
+  - handshake failures spike,
+  - decrypt failures > baseline,
+  - replay detections > 0.
+- Metrics endpoint restricted to trusted network or protected scrape path.
@@ -4,11 +4,27 @@
  * Loads .env from parent (NATS_TOKEN, REGISTRY_URL, OPENAI_API_KEY).
  */
 
-import { connect } from "nats";
 import { Agent, InMemory, Tool } from "@nebulaos/core";
 import { OpenAI } from "@nebulaos/openai";
 import { MeshClient } from "@meshprotocol/sdk";
 import { z } from "zod";
+import { createDataPlaneClient, parseGrpcEndpoint } from "../shared/dataplane.js";
+import {
+  buildHandshakePayload,
+  createX25519Ephemeral,
+  decryptChunk,
+  deriveSessionKey,
+  encryptChunk,
+  exportEd25519PublicKeyBase64,
+  exportX25519PublicKeyBase64,
+  exportX25519PublicKeyBytes,
+  importX25519PublicKeyBase64,
+  loadEd25519PrivateKeyFromBase64,
+  publicKeyFromBase64,
+  signEd25519,
+  verifyEd25519,
+} from "../shared/security.js";
+import crypto from "node:crypto";
 
 const DEFAULT_HOST = "nats.meshprotocol.dev";
 const DEFAULT_PORT = "4222";
@@ -39,10 +55,15 @@ const NATS_SERVER = getNatsServerUrl();
 const NATS_TOKEN = (process.env.NATS_TOKEN || "").trim();
 const REGISTRY_URL = process.env.REGISTRY_URL || "https://api.meshprotocol.dev";
 const CONSUMER_DID = "did:mesh:agent:public-demo-consumer";
-const TASK_PREFIX = "mesh.tasks.";
-const RESULT_PREFIX = "mesh.results.";
 const REQUEST_TIMEOUT_MS = 25_000;
 const RESULT_TIMEOUT_MS = 15_000;
+const CONSUMER_ED25519_PRIVATE_KEY_BASE64 = (process.env.CONSUMER_ED25519_PRIVATE_KEY_BASE64 || "").trim();
+const ALLOW_LEGACY_DATAPLANE = (process.env.ALLOW_LEGACY_DATAPLANE || "").trim() === "1";
+const DATAPLANE_ALLOW_INSECURE = (process.env.DATAPLANE_ALLOW_INSECURE || "").trim() === "1";
+const DATAPLANE_TLS_CA_CERT_PATH = (process.env.DATAPLANE_TLS_CA_CERT_PATH || "").trim();
+const DATAPLANE_TLS_CLIENT_CERT_PATH = (process.env.DATAPLANE_TLS_CLIENT_CERT_PATH || "").trim();
+const DATAPLANE_TLS_CLIENT_KEY_PATH = (process.env.DATAPLANE_TLS_CLIENT_KEY_PATH || "").trim();
+const DATAPLANE_TLS_SERVER_NAME = (process.env.DATAPLANE_TLS_SERVER_NAME || "").trim();
 
 if (!process.env.OPENAI_API_KEY) {
   console.error("[Consumer] OPENAI_API_KEY is required (set in .env).");
@@ -52,14 +73,19 @@ if (!process.env.NATS_TOKEN) {
   console.error("[Consumer] NATS_TOKEN is required for public mesh (set in .env).");
   process.exit(1);
 }
+if (!CONSUMER_ED25519_PRIVATE_KEY_BASE64) {
+  console.error("[Consumer] CONSUMER_ED25519_PRIVATE_KEY_BASE64 is required.");
+  process.exit(1);
+}
 
 async function main() {
   console.log("[Consumer] Connecting to PUBLIC mesh (meshprotocol.dev)...\n");
 
-  const natsForTasks = await connect({
-    servers: NATS_SERVER,
-    token: NATS_TOKEN || undefined,
-  });
+  const consumerPrivateKey = loadEd25519PrivateKeyFromBase64(CONSUMER_ED25519_PRIVATE_KEY_BASE64);
+  const consumerPublicKey = crypto.createPublicKey(consumerPrivateKey);
+  const consumerPublicKeyBase64 = exportEd25519PublicKeyBase64(consumerPublicKey);
+  const consumerStaticKeyAgreement = createX25519Ephemeral();
+  const consumerKeyAgreementPublicBase64 = exportX25519PublicKeyBase64(consumerStaticKeyAgreement.publicKey);
 
   const mesh = new MeshClient({
     natsUrl: NATS_SERVER,
@@ -69,6 +95,43 @@ async function main() {
     natsAuth: NATS_TOKEN ? { token: NATS_TOKEN } : undefined,
   });
 
+  const consumerCard = {
+    metadata: {
+      id: CONSUMER_DID,
+      name: "PublicDemoConsumer",
+      version: "1.0.0",
+      owner: "did:mesh:org:demo",
+      did_document: {
+        id: CONSUMER_DID,
+        verification_method: [
+          {
+            id: `${CONSUMER_DID}#key-1`,
+            type: "Ed25519VerificationKey2020",
+            controller: CONSUMER_DID,
+            public_key_base64: consumerPublicKeyBase64,
+          },
+        ],
+        key_agreement: [
+          {
+            id: `${CONSUMER_DID}#key-agreement-1`,
+            type: "X25519KeyAgreementKey2020",
+            controller: CONSUMER_DID,
+            public_key_base64: consumerKeyAgreementPublicBase64,
+          },
+        ],
+      },
+    },
+    spec: {
+      domains: { primary: ["demo", "consumer"] },
+      capabilities: [{ id: "requester", description: "Consumer identity registration only" }],
+      endpoints: {
+        control_plane: { nats_subject: "mesh.agents.demo.consumer" },
+        data_plane: { grpc: "grpc://127.0.0.1:0" },
+      },
+    },
+  };
+  await mesh.register(consumerCard, { status: "active" });
+
   const requestMathFromMeshTool = new Tool({
     id: "request_math_from_mesh",
     description: "Sends a math/calculation question to a specialist on the mesh. Use this when the user asks for any math operation.",
@@ -88,41 +151,131 @@ async function main() {
       }
 
       const sessionId = result.sessionId;
-      const taskSubject = TASK_PREFIX + sessionId;
-      const resultSubject = RESULT_PREFIX + sessionId;
-
-      await new Promise((r) => setTimeout(r, 300));
+      const providerDid = result.parties.provider;
+      const providerInfoResp = await fetch(`${REGISTRY_URL.replace(/\/$/, "")}/agents/${encodeURIComponent(providerDid)}`);
+      if (!providerInfoResp.ok) {
+        throw new Error(`Provider registry lookup failed: ${providerInfoResp.status}`);
+      }
+      const providerInfo = await providerInfoResp.json();
+      const capability = providerInfo?.card?.metadata?.annotations?.dataplane_capability;
+      if (capability !== "v1-e2e" && !ALLOW_LEGACY_DATAPLANE) {
+        throw new Error("Provider does not advertise dataplane_capability=v1-e2e");
+      }
+      const providerGrpcEndpoint = parseGrpcEndpoint(providerInfo?.card?.spec?.endpoints?.data_plane?.grpc || "");
+      const providerPubKeyB64 = providerInfo?.card?.metadata?.did_document?.verification_method?.[0]?.public_key_base64;
+      const providerAgreementKeyB64 = providerInfo?.card?.metadata?.did_document?.key_agreement?.[0]?.public_key_base64;
+      if (!providerPubKeyB64) {
+        throw new Error("Provider DID document missing public key");
+      }
+      if (!providerAgreementKeyB64) {
+        throw new Error("Provider DID document missing keyAgreement X25519 key");
+      }
+      const providerPublicKey = publicKeyFromBase64(providerPubKeyB64);
+      const dpClient = createDataPlaneClient(providerGrpcEndpoint, {
+        insecure: DATAPLANE_ALLOW_INSECURE,
+        caCertPath: DATAPLANE_TLS_CA_CERT_PATH,
+        clientCertPath: DATAPLANE_TLS_CLIENT_CERT_PATH,
+        clientKeyPath: DATAPLANE_TLS_CLIENT_KEY_PATH,
+        serverName: DATAPLANE_TLS_SERVER_NAME,
+      });
+      const consumerEph = createX25519Ephemeral();
+      const consumerEphPub = exportX25519PublicKeyBytes(consumerEph.publicKey);
+      // Ensures provider DID exposes a valid X25519 key for agreement.
+      const providerAgreementKey = importX25519PublicKeyBase64(providerAgreementKeyB64);
+      const didSharedSecret = crypto.diffieHellman({
+        privateKey: consumerEph.privateKey,
+        publicKey: providerAgreementKey,
+      });
+      if (!didSharedSecret || didSharedSecret.length === 0) {
+        throw new Error("Unable to derive shared secret using provider DID key agreement");
+      }
+      const consumerSigPayload = buildHandshakePayload(sessionId, CONSUMER_DID, consumerEphPub);
+      const consumerSig = signEd25519(consumerPrivateKey, consumerSigPayload);
 
-      natsForTasks.publish(
-        taskSubject,
-        new TextEncoder().encode(JSON.stringify({ description: question }))
+      const handshakeResp = await new Promise((resolve, reject) => {
+        dpClient.Handshake({
+          session_id: sessionId,
+          session_token: result.sessionToken,
+          consumer_ephemeral_pub: consumerEphPub,
+          consumer_did: CONSUMER_DID,
+          consumer_did_signature: consumerSig,
+        }, (err, response) => (err ? reject(err) : resolve(response)));
+      });
+      const providerSigPayload = buildHandshakePayload(
+        sessionId,
+        handshakeResp.provider_did,
+        handshakeResp.provider_ephemeral_pub
       );
+      const providerSigValid = verifyEd25519(
+        providerPublicKey,
+        providerSigPayload,
+        Buffer.from(handshakeResp.provider_did_signature)
+      );
+      if (!providerSigValid) {
+        throw new Error("Provider handshake signature invalid");
+      }
+      const sessionKey = deriveSessionKey({
+        privateKey: consumerEph.privateKey,
+        peerPublicKeyBytes: handshakeResp.provider_ephemeral_pub,
+        sessionId,
+      });
 
-      const answer = await new Promise((resolve, reject) => {
-        const timeout = setTimeout(() => {
-          sub.unsubscribe();
-          reject(new Error("Timeout waiting for specialist response"));
-        }, RESULT_TIMEOUT_MS);
-
-        const sub = natsForTasks.subscribe(resultSubject, {
-          max: 1,
-          callback: (err, msg) => {
-            clearTimeout(timeout);
-            sub.unsubscribe();
+      await new Promise((resolve, reject) => {
+        const call = dpClient.Transfer(
+          { "x-session-id": sessionId },
+          (err, ack) => {
             if (err) {
               reject(err);
               return;
             }
-            try {
-              const data = JSON.parse(new TextDecoder().decode(msg.data));
-              resolve(data.result ?? data.error ?? "(no response)");
-            } catch {
-              resolve("(invalid response)");
+            if (!ack?.accepted) {
+              reject(new Error(ack?.error_message || "provider rejected transfer"));
+              return;
             }
-          },
+            resolve();
+          }
+        );
+        const encrypted = encryptChunk({
+          key: sessionKey,
+          sequence: 1,
+          payloadBuffer: Buffer.from(JSON.stringify({ description: question }), "utf8"),
         });
+        call.write({
+          ciphertext: encrypted.ciphertext,
+          nonce: encrypted.nonce,
+          sequence: encrypted.sequence,
+          is_final: encrypted.is_final,
+          algorithm: encrypted.algorithm,
+        });
+        call.end();
       });
 
+      const answer = await new Promise((resolve, reject) => {
+        const timeout = setTimeout(() => {
+          reject(new Error("Timeout waiting for specialist response"));
+        }, RESULT_TIMEOUT_MS);
+        const stream = dpClient.Result({ session_id: sessionId });
+        stream.on("data", (chunk) => {
+          clearTimeout(timeout);
+          try {
+            const plaintext = decryptChunk({ key: sessionKey, chunk });
+            const data = JSON.parse(plaintext.toString("utf8"));
+            resolve(data.result ?? data.error ?? "(no response)");
+          } catch {
+            resolve("(invalid response)");
+          }
+        });
+        stream.on("error", (err) => {
+          clearTimeout(timeout);
+          reject(err);
+        });
+        stream.on("end", () => {
+          clearTimeout(timeout);
+        });
+      });
+
+      dpClient.close?.();
+
       return { success: true, result: answer };
     },
   });
@@ -151,7 +304,6 @@ async function main() {
 
   console.log("[Consumer] Response:", exec.content ?? "(empty)");
 
-  await natsForTasks.close();
   await mesh.close();
   console.log("\n[Consumer] Done.");
 }