Fix download for DDO v5 assets (provider-initialize + policy-server probe)

CLAUDE.md

@@ -168,7 +168,7 @@ uses directly. `destroyProviders()` tears providers down on exit (next to `stopP
 - **Publish** (`publish`, `publishAlgo`): read a JSON DDO file, then `createAssetUtil` with `asset.indexedMetadata.nft.name/symbol` and `asset.services[0].files.files`. `--encrypt` (default `true`) controls DDO encryption. See `metadata/*.json` for the expected DDO shape.
 - **Edit** (`editAsset`): resolve the DDO via `waitForIndexer`, shallow-merge the top-level keys from the update JSON into the asset, then `updateAssetMetadata`.
 - **allowAlgo / disallowAlgo**: mutate `services[0].compute.publisherTrustedAlgorithms` (checks signer is the NFT owner and the service is a `compute` service; computes container + files checksums via `ProviderInstance.checkDidFiles` / `getHash`) and re-publish metadata. (`disallowAlgo` exists on `Commands` but is not registered as a CLI command.)
-- **Download/consume** (`download`): resolve DDO → `orderAsset` (buys a datatoken) → `tx.wait()` → `ProviderInstance.getDownloadUrl` → `downloadFile` (streams to disk, filename from `content-disposition` when present).
+- **Download/consume** (`download`): resolve DDO → look up the target service by id (errors if the `serviceId` is not in the DDO, instead of silently falling back to `services[0]`) → for **DDO version ≥ 5.0.0** run a provider-initialize step (`Commands.initializeProvider` → `ProviderInstance.initialize`, plus an SSI/policy-server verification via `ProviderInstance.initializePSVerification` when `SSI_WALLET_API` is set) then fetch the policy-server object (`getPolicyServerOBJ`, which now returns `null` when the node reports the policy server is not configured) → `orderAsset` (buys a datatoken) → `tx.wait()` → `ProviderInstance.getDownloadUrl` → `downloadFile` (streams to disk, filename from `content-disposition` when present). Each step catches its own error, prints an actionable message, and returns rather than throwing.
 
 ### Compute flow
 

README.md

@@ -335,7 +335,9 @@ Notes when switching nodes:
   (Order of `--did` and `--folder` does not matter.)
 
 - **Rules:**
-  serviceId is optional. If omitted, the CLI defaults to the first available download service.
+  serviceId is optional. If omitted, the CLI defaults to the first service listed in the DDO (`services[0]`). If you pass a `serviceId` that does not exist in the DDO, the command now fails fast with a clear error instead of silently ordering the first service.
+
+  For **v5 DDOs** (version ≥ 5.0.0) the download first runs a provider-initialization step against the asset's service endpoint. When `SSI_WALLET_API` is set (see the env vars above) this also performs the SSI / policy-server verification flow; when the target node reports it has no policy server configured, that step is skipped automatically.
 
 ---
 

src/commands.ts

@@ -30,6 +30,7 @@
   ConfigHelper,
   Datatoken,
   ProviderInstance,
+  ProviderInitialize,
   amountToUnits,
   getHash,
   orderAsset,
@@ -58,6 +59,7 @@
 import {
   getPolicyServerOBJ,
   getPolicyServerOBJs,
+  isPolicyServerConfigured,
   isVersionGte,
 } from "./policyServerHelper.js";
 import {
@@ -105,7 +107,7 @@
  * Drains a (possibly endless) log stream into a string. Chunks already received
  * when the read is aborted are kept — a bounded read is the normal way this ends.
  */
 async function drainLogStream(stream: any): Promise<string> {
   const chunks: Uint8Array[] = [];
   if (!stream[Symbol.asyncIterator]) {
     return await new Response(stream).text();
@@ -323,7 +325,7 @@
         indexedMetadata.nft.name,
         indexedMetadata.nft.symbol,
         this.signer,
         (services[0].files as any).files ?? services[0].files,
         asset,
         this.oceanNodeUrl,
         this.config,
@@ -364,7 +366,7 @@
         indexedMetadata.nft.name,
         indexedMetadata.nft.symbol,
         this.signer,
         (services[0].files as any).files ?? services[0].files,
         algoAsset,
         this.oceanNodeUrl,
         this.config,
@@ -440,6 +442,57 @@
     } else console.log(util.inspect(resolvedDDO, false, null, true));
   }
 
+  private async initializeProvider(
+    asset: Asset,
+    serviceId: string,
+    accountId: string,
+    providerUrl: string,
+  ): Promise<ProviderInitialize> {
+    // Only run SSI/policy-server verification when a wallet is configured AND
+    // the node confirms it has a policy server. This mirrors getPolicyServerOBJ's
+    // skip behavior, so a download against a node without a policy server
+    // proceeds instead of failing in initializePSVerification.
+    if (
+      process.env.SSI_WALLET_API?.trim() &&
+      (await isPolicyServerConfigured(providerUrl))
+    ) {
+      const command = {
+        documentId: asset.id,
+        serviceId,
+        consumerAddress: accountId,
+        policyServer: {
+          sessionId: "",
+          successRedirectUri: "",
+          errorRedirectUri: "",
+          responseRedirectUri: "",
+          presentationDefinitionUri: "",
+        },
+      };
+      const initializePs = await ProviderInstance.initializePSVerification(
+        providerUrl,
+        this.signer,
+        command,
+      );
+      if (!initializePs?.success) {
+        throw new Error(
+          `Provider initialization failed: ${initializePs?.error || "Policy Server verification failed"}`,
+        );
+      }
+    }
+    try {
+      return await ProviderInstance.initialize(
+        asset.id,
+        serviceId,
+        0,
+        accountId,
+        providerUrl,
+      );
+    } catch (error: unknown) {
+      const message = error instanceof Error ? error.message : String(error);
+      throw new Error(message.replace(/^Error:\s*/i, ""), { cause: error });
+    }
+  }
+
   public async download(args: string[]) {
     const did = args[1];
     const dataDdo = await this.aquarius.waitForIndexer(
@@ -460,20 +513,40 @@
     const ddoInstance = DDOManager.getDDOClass(dataDdo);
     const { services, version } = ddoInstance.getDDOFields();
     const serviceId = args[3] ? args[3] : services[0].id;
+    const service = services.find((s) => s.id === serviceId);
+    if (!service) {
+      console.error(
+        chalk.red(`Service ID "${serviceId}" not found in DDO ${did}.`),
+      );
+      return;
+    }
+
     let policyServer = null;
-    try {
-      if (isVersionGte(version, "5.0.0")) {
+    if (isVersionGte(version, "5.0.0")) {
+      try {
+        await this.initializeProvider(
+          dataDdo,
+          serviceId,
+          await this.signer.getAddress(),
+          service.serviceEndpoint || this.oceanNodeUrl,
+        );
+      } catch (error: unknown) {
+        const message = error instanceof Error ? error.message : String(error);
+        console.error(chalk.red("Error initializing Provider:"), message);
+        return;
+      }
+      try {
         policyServer = await getPolicyServerOBJ(
           dataDdo,
           serviceId,
           this.signer,
           this.oceanNodeUrl,
         );
+      } catch (error: unknown) {
+        const message = error instanceof Error ? error.message : String(error);
+        console.error(chalk.red("Error getting Policy Server Object:"), message);
+        return;
       }
-    } catch (error) {
-      throw new Error("Error getting Policy Server Object: " + error.message, {
-        cause: error,
-      });
     }
     const datatoken = new Datatoken(
       this.signer,
@@ -482,21 +555,28 @@
     );
     // Order the same service that policy retrieval and getDownloadUrl target.
     const serviceIndex = services.findIndex((s) => s.id === serviceId);
-    const tx = await this.orderWithRetry(() =>
-      orderAsset(
-        dataDdo,
-        this.signer,
-        this.config,
-        datatoken,
-        this.oceanNodeUrl,
-        undefined, // consumerAddress
-        undefined, // consumeMarketOrderFee
-        undefined, // providerFees
-        undefined, // consumeMarketFixedSwapFee
-        undefined, // datatokenIndex
-        serviceIndex < 0 ? 0 : serviceIndex,
-      ),
-    );
+    let tx;
+    try {
+      tx = await this.orderWithRetry(() =>
+        orderAsset(
+          dataDdo,
+          this.signer,
+          this.config,
+          datatoken,
+          this.oceanNodeUrl,
+          undefined, // consumerAddress
+          undefined, // consumeMarketOrderFee
+          undefined, // providerFees
+          undefined, // consumeMarketFixedSwapFee
+          undefined, // datatokenIndex
+          serviceIndex < 0 ? 0 : serviceIndex,
+        ),
+      );
+    } catch (error: unknown) {
+      const message = error instanceof Error ? error.message : String(error);
+      console.error(chalk.red("Error ordering asset:"), message);
+      return;
+    }
 
     if (!tx) {
       console.error(
@@ -628,7 +708,7 @@
       ) {
         const expectedAlgoServiceId = algoServiceIdInput.trim();
         const matchAlgoSvc = servicesAlgo.find(
           (s: any) => s.id === expectedAlgoServiceId,
         );
         if (!matchAlgoSvc) {
           console.error(
@@ -666,7 +746,7 @@
         const expectedServiceId = inputServices[i];
         if (expectedServiceId) {
           const match = servicesDdo.find(
             (s: any) => s.id === expectedServiceId,
           );
           if (!match) {
             console.error(
@@ -946,7 +1026,7 @@
       ) {
         const expectedAlgoServiceId = algoServiceIdInput.trim();
         const matchAlgoSvc = servicesAlgo.find(
           (s: any) => s.id === expectedAlgoServiceId,
         );
         if (!matchAlgoSvc) {
           console.error(
@@ -958,7 +1038,7 @@
         chosenAlgoServiceId = expectedAlgoServiceId;
       }
       algoServiceIndex = servicesAlgo.findIndex(
         (s: any) => s.id === chosenAlgoServiceId,
       );
       if (algoServiceIndex < 0) {
         console.error(
@@ -998,7 +1078,7 @@
         const expectedServiceId = inputServices[i];
         if (expectedServiceId) {
           const match = servicesDdo.find(
             (s: any) => s.id === expectedServiceId,
           );
           if (!match) {
             console.error(
@@ -1016,7 +1096,7 @@
         providerURI = servicesDdo[0].serviceEndpoint;
       }
       const chosenServiceIndex = servicesDdo.findIndex(
         (s: any) => s.id === chosenServiceId,
       );
       if (chosenServiceIndex < 0) {
         console.error(
@@ -1480,7 +1560,7 @@
       ) {
         const expectedAlgoServiceId = algoServiceIdInput.trim();
         const matchAlgoSvc = servicesAlgo.find(
           (s: any) => s.id === expectedAlgoServiceId,
         );
         if (!matchAlgoSvc) {
           console.error(

src/policyServerHelper.ts

@@ -13,6 +13,32 @@ import {
 import axios from "axios";
 import { Signer } from "ethers";
 
+// Bounded timeout for the node `status` probe. Without it an unresponsive node
+// would hang the probe (and any download/compute waiting on it) indefinitely.
+const PS_STATUS_PROBE_TIMEOUT_MS = 10_000;
+
+/**
+ * Probe whether the target node has a policy server configured, via the
+ * `status` directCommand. Returns `true` only when the node explicitly reports
+ * `isPSConfigured === true`. On a `false` report, a probe error, or a timeout it
+ * returns `false`, so callers can skip policy-server verification and proceed
+ * rather than hanging or hard-failing on an unresponsive node.
+ */
+export async function isPolicyServerConfigured(
+  providerUrl: string,
+): Promise<boolean> {
+  try {
+    const statusResponse = await axios.post(
+      `${providerUrl}/directCommand`,
+      { command: "status" },
+      { timeout: PS_STATUS_PROBE_TIMEOUT_MS },
+    );
+    return statusResponse.data?.isPSConfigured === true;
+  } catch {
+    return false;
+  }
+}
+
 // Semver-aware "version >= minimum" comparison (numeric, dot-separated). Avoids
 // the lexicographic pitfalls of comparing version strings directly (e.g.
 // '5.10.0' < '5.9.0' as strings). A missing/empty version is treated as below
@@ -274,13 +300,35 @@ export function extractURLSearchParams(
   return params;
 }
 
+/**
+ * Resolve the policy-server object for a single asset/service.
+ *
+ * Returns `null` when policy-server support is unavailable — i.e. the node
+ * reports it has no policy server configured (`isPSConfigured !== true`) — so
+ * callers must treat `null` as "no policy server" and proceed without one. A
+ * probe that fails or times out falls through to the normal flow instead of
+ * masking a real error with `null`.
+ */
 export async function getPolicyServerOBJ(
   ddo: Asset,
   serviceId: string,
   signer: Signer,
   providerUrl: string,
-): Promise<PolicyServerInitiateActionData> {
+): Promise<PolicyServerInitiateActionData | null> {
   try {
+    try {
+      const statusResponse = await axios.post(
+        `${providerUrl}/directCommand`,
+        { command: "status" },
+        { timeout: PS_STATUS_PROBE_TIMEOUT_MS },
+      );
+      if (statusResponse.data?.isPSConfigured !== true) {
+        return null;
+      }
+    } catch {
+      // Node did not answer the status probe; fall through and attempt the
+      // normal flow rather than masking a real error with a null.
+    }
     const accountId = await signer.getAddress();
     const presentationResult = await requestCredentialPresentation(
       ddo,
@@ -380,6 +428,16 @@ export async function getPolicyServerOBJ(
   }
 }
 
+/**
+ * Resolve policy-server objects for a set of datasets plus an optional
+ * algorithm (compute flows).
+ *
+ * Returns `null` when policy-server support is unavailable for the job — any
+ * entry below DDO v5, or any entry whose per-asset lookup yields `null` (node
+ * has no policy server configured). Callers must treat `null` as "no policy
+ * server" and pass it straight through to the provider (which accepts a
+ * nullable `policyServer`).
+ */
 export async function getPolicyServerOBJs(
   ddos: {
     documentId: string;
@@ -410,6 +468,9 @@ export async function getPolicyServerOBJs(
         signer,
         providerUrl,
       );
+      if (!result) {
+        return null;
+      }
       results.push({
         ...result,
         documentId: ddo.documentId,
@@ -430,6 +491,9 @@ export async function getPolicyServerOBJs(
           signer,
           providerUrl,
         );
+        if (!algoResult) {
+          return null;
+        }
         results.push({
           ...algoResult,
           documentId: algo.documentId,