feat(telemetry): include org_id in heartbeat payload (v9.1 #2277)

CHANGELOG.md

@@ -7,6 +7,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **`org_id` field in the telemetry heartbeat body (v9.1 preflight, #2277).**
+  Brings Java SDK telemetry up to parity with the platform's
+  `startup_telemetry.go` emitter — every heartbeat now identifies which
+  deployment-organization emitted it. Two sources in precedence order:
+  1. The `ORG_ID` env var when set (the operator's explicit configuration
+     on self-hosted deployments, or the `cs_<uuid>` tenant identifier on
+     Community SaaS).
+  2. Otherwise the `local-dev-org` sentinel.
+
+  Exposed as `TelemetryReporter.telemetryOrgId()` +
+  `TelemetryReporter.ORG_ID_LOCAL_DEV_SENTINEL`. The receiver already
+  accepts the field with `omitempty` for backward compat with pre-v8.1
+  SDKs that don't send it. Honors `AXONFLOW_TELEMETRY=off` like every
+  other heartbeat field. See `axonflow-landing/content/privacy.html`
+  for the customer-facing commitment that covers this field.
+
+### Changed
+
+- **Telemetry-enabled log line** softened from "anonymous telemetry
+  enabled" to "telemetry enabled" to stay coherent with the v9.1
+  `org_id` addition (the operator-supplied `ORG_ID` on self-hosted is
+  not anonymized; only the `instance_id` and `cs_<uuid>` Community
+  SaaS identifier remain anonymous-by-design). `HeartbeatState` and
+  `TelemetryReporter` JavaDoc softened similarly.
+
 ### Fixed
 
 - **README "Retry Configuration" section.** Removed three `RetryConfig.Builder`

runtime-e2e/v91_org_id_telemetry/README.md

@@ -0,0 +1,66 @@
+# Runtime proof — `org_id` in SDK telemetry payload (v9.1)
+
+Verifies the v9.1 contract for the Java SDK: every telemetry heartbeat
+body carries an `org_id` field, populated from the `ORG_ID` env var
+with a `local-dev-org` sentinel fallback. Issue #2277.
+
+## Usage
+
+Build the SDK first:
+
+```sh
+mvn package -DskipTests
+
+# ORG_ID set — operator-supplied (self-hosted) or cs_<uuid>:
+ORG_ID=acme-corp java -cp "target/axonflow-sdk-8.1.0.jar:target/dependency/*" \
+  runtime-e2e/v91_org_id_telemetry/V91OrgIdTelemetryTest.java
+
+# ORG_ID unset — local-dev-org sentinel:
+unset ORG_ID && java -cp "target/axonflow-sdk-8.1.0.jar:target/dependency/*" \
+  runtime-e2e/v91_org_id_telemetry/V91OrgIdTelemetryTest.java
+```
+
+(Depending on Maven layout, you may need `mvn dependency:copy-dependencies`
+to populate `target/dependency/`.)
+
+Expected output:
+
+```
+PASS: telemetry wire payload carries org_id="acme-corp" (expected="acme-corp")
+Wire body: {"telemetry_type":"sdk","sdk":"java", ... ,"org_id":"acme-corp"}
+```
+
+## CI coverage
+
+Functional E2E equivalent runs in CI via WireMock-based tests in
+`src/test/java/com/getaxonflow/sdk/telemetry/TelemetryReporterTest.java`:
+
+- `v9.1: telemetryOrgId returns ORG_ID env when set`
+- `v9.1: telemetryOrgId returns local-dev-org sentinel when ORG_ID unset`
+- `v9.1: telemetryOrgId treats empty ORG_ID as unset`
+- `v9.1: telemetryOrgId passes through cs_<uuid> Community SaaS tenant identifier`
+- `v9.1: buildPayload includes ORG_ID env on the wire`
+- `v9.1: buildPayload emits local-dev-org sentinel when ORG_ID unset`
+- `v9.1: buildPayload passes through cs_<uuid> on the wire`
+- `v9.1: functional E2E — ORG_ID arrives on the wire at the receiver (WireMock real HTTP)`
+- `v9.1: functional E2E — sentinel arrives on the wire when ORG_ID unset`
+
+## Mutation proof
+
+Remove the `root.put("org_id", telemetryOrgId());` line in
+`TelemetryReporter.buildPayload` and rerun. The proof exits with
+`FAIL: wire org_id = "<MISSING>", want "<expected>"` and JsonNode
+returns missing-node fallback `""`.
+
+## Cross-SDK parity
+
+Companion runtime-e2e tests live under the same subdirectory in the
+other 4 SDKs:
+
+- `axonflow-sdk-go/runtime-e2e/v91_org_id_telemetry/`
+- `axonflow-sdk-python/runtime-e2e/v91_org_id_telemetry/`
+- `axonflow-sdk-typescript/runtime-e2e/v91_org_id_telemetry/`
+- `axonflow-sdk-rust/runtime-e2e/v91_org_id_telemetry/`
+
+All five SDKs emit `org_id` with the same wire name, same sentinel
+value (`local-dev-org`), and the same precedence (env → sentinel).

runtime-e2e/v91_org_id_telemetry/V91OrgIdTelemetryTest.java

@@ -0,0 +1,89 @@
+/*
+ * runtime-e2e/v91_org_id_telemetry/V91OrgIdTelemetryTest.java
+ *
+ * Real-wire test of the SDK's v9.1 org_id telemetry field (#2277).
+ *
+ * Spins up a tiny in-process HttpServer that pretends to be the
+ * checkpoint receiver, inspects the raw JSON body for the org_id
+ * field, and exits with the verdict. Bytes flow real → real through
+ * the JDK's HttpServer + the SDK's outbound OkHttpClient.
+ *
+ * Run:
+ *   # ORG_ID set — operator-supplied or cs_<uuid>:
+ *   ORG_ID=acme-corp java -cp "<sdk-jar>:<deps>" \
+ *     runtime-e2e/v91_org_id_telemetry/V91OrgIdTelemetryTest.java
+ *
+ *   # ORG_ID unset — sentinel:
+ *   unset ORG_ID && java -cp "<sdk-jar>:<deps>" \
+ *     runtime-e2e/v91_org_id_telemetry/V91OrgIdTelemetryTest.java
+ *
+ * Sister coverage runs in CI via TelemetryReporterTest's
+ * functional-E2E test (uses WireMock — equivalent shape).
+ */
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.getaxonflow.sdk.telemetry.TelemetryReporter;
+import com.sun.net.httpserver.HttpServer;
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class V91OrgIdTelemetryTest {
+  public static void main(String[] args) throws Exception {
+    String orgIdEnv = System.getenv("ORG_ID");
+    String expected = (orgIdEnv == null || orgIdEnv.isEmpty()) ? "local-dev-org" : orgIdEnv;
+
+    AtomicReference<String> captured = new AtomicReference<>("");
+
+    HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    server.createContext("/v1/ping", exchange -> {
+      try (ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
+        exchange.getRequestBody().transferTo(bos);
+        captured.set(bos.toString(StandardCharsets.UTF_8));
+      }
+      byte[] resp = "{\"latest_version\":null,\"alerts\":[]}".getBytes(StandardCharsets.UTF_8);
+      exchange.getResponseHeaders().add("Content-Type", "application/json");
+      exchange.sendResponseHeaders(200, resp.length);
+      exchange.getResponseBody().write(resp);
+      exchange.close();
+    });
+    server.createContext("/health", exchange -> {
+      byte[] resp = "{\"version\":\"8.0.0-runtime-e2e\"}".getBytes(StandardCharsets.UTF_8);
+      exchange.getResponseHeaders().add("Content-Type", "application/json");
+      exchange.sendResponseHeaders(200, resp.length);
+      exchange.getResponseBody().write(resp);
+      exchange.close();
+    });
+    server.start();
+
+    int port = server.getAddress().getPort();
+    String checkpoint = "http://127.0.0.1:" + port + "/v1/ping";
+    String agent = "http://127.0.0.1:" + port;
+
+    System.out.println("Asserting wire org_id = " + expected);
+
+    TelemetryReporter.sendPing("production", agent, false, null, checkpoint);
+    Thread.sleep(2000);
+
+    server.stop(0);
+
+    String body = captured.get();
+    if (body.isEmpty()) {
+      System.err.println("FAIL: no telemetry body captured");
+      System.exit(1);
+    }
+
+    ObjectMapper mapper = new ObjectMapper();
+    JsonNode root = mapper.readTree(body);
+    String got = root.has("org_id") ? root.get("org_id").asText() : "<MISSING>";
+
+    if (!expected.equals(got)) {
+      System.err.println("FAIL: wire org_id = \"" + got + "\", want \"" + expected + "\"");
+      System.err.println("Body: " + body);
+      System.exit(1);
+    }
+    System.out.println("PASS: telemetry wire payload carries org_id=\"" + got + "\" (expected=\"" + expected + "\")");
+    System.out.println("Wire body: " + body);
+  }
+}

src/main/java/com/getaxonflow/sdk/telemetry/HeartbeatState.java

@@ -22,7 +22,7 @@
  * <p>Implements the cross-SDK contract:
  *
  * <pre>
- *   AxonFlow emits at most one anonymous heartbeat per environment every
+ *   AxonFlow emits at most one heartbeat per environment every
  *   7 days during SDK activity.
  * </pre>
  *

src/main/java/com/getaxonflow/sdk/telemetry/TelemetryReporter.java

@@ -65,7 +65,7 @@ public class TelemetryReporter {
   private static final MediaType JSON = MediaType.get("application/json; charset=utf-8");
 
   /**
-   * Sends an anonymous telemetry ping synchronously (blocks until the round-trip completes).
+   * Sends a telemetry ping synchronously (blocks until the round-trip completes).
    *
    * @param mode the deployment mode (e.g. "production", "sandbox")
    * @param sdkEndpoint the configured SDK endpoint, used to detect platform version via /health
@@ -113,7 +113,7 @@ static void sendPing(
   public static boolean sendPingNow(
       String mode, String sdkEndpoint, boolean debug, String checkpointUrl) {
     logger.info(
-        "AxonFlow: anonymous telemetry enabled. Opt out: AXONFLOW_TELEMETRY=off | https://docs.getaxonflow.com/docs/telemetry");
+        "AxonFlow: telemetry enabled. Opt out: AXONFLOW_TELEMETRY=off | https://docs.getaxonflow.com/docs/telemetry");
 
     String endpoint =
         (checkpointUrl != null && !checkpointUrl.isEmpty()) ? checkpointUrl : DEFAULT_ENDPOINT;
@@ -244,13 +244,40 @@ static String buildPayload(
         root.put("stream", "sandbox");
       }
 
+      // v9.1 deployment-organization identifier (#2277). Two sources, precedence order:
+      // ORG_ID env (operator-supplied on self-hosted, or cs_<uuid> on Community SaaS) or
+      // the "local-dev-org" sentinel. Always emitted. See axonflow-landing/content/privacy.html
+      // for the customer-facing commitment that covers this field.
+      root.put("org_id", telemetryOrgId());
+
       return mapper.writeValueAsString(root);
     } catch (Exception e) {
       // Fallback minimal payload
       return "{\"sdk\":\"java\",\"sdk_version\":\"" + AxonFlowConfig.SDK_VERSION + "\"}";
     }
   }
 
+  /**
+   * Sentinel emitted on the telemetry wire when {@code ORG_ID} is unset — the
+   * default-config Community-mode developer case. See #2277.
+   */
+  public static final String ORG_ID_LOCAL_DEV_SENTINEL = "local-dev-org";
+
+  /**
+   * Returns the {@code org_id} value to emit on the next telemetry ping. Reads
+   * {@code ORG_ID} from the environment (the operator's explicit configuration for
+   * self-hosted deployments, or the {@code cs_<uuid>} tenant identifier on Community
+   * SaaS) and falls back to {@link #ORG_ID_LOCAL_DEV_SENTINEL} when unset. Always
+   * returns a non-empty string. See #2277.
+   */
+  static String telemetryOrgId() {
+    String value = System.getenv("ORG_ID");
+    if (value == null || value.isEmpty()) {
+      return ORG_ID_LOCAL_DEV_SENTINEL;
+    }
+    return value;
+  }
+
   /**
    * Endpoint type classifications for telemetry. See issue #1525.
    *

src/test/java/com/getaxonflow/sdk/telemetry/TelemetryReporterTest.java

@@ -26,6 +26,8 @@
 import com.github.tomakehurst.wiremock.junit5.WireMockTest;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.junitpioneer.jupiter.ClearEnvironmentVariable;
+import org.junitpioneer.jupiter.SetEnvironmentVariable;
 
 @DisplayName("TelemetryReporter")
 @WireMockTest
@@ -429,4 +431,108 @@ void testPayloadDeploymentModeEnterprise(WireMockRuntimeInfo wmRuntimeInfo) thro
     // enterprise mode is not "sandbox", so stream is omitted
     assertThat(body.has("stream")).isFalse();
   }
+
+  // --- v9.1 org_id tests (issue #2277) ---
+
+  @Test
+  @SetEnvironmentVariable(key = "ORG_ID", value = "acme-corp")
+  @DisplayName("v9.1: telemetryOrgId returns ORG_ID env when set (operator-supplied)")
+  void testTelemetryOrgIdEnvWins() {
+    assertThat(TelemetryReporter.telemetryOrgId()).isEqualTo("acme-corp");
+  }
+
+  @Test
+  @ClearEnvironmentVariable(key = "ORG_ID")
+  @DisplayName("v9.1: telemetryOrgId returns local-dev-org sentinel when ORG_ID unset")
+  void testTelemetryOrgIdSentinelWhenUnset() {
+    assertThat(TelemetryReporter.telemetryOrgId())
+        .isEqualTo(TelemetryReporter.ORG_ID_LOCAL_DEV_SENTINEL);
+    assertThat(TelemetryReporter.ORG_ID_LOCAL_DEV_SENTINEL).isEqualTo("local-dev-org");
+  }
+
+  @Test
+  @SetEnvironmentVariable(key = "ORG_ID", value = "")
+  @DisplayName("v9.1: telemetryOrgId treats empty ORG_ID as unset")
+  void testTelemetryOrgIdEmptyFallsThrough() {
+    assertThat(TelemetryReporter.telemetryOrgId())
+        .isEqualTo(TelemetryReporter.ORG_ID_LOCAL_DEV_SENTINEL);
+  }
+
+  @Test
+  @SetEnvironmentVariable(key = "ORG_ID", value = "cs_e3a4b5c6-d7e8-4f90-a1b2-c3d4e5f6a7b8")
+  @DisplayName("v9.1: telemetryOrgId passes through cs_<uuid> Community SaaS tenant identifier")
+  void testTelemetryOrgIdCsPrefixedPassesThrough() {
+    assertThat(TelemetryReporter.telemetryOrgId())
+        .isEqualTo("cs_e3a4b5c6-d7e8-4f90-a1b2-c3d4e5f6a7b8");
+  }
+
+  @Test
+  @SetEnvironmentVariable(key = "ORG_ID", value = "acme-corp")
+  @DisplayName("v9.1: buildPayload includes ORG_ID env on the wire")
+  void testPayloadIncludesOrgIdFromEnv() throws Exception {
+    String payload = TelemetryReporter.buildPayload("production", null);
+    JsonNode root = objectMapper.readTree(payload);
+    assertThat(root.get("org_id").asText()).isEqualTo("acme-corp");
+    // Wire-literal substring assertion defends against tag-removal mutations.
+    assertThat(payload).contains("\"org_id\":\"acme-corp\"");
+  }
+
+  @Test
+  @ClearEnvironmentVariable(key = "ORG_ID")
+  @DisplayName("v9.1: buildPayload emits local-dev-org sentinel when ORG_ID unset")
+  void testPayloadIncludesSentinelWhenUnset() throws Exception {
+    String payload = TelemetryReporter.buildPayload("production", null);
+    JsonNode root = objectMapper.readTree(payload);
+    assertThat(root.get("org_id").asText()).isEqualTo("local-dev-org");
+    assertThat(payload).contains("\"org_id\":\"local-dev-org\"");
+  }
+
+  @Test
+  @SetEnvironmentVariable(key = "ORG_ID", value = "cs_f29e9c5c-5c5b-4e0d-8e0d-aabbccddeeff")
+  @DisplayName("v9.1: buildPayload passes through cs_<uuid> on the wire")
+  void testPayloadIncludesCsPrefixedTenant() throws Exception {
+    String payload = TelemetryReporter.buildPayload("production", null);
+    JsonNode root = objectMapper.readTree(payload);
+    assertThat(root.get("org_id").asText())
+        .isEqualTo("cs_f29e9c5c-5c5b-4e0d-8e0d-aabbccddeeff");
+    assertThat(payload).contains("\"org_id\":\"cs_f29e9c5c-5c5b-4e0d-8e0d-aabbccddeeff\"");
+  }
+
+  @Test
+  @SetEnvironmentVariable(key = "ORG_ID", value = "acme-corp")
+  @DisplayName(
+      "v9.1: functional E2E — ORG_ID arrives on the wire at the receiver (WireMock real HTTP)")
+  void testOrgIdReachesReceiverOverHttp(WireMockRuntimeInfo wmRuntimeInfo) throws Exception {
+    stubFor(post("/v1/ping").willReturn(ok()));
+    String customUrl = wmRuntimeInfo.getHttpBaseUrl() + "/v1/ping";
+
+    TelemetryReporter.sendPing(
+        "production", "http://localhost:8080", false, null, customUrl);
+    Thread.sleep(2000);
+
+    var requests = WireMock.findAll(postRequestedFor(urlEqualTo("/v1/ping")));
+    assertThat(requests).hasSize(1);
+    String body = requests.get(0).getBodyAsString();
+    JsonNode root = objectMapper.readTree(body);
+    assertThat(root.get("org_id").asText()).isEqualTo("acme-corp");
+    assertThat(body).contains("\"org_id\":\"acme-corp\"");
+  }
+
+  @Test
+  @ClearEnvironmentVariable(key = "ORG_ID")
+  @DisplayName("v9.1: functional E2E — sentinel arrives on the wire when ORG_ID unset")
+  void testOrgIdSentinelReachesReceiver(WireMockRuntimeInfo wmRuntimeInfo) throws Exception {
+    stubFor(post("/v1/ping").willReturn(ok()));
+    String customUrl = wmRuntimeInfo.getHttpBaseUrl() + "/v1/ping";
+
+    TelemetryReporter.sendPing(
+        "production", "http://localhost:8080", false, null, customUrl);
+    Thread.sleep(2000);
+
+    var requests = WireMock.findAll(postRequestedFor(urlEqualTo("/v1/ping")));
+    assertThat(requests).hasSize(1);
+    String body = requests.get(0).getBodyAsString();
+    JsonNode root = objectMapper.readTree(body);
+    assertThat(root.get("org_id").asText()).isEqualTo("local-dev-org");
+  }
 }