Repository: LangStream/langstream (language: python)

Issue:
[feature] Add a way to configure init jobs to setup tables on Vector Databases and other resources (implements Assets)
Since the beginning of the design of LangStream we thought about adding the concept of "assets".

Now that people are using more and more LS it is clear that setting up third party resources must be handled by the LS application as well.

LS already creates the topics on Kafka/Pulsar but it cannot create the tables on Cassandra or the Indexes on Pinecone/ElastichSeach.

This issue a feature request for implement some kind of "initialisation jobs" that are executed at deployment time:
- when you deploy the application for the first time
- when you update the application
- when you delete the application

But it would be awkward to ask developers to create "jobs", because you have to deal with the state of the resource.
For instance you want to create the table only if it exists.


configuration.yaml

```
assets:
  - type: cassandra-table
     datasource: MyAstraDataSource
     creation-mode: create-if-not-exists|none
     drop-mode: drop-on-undeploy|none
     config:
          tableName: products
          keyspace: vsearch
          statements:
                - "CREATE TABLE products.. .."
                - "CREATE INDEX .... "
  - type: pinecone-index
     datasource: MyPineconeDataSource
     creation-mode: create-if-not-exists|none
     drop-mode: drop-on-undeploy|none
     config:
          indexName: products
          ...
  - type: jdbc-table
     datasource: MyPostGresDataSource
     creation-mode: create-if-not-exists|none
     drop-mode: drop-on-undeploy|none
     config:
          driverClassName: org.postgres.Driver
          tableName: products
          statements:
                - "CREATE TABLE products.. .."
                - "CREATE INDEX .... "
```

                 




 


Discussion / hints:

Fix patch:
diff --git a/langstream-admin-client/src/main/java/ai/langstream/admin/client/AdminClient.java b/langstream-admin-client/src/main/java/ai/langstream/admin/client/AdminClient.java
index df0db0f35..56f93e62d 100644
--- a/langstream-admin-client/src/main/java/ai/langstream/admin/client/AdminClient.java
+++ b/langstream-admin-client/src/main/java/ai/langstream/admin/client/AdminClient.java
@@ -289,7 +289,7 @@ public String getCodeInfo(String application, String codeArchiveId) {
     }
 
     @Override
-    public void close() throws Exception {
+    public void close() {
         if (executorService != null) {
             executorService.shutdownNow();
         }
diff --git a/langstream-api/src/main/java/ai/langstream/api/model/AssetDefinition.java b/langstream-api/src/main/java/ai/langstream/api/model/AssetDefinition.java
new file mode 100644
index 000000000..1acb37f94
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/model/AssetDefinition.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
+import java.util.Objects;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+@Getter
+@Setter
+@ToString
+@EqualsAndHashCode
+public class AssetDefinition {
+
+    public static final String CREATE_MODE_NONE = "none";
+    public static final String CREATE_MODE_CREATE_IF_NOT_EXISTS = "create-if-not-exists";
+
+    public AssetDefinition() {
+        creationMode = CREATE_MODE_NONE;
+    }
+
+    public AssetDefinition(
+            String id,
+            String name,
+            String creationMode,
+            String assetType,
+            Map<String, Object> config) {
+        this();
+        this.id = id;
+        this.assetType = assetType;
+        this.name = name;
+        this.config = config;
+        this.creationMode = Objects.requireNonNullElse(creationMode, CREATE_MODE_NONE);
+        validateCreationMode();
+    }
+
+    private String id;
+    private String name;
+
+    @JsonProperty("creation-mode")
+    private String creationMode;
+
+    private Map<String, Object> config;
+
+    @JsonProperty("asset-type")
+    private String assetType;
+
+    private void validateCreationMode() {
+        switch (creationMode) {
+            case CREATE_MODE_NONE:
+            case CREATE_MODE_CREATE_IF_NOT_EXISTS:
+                break;
+            default:
+                throw new IllegalArgumentException("Invalid creation mode: " + creationMode);
+        }
+    }
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/model/Module.java b/langstream-api/src/main/java/ai/langstream/api/model/Module.java
index faf39342f..842c2571b 100644
--- a/langstream-api/src/main/java/ai/langstream/api/model/Module.java
+++ b/langstream-api/src/main/java/ai/langstream/api/model/Module.java
@@ -15,8 +15,11 @@
  */
 package ai.langstream.api.model;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
@@ -29,6 +32,8 @@ public class Module {
     private final Map<String, Pipeline> pipelines = new HashMap<>();
 
     private final Map<String, TopicDefinition> topics = new HashMap<>();
+    // the order of assets is important, there may be some dependencies between them
+    private final List<AssetDefinition> assets = new ArrayList<>();
 
     public Module(String id) {
         this.id = id;
@@ -44,6 +49,25 @@ public Pipeline addPipeline(String pipelineId) {
         return p;
     }
 
+    public AssetDefinition addAsset(AssetDefinition assetDefinition) {
+        final String assetId = assetDefinition.getId();
+        AssetDefinition existing =
+                assets.stream()
+                        .filter(a -> Objects.equals(a.getId(), assetId))
+                        .findFirst()
+                        .orElse(null);
+
+        if (existing != null) {
+            if (!existing.equals(assetDefinition)) {
+                throw new IllegalArgumentException(
+                        "Asset " + assetId + " already exists in module " + id);
+            }
+            return existing;
+        }
+        assets.add(assetDefinition);
+        return assetDefinition;
+    }
+
     public TopicDefinition addTopic(TopicDefinition topicDefinition) {
         final String topicName = topicDefinition.getName();
         TopicDefinition existing = topics.get(topicName);
@@ -52,7 +76,7 @@ public TopicDefinition addTopic(TopicDefinition topicDefinition) {
             // but only if the definition is the same
             if (!existing.equals(topicDefinition)) {
                 throw new IllegalArgumentException(
-                        "Pipeline " + topicName + " already exists in module " + id);
+                        "Topic " + topicName + " already exists in module " + id);
             }
             return existing;
         }
diff --git a/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManager.java b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManager.java
new file mode 100644
index 000000000..8d46d7a4e
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManager.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.runner.assets;
+
+import ai.langstream.api.model.AssetDefinition;
+
+/** Body of the agent */
+public interface AssetManager {
+
+    boolean assetExists(AssetDefinition assetDefinition) throws Exception;
+
+    void deployAsset(AssetDefinition assetDefinition) throws Exception;
+
+    default void close() throws Exception {}
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerAndLoader.java b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerAndLoader.java
new file mode 100644
index 000000000..6a8ad7698
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerAndLoader.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.runner.assets;
+
+import ai.langstream.api.model.AssetDefinition;
+
+public record AssetManagerAndLoader(AssetManager agentCode, ClassLoader classLoader) {
+
+    public void executeWithContextClassloader(RunnableWithException code) throws Exception {
+        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(classLoader);
+            code.run(agentCode);
+        } finally {
+            Thread.currentThread().setContextClassLoader(contextClassLoader);
+        }
+    }
+
+    public <T> T callWithContextClassloader(CallableWithException code) throws Exception {
+        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+        try {
+            Thread.currentThread().setContextClassLoader(classLoader);
+            return (T) code.call(agentCode);
+        } finally {
+            Thread.currentThread().setContextClassLoader(contextClassLoader);
+        }
+    }
+
+    public AssetManager asAssetManager() {
+
+        return new AssetManager() {
+
+            @Override
+            public boolean assetExists(AssetDefinition assetDefinition) throws Exception {
+                return callWithContextClassloader(
+                        agentCode -> agentCode.assetExists(assetDefinition));
+            }
+
+            @Override
+            public void deployAsset(AssetDefinition assetDefinition) throws Exception {
+                executeWithContextClassloader(agentCode -> agentCode.deployAsset(assetDefinition));
+            }
+
+            @Override
+            public void close() throws Exception {
+                executeWithContextClassloader(agentCode -> agentCode.close());
+            }
+        };
+    }
+
+    public void close() throws Exception {
+        executeWithContextClassloader(agentCode -> agentCode.close());
+    }
+
+    @FunctionalInterface
+    public interface RunnableWithException {
+        void run(AssetManager agent) throws Exception;
+    }
+
+    @FunctionalInterface
+    public interface CallableWithException {
+        Object call(AssetManager agent) throws Exception;
+    }
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerProvider.java b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerProvider.java
new file mode 100644
index 000000000..b7413d710
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerProvider.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.runner.assets;
+
+/** Factory */
+public interface AssetManagerProvider {
+
+    boolean supports(String assetType);
+
+    /**
+     * Create a new AssetManager for the given asset type
+     *
+     * @return the new AssetManager
+     */
+    AssetManager createInstance(String assetType);
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerRegistry.java b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerRegistry.java
new file mode 100644
index 000000000..812ca21e8
--- /dev/null
+++ b/langstream-api/src/main/java/ai/langstream/api/runner/assets/AssetManagerRegistry.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.api.runner.assets;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * The runtime registry is a singleton that holds all the runtime information about the possible
+ * implementations of the LangStream API.
+ */
+@Slf4j
+public class AssetManagerRegistry {
+
+    private AgentPackageLoader assetManagerPackageLoader;
+
+    public AssetManagerRegistry() {}
+
+    public interface AssetPackage {
+        String getName();
+
+        ClassLoader getClassloader();
+    }
+
+    public interface AgentPackageLoader {
+        AssetPackage loadPackageForAsset(String assetType) throws Exception;
+
+        List<? extends ClassLoader> getAllClassloaders() throws Exception;
+
+        ClassLoader getCustomCodeClassloader();
+    }
+
+    @SneakyThrows
+    public AssetManagerAndLoader getAssetManager(String assetType) {
+        log.info("Loading AssetManager code for type {}", assetType);
+        Objects.requireNonNull(assetType, "assetType cannot be null");
+
+        ClassLoader customCodeClassloader =
+                assetManagerPackageLoader != null
+                        ? assetManagerPackageLoader.getCustomCodeClassloader()
+                        : AssetManagerRegistry.class.getClassLoader();
+        // always allow to find the system agents
+        AssetManagerAndLoader agentCodeProviderProviderFromSystem =
+                loadFromClassloader(assetType, customCodeClassloader);
+        if (agentCodeProviderProviderFromSystem != null) {
+            log.info("The agent {} is loaded from the system classpath", assetType);
+            return agentCodeProviderProviderFromSystem;
+        }
+
+        if (assetManagerPackageLoader != null) {
+            AssetPackage assetPackage = assetManagerPackageLoader.loadPackageForAsset(assetType);
+
+            if (assetPackage != null) {
+                log.info("Found the package the agent belongs to: {}", assetPackage.getName());
+                AssetManagerAndLoader agentCodeProviderProvider =
+                        loadFromClassloader(assetType, assetPackage.getClassloader());
+                if (agentCodeProviderProvider == null) {
+                    throw new RuntimeException(
+                            "Package "
+                                    + assetPackage.getName()
+                                    + " declared to support agent type "
+                                    + assetType
+                                    + " but no agent found");
+                }
+                return agentCodeProviderProvider;
+            }
+
+            log.info("No agent found in the package, let's try to find it among all the packages");
+
+            // we are not lucky, let's try to find the agent in all the packages
+            List<ClassLoader> candidateClassloaders =
+                    new ArrayList<>(assetManagerPackageLoader.getAllClassloaders());
+
+            for (ClassLoader classLoader : candidateClassloaders) {
+                AssetManagerAndLoader agentCodeProviderProvider =
+                        loadFromClassloader(assetType, classLoader);
+                if (agentCodeProviderProvider != null) {
+                    return agentCodeProviderProvider;
+                }
+            }
+        }
+
+        throw new RuntimeException("No AgentCodeProvider found for type " + assetType);
+    }
+
+    private static AssetManagerAndLoader loadFromClassloader(
+            String assetType, ClassLoader classLoader) {
+        ServiceLoader<AssetManagerProvider> loader =
+                ServiceLoader.load(AssetManagerProvider.class, classLoader);
+        Optional<ServiceLoader.Provider<AssetManagerProvider>> agentCodeProviderProvider =
+                loader.stream().filter(p -> p.get().supports(assetType)).findFirst();
+
+        return agentCodeProviderProvider
+                .map(
+                        provider ->
+                                new AssetManagerAndLoader(
+                                        provider.get().createInstance(assetType), classLoader))
+                .orElse(null);
+    }
+
+    public void setAssetManagerPackageLoader(AgentPackageLoader loader) {
+        this.assetManagerPackageLoader = loader;
+    }
+}
diff --git a/langstream-api/src/main/java/ai/langstream/api/runtime/DeployContext.java b/langstream-api/src/main/java/ai/langstream/api/runtime/DeployContext.java
index 4221a41b5..b97deff39 100644
--- a/langstream-api/src/main/java/ai/langstream/api/runtime/DeployContext.java
+++ b/langstream-api/src/main/java/ai/langstream/api/runtime/DeployContext.java
@@ -15,10 +15,20 @@
  */
 package ai.langstream.api.runtime;
 
+import ai.langstream.api.runner.assets.AssetManagerRegistry;
 import ai.langstream.api.webservice.application.ApplicationCodeInfo;
 
 public interface DeployContext extends AutoCloseable {
 
-    ApplicationCodeInfo getApplicationCodeInfo(
-            String tenant, String applicationId, String codeArchiveId);
+    default ApplicationCodeInfo getApplicationCodeInfo(
+            String tenant, String applicationId, String codeArchiveId) {
+        throw new UnsupportedOperationException();
+    }
+
+    default AssetManagerRegistry getAssetManagerRegistry() {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    default void close() {}
 }
diff --git a/langstream-api/src/main/java/ai/langstream/api/runtime/ExecutionPlan.java b/langstream-api/src/main/java/ai/langstream/api/runtime/ExecutionPlan.java
index 0811b07e3..b9892721e 100644
--- a/langstream-api/src/main/java/ai/langstream/api/runtime/ExecutionPlan.java
+++ b/langstream-api/src/main/java/ai/langstream/api/runtime/ExecutionPlan.java
@@ -16,6 +16,7 @@
 package ai.langstream.api.runtime;
 
 import ai.langstream.api.model.Application;
+import ai.langstream.api.model.AssetDefinition;
 import ai.langstream.api.model.Connection;
 import ai.langstream.api.model.Module;
 import ai.langstream.api.model.TopicDefinition;
@@ -32,6 +33,7 @@
 public final class ExecutionPlan {
 
     private final Map<TopicDefinition, Topic> topics = new HashMap<>();
+    private final List<AssetDefinition> assets = new ArrayList<>();
     private final Map<String, AgentNode> agents = new HashMap<>();
     private final String applicationId;
     private final Application application;
@@ -145,4 +147,12 @@ public TopicDefinition getTopicDefinitionByName(String name) {
                 .map(Map.Entry::getKey)
                 .orElse(null);
     }
+
+    public List<AssetDefinition> getAssets() {
+        return assets;
+    }
+
+    public void registerAsset(AssetDefinition asset) {
+        assets.add(asset);
+    }
 }
diff --git a/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java b/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java
index 375b09da4..63ae4abe2 100644
--- a/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java
+++ b/langstream-core/src/main/java/ai/langstream/impl/common/ApplicationPlaceholderResolver.java
@@ -101,6 +101,14 @@ private static Map<String, Module> resolveModules(
                                 newTopics.put(resolveValue(context, topicName), definition);
                             });
             module.replaceTopics(newTopics);
+            if (module.getAssets() != null) {
+                module.getAssets()
+                        .forEach(
+                                asset -> {
+                                    asset.setConfig(resolveMap(context, asset.getConfig()));
+                                });
+            }
+
             for (Map.Entry<String, Pipeline> pipelineEntry : module.getPipelines().entrySet()) {
                 final Pipeline pipeline = pipelineEntry.getValue();
                 List<AgentConfiguration> newAgents = new ArrayList<>();
diff --git a/langstream-core/src/main/java/ai/langstream/impl/common/BasicClusterRuntime.java b/langstream-core/src/main/java/ai/langstream/impl/common/BasicClusterRuntime.java
index 5b6cb1e87..a34de1292 100644
--- a/langstream-core/src/main/java/ai/langstream/impl/common/BasicClusterRuntime.java
+++ b/langstream-core/src/main/java/ai/langstream/impl/common/BasicClusterRuntime.java
@@ -17,6 +17,7 @@
 
 import ai.langstream.api.model.AgentConfiguration;
 import ai.langstream.api.model.Application;
+import ai.langstream.api.model.AssetDefinition;
 import ai.langstream.api.model.Connection;
 import ai.langstream.api.model.Module;
 import ai.langstream.api.model.Pipeline;
@@ -56,6 +57,8 @@ public ExecutionPlan buildExecutionPlan(
 
         detectTopics(result, streamingClusterRuntime);
 
+        detectAssets(result);
+
         detectAgents(result, streamingClusterRuntime, pluginsRegistry);
 
         validateExecutionPlan(result, streamingClusterRuntime);
@@ -85,6 +88,19 @@ protected void detectTopics(
         }
     }
 
+    /** Detects assets that are explicitly defined in the application instance. */
+    protected void detectAssets(ExecutionPlan result) {
+        Application applicationInstance = result.getApplication();
+        for (Module module : applicationInstance.getModules().values()) {
+            if (module.getAssets() != null) {
+                // the order is important, there may be some dependencies between them
+                for (AssetDefinition asset : module.getAssets()) {
+                    result.registerAsset(asset);
+                }
+            }
+        }
+    }
+
     /**
      * Detects the Agaents and build connections to them. This operation may implicitly declare
      * additional topics.
diff --git a/langstream-core/src/main/java/ai/langstream/impl/parser/ModelBuilder.java b/langstream-core/src/main/java/ai/langstream/impl/parser/ModelBuilder.java
index cfeba17c7..b1e7add24 100644
--- a/langstream-core/src/main/java/ai/langstream/impl/parser/ModelBuilder.java
+++ b/langstream-core/src/main/java/ai/langstream/impl/parser/ModelBuilder.java
@@ -21,6 +21,7 @@
 
 import ai.langstream.api.model.AgentConfiguration;
 import ai.langstream.api.model.Application;
+import ai.langstream.api.model.AssetDefinition;
 import ai.langstream.api.model.ComputeCluster;
 import ai.langstream.api.model.Connection;
 import ai.langstream.api.model.Dependency;
@@ -430,6 +431,28 @@ private static void parsePipelineFile(String filename, String content, Applicati
             }
         }
 
+        if (pipelineConfiguration.getAssets() != null) {
+            for (AssetDefinitionModel assetDefinition : pipelineConfiguration.getAssets()) {
+                String assetId =
+                        Objects.requireNonNullElse(
+                                assetDefinition.getId(), assetDefinition.getName());
+                if (assetId == null || assetId.isEmpty()) {
+                    throw new IllegalArgumentException("Asset id or name are required");
+                }
+                String name = assetDefinition.getName();
+                if (assetDefinition.getName() == null) {
+                    name = assetDefinition.getId();
+                }
+                module.addAsset(
+                        new AssetDefinition(
+                                assetId,
+                                name,
+                                assetDefinition.getCreationMode(),
+                                assetDefinition.getAssetType(),
+                                assetDefinition.getConfig()));
+            }
+        }
+
         int autoId = 1;
         if (pipelineConfiguration.getPipeline() != null) {
             for (AgentModel agent : pipelineConfiguration.getPipeline()) {
@@ -560,6 +583,8 @@ public static final class PipelineFileModel {
 
         private List<TopicDefinitionModel> topics = new ArrayList<>();
 
+        private List<AssetDefinitionModel> assets = new ArrayList<>();
+
         private ResourcesSpec resources;
         private ErrorsSpec errors;
     }
@@ -582,6 +607,22 @@ public static final class TopicDefinitionModel {
         private Map<String, Object> config;
     }
 
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static final class AssetDefinitionModel {
+        private String id;
+        private String name;
+
+        @JsonProperty("creation-mode")
+        private String creationMode;
+
+        @JsonProperty("asset-type")
+        private String assetType;
+
+        private Map<String, Object> config;
+    }
+
     @Data
     @NoArgsConstructor
     @AllArgsConstructor
diff --git a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntime.java b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntime.java
index c480f1e86..2e5ea6a93 100644
--- a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntime.java
+++ b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/main/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntime.java
@@ -15,7 +15,10 @@
  */
 package ai.langstream.runtime.impl.k8s;
 
+import ai.langstream.api.model.Application;
+import ai.langstream.api.model.AssetDefinition;
 import ai.langstream.api.model.ErrorsSpec;
+import ai.langstream.api.model.Resource;
 import ai.langstream.api.model.StreamingCluster;
 import ai.langstream.api.runtime.AgentNode;
 import ai.langstream.api.runtime.DeployContext;
@@ -89,20 +92,30 @@ public void initialize(Map<String, Object> configuration) {
     @SneakyThrows
     public Object deploy(
             String tenant,
-            ExecutionPlan applicationInstance,
+            ExecutionPlan executionPlan,
             StreamingClusterRuntime streamingClusterRuntime,
             String codeStorageArchiveId,
             DeployContext deployContext) {
-        streamingClusterRuntime.deploy(applicationInstance);
+
+        // this code is executed in the application setup job, that is a process with admin
+        // privileges
+        // this code must never run custom code imported from the application (like Jars or Py).
+
+        // deploy topics
+        streamingClusterRuntime.deploy(executionPlan);
+
+        List<Map<String, Object>> assets = collectAssets(executionPlan);
+
         List<AgentCustomResource> agentCustomResources = new ArrayList<>();
         List<Secret> secrets = new ArrayList<>();
         collectAgentCustomResourcesAndSecrets(
                 tenant,
                 agentCustomResources,
                 secrets,
-                applicationInstance,
+                executionPlan,
                 streamingClusterRuntime,
-                codeStorageArchiveId);
+                codeStorageArchiveId,
+                assets);
         final String namespace = computeNamespace(tenant);
 
         for (Secret secret : secrets) {
@@ -117,7 +130,7 @@ public Object deploy(
                 final String codeArchiveId =
                         tryKeepPreviousCodeArchiveId(
                                 tenant,
-                                applicationInstance.getApplicationId(),
+                                executionPlan.getApplicationId(),
                                 codeStorageArchiveId,
                                 existing.getSpec().getCodeArchiveId(),
                                 deployContext);
@@ -131,6 +144,51 @@ public Object deploy(
         return null;
     }
 
+    private static Map<String, Object> planAsset(
+            AssetDefinition assetDefinition, Application application) {
+        Map<String, Resource> resources = application.getResources();
+        Map<String, Object> asset = new HashMap<>();
+        asset.put("id", assetDefinition.getId());
+        asset.put("name", assetDefinition.getName());
+        asset.put("asset-type", assetDefinition.getAssetType());
+        asset.put("creation-mode", assetDefinition.getCreationMode());
+        Map<String, Object> configuration = new HashMap<>();
+        if (assetDefinition.getConfig() != null) {
+            assetDefinition
+                    .getConfig()
+                    .forEach(
+                            (key, value) -> {
+                                // automatically resolve resource references
+                                // should we do it depending on the asset type ?
+                                if ("datasource".equals(key)
+                                        && value instanceof String resourceId) {
+                                    Resource resource = resources.get(resourceId);
+                                    if (resource != null) {
+                                        value = resource;
+                                    }
+                                }
+                                configuration.put(key, value);
+                            });
+        }
+        asset.put("config", configuration);
+        return asset;
+    }
+
+    private List<Map<String, Object>> collectAssets(ExecutionPlan executionPlan) {
+        List<Map<String, Object>> assets = new ArrayList<>();
+        if (executionPlan.getAssets() != null) {
+            executionPlan
+                    .getAssets()
+                    .forEach(
+                            (assetDefinition) -> {
+                                Map<String, Object> asset =
+                                        planAsset(assetDefinition, executionPlan.getApplication());
+                                assets.add(asset);
+                            });
+        }
+        return assets;
+    }
+
     static String tryKeepPreviousCodeArchiveId(
             String tenant,
             String applicationId,
@@ -222,7 +280,8 @@ private void collectAgentCustomResourcesAndSecrets(
             List<Secret> secrets,
             ExecutionPlan applicationInstance,
             StreamingClusterRuntime streamingClusterRuntime,
-            String codeStorageArchiveId) {
+            String codeStorageArchiveId,
+            List<Map<String, Object>> assets) {
         for (AgentNode agentImplementation : applicationInstance.getAgents().values()) {
             collectAgentCustomResourceAndSecret(
                     tenant,
@@ -231,7 +290,8 @@ private void collectAgentCustomResourcesAndSecrets(
                     agentImplementation,
                     streamingClusterRuntime,
                     applicationInstance,
-                    codeStorageArchiveId);
+                    codeStorageArchiveId,
+                    assets);
         }
     }
 
@@ -243,7 +303,8 @@ private void collectAgentCustomResourceAndSecret(
             AgentNode agent,
             StreamingClusterRuntime streamingClusterRuntime,
             ExecutionPlan applicationInstance,
-            String codeStorageArchiveId) {
+            String codeStorageArchiveId,
+            List<Map<String, Object>> assets) {
         log.info(
                 "Building configuration for Agent {}, codeStorageArchiveId {}",
                 agent,
@@ -298,7 +359,8 @@ private void collectAgentCustomResourceAndSecret(
                                 defaultAgentImplementation.getAgentType(),
                                 defaultAgentImplementation.getConfiguration(),
                                 errorsConfiguration),
-                        streamingCluster);
+                        streamingCluster,
+                        assets);
 
         final Secret secret =
                 AgentResourcesFactory.generateAgentSecret(
@@ -364,7 +426,8 @@ public void delete(
                 secrets,
                 applicationInstance,
                 streamingClusterRuntime,
-                codeStorageArchiveId);
+                codeStorageArchiveId,
+                List.of()); // no assets here
         final String namespace = computeNamespace(tenant);
 
         for (Secret secret : secrets) {
diff --git a/langstream-runtime/langstream-runtime-api/src/main/java/ai/langstream/runtime/api/agent/RuntimePodConfiguration.java b/langstream-runtime/langstream-runtime-api/src/main/java/ai/langstream/runtime/api/agent/RuntimePodConfiguration.java
index e29da6235..8bb1b97f3 100644
--- a/langstream-runtime/langstream-runtime-api/src/main/java/ai/langstream/runtime/api/agent/RuntimePodConfiguration.java
+++ b/langstream-runtime/langstream-runtime-api/src/main/java/ai/langstream/runtime/api/agent/RuntimePodConfiguration.java
@@ -16,10 +16,12 @@
 package ai.langstream.runtime.api.agent;
 
 import ai.langstream.api.model.StreamingCluster;
+import java.util.List;
 import java.util.Map;
 
 public record RuntimePodConfiguration(
         Map<String, Object> input,
         Map<String, Object> output,
         AgentSpec agent,
-        StreamingCluster streamingCluster) {}
+        StreamingCluster streamingCluster,
+        List<Map<String, Object>> assets) {}
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java
index 63150738d..84f63ccc4 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java
+++ b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/AgentRunner.java
@@ -19,6 +19,10 @@
 import static ai.langstream.api.model.ErrorsSpec.FAIL;
 import static ai.langstream.api.model.ErrorsSpec.SKIP;
 
+import ai.langstream.api.model.AssetDefinition;
+import ai.langstream.api.runner.assets.AssetManager;
+import ai.langstream.api.runner.assets.AssetManagerAndLoader;
+import ai.langstream.api.runner.assets.AssetManagerRegistry;
 import ai.langstream.api.runner.code.AgentCode;
 import ai.langstream.api.runner.code.AgentCodeAndLoader;
 import ai.langstream.api.runner.code.AgentCodeRegistry;
@@ -158,6 +162,12 @@ public void run(
         try (NarFileHandler narFileHandler =
                 new NarFileHandler(agentsDirectory, customLibClassloader)) {
             narFileHandler.scan();
+
+            AssetManagerRegistry assetManagerRegistry = new AssetManagerRegistry();
+            assetManagerRegistry.setAssetManagerPackageLoader(narFileHandler);
+
+            deployAssets(configuration, assetManagerRegistry);
+
             AgentCodeRegistry agentCodeRegistry = new AgentCodeRegistry();
             agentCodeRegistry.setAgentPackageLoader(narFileHandler);
 
@@ -165,7 +175,6 @@ public void run(
                     TOPIC_CONNECTIONS_REGISTRY.getTopicConnectionsRuntime(
                             configuration.streamingCluster());
 
-            log.info("TopicConnectionsRuntime {}", topicConnectionsRuntime);
             try {
                 AgentCodeAndLoader agentCode = initAgent(configuration, agentCodeRegistry);
                 Server server = null;
@@ -969,4 +978,48 @@ public TopicConnectionProvider getTopicConnectionProvider() {
             return topicConnectionProvider;
         }
     }
+
+    private void deployAssets(
+            RuntimePodConfiguration configuration, AssetManagerRegistry assetManagerRegistry)
+            throws Exception {
+        if (configuration.assets() != null) {
+            for (Map<String, Object> assetDefinition : configuration.assets()) {
+                AssetDefinition asset = MAPPER.convertValue(assetDefinition, AssetDefinition.class);
+                deployAsset(asset, assetManagerRegistry);
+            }
+        }
+    }
+
+    private void deployAsset(AssetDefinition asset, AssetManagerRegistry assetManagerRegistry)
+            throws Exception {
+        log.info("Deploying asset {}", asset);
+        AssetManagerAndLoader assetManager =
+                assetManagerRegistry.getAssetManager(asset.getAssetType());
+        if (assetManager == null) {
+            throw new RuntimeException(
+                    "No asset manager found for asset type " + asset.getAssetType());
+        }
+        try {
+            String creationMode = asset.getCreationMode();
+            switch (creationMode) {
+                case AssetDefinition.CREATE_MODE_CREATE_IF_NOT_EXISTS -> {
+                    AssetManager assetManagerImpl = assetManager.asAssetManager();
+                    boolean exists = assetManagerImpl.assetExists(asset);
+
+                    if (!exists) {
+                        log.info(
+                                "Asset {}  of type {} needs to be created",
+                                asset.getName(),
+                                asset.getAssetType());
+                        assetManagerImpl.deployAsset(asset);
+                    }
+                }
+                case AssetDefinition.CREATE_MODE_NONE -> {
+                    return;
+                }
+            }
+        } finally {
+            assetManager.close();
+        }
+    }
 }
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/nar/NarFileHandler.java b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/nar/NarFileHandler.java
index 5b39fe1e1..03e945438 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/nar/NarFileHandler.java
+++ b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/agent/nar/NarFileHandler.java
@@ -17,6 +17,7 @@
 
 import ai.langstream.api.codestorage.GenericZipFileArchiveFile;
 import ai.langstream.api.codestorage.LocalZipFileArchiveFile;
+import ai.langstream.api.runner.assets.AssetManagerRegistry;
 import ai.langstream.api.runner.code.AgentCodeRegistry;
 import java.io.BufferedReader;
 import java.io.InputStream;
@@ -39,7 +40,10 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class NarFileHandler implements AutoCloseable, AgentCodeRegistry.AgentPackageLoader {
+public class NarFileHandler
+        implements AutoCloseable,
+                AgentCodeRegistry.AgentPackageLoader,
+                AssetManagerRegistry.AgentPackageLoader {
 
     private final Path packagesDirectory;
     private final Path temporaryDirectory;
@@ -93,14 +97,17 @@ public void close() {
     final class PackageMetadata {
         private final Path nar;
         private final String name;
-        private final Set<String> agents;
+        private final Set<String> agentTypes;
+        private final Set<String> assetTypes;
         private Path directory;
         private URLClassLoader classLoader;
 
-        public PackageMetadata(Path nar, String name, Set<String> agents) {
+        public PackageMetadata(
+                Path nar, String name, Set<String> agentTypes, Set<String> assetTypes) {
             this.nar = nar;
             this.name = name;
-            this.agents = agents;
+            this.agentTypes = agentTypes;
+            this.assetTypes = assetTypes;
         }
 
         public void unpack() throws Exception {
@@ -131,40 +138,91 @@ public void handleNarFile(Path narFile) throws Exception {
 
         // first of all we look for an index file
         try (ZipFile zipFile = new ZipFile(narFile.toFile())) {
-            ZipEntry entry = zipFile.getEntry("META-INF/ai.langstream.agents.index");
-            if (entry != null) {
-                InputStream inputStream = zipFile.getInputStream(entry);
+
+            List<String> agents = List.of();
+            List<String> assetTypes = List.of();
+
+            ZipEntry entryAgentsIndex = zipFile.getEntry("META-INF/ai.langstream.agents.index");
+            if (entryAgentsIndex != null) {
+                InputStream inputStream = zipFile.getInputStream(entryAgentsIndex);
                 byte[] bytes = inputStream.readAllBytes();
                 String string = new String(bytes, StandardCharsets.UTF_8);
                 BufferedReader reader = new BufferedReader(new StringReader(string));
-                List<String> agents =
-                        reader.lines().filter(s -> !s.isBlank() && !s.startsWith("#")).toList();
+                agents = reader.lines().filter(s -> !s.isBlank() && !s.startsWith("#")).toList();
                 log.info(
-                        "The file {} contains a static index, skipping the unpacking. It is expected that handles these agents: {}",
+                        "The file {} contains a static agents index, skipping the unpacking. It is expected that handles these agents: {}",
                         narFile,
                         agents);
+            }
+
+            ZipEntry entryAssetsIndex = zipFile.getEntry("META-INF/ai.langstream.assets.index");
+            if (entryAssetsIndex != null) {
+                InputStream inputStream = zipFile.getInputStream(entryAssetsIndex);
+                byte[] bytes = inputStream.readAllBytes();
+                String string = new String(bytes, StandardCharsets.UTF_8);
+                BufferedReader reader = new BufferedReader(new StringReader(string));
+                assetTypes =
+                        reader.lines().filter(s -> !s.isBlank() && !s.startsWith("#")).toList();
+                log.info(
+                        "The file {} contains a static assetTypes index, skipping the unpacking. It is expected that handles these assetTypes: {}",
+                        narFile,
+                        assetTypes);
+            }
+
+            if (!agents.isEmpty() || !assetTypes.isEmpty()) {
                 PackageMetadata metadata =
-                        new PackageMetadata(narFile, filename, Set.copyOf(agents));
+                        new PackageMetadata(
+                                narFile, filename, Set.copyOf(agents), Set.copyOf(assetTypes));
                 packages.put(filename, metadata);
                 return;
             }
 
-            ZipEntry serviceProvider =
+            ZipEntry serviceProviderForAgents =
                     zipFile.getEntry(
                             "META-INF/services/ai.langstream.api.runner.code.AgentCodeProvider");
-            if (serviceProvider == null) {
+            ZipEntry serviceProviderForAssets =
+                    zipFile.getEntry(
+                            "META-INF/services/ai.langstream.api.runner.assets.AssetManagerProvider");
+            if (serviceProviderForAgents == null && serviceProviderForAssets == null) {
                 log.info(
-                        "The file {} does not contain any AgentCodeProvider, skipping the file",
+                        "The file {} does not contain any AgentCodeProvider/AssetManagerProvider, skipping the file",
                         narFile);
                 return;
             }
         }
 
-        log.info("The file {} does not contain an index, still adding the file", narFile);
-        PackageMetadata metadata = new PackageMetadata(narFile, filename, null);
+        log.info("The file {} does not contain any index, still adding the file", narFile);
+        PackageMetadata metadata = new PackageMetadata(narFile, filename, null, null);
         packages.put(filename, metadata);
     }
 
+    @Override
+    public AssetManagerRegistry.AssetPackage loadPackageForAsset(String assetType)
+            throws Exception {
+        PackageMetadata packageForAssetType = getPackageForAssetType(assetType);
+        if (packageForAssetType == null) {
+            return null;
+        }
+        URLClassLoader classLoader =
+                createClassloaderForPackage(customCodeClassloader, packageForAssetType);
+        log.info(
+                "For package {}, classloader {}, parent {}",
+                packageForAssetType.getName(),
+                classLoader,
+                classLoader.getParent());
+        return new AssetManagerRegistry.AssetPackage() {
+            @Override
+            public ClassLoader getClassloader() {
+                return classLoader;
+            }
+
+            @Override
+            public String getName() {
+                return packageForAssetType.getName();
+            }
+        };
+    }
+
     @Override
     @SneakyThrows
     public AgentCodeRegistry.AgentPackage loadPackageForAgent(String agentType) {
@@ -194,7 +252,14 @@ public String getName() {
 
     public PackageMetadata getPackageForAgentType(String name) {
         return packages.values().stream()
-                .filter(p -> p.agents != null && p.agents.contains(name))
+                .filter(p -> p.agentTypes != null && p.agentTypes.contains(name))
+                .findFirst()
+                .orElse(null);
+    }
+
+    public PackageMetadata getPackageForAssetType(String name) {
+        return packages.values().stream()
+                .filter(p -> p.assetTypes != null && p.assetTypes.contains(name))
                 .findFirst()
                 .orElse(null);
     }
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/deployer/RuntimeDeployer.java b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/deployer/RuntimeDeployer.java
index c8046d9f3..7fcd7d545 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/deployer/RuntimeDeployer.java
+++ b/langstream-runtime/langstream-runtime-impl/src/main/java/ai/langstream/runtime/deployer/RuntimeDeployer.java
@@ -138,7 +138,7 @@ public ApplicationCodeInfo getApplicationCodeInfo(
         }
 
         @Override
-        public void close() throws Exception {
+        public void close() {
             if (adminClient != null) {
                 adminClient.close();
             }


Test patch:
diff --git a/langstream-core/src/test/java/ai/langstream/model/parser/AssetsTest.java b/langstream-core/src/test/java/ai/langstream/model/parser/AssetsTest.java
new file mode 100644
index 000000000..4a3c0dea8
--- /dev/null
+++ b/langstream-core/src/test/java/ai/langstream/model/parser/AssetsTest.java
@@ -0,0 +1,84 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.model.parser;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import ai.langstream.api.model.Application;
+import ai.langstream.api.model.AssetDefinition;
+import ai.langstream.api.model.Module;
+import ai.langstream.impl.parser.ModelBuilder;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+public class AssetsTest {
+
+    @Test
+    public void testConfigureAssets() throws Exception {
+        Application applicationInstance =
+                ModelBuilder.buildApplicationInstance(
+                                Map.of(
+                                        "module.yaml",
+                                        """
+                                module: "module-1"
+                                id: "pipeline-1"
+                                assets:
+                                  - name: "Cassandra table"
+                                    asset-type: "cassandra-table"
+                                    creation-mode: create-if-not-exists
+                                    config:
+                                      table-name: "products"
+                                      key-space: "my_keyspace"
+                                      statements:
+                                         - "CREATE TABLE my_keyspace.products (id text PRIMARY KEY, name text, price int)"
+                                         - "CREATE INDEX ON my_keyspace.products (name, price)"
+                                pipeline:
+                                  - name: "step1"
+                                    type: "noop"
+                                """),
+                                buildInstanceYaml(),
+                                null)
+                        .getApplication();
+        Module module = applicationInstance.getModule("module-1");
+        assertEquals(1, module.getAssets().size());
+        AssetDefinition asset = module.getAssets().get(0);
+        assertEquals("Cassandra table", asset.getName());
+        assertEquals("create-if-not-exists", asset.getCreationMode());
+        assertEquals("cassandra-table", asset.getAssetType());
+        assertEquals(
+                Map.of(
+                        "table-name",
+                        "products",
+                        "key-space",
+                        "my_keyspace",
+                        "statements",
+                        List.of(
+                                "CREATE TABLE my_keyspace.products (id text PRIMARY KEY, name text, price int)",
+                                "CREATE INDEX ON my_keyspace.products (name, price)")),
+                asset.getConfig());
+    }
+
+    private static String buildInstanceYaml() {
+        return """
+                instance:
+                  streamingCluster:
+                    type: "noop"
+                  computeCluster:
+                    type: "none"
+                """;
+    }
+}
diff --git a/langstream-k8s-deployer/langstream-k8s-deployer-operator/src/test/java/ai/langstream/deployer/k8s/controllers/AgentControllerIT.java b/langstream-k8s-deployer/langstream-k8s-deployer-operator/src/test/java/ai/langstream/deployer/k8s/controllers/AgentControllerIT.java
index edfb94447..3c6298891 100644
--- a/langstream-k8s-deployer/langstream-k8s-deployer-operator/src/test/java/ai/langstream/deployer/k8s/controllers/AgentControllerIT.java
+++ b/langstream-k8s-deployer/langstream-k8s-deployer-operator/src/test/java/ai/langstream/deployer/k8s/controllers/AgentControllerIT.java
@@ -70,7 +70,8 @@ void testAgentController() {
                                                 "fn-type",
                                                 Map.of("config", true),
                                                 Map.of()),
-                                        new StreamingCluster("noop", Map.of("config", true)))))
+                                        new StreamingCluster("noop", Map.of("config", true)),
+                                        null)))
                 .inNamespace(namespace)
                 .serverSideApply();
 
diff --git a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java
index 9ab29b71e..c0fef2e47 100644
--- a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java
+++ b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeDockerTest.java
@@ -315,9 +315,6 @@ public ApplicationCodeInfo getApplicationCodeInfo(
                                         "java-%s-%s-%s".formatted(tenant, applicationId, last));
                                 return new ApplicationCodeInfo(digests);
                             }
-
-                            @Override
-                            public void close() throws Exception {}
                         });
 
         ExecutionPlan implementation = deployer.createImplementation("app", applicationInstance);
diff --git a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeTest.java b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeTest.java
index 1709601a5..49bb6f58a 100644
--- a/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeTest.java
+++ b/langstream-k8s-runtime/langstream-k8s-runtime-core/src/test/java/ai/langstream/runtime/impl/k8s/KubernetesClusterRuntimeTest.java
@@ -126,9 +126,6 @@ public ApplicationCodeInfo getApplicationCodeInfo(
                         }
                         return mapped;
                     }
-
-                    @Override
-                    public void close() throws Exception {}
                 };
 
         final String result =
diff --git a/langstream-k8s-storage/src/test/java/ai/langstream/impl/storage/k8s/apps/KubernetesApplicationStoreLogsTest.java b/langstream-k8s-storage/src/test/java/ai/langstream/impl/storage/k8s/apps/KubernetesApplicationStoreLogsTest.java
index 8056a5f32..f159a0a23 100644
--- a/langstream-k8s-storage/src/test/java/ai/langstream/impl/storage/k8s/apps/KubernetesApplicationStoreLogsTest.java
+++ b/langstream-k8s-storage/src/test/java/ai/langstream/impl/storage/k8s/apps/KubernetesApplicationStoreLogsTest.java
@@ -149,7 +149,8 @@ private static AgentCustomResource agentCustomResource(
                                                 "fn-type",
                                                 Map.of("config", true),
                                                 Map.of()),
-                                        new StreamingCluster("noop", Map.of("config", true)))))
+                                        new StreamingCluster("noop", Map.of("config", true)),
+                                        null)))
                 .inNamespace("langstream-" + tenant)
                 .serverSideApply();
 
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java
index 882050f55..1016a4b1d 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/AbstractApplicationRunner.java
@@ -111,13 +111,26 @@ protected ApplicationRuntime deployApplication(
             String instance,
             String... expectedAgents)
             throws Exception {
+        return deployApplicationWithSecrets(
+                tenant, appId, application, instance, null, expectedAgents);
+    }
+
+    protected ApplicationRuntime deployApplicationWithSecrets(
+            String tenant,
+            String appId,
+            Map<String, String> application,
+            String instance,
+            String secretsContents,
+            String... expectedAgents)
+            throws Exception {
 
         kubeServer.spyAgentCustomResources(tenant, expectedAgents);
         final Map<String, Secret> secrets =
                 kubeServer.spyAgentCustomResourcesSecrets(tenant, expectedAgents);
 
         Application applicationInstance =
-                ModelBuilder.buildApplicationInstance(application, instance, null).getApplication();
+                ModelBuilder.buildApplicationInstance(application, instance, secretsContents)
+                        .getApplication();
 
         ExecutionPlan implementation =
                 applicationDeployer.createImplementation(appId, applicationInstance);
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/assets/DeployAssetsTest.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/assets/DeployAssetsTest.java
new file mode 100644
index 000000000..8f8c870c9
--- /dev/null
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/assets/DeployAssetsTest.java
@@ -0,0 +1,119 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.assets;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import ai.langstream.AbstractApplicationRunner;
+import ai.langstream.api.model.AssetDefinition;
+import ai.langstream.mockagents.MockAssetManagerCodeProvider;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+class DeployAssetsTest extends AbstractApplicationRunner {
+
+    @Test
+    public void testDeployAsset() throws Exception {
+        String tenant = "tenant";
+        String[] expectedAgents = {"app-step1"};
+
+        String secrets =
+                """
+                            secrets:
+                              - id: "the-secret"
+                                data:
+                                   password: "bar"
+                            """;
+        Map<String, String> application =
+                Map.of(
+                        "configuration.yaml",
+                        """
+                            configuration:
+                               resources:
+                                    - type: "datasource"
+                                      name: "the-resource"
+                                      configuration:
+                                         foo: "{{{secrets.the-secret.password}}}"
+                            """,
+                        "module.yaml",
+                        """
+                        module: "module-1"
+                        id: "pipeline-1"
+                        assets:
+                          - name: "my-table"
+                            creation-mode: create-if-not-exists
+                            asset-type: "mock-database-resource"
+                            config:
+                                table: "{{{globals.table-name}}}"
+                                datasource: "the-resource"
+                        topics:
+                          - name: "input-topic"
+                            creation-mode: create-if-not-exists
+                          - name: "output-topic"
+                            creation-mode: create-if-not-exists
+                        pipeline:
+                          - name: "identity"
+                            id: "step1"
+                            type: "identity"
+                            input: "input-topic"
+                            output: "output-topic"
+                        """);
+        try (ApplicationRuntime applicationRuntime =
+                deployApplicationWithSecrets(
+                        tenant, "app", application, buildInstanceYaml(), secrets, expectedAgents)) {
+            try (KafkaProducer<String, String> producer = createProducer();
+                    KafkaConsumer<String, String> consumer = createConsumer("output-topic")) {
+                sendMessage("input-topic", "test", producer);
+                executeAgentRunners(applicationRuntime);
+                waitForMessages(consumer, List.of("test"));
+            }
+
+            CopyOnWriteArrayList<AssetDefinition> deployedAssets =
+                    MockAssetManagerCodeProvider.MockDatabaseResourceAssetManager.DEPLOYED_ASSETS;
+            assertEquals(1, deployedAssets.size());
+            AssetDefinition deployedAsset = deployedAssets.get(0);
+            assertEquals("my-table", deployedAsset.getConfig().get("table"));
+            Map<String, Object> datasource =
+                    (Map<String, Object>) deployedAsset.getConfig().get("datasource");
+            Map<String, Object> datasourceConfiguration =
+                    (Map<String, Object>) datasource.get("configuration");
+            assertEquals("bar", datasourceConfiguration.get("foo"));
+        }
+    }
+
+    @Override
+    protected String buildInstanceYaml() {
+        return """
+                instance:
+                  globals:
+                     table-name: "my-table"
+                  streamingCluster:
+                    type: "kafka"
+                    configuration:
+                      admin:
+                        bootstrap.servers: "%s"
+                  computeCluster:
+                     type: "kubernetes"
+                """
+                .formatted(kafkaContainer.getBootstrapServers());
+    }
+}
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/mockagents/MockAssetManagerCodeProvider.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/mockagents/MockAssetManagerCodeProvider.java
new file mode 100644
index 000000000..340acd655
--- /dev/null
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/mockagents/MockAssetManagerCodeProvider.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.mockagents;
+
+import ai.langstream.api.model.AssetDefinition;
+import ai.langstream.api.runner.assets.AssetManager;
+import ai.langstream.api.runner.assets.AssetManagerProvider;
+import java.util.concurrent.CopyOnWriteArrayList;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class MockAssetManagerCodeProvider implements AssetManagerProvider {
+    @Override
+    public boolean supports(String agentType) {
+        return "mock-database-resource".equals(agentType);
+    }
+
+    @Override
+    public AssetManager createInstance(String agentType) {
+        switch (agentType) {
+            case "mock-database-resource":
+                return new MockDatabaseResourceAssetManager();
+            default:
+                throw new IllegalStateException();
+        }
+    }
+
+    public static class MockDatabaseResourceAssetManager implements AssetManager {
+
+        public static CopyOnWriteArrayList<AssetDefinition> DEPLOYED_ASSETS =
+                new CopyOnWriteArrayList<>();
+
+        @Override
+        public synchronized boolean assetExists(AssetDefinition assetDefinition) throws Exception {
+            return DEPLOYED_ASSETS.stream()
+                    .anyMatch(a -> a.getId().equals(assetDefinition.getId()));
+        }
+
+        @Override
+        public synchronized void deployAsset(AssetDefinition assetDefinition) throws Exception {
+            log.info("Deploying asset {}", assetDefinition);
+            DEPLOYED_ASSETS.add(assetDefinition);
+        }
+    }
+}
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/LoadAgentCodeTest.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/LoadAgentCodeTest.java
similarity index 98%
rename from langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/LoadAgentCodeTest.java
rename to langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/LoadAgentCodeTest.java
index 37699dcd2..636421821 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/kafka/LoadAgentCodeTest.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/LoadAgentCodeTest.java
@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package ai.langstream.kafka;
+package ai.langstream.runtime;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/LoadAssertManagerCodeTest.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/LoadAssertManagerCodeTest.java
new file mode 100644
index 000000000..fc04797c4
--- /dev/null
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/LoadAssertManagerCodeTest.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.runtime;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+import ai.langstream.api.model.AssetDefinition;
+import ai.langstream.api.runner.assets.AssetManager;
+import ai.langstream.api.runner.assets.AssetManagerRegistry;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+
+@Slf4j
+class LoadAssertManagerCodeTest {
+
+    @Test
+    public void testLoadNoop() throws Exception {
+        AssetManagerRegistry registry = new AssetManagerRegistry();
+        AssetDefinition assetDefinition = new AssetDefinition();
+        assetDefinition.setAssetType("mock-database-resource");
+        AssetManager assetManager =
+                registry.getAssetManager(assetDefinition.getAssetType()).agentCode();
+
+        assertFalse(assetManager.assetExists(assetDefinition));
+        assetManager.deployAsset(assetDefinition);
+    }
+}
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java
index 6fd80214d..2ea1d803a 100644
--- a/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java
+++ b/langstream-runtime/langstream-runtime-impl/src/test/java/ai/langstream/runtime/agent/AgentRunnerStarterTest.java
@@ -38,7 +38,8 @@ public void test() throws Exception {
         String podRuntimeFile =
                 Files.createTempFile("langstream", ".json").toFile().getAbsolutePath();
         mapper.writeValue(
-                new File(podRuntimeFile), new RuntimePodConfiguration(null, null, null, null));
+                new File(podRuntimeFile),
+                new RuntimePodConfiguration(null, null, null, null, null));
         String codeDir =
                 Files.createTempDirectory("langstream-cli-test").toFile().getAbsolutePath();
         String agentsDir =
diff --git a/langstream-runtime/langstream-runtime-impl/src/test/resources/META-INF/services/ai.langstream.api.runner.assets.AssetManagerProvider b/langstream-runtime/langstream-runtime-impl/src/test/resources/META-INF/services/ai.langstream.api.runner.assets.AssetManagerProvider
new file mode 100644
index 000000000..2686b64ec
--- /dev/null
+++ b/langstream-runtime/langstream-runtime-impl/src/test/resources/META-INF/services/ai.langstream.api.runner.assets.AssetManagerProvider
@@ -0,0 +1,1 @@
+ai.langstream.mockagents.MockAssetManagerCodeProvider
\ No newline at end of file