feat(provider-contract): define materializable L2 graph blueprints

packages/external-provider-contract/test/l2-execution-plan.test.mjs

@@ -1,9 +1,13 @@
 import assert from "node:assert/strict";
+import { createHash } from "node:crypto";
 import {
   L2_EXECUTION_PLAN_SCHEMA_VERSION,
+  L2_GRAPH_BLUEPRINT_SCHEMA_VERSION,
+  L2_MAPPING_RUNTIME_SCHEMA_VERSION,
   L2_MATERIALIZATION_RECEIPT_SCHEMA_VERSION,
   attestL2ExecutionPlanMaterialization,
   compileL2ExecutionPlan,
+  compileL2GraphBlueprint,
   instantiateL2Connection,
   validateL2ExecutionPlan,
 } from "../src/index.mjs";
@@ -30,6 +34,58 @@ assert.match(positionsPlan.executionPlanDigest, /^sha256:[a-f0-9]{64}$/);
 assert.match(positionsPlan.artifacts.packageDigest, /^sha256:[a-f0-9]{64}$/);
 assert.match(positionsPlan.artifacts.mappingContractDigest, /^sha256:[a-f0-9]{64}$/);
 assert.match(positionsPlan.artifacts.telemetryRegistryDigest, /^sha256:[a-f0-9]{64}$/);
+assert.equal(positionsPlan.graphBlueprint.schemaVersion, L2_GRAPH_BLUEPRINT_SCHEMA_VERSION);
+assert.equal(
+  positionsPlan.graphBlueprint.runtime.mappingRuntime.schemaVersion,
+  L2_MAPPING_RUNTIME_SCHEMA_VERSION,
+);
+assert.match(positionsPlan.graphBlueprint.blueprintDigest, /^sha256:[a-f0-9]{64}$/);
+assert.equal(compileL2GraphBlueprint(positionsPlan), positionsPlan.graphBlueprint);
+assert.deepEqual(
+  positionsPlan.graphBlueprint.nodes.map(({ id, stepId, runtimeKind, order }) => ({
+    id,
+    stepId,
+    runtimeKind,
+    order,
+  })),
+  [
+    { id: "collection.trigger.manual", stepId: "collection.trigger", runtimeKind: "ndc.manual-trigger", order: 0 },
+    { id: "collection.trigger.interval", stepId: "collection.trigger", runtimeKind: "ndc.interval-trigger", order: 1 },
+    { id: "provider.fetch-monitoring-config", stepId: "provider.fetch-monitoring-config", runtimeKind: "ndc.provider-request", order: 2 },
+    { id: "provider.fetch-units", stepId: "provider.fetch-units", runtimeKind: "ndc.provider-request", order: 3 },
+    { id: "provider.extract-units", stepId: "provider.extract-units", runtimeKind: "ndc.extract-items", order: 4 },
+    { id: "ontology.map", stepId: "ontology.map", runtimeKind: "ndc.semantic-mapping", order: 5 },
+    { id: "data-product.publish", stepId: "data-product.publish", runtimeKind: "ndc.data-product-publish", order: 6 },
+  ],
+);
+assert.deepEqual(
+  positionsPlan.graphBlueprint.edges.map(({ from, to }) => ({ from, to })),
+  [
+    { from: "collection.trigger.manual", to: "provider.fetch-monitoring-config" },
+    { from: "collection.trigger.interval", to: "provider.fetch-monitoring-config" },
+    { from: "provider.fetch-monitoring-config", to: "provider.fetch-units" },
+    { from: "provider.fetch-units", to: "provider.extract-units" },
+    { from: "provider.extract-units", to: "ontology.map" },
+    { from: "ontology.map", to: "data-product.publish" },
+  ],
+);
+assert.deepEqual(positionsPlan.graphBlueprint.runtime.mappingRuntime.derivationKinds, [
+  "bounded_readings",
+  "ordered_rules",
+]);
+assert.deepEqual(positionsPlan.graphBlueprint.runtime.retry, {
+  maxAttempts: 4,
+  backoff: "fixed_delay",
+  delayMs: 2000,
+});
+assert.equal(
+  JSON.stringify(positionsPlan.graphBlueprint).includes("ndc-credref:"),
+  false,
+);
+assert.equal(
+  positionsPlan.graphBlueprint.nodes.every((node) => !node.runtimeKind.includes("gelios")),
+  true,
+);
 assert.deepEqual(positionsPlan.steps.map((step) => step.kind), [
   "collection_trigger",
   "provider_request",
@@ -54,6 +110,7 @@ assert.deepEqual(
 );
 assert.equal(Object.isFrozen(positionsPlan), true);
 assert.equal(Object.isFrozen(positionsPlan.steps), true);
+assert.equal(Object.isFrozen(positionsPlan.graphBlueprint), true);
 
 const clonedPlan = compileL2ExecutionPlan(
   structuredClone(geliosProviderPackageV8),
@@ -77,6 +134,28 @@ const profilePlan = compileL2ExecutionPlan(geliosProviderPackageV8, profileConne
 assert.deepEqual(validateL2ExecutionPlan(profilePlan), { ok: true, errors: [] });
 assert.equal(profilePlan.artifacts.telemetryRegistryDigest, undefined);
 assert.notEqual(profilePlan.executionPlanDigest, positionsPlan.executionPlanDigest);
+assert.deepEqual(profilePlan.graphBlueprint.runtime.mappingRuntime.derivationKinds, []);
+
+for (const collectionProfile of geliosProviderPackageV8.collectionProfiles) {
+  const connection = instantiateL2Connection(geliosProviderPackageV8, {
+    tenantId: "tenant-all-profiles-fixture",
+    connectionId: `connection-${collectionProfile.id.replaceAll(".", "-")}`,
+    collectionProfileId: collectionProfile.id,
+    providerCredentialRef: "ndc-credref:provider-all-profiles-fixture",
+  });
+  const plan = compileL2ExecutionPlan(
+    geliosProviderPackageV8,
+    connection,
+    collectionProfile.dataProductId === "fleet.positions.current.v5"
+      ? { telemetryFieldRegistry: geliosTelemetryFieldRegistryV1 }
+      : {},
+  );
+  assert.deepEqual(validateL2ExecutionPlan(plan), { ok: true, errors: [] });
+  assert.equal(JSON.stringify(plan.graphBlueprint.runtime.mappingRuntime).includes("gelios"), false);
+  const expectedEntrypoints = collectionProfile.mode === "realtime" ? 2 : 1;
+  assert.equal(plan.graphBlueprint.nodes.length, plan.steps.length + expectedEntrypoints - 1);
+  assert.equal(plan.graphBlueprint.edges.length, plan.steps.length + expectedEntrypoints - 2);
+}
 
 const secondConnection = instantiateL2Connection(geliosProviderPackageV8, {
   tenantId: "tenant-compiler-fixture",
@@ -115,7 +194,39 @@ assert.equal(
   true,
 );
 
+const authorityEscalatedPlan = structuredClone(positionsPlan);
+authorityEscalatedPlan.graphBlueprint.runtime.target = "provider_specific_runtime";
+authorityEscalatedPlan.graphBlueprint.blueprintDigest = digestWithout(
+  authorityEscalatedPlan.graphBlueprint,
+  "blueprintDigest",
+);
+authorityEscalatedPlan.executionPlanDigest = digestWithout(
+  authorityEscalatedPlan,
+  "executionPlanDigest",
+);
+assert.equal(
+  validateL2ExecutionPlan(authorityEscalatedPlan).errors.includes("executionPlan.graphBlueprint_mismatch"),
+  true,
+);
+
 const serialized = JSON.stringify(positionsPlan);
 assert.equal(serialized.includes("Bearer "), false);
 assert.equal(serialized.includes("access_token"), false);
 assert.equal(serialized.includes("refresh_token"), false);
+
+function digestWithout(value, key) {
+  const descriptor = structuredClone(value);
+  delete descriptor[key];
+  return `sha256:${createHash("sha256").update(JSON.stringify(stableValue(descriptor)), "utf8").digest("hex")}`;
+}
+
+function stableValue(value) {
+  if (Array.isArray(value)) return value.map(stableValue);
+  if (!value || typeof value !== "object") return value;
+  return Object.fromEntries(
+    Object.keys(value)
+      .filter((key) => value[key] !== undefined)
+      .sort()
+      .map((key) => [key, stableValue(value[key])]),
+  );
+}