chore(core): properly handle dashboard header button on small screens

Closes https://github.com/kestra-io/kestra/issues/13838.

core/build.gradle

@@ -24,6 +24,9 @@ dependencies {
     // reactor
     api "io.projectreactor:reactor-core"
 
+    // awaitility
+    api "org.awaitility:awaitility"
+
     // micronaut
     api "io.micronaut.data:micronaut-data-model"
     implementation "io.micronaut:micronaut-http-server-netty"

core/src/main/java/io/kestra/core/assets/AssetManagerFactory.java

@@ -0,0 +1,25 @@
+package io.kestra.core.assets;
+
+import io.kestra.core.models.assets.Asset;
+import io.kestra.core.runners.AssetEmitter;
+import jakarta.inject.Singleton;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Singleton
+public class AssetManagerFactory {
+    public AssetEmitter of(boolean enabled) {
+        return new AssetEmitter() {
+            @Override
+            public void upsert(Asset asset) {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public List<Asset> outputs() {
+                return new ArrayList<>();
+            }
+        };
+    }
+}

core/src/main/java/io/kestra/core/assets/AssetService.java

@@ -0,0 +1,31 @@
+package io.kestra.core.assets;
+
+import io.kestra.core.models.assets.Asset;
+import io.kestra.core.models.assets.AssetIdentifier;
+import io.kestra.core.models.assets.AssetUser;
+import io.kestra.core.queues.QueueException;
+import io.micronaut.context.annotation.Secondary;
+import jakarta.inject.Singleton;
+
+import java.util.List;
+
+public interface AssetService {
+
+    void asyncUpsert(AssetUser assetUser, Asset asset) throws QueueException;
+
+    void assetLineage(AssetUser assetUser, List<AssetIdentifier> inputs, List<AssetIdentifier> outputs) throws QueueException;
+
+    @Singleton
+    @Secondary
+    class NoopAssetService implements AssetService {
+        @Override
+        public void asyncUpsert(AssetUser assetUser, Asset asset) throws QueueException {
+            // no-op
+        }
+
+        @Override
+        public void assetLineage(AssetUser assetUser, List<AssetIdentifier> inputs, List<AssetIdentifier> outputs) {
+            // no-op
+        }
+    }
+}

core/src/main/java/io/kestra/core/docs/JsonSchemaGenerator.java

@@ -22,6 +22,7 @@ import com.github.victools.jsonschema.module.swagger2.Swagger2Module;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ScheduleCondition;
 import io.kestra.core.models.dashboards.DataFilter;
@@ -63,7 +64,7 @@ import static io.kestra.core.serializers.JacksonMapper.MAP_TYPE_REFERENCE;
 @Singleton
 @Slf4j
 public class JsonSchemaGenerator {
-    
+
     private static final List<Class<?>> TYPES_RESOLVED_AS_STRING = List.of(Duration.class, LocalTime.class, LocalDate.class, LocalDateTime.class, ZonedDateTime.class, OffsetDateTime.class, OffsetTime.class);
     private static final List<Class<?>> SUBTYPE_RESOLUTION_EXCLUSION_FOR_PLUGIN_SCHEMA = List.of(Task.class, AbstractTrigger.class);
 
@@ -276,10 +277,10 @@ public class JsonSchemaGenerator {
             .with(Option.DEFINITION_FOR_MAIN_SCHEMA)
             .with(Option.PLAIN_DEFINITION_KEYS)
             .with(Option.ALLOF_CLEANUP_AT_THE_END);
-            
-        // HACK: Registered a custom JsonUnwrappedDefinitionProvider prior to the JacksonModule 
+
+        // HACK: Registered a custom JsonUnwrappedDefinitionProvider prior to the JacksonModule
         // to be able to return an CustomDefinition with an empty node when the ResolvedType can't be found.
-        builder.forTypesInGeneral().withCustomDefinitionProvider(new JsonUnwrappedDefinitionProvider(){
+        builder.forTypesInGeneral().withCustomDefinitionProvider(new JsonUnwrappedDefinitionProvider() {
             @Override
             public CustomDefinition provideCustomSchemaDefinition(ResolvedType javaType, SchemaGenerationContext context) {
                 try {
@@ -321,7 +322,7 @@ public class JsonSchemaGenerator {
         // inline some type
         builder.forTypesInGeneral()
             .withCustomDefinitionProvider(new CustomDefinitionProviderV2() {
-                
+
                 @Override
                 public CustomDefinition provideCustomSchemaDefinition(ResolvedType javaType, SchemaGenerationContext context) {
                     if (javaType.isInstanceOf(Map.class) || javaType.isInstanceOf(Enum.class)) {
@@ -589,7 +590,8 @@ public class JsonSchemaGenerator {
         // The `const` property is used by editors for auto-completion based on that schema.
         builder.forTypesInGeneral().withTypeAttributeOverride((collectedTypeAttributes, scope, context) -> {
             final Class<?> pluginType = scope.getType().getErasedType();
-            if (pluginType.getAnnotation(Plugin.class) != null) {
+            Plugin pluginAnnotation = pluginType.getAnnotation(Plugin.class);
+            if (pluginAnnotation != null) {
                 ObjectNode properties = (ObjectNode) collectedTypeAttributes.get("properties");
                 if (properties != null) {
                     properties.set("type", context.getGeneratorConfig().createObjectNode()
@@ -764,6 +766,14 @@ public class JsonSchemaGenerator {
                         consumer.accept(typeContext.resolve(clz));
                     }
                 }).toList();
+        } else if (declaredType.getErasedType() == Asset.class) {
+            return getRegisteredPlugins()
+                .stream()
+                .flatMap(registeredPlugin -> registeredPlugin.getAssets().stream())
+                .filter(p -> allowedPluginTypes.isEmpty() || allowedPluginTypes.contains(p.getName()))
+                .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
+                .map(typeContext::resolve)
+                .toList();
         }
 
         return null;

core/src/main/java/io/kestra/core/models/QueryFilter.java

@@ -103,12 +103,48 @@ public record QueryFilter(
                 return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.IN, Op.NOT_IN, Op.CONTAINS);
             }
         },
+        METADATA("metadata") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.IN, Op.NOT_IN, Op.CONTAINS);
+            }
+        },
         FLOW_ID("flowId") {
             @Override
             public List<Op> supportedOp() {
                 return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX, Op.IN, Op.NOT_IN, Op.PREFIX);
             }
         },
+        FLOW_REVISION("flowRevision") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.IN, Op.NOT_IN);
+            }
+        },
+        ID("id") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX);
+            }
+        },
+        ASSET_ID("assetId") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX);
+            }
+        },
+        TYPE("type") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX);
+            }
+        },
+        CREATED("created") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.GREATER_THAN_OR_EQUAL_TO, Op.GREATER_THAN, Op.LESS_THAN_OR_EQUAL_TO, Op.LESS_THAN, Op.EQUALS, Op.NOT_EQUALS);
+            }
+        },
         UPDATED("updated") {
             @Override
             public List<Op> supportedOp() {
@@ -163,6 +199,18 @@ public record QueryFilter(
                 return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.IN, Op.NOT_IN);
             }
         },
+        TASK_ID("taskId") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.IN, Op.NOT_IN);
+            }
+        },
+        TASK_RUN_ID("taskRunId") {
+            @Override
+            public List<Op> supportedOp() {
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.IN, Op.NOT_IN);
+            }
+        },
         CHILD_FILTER("childFilter") {
             @Override
             public List<Op> supportedOp() {
@@ -312,6 +360,34 @@ public record QueryFilter(
                     Field.UPDATED
                 );
             }
+        },
+        ASSET {
+            @Override
+            public List<Field> supportedField() {
+                return List.of(
+                    Field.QUERY,
+                    Field.ID,
+                    Field.TYPE,
+                    Field.NAMESPACE,
+                    Field.METADATA,
+                    Field.UPDATED
+                );
+            }
+        },
+        ASSET_USAGE {
+            @Override
+            public List<Field> supportedField() {
+                return List.of(
+                    Field.ASSET_ID,
+                    Field.NAMESPACE,
+                    Field.FLOW_ID,
+                    Field.FLOW_REVISION,
+                    Field.EXECUTION_ID,
+                    Field.TASK_ID,
+                    Field.TASK_RUN_ID,
+                    Field.CREATED
+                );
+            }
         };
 
         public abstract List<Field> supportedField();

core/src/main/java/io/kestra/core/models/assets/Asset.java

@@ -0,0 +1,111 @@
+package io.kestra.core.models.assets;
+
+import com.fasterxml.jackson.annotation.JsonAnySetter;
+import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.Plugin;
+import io.kestra.core.utils.IdUtils;
+import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.annotation.Nullable;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.Pattern;
+import jakarta.validation.constraints.Size;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+import java.util.*;
+
+@Getter
+@NoArgsConstructor
+public abstract class Asset implements HasUID, DeletedInterface, Plugin {
+    @Hidden
+    @Pattern(regexp = "^[a-z0-9][a-z0-9_-]*")
+    protected String tenantId;
+
+    @Pattern(regexp = "^[a-z0-9][a-z0-9._-]*")
+    @Size(min = 1, max = 150)
+    protected String namespace;
+
+    @NotBlank
+    @Pattern(regexp = "^[a-zA-Z0-9][a-zA-Z0-9._-]*")
+    @Size(min = 1, max = 150)
+    protected String id;
+
+    @NotBlank
+    protected String type;
+
+    protected String displayName;
+
+    protected String description;
+
+    protected Map<String, Object> metadata;
+
+    @Nullable
+    @Hidden
+    private Instant created;
+
+    @Nullable
+    @Hidden
+    private Instant updated;
+
+    @Hidden
+    private boolean deleted;
+
+    public Asset(
+        String tenantId,
+        String namespace,
+        String id,
+        String type,
+        String displayName,
+        String description,
+        Map<String, Object> metadata,
+        Instant created,
+        Instant updated,
+        boolean deleted
+    ) {
+        this.tenantId = tenantId;
+        this.namespace = namespace;
+        this.id = id;
+        this.type = type;
+        this.displayName = displayName;
+        this.description = description;
+        this.metadata = Optional.ofNullable(metadata).map(HashMap::new).orElse(new HashMap<>());
+        Instant now = Instant.now();
+        this.created = Optional.ofNullable(created).orElse(now);
+        this.updated = Optional.ofNullable(updated).orElse(now);
+        this.deleted = deleted;
+    }
+
+    public <T extends Asset> T toUpdated() {
+        if (this.created == null) {
+            this.created = Instant.now();
+        }
+        this.updated = Instant.now();
+        return (T) this;
+    }
+
+    public Asset toDeleted() {
+        this.deleted = true;
+        return this;
+    }
+
+    @JsonAnySetter
+    public void setMetadata(String name, Object value) {
+        metadata.put(name, value);
+    }
+
+    @Override
+    public String uid() {
+        return Asset.uid(tenantId, id);
+    }
+
+    public static String uid(String tenantId, String id) {
+        return IdUtils.fromParts(tenantId, id);
+    }
+
+    public Asset withTenantId(String tenantId) {
+        this.tenantId = tenantId;
+        return this;
+    }
+}

core/src/main/java/io/kestra/core/models/assets/AssetIdentifier.java

@@ -0,0 +1,19 @@
+package io.kestra.core.models.assets;
+
+import io.kestra.core.utils.IdUtils;
+import io.swagger.v3.oas.annotations.Hidden;
+
+public record AssetIdentifier(@Hidden String tenantId, @Hidden String namespace, String id){
+
+    public AssetIdentifier withTenantId(String tenantId) {
+        return new AssetIdentifier(tenantId, this.namespace, this.id);
+    }
+
+    public String uid() {
+        return IdUtils.fromParts(tenantId, id);
+    }
+
+    public static AssetIdentifier of(Asset asset) {
+        return new AssetIdentifier(asset.getTenantId(), asset.getNamespace(), asset.getId());
+    }
+}

core/src/main/java/io/kestra/core/models/assets/AssetUser.java

@@ -0,0 +1,18 @@
+package io.kestra.core.models.assets;
+
+import io.kestra.core.models.HasUID;
+import io.kestra.core.models.flows.FlowId;
+import io.kestra.core.utils.IdUtils;
+
+/**
+ * Represents an entity that used an asset
+ */
+public record AssetUser(String tenantId, String namespace, String flowId, Integer flowRevision, String executionId, String taskId, String taskRunId) implements HasUID {
+    public String uid() {
+        return IdUtils.fromParts(tenantId, namespace, flowId, String.valueOf(flowRevision), executionId, taskRunId);
+    }
+
+    public FlowId toFlowId() {
+        return FlowId.of(tenantId, namespace, flowId, flowRevision);
+    }
+}

core/src/main/java/io/kestra/core/models/assets/AssetsDeclaration.java

@@ -0,0 +1,22 @@
+package io.kestra.core.models.assets;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import io.micronaut.core.annotation.Introspected;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+import java.util.Optional;
+
+@Getter
+public class AssetsDeclaration extends AssetsInOut {
+    private boolean enableAuto;
+
+    @JsonCreator
+    public AssetsDeclaration(Boolean enableAuto, List<AssetIdentifier> inputs, List<Asset> outputs) {
+        super(inputs, outputs);
+
+        this.enableAuto = Optional.ofNullable(enableAuto).orElse(false);
+    }
+}

core/src/main/java/io/kestra/core/models/assets/AssetsInOut.java

@@ -0,0 +1,21 @@
+package io.kestra.core.models.assets;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import lombok.Getter;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+@Getter
+public class AssetsInOut {
+    private List<AssetIdentifier> inputs;
+
+    private List<Asset> outputs;
+
+    @JsonCreator
+    public AssetsInOut(List<AssetIdentifier> inputs, List<Asset> outputs) {
+        this.inputs = Optional.ofNullable(inputs).orElse(Collections.emptyList());
+        this.outputs = Optional.ofNullable(outputs).orElse(Collections.emptyList());
+    }
+}

core/src/main/java/io/kestra/core/models/assets/Custom.java

@@ -0,0 +1,32 @@
+package io.kestra.core.models.assets;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import io.kestra.core.models.annotations.Plugin;
+import io.swagger.v3.oas.annotations.Hidden;
+import lombok.Builder;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+import java.util.Map;
+
+@NoArgsConstructor
+@Plugin
+@Hidden
+public class Custom extends Asset {
+    @Builder
+    @JsonCreator
+    public Custom(
+        String tenantId,
+        String namespace,
+        String id,
+        String type,
+        String displayName,
+        String description,
+        Map<String, Object> metadata,
+        Instant created,
+        Instant updated,
+        boolean deleted
+    ) {
+        super(tenantId, namespace, id, type, displayName, description, metadata, created, updated, deleted);
+    }
+}

core/src/main/java/io/kestra/core/models/assets/External.java

@@ -0,0 +1,32 @@
+package io.kestra.core.models.assets;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import io.kestra.core.models.annotations.Plugin;
+import io.swagger.v3.oas.annotations.Hidden;
+import lombok.Builder;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+import java.util.Map;
+
+@NoArgsConstructor
+@Plugin
+public class External extends Asset {
+    public static final String ASSET_TYPE = External.class.getName();
+
+    @Builder
+    @JsonCreator
+    public External(
+        String tenantId,
+        String namespace,
+        String id,
+        String displayName,
+        String description,
+        Map<String, Object> metadata,
+        Instant created,
+        Instant updated,
+        boolean deleted
+    ) {
+        super(tenantId, namespace, id, ASSET_TYPE, displayName, description, metadata, created, updated, deleted);
+    }
+}

core/src/main/java/io/kestra/core/models/executions/TaskRun.java

@@ -2,6 +2,7 @@ package io.kestra.core.models.executions;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.models.TenantInterface;
+import io.kestra.core.models.assets.AssetsInOut;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
@@ -57,6 +58,10 @@ public class TaskRun implements TenantInterface {
     @Schema(implementation = Object.class)
     Variables outputs;
 
+    @With
+    @Nullable
+    AssetsInOut assets;
+
     @NotNull
     State state;
 
@@ -87,6 +92,7 @@ public class TaskRun implements TenantInterface {
             this.value,
             this.attempts,
             this.outputs,
+            this.assets,
             this.state.withState(state),
             this.iteration,
             this.dynamic,
@@ -114,6 +120,7 @@ public class TaskRun implements TenantInterface {
             this.value,
             newAttempts,
             this.outputs,
+            this.assets,
             this.state.withState(state),
             this.iteration,
             this.dynamic,
@@ -137,6 +144,7 @@ public class TaskRun implements TenantInterface {
             this.value,
             newAttempts,
             this.outputs,
+            this.assets,
             this.state.withState(State.Type.FAILED),
             this.iteration,
             this.dynamic,
@@ -156,6 +164,7 @@ public class TaskRun implements TenantInterface {
             .value(this.getValue())
             .attempts(this.getAttempts())
             .outputs(this.getOutputs())
+            .assets(this.getAssets())
             .state(state == null ? this.getState() : state)
             .iteration(this.getIteration())
             .build();
@@ -238,6 +247,7 @@ public class TaskRun implements TenantInterface {
             ", parentTaskRunId=" + this.getParentTaskRunId() +
             ", state=" + this.getState().getCurrent().toString() +
             ", outputs=" + this.getOutputs() +
+            ", assets=" + this.getAssets() +
             ", attempts=" + this.getAttempts() +
             ")";
     }

core/src/main/java/io/kestra/core/models/tasks/Task.java

@@ -5,11 +5,13 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.assets.AssetsDeclaration;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.core.flow.WorkingDirectory;
+import jakarta.annotation.Nullable;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.Size;
 import lombok.Builder;
@@ -78,6 +80,11 @@ abstract public class Task implements TaskInterface {
     @Valid
     private Cache taskCache;
 
+    @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
+    @Valid
+    @Nullable
+    private Property<AssetsDeclaration> assets;
+
     public Optional<Task> findById(String id) {
         if (this.getId().equals(id)) {
             return Optional.of(this);

core/src/main/java/io/kestra/core/models/tasks/runners/TaskLogLineMatcher.java

@@ -1,10 +1,13 @@
 package io.kestra.core.models.tasks.runners;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.executions.AbstractMetricEntry;
+import io.kestra.core.queues.QueueException;
+import io.kestra.core.runners.AssetEmitter;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
-import jakarta.inject.Singleton;
 import org.slf4j.Logger;
 import org.slf4j.event.Level;
 import org.slf4j.spi.LoggingEventBuilder;
@@ -18,6 +21,7 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import static io.kestra.core.runners.RunContextLogger.ORIGINAL_TIMESTAMP_KEY;
+import static io.kestra.core.utils.Rethrow.throwConsumer;
 
 /**
  * Service for matching and capturing structured data from task execution logs.
@@ -76,6 +80,18 @@ public class TaskLogLineMatcher {
                 }
             });
         }
+
+        if (match.assets() != null) {
+            try {
+                AssetEmitter assetEmitter = runContext.assets();
+                match.assets().forEach(throwConsumer(assetEmitter::upsert));
+            } catch (IllegalVariableEvaluationException e) {
+                logger.warn("Unable to get asset emitter for log '{}'", data, e);
+            } catch (QueueException e) {
+                logger.warn("Unable to emit asset for log '{}'", data, e);
+            }
+        }
+
         return match;
     }
 
@@ -94,8 +110,9 @@ public class TaskLogLineMatcher {
     public record TaskLogMatch(
         Map<String, Object> outputs,
         List<AbstractMetricEntry<?>> metrics,
-        List<LogLine> logs
-    ) {
+        List<LogLine> logs,
+        List<Asset> assets
+        ) {
         @Override
         public Map<String, Object> outputs() {
             return Optional.ofNullable(outputs).orElse(Map.of());

core/src/main/java/io/kestra/core/models/triggers/AbstractTrigger.java

@@ -6,8 +6,10 @@ import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.assets.AssetsDeclaration;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
 import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
@@ -88,6 +90,9 @@ abstract public class AbstractTrigger implements TriggerInterface {
     )
     private boolean allowConcurrent = false;
 
+    @PluginProperty(hidden = true, group = PluginProperty.CORE_GROUP)
+    private Property<AssetsDeclaration> assets;
+
     /**
      * For backward compatibility: we rename minLogLevel to logLevel.
      * @deprecated use {@link #logLevel} instead

core/src/main/java/io/kestra/core/plugins/DefaultPluginRegistry.java

@@ -1,6 +1,7 @@
 package io.kestra.core.plugins;
 
 import io.kestra.core.models.Plugin;
+import io.kestra.core.models.assets.Asset;
 import jakarta.annotation.Nullable;
 import jakarta.validation.constraints.NotNull;
 import org.slf4j.Logger;

core/src/main/java/io/kestra/core/plugins/PluginModule.java

@@ -2,6 +2,7 @@ package io.kestra.core.plugins;
 
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import io.kestra.core.app.AppPluginInterface;
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.dashboards.DataFilter;
 import io.kestra.core.models.dashboards.DataFilterKPI;
@@ -11,6 +12,7 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.logs.LogExporter;
 import io.kestra.core.models.tasks.runners.TaskRunner;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.plugins.serdes.AssetDeserializer;
 import io.kestra.core.plugins.serdes.PluginDeserializer;
 import io.kestra.core.secret.SecretPluginInterface;
 import io.kestra.core.storages.StorageInterface;
@@ -45,5 +47,6 @@ public class PluginModule extends SimpleModule {
         addDeserializer(SecretPluginInterface.class, new PluginDeserializer<>());
         addDeserializer(AppPluginInterface.class, new PluginDeserializer<>());
         addDeserializer(LogExporter.class, new PluginDeserializer<>());
+        addDeserializer(Asset.class, new AssetDeserializer());
     }
 }

core/src/main/java/io/kestra/core/plugins/PluginScanner.java

@@ -3,6 +3,7 @@ package io.kestra.core.plugins;
 import io.kestra.core.app.AppBlockInterface;
 import io.kestra.core.app.AppPluginInterface;
 import io.kestra.core.models.Plugin;
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.dashboards.DataFilter;
 import io.kestra.core.models.dashboards.DataFilterKPI;
@@ -108,6 +109,7 @@ public class PluginScanner {
         List<Class<? extends StorageInterface>> storages = new ArrayList<>();
         List<Class<? extends SecretPluginInterface>> secrets = new ArrayList<>();
         List<Class<? extends TaskRunner<?>>> taskRunners = new ArrayList<>();
+        List<Class<? extends Asset>> assets = new ArrayList<>();
         List<Class<? extends AppPluginInterface>> apps = new ArrayList<>();
         List<Class<? extends AppBlockInterface>> appBlocks = new ArrayList<>();
         List<Class<? extends Chart<?>>> charts = new ArrayList<>();
@@ -155,6 +157,10 @@ public class PluginScanner {
                         //noinspection unchecked
                         taskRunners.add((Class<? extends TaskRunner<?>>) runner.getClass());
                     }
+                    case Asset asset -> {
+                        log.debug("Loading Asset plugin: '{}'", plugin.getClass());
+                        assets.add(asset.getClass());
+                    }
                     case AppPluginInterface app -> {
                         log.debug("Loading App plugin: '{}'", plugin.getClass());
                         apps.add(app.getClass());
@@ -223,6 +229,7 @@ public class PluginScanner {
             .conditions(conditions)
             .storages(storages)
             .secrets(secrets)
+            .assets(assets)
             .apps(apps)
             .appBlocks(appBlocks)
             .taskRunners(taskRunners)

core/src/main/java/io/kestra/core/plugins/RegisteredPlugin.java

@@ -3,6 +3,7 @@ package io.kestra.core.plugins;
 import io.kestra.core.app.AppBlockInterface;
 import io.kestra.core.app.AppPluginInterface;
 import io.kestra.core.models.annotations.PluginSubGroup;
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.dashboards.DataFilter;
 import io.kestra.core.models.dashboards.DataFilterKPI;
@@ -39,6 +40,7 @@ public class RegisteredPlugin {
     public static final String STORAGES_GROUP_NAME = "storages";
     public static final String SECRETS_GROUP_NAME = "secrets";
     public static final String TASK_RUNNERS_GROUP_NAME = "task-runners";
+    public static final String ASSETS_GROUP_NAME = "assets";
     public static final String APPS_GROUP_NAME = "apps";
     public static final String APP_BLOCKS_GROUP_NAME = "app-blocks";
     public static final String CHARTS_GROUP_NAME = "charts";
@@ -56,6 +58,7 @@ public class RegisteredPlugin {
     private final List<Class<? extends StorageInterface>> storages;
     private final List<Class<? extends SecretPluginInterface>> secrets;
     private final List<Class<? extends TaskRunner<?>>> taskRunners;
+    private final List<Class<? extends Asset>> assets;
     private final List<Class<? extends AppPluginInterface>> apps;
     private final List<Class<? extends AppBlockInterface>> appBlocks;
     private final List<Class<? extends Chart<?>>> charts;
@@ -74,6 +77,7 @@ public class RegisteredPlugin {
             !storages.isEmpty() ||
             !secrets.isEmpty() ||
             !taskRunners.isEmpty() ||
+            !assets.isEmpty()  ||
             !apps.isEmpty()  ||
             !appBlocks.isEmpty() ||
             !charts.isEmpty() ||
@@ -145,6 +149,10 @@ public class RegisteredPlugin {
             return AppPluginInterface.class;
         }
 
+        if (this.getAssets().stream().anyMatch(r -> r.getName().equals(cls))) {
+            return Asset.class;
+        }
+
         if (this.getLogExporters().stream().anyMatch(r -> r.getName().equals(cls))) {
             return LogExporter.class;
         }
@@ -180,6 +188,7 @@ public class RegisteredPlugin {
         result.put(STORAGES_GROUP_NAME, Arrays.asList(this.getStorages().toArray(Class[]::new)));
         result.put(SECRETS_GROUP_NAME, Arrays.asList(this.getSecrets().toArray(Class[]::new)));
         result.put(TASK_RUNNERS_GROUP_NAME, Arrays.asList(this.getTaskRunners().toArray(Class[]::new)));
+        result.put(ASSETS_GROUP_NAME, Arrays.asList(this.getAssets().toArray(Class[]::new)));
         result.put(APPS_GROUP_NAME, Arrays.asList(this.getApps().toArray(Class[]::new)));
         result.put(APP_BLOCKS_GROUP_NAME, Arrays.asList(this.getAppBlocks().toArray(Class[]::new)));
         result.put(CHARTS_GROUP_NAME, Arrays.asList(this.getCharts().toArray(Class[]::new)));
@@ -359,6 +368,12 @@ public class RegisteredPlugin {
             b.append("] ");
         }
 
+        if (!this.getAssets().isEmpty()) {
+            b.append("[Assets: ");
+            b.append(this.getAssets().stream().map(Class::getName).collect(Collectors.joining(", ")));
+            b.append("] ");
+        }
+
         if (!this.getApps().isEmpty()) {
             b.append("[Apps: ");
             b.append(this.getApps().stream().map(Class::getName).collect(Collectors.joining(", ")));

core/src/main/java/io/kestra/core/plugins/serdes/AssetDeserializer.java

@@ -0,0 +1,16 @@
+package io.kestra.core.plugins.serdes;
+
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import io.kestra.core.models.Plugin;
+import io.kestra.core.models.assets.Asset;
+import io.kestra.core.models.assets.Custom;
+
+/**
+ * Specific {@link JsonDeserializer} for deserializing {@link Asset}.
+ */
+public final class AssetDeserializer extends PluginDeserializer<Asset> {
+    @Override
+    protected Class<? extends Plugin> fallbackClass() {
+        return Custom.class;
+    }
+}

core/src/main/java/io/kestra/core/plugins/serdes/PluginDeserializer.java

@@ -29,7 +29,7 @@ import java.util.Optional;
  * The {@link PluginDeserializer} uses the {@link PluginRegistry} to found the plugin class corresponding to
  * a plugin type.
  */
-public final class PluginDeserializer<T extends Plugin> extends JsonDeserializer<T> {
+public class PluginDeserializer<T extends Plugin> extends JsonDeserializer<T> {
 
     private static final Logger log = LoggerFactory.getLogger(PluginDeserializer.class);
 
@@ -93,6 +93,10 @@ public final class PluginDeserializer<T extends Plugin> extends JsonDeserializer
                 identifier
             );
             pluginType = pluginRegistry.findClassByIdentifier(identifier);
+
+            if (pluginType == null) {
+                pluginType = fallbackClass();
+            }
         }
 
         if (pluginType == null) {
@@ -153,4 +157,8 @@ public final class PluginDeserializer<T extends Plugin> extends JsonDeserializer
 
         return isVersioningSupported && version != null && !version.isEmpty() ? type + ":" + version : type;
     }
+
+    protected Class<? extends Plugin> fallbackClass() {
+        return null;
+    }
 }

core/src/main/java/io/kestra/core/runners/AssetEmitter.java

@@ -0,0 +1,12 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.assets.Asset;
+import io.kestra.core.queues.QueueException;
+
+import java.util.List;
+
+public interface AssetEmitter {
+    void upsert(Asset asset) throws QueueException;
+
+    List<Asset> outputs();
+}

core/src/main/java/io/kestra/core/runners/DefaultRunContext.java

@@ -6,11 +6,13 @@ import com.google.common.base.CaseFormat;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.assets.AssetsDeclaration;
 import io.kestra.core.models.Plugin;
 import io.kestra.core.models.executions.AbstractMetricEntry;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.assets.AssetManagerFactory;
 import io.kestra.core.plugins.PluginConfigurations;
 import io.kestra.core.services.KVStoreService;
 import io.kestra.core.storages.Storage;
@@ -54,6 +56,7 @@ public class DefaultRunContext extends RunContext {
     private MetricRegistry meterRegistry;
     private VersionProvider version;
     private KVStoreService kvStoreService;
+    private AssetManagerFactory assetManagerFactory;
     private Optional<String> secretKey;
     private WorkingDir workingDir;
     private Validator validator;
@@ -73,6 +76,8 @@ public class DefaultRunContext extends RunContext {
     private Task task;
     private AbstractTrigger trigger;
 
+    private volatile AssetEmitter assetEmitter;
+
     private final AtomicBoolean isInitialized = new AtomicBoolean(false);
 
 
@@ -161,6 +166,7 @@ public class DefaultRunContext extends RunContext {
             this.secretKey = applicationContext.getProperty("kestra.encryption.secret-key", String.class);
             this.validator = applicationContext.getBean(Validator.class);
             this.localPath = applicationContext.getBean(LocalPathFactory.class).createLocalPath(this);
+            this.assetManagerFactory = applicationContext.getBean(AssetManagerFactory.class);
         }
     }
 
@@ -537,6 +543,23 @@ public class DefaultRunContext extends RunContext {
         return flow != null ? flow.get("tenantId") : null;
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public TaskRunInfo taskRunInfo() {
+        Optional<Map<String, Object>> maybeTaskRunMap = Optional.ofNullable(this.getVariables().get("taskrun"))
+            .map(Map.class::cast);
+        return new TaskRunInfo(
+            (String) this.getVariables().get("executionId"),
+            (String) this.getVariables().get("taskId"),
+            maybeTaskRunMap.map(m -> (String) m.get("id"))
+                .orElse(null),
+            maybeTaskRunMap.map(m -> (String) m.get("value"))
+                .orElse(null)
+        );
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -545,12 +568,7 @@ public class DefaultRunContext extends RunContext {
     public FlowInfo flowInfo() {
         Map<String, Object> flow = (Map<String, Object>) this.getVariables().get("flow");
         // normally only tests should not have the flow variable
-        return flow == null ? new FlowInfo(null, null, null, null) : new FlowInfo(
-            (String) flow.get("tenantId"),
-            (String) flow.get("namespace"),
-            (String) flow.get("id"),
-            (Integer) flow.get("revision")
-        );
+        return flow == null ? new FlowInfo(null, null, null, null) : FlowInfo.from(flow);
     }
 
     /**
@@ -594,6 +612,25 @@ public class DefaultRunContext extends RunContext {
         return new AclCheckerImpl(this.applicationContext, flowInfo());
     }
 
+    @Override
+    public AssetEmitter assets() throws IllegalVariableEvaluationException {
+        if (this.assetEmitter == null) {
+            synchronized (this) {
+                if (this.assetEmitter == null) {
+                    this.assetEmitter = assetManagerFactory.of(
+                        Optional.ofNullable(task).map(Task::getAssets)
+                            .or(() -> Optional.ofNullable(trigger).map(AbstractTrigger::getAssets))
+                            .flatMap(throwFunction(asset -> this.render(asset).as(AssetsDeclaration.class)))
+                            .map(AssetsDeclaration::isEnableAuto)
+                            .orElse(false)
+                    );
+                }
+            }
+        }
+
+        return this.assetEmitter;
+    }
+
     @Override
     public LocalPath localPath() {
         return localPath;

core/src/main/java/io/kestra/core/runners/RunContext.java

@@ -143,6 +143,8 @@ public abstract class RunContext implements PropertyContext {
     @Deprecated(forRemoval = true)
     public abstract String tenantId();
 
+    public abstract TaskRunInfo taskRunInfo();
+
     public abstract FlowInfo flowInfo();
 
     /**
@@ -190,7 +192,19 @@ public abstract class RunContext implements PropertyContext {
      */
     public abstract LocalPath localPath();
 
+    public record TaskRunInfo(String executionId, String taskId, String taskRunId, Object value) {
+
+    }
+
     public record FlowInfo(String tenantId, String namespace, String id, Integer revision) {
+        public static FlowInfo from(Map<String, Object> flowInfoMap) {
+            return new FlowInfo(
+                (String) flowInfoMap.get("tenantId"),
+                (String) flowInfoMap.get("namespace"),
+                (String) flowInfoMap.get("id"),
+                (Integer) flowInfoMap.get("revision")
+            );
+        }
     }
 
     /**
@@ -206,6 +220,11 @@ public abstract class RunContext implements PropertyContext {
      */
     public abstract AclChecker acl();
 
+    /**
+     * Get access to the Assets handler.
+     */
+    public abstract AssetEmitter assets() throws IllegalVariableEvaluationException;
+
     /**
      * Clone this run context for a specific plugin.
      * @return a new run context with the plugin configuration of the given plugin.

core/src/main/java/io/kestra/core/runners/WorkerTaskResult.java

@@ -21,8 +21,7 @@ public class WorkerTaskResult implements HasUID {
     List<TaskRun> dynamicTaskRuns;
 
     public WorkerTaskResult(TaskRun taskRun) {
-        this.taskRun = taskRun;
-        this.dynamicTaskRuns = new ArrayList<>();
+        this(taskRun, new ArrayList<>(1)); // there are usually very few dynamic task runs, so we init the list with a capacity of 1
     }
 
     /**

core/src/main/java/io/kestra/core/test/flow/TaskFixture.java

@@ -1,5 +1,6 @@
 package io.kestra.core.test.flow;
 
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.property.Property;
 import jakarta.validation.constraints.NotNull;
@@ -8,6 +9,7 @@ import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 
+import java.util.List;
 import java.util.Map;
 
 @Getter
@@ -25,5 +27,7 @@ public class TaskFixture {
 
     private Map<String, Object> outputs;
 
+    private List<Asset> assets;
+
     private Property<String> description;
 }

core/src/main/java/io/kestra/core/validations/validator/FlowValidator.java

@@ -19,15 +19,7 @@ import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 
 import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -155,6 +147,8 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
             .map(task -> task.getId())
             .collect(Collectors.toList());
 
+        violations.addAll(assetsViolations(allTasks));
+
         if (!invalidTasks.isEmpty()) {
             violations.add("Invalid output reference: use outputs[key-name] instead of outputs.key-name — keys with dashes require bracket notation, offending tasks:" +
                 " [" + String.join(", ", invalidTasks) + "]");
@@ -181,6 +175,12 @@ public class FlowValidator implements ConstraintValidator<FlowValidation, Flow>
         }
     }
 
+    protected List<String> assetsViolations(List<Task> allTasks) {
+        return allTasks.stream().filter(task -> task.getAssets() != null)
+            .map(taskWithAssets -> "Task '" + taskWithAssets.getId() + "' can't have any `assets` because assets are only available in Enterprise Edition.")
+            .toList();
+    }
+
     private static boolean checkObjectFieldsWithPatterns(Object object, List<Pattern> patterns) {
         if (object == null) {
             return true;

core/src/test/java/io/kestra/core/docs/ClassPluginDocumentationTest.java

@@ -103,7 +103,7 @@ class ClassPluginDocumentationTest {
             PluginClassAndMetadata<AbstractTrigger> metadata = PluginClassAndMetadata.create(scan, Schedule.class, AbstractTrigger.class, null);
             ClassPluginDocumentation<? extends AbstractTrigger> doc = ClassPluginDocumentation.of(jsonSchemaGenerator, metadata, scan.version(), true);
 
-            assertThat(doc.getDefs().size()).isEqualTo(20);
+            assertThat(doc.getDefs()).hasSize(23);
             assertThat(doc.getDocLicense()).isNull();
 
             assertThat(((Map<String, Object>) doc.getDefs().get("io.kestra.core.models.tasks.WorkerGroup")).get("type")).isEqualTo("object");
@@ -142,9 +142,9 @@ class ClassPluginDocumentationTest {
             ClassPluginDocumentation<? extends DynamicPropertyExampleTask> doc = ClassPluginDocumentation.of(jsonSchemaGenerator, metadata, scan.version(), true);
 
             assertThat(doc.getCls()).isEqualTo("io.kestra.core.models.property.DynamicPropertyExampleTask");
-            assertThat(doc.getDefs()).hasSize(6);
+            assertThat(doc.getDefs()).hasSize(9);
             Map<String, Object> properties = (Map<String, Object>) doc.getPropertiesSchema().get("properties");
-            assertThat(properties).hasSize(21);
+            assertThat(properties).hasSize(22);
 
             Map<String, Object> number = (Map<String, Object>) properties.get("number");
             assertThat(number.get("anyOf")).isNotNull();

core/src/test/java/io/kestra/core/docs/JsonSchemaGeneratorTest.java

@@ -47,7 +47,6 @@ import static org.hamcrest.Matchers.*;
 @KestraTest
 class JsonSchemaGeneratorTest {
 
-
     @Inject
     JsonSchemaGenerator jsonSchemaGenerator;
 
@@ -346,7 +345,7 @@ class JsonSchemaGeneratorTest {
     void pluginSchemaShouldNotResolveTaskAndTriggerSubtypes() {
         Map<String, Object> generate = jsonSchemaGenerator.properties(null, TaskWithSubTaskAndSubTrigger.class);
         var definitions = (Map<String, Map<String, Object>>) generate.get("$defs");
-        assertThat(definitions.size(), is(27));
+        assertThat(definitions.size(), is(30));
     }
 
     @SuppressWarnings("unchecked")

core/src/test/java/io/kestra/core/plugins/AdditionalPluginTest.java

@@ -15,13 +15,11 @@ import jakarta.inject.Inject;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
-import org.hamcrest.MatcherAssert;
 import org.junit.jupiter.api.Test;
 
 import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.hamcrest.Matchers.is;
 
 @KestraTest(startRunner = true)
 class AdditionalPluginTest {
@@ -43,7 +41,7 @@ class AdditionalPluginTest {
     void shouldResolveAdditionalPluginSubtypes() {
         Map<String, Object> generate = jsonSchemaGenerator.properties(null, AdditionalPluginTest.AdditionalPluginTestTask.class);
         var definitions = (Map<String, Map<String, Object>>) generate.get("$defs");
-        assertThat(definitions).hasSize(7);
+        assertThat(definitions).hasSize(10);
         assertThat(definitions).containsKey("io.kestra.core.plugins.AdditionalPluginTest-AdditionalPluginTest1");
         assertThat(definitions).containsKey("io.kestra.core.plugins.AdditionalPluginTest-AdditionalPluginTest2");
     }

core/src/test/java/io/kestra/core/plugins/serdes/PluginDeserializerTest.java

@@ -95,4 +95,4 @@ class PluginDeserializerTest {
 
     public record TestPlugin(String type) implements Plugin {
     }
-}
+}

core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java

@@ -558,4 +558,4 @@ public abstract class AbstractRunnerTest {
     public void shouldCallTasksAfterListener(Execution execution) {
         afterExecutionTestCase.shouldCallTasksAfterListener(execution);
     }
-}
+}

core/src/test/java/io/kestra/core/runners/test/AssetEmitter.java

@@ -0,0 +1,29 @@
+package io.kestra.core.runners.test;
+
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.assets.Asset;
+import io.kestra.core.models.tasks.*;
+import io.kestra.core.runners.RunContext;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Plugin
+public class AssetEmitter extends Task implements RunnableTask<VoidOutput> {
+    @NotNull
+    @PluginProperty
+    private Asset assetToEmit;
+
+
+    @Override
+    public VoidOutput run(RunContext runContext) throws Exception {
+        runContext.assets().upsert(assetToEmit);
+        return null;
+    }
+}

core/src/test/java/io/kestra/core/validations/FlowValidationTest.java

@@ -1,5 +1,7 @@
 package io.kestra.core.validations;
 
+import io.kestra.core.models.assets.AssetIdentifier;
+import io.kestra.core.models.assets.AssetsDeclaration;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.GenericFlow;
 import io.kestra.core.models.validations.ModelValidator;
@@ -7,7 +9,9 @@ import io.kestra.core.serializers.YamlParser;
 import io.kestra.core.tenant.TenantService;
 import io.kestra.core.utils.TestsUtils;
 import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.plugin.core.log.Log;
 import jakarta.inject.Inject;
+import jakarta.validation.ConstraintViolation;
 import org.junit.jupiter.api.Test;
 import io.kestra.core.models.validations.ValidateConstraintViolation;
 import io.kestra.core.services.FlowService;
@@ -229,6 +233,31 @@ class FlowValidationTest {
         assertThat(validate.get().getMessage()).contains("Duplicate preconditions with id [flows]");
     }
 
+    @Test
+    void eeAllowsDefiningAssets() {
+        Flow flow = Flow.builder()
+            .id(TestsUtils.randomString())
+            .namespace(TestsUtils.randomNamespace())
+            .tasks(List.of(
+                Log.builder()
+                    .id("log")
+                    .type(Log.class.getName())
+                    .message("any")
+                    .assets(io.kestra.core.models.property.Property.ofValue(
+                        new AssetsDeclaration(true, List.of(new AssetIdentifier(null, null, "anyId")), null))
+                    )
+                    .build()
+            ))
+            .build();
+
+        Optional<ConstraintViolationException> violations = modelValidator.isValid(flow);
+
+        assertThat(violations.isPresent()).isEqualTo(true);
+        assertThat(violations.get().getConstraintViolations().stream().map(ConstraintViolation::getMessage)).satisfiesExactly(
+            message -> assertThat(message).contains("Task 'log' can't have any `assets` because assets are only available in Enterprise Edition.")
+        );
+    };
+
     private Flow parse(String path) {
         URL resource = TestsUtils.class.getClassLoader().getResource(path);
         assert resource != null;
@@ -237,4 +266,4 @@ class FlowValidationTest {
 
         return YamlParser.parse(file, Flow.class);
     }
-}
+}

executor/src/main/java/io/kestra/executor/ExecutorService.java

@@ -1,9 +1,14 @@
 package io.kestra.executor;
 
+import io.kestra.core.assets.AssetService;
 import io.kestra.core.debug.Breakpoint;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.Label;
+import io.kestra.core.models.assets.AssetIdentifier;
+import io.kestra.core.models.assets.AssetUser;
+import io.kestra.core.models.assets.AssetsDeclaration;
+import io.kestra.core.models.assets.AssetsInOut;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.FlowWithSource;
@@ -96,6 +101,12 @@ public class ExecutorService {
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     private QueueInterface<LogEntry> logQueue;
 
+    @Inject
+    private AssetService assetService;
+
+    @Inject
+    private RunContextInitializer runContextInitializer;
+
     protected FlowMetaStoreInterface flowExecutorInterface() {
         // bean is injected late, so we need to wait
         if (this.flowExecutorInterface == null) {
@@ -896,21 +907,35 @@ public class ExecutorService {
         boolean hasMockedWorkerTask = false;
         record FixtureAndTaskRun(TaskFixture fixture, TaskRun taskRun) {}
         if (executor.getExecution().getFixtures() != null) {
-            RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
+            RunContext runContext = runContextInitializer.forExecutor((DefaultRunContext) runContextFactory.of(
+                executor.getFlow(),
+                executor.getExecution()
+            ));
             List<WorkerTaskResult> workerTaskResults = executor.getExecution()
                 .getTaskRunList()
                 .stream()
                 .filter(taskRun -> taskRun.getState().getCurrent().isCreated())
                 .flatMap(taskRun -> executor.getExecution().getFixtureForTaskRun(taskRun).stream().map(fixture -> new FixtureAndTaskRun(fixture, taskRun)))
-                .map(throwFunction(fixtureAndTaskRun -> WorkerTaskResult.builder()
-                    .taskRun(fixtureAndTaskRun.taskRun()
-                        .withState(Optional.ofNullable(fixtureAndTaskRun.fixture().getState()).orElse(State.Type.SUCCESS))
-                        .withOutputs(
-                            variablesService.of(StorageContext.forTask(fixtureAndTaskRun.taskRun),
-                                fixtureAndTaskRun.fixture().getOutputs() == null ? null : runContext.render(fixtureAndTaskRun.fixture().getOutputs()))
-                        )
-                    )
-                    .build()
+                .map(throwFunction(fixtureAndTaskRun -> {
+                    Optional<AssetsDeclaration> renderedAssetsDeclaration = runContext.render(executor.getFlow().findTaskByTaskId(fixtureAndTaskRun.taskRun.getTaskId()).getAssets()).as(AssetsDeclaration.class);
+                        return WorkerTaskResult.builder()
+                            .taskRun(fixtureAndTaskRun.taskRun()
+                                .withState(Optional.ofNullable(fixtureAndTaskRun.fixture().getState()).orElse(State.Type.SUCCESS))
+                                .withOutputs(
+                                    variablesService.of(StorageContext.forTask(fixtureAndTaskRun.taskRun),
+                                        fixtureAndTaskRun.fixture().getOutputs() == null ? null : runContext.render(fixtureAndTaskRun.fixture().getOutputs()))
+                                )
+                                .withAssets(new AssetsInOut(
+                                    renderedAssetsDeclaration.map(AssetsDeclaration::getInputs).orElse(Collections.emptyList()).stream()
+                                        .map(assetIdentifier -> assetIdentifier.withTenantId(executor.getFlow().getTenantId()))
+                                        .toList(),
+                                    fixtureAndTaskRun.fixture().getAssets() == null ? null : fixtureAndTaskRun.fixture().getAssets().stream()
+                                        .map(asset -> asset.withTenantId(executor.getFlow().getTenantId()))
+                                        .toList()
+                                ))
+                            )
+                            .build();
+                    }
                 ))
                 .toList();
 
@@ -1172,6 +1197,47 @@ public class ExecutorService {
                     metricRegistry.tags(workerTaskResult)
                 )
                 .record(taskRun.getState().getDurationOrComputeIt());
+
+            if (
+                !taskRun.getState().isFailed()
+                    && taskRun.getAssets() != null &&
+                    (!taskRun.getAssets().getInputs().isEmpty() || !taskRun.getAssets().getOutputs().isEmpty())
+            ) {
+                AssetUser assetUser = new AssetUser(
+                    taskRun.getTenantId(),
+                    taskRun.getNamespace(),
+                    taskRun.getFlowId(),
+                    newExecution.getFlowRevision(),
+                    taskRun.getExecutionId(),
+                    taskRun.getTaskId(),
+                    taskRun.getId()
+                );
+
+                List<AssetIdentifier> outputIdentifiers = taskRun.getAssets().getOutputs().stream()
+                    .map(asset -> asset.withTenantId(taskRun.getTenantId()))
+                    .map(AssetIdentifier::of)
+                    .toList();
+                List<AssetIdentifier> inputAssets = taskRun.getAssets().getInputs().stream()
+                    .map(assetIdentifier -> assetIdentifier.withTenantId(taskRun.getTenantId()))
+                    .toList();
+                try {
+                    assetService.assetLineage(
+                        assetUser,
+                        inputAssets,
+                        outputIdentifiers
+                    );
+                } catch (QueueException e) {
+                    log.warn("Unable to submit asset lineage event for {} -> {}", inputAssets, outputIdentifiers, e);
+                }
+
+                taskRun.getAssets().getOutputs().forEach(asset -> {
+                    try {
+                        assetService.asyncUpsert(assetUser, asset);
+                    } catch (QueueException e) {
+                        log.warn("Unable to submit asset upsert event for asset {}", asset.getId(), e);
+                    }
+                });
+            }
         }
     }
 

jdbc-h2/src/main/resources/migrations/h2/V1_52__assets_queues.sql

@@ -0,0 +1,22 @@
+ALTER TABLE queues ALTER COLUMN "type" ENUM(
+    'io.kestra.core.models.executions.Execution',
+    'io.kestra.core.models.templates.Template',
+    'io.kestra.core.models.executions.ExecutionKilled',
+    'io.kestra.core.runners.WorkerJob',
+    'io.kestra.core.runners.WorkerTaskResult',
+    'io.kestra.core.runners.WorkerInstance',
+    'io.kestra.core.runners.WorkerTaskRunning',
+    'io.kestra.core.models.executions.LogEntry',
+    'io.kestra.core.models.triggers.Trigger',
+    'io.kestra.ee.models.audits.AuditLog',
+    'io.kestra.core.models.executions.MetricEntry',
+    'io.kestra.core.runners.WorkerTriggerResult',
+    'io.kestra.core.runners.SubflowExecutionResult',
+    'io.kestra.core.server.ClusterEvent',
+    'io.kestra.core.runners.SubflowExecutionEnd',
+    'io.kestra.core.models.flows.FlowInterface',
+    'io.kestra.core.runners.MultipleConditionEvent',
+    'io.kestra.ee.assets.AssetLineageEvent',
+    'io.kestra.ee.assets.AssetUpsertCommand',
+    'io.kestra.ee.assets.AssetStateEvent'
+) NOT NULL

jdbc-mysql/src/main/resources/migrations/mysql/V1_53__assets_queues.sql

@@ -0,0 +1,22 @@
+ALTER TABLE queues MODIFY COLUMN `type` ENUM(
+    'io.kestra.core.models.executions.Execution',
+    'io.kestra.core.models.templates.Template',
+    'io.kestra.core.models.executions.ExecutionKilled',
+    'io.kestra.core.runners.WorkerJob',
+    'io.kestra.core.runners.WorkerTaskResult',
+    'io.kestra.core.runners.WorkerInstance',
+    'io.kestra.core.runners.WorkerTaskRunning',
+    'io.kestra.core.models.executions.LogEntry',
+    'io.kestra.core.models.triggers.Trigger',
+    'io.kestra.ee.models.audits.AuditLog',
+    'io.kestra.core.models.executions.MetricEntry',
+    'io.kestra.core.runners.WorkerTriggerResult',
+    'io.kestra.core.runners.SubflowExecutionResult',
+    'io.kestra.core.server.ClusterEvent',
+    'io.kestra.core.runners.SubflowExecutionEnd',
+    'io.kestra.core.models.flows.FlowInterface',
+    'io.kestra.core.runners.MultipleConditionEvent',
+    'io.kestra.ee.assets.AssetLineageEvent',
+    'io.kestra.ee.assets.AssetUpsertCommand',
+    'io.kestra.ee.assets.AssetStateEvent'
+    ) NOT NULL;

jdbc-postgres/src/main/resources/migrations/postgres/V1_52__assets_queues.sql

@@ -0,0 +1,3 @@
+ALTER TYPE queue_type ADD VALUE IF NOT EXISTS 'io.kestra.ee.assets.AssetLineageEvent';
+ALTER TYPE queue_type ADD VALUE IF NOT EXISTS 'io.kestra.ee.assets.AssetUpsertCommand';
+ALTER TYPE queue_type ADD VALUE IF NOT EXISTS 'io.kestra.ee.assets.AssetStateEvent';

jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcRepository.java

@@ -328,6 +328,10 @@ public abstract class AbstractJdbcRepository {
             return applyTriggerStateCondition(value, operation);
         }
 
+        if (field.equals(QueryFilter.Field.METADATA)) {
+            return findMetadataCondition((Map<?, ?>) value, operation);
+        }
+
         // Convert the field name to lowercase and quote it
         Name columnName = getColumnName(field);
 
@@ -380,6 +384,10 @@ public abstract class AbstractJdbcRepository {
         throw new InvalidQueryFiltersException("Unsupported operation: " + operation);
     }
 
+    protected Condition findMetadataCondition(Map<?, ?> metadata, QueryFilter.Op operation) {
+        throw new InvalidQueryFiltersException("Unsupported operation: " + operation);
+    }
+
     // Generate the condition for Field.STATE
     @SuppressWarnings("unchecked")
     private Condition generateStateCondition(Object value, QueryFilter.Op operation) {

tests/src/main/java/io/kestra/core/utils/TestsUtils.java

@@ -74,7 +74,7 @@ abstract public class TestsUtils {
      * @param prefix
      * @return
      */
-    private static String randomString(String... prefix) {
+    public static String randomString(String... prefix) {
         if (prefix.length == 0) {
             prefix = new String[]{String.join("-", stackTraceToParts())};
         }

ui/src/components/dashboard/components/selector/Selector.vue

@@ -1,7 +1,7 @@
 <template>
     <el-dropdown trigger="click" hideOnClick placement="bottom-end">
-        <el-button :icon="Menu" class="selected">
-            <span class="text-truncate">
+        <el-button :icon="ChartLineVariant" class="selected">
+            <span v-if="!verticalLayout" class="text-truncate">
                 {{ selected ?? t("dashboards.default") }}
             </span>
         </el-button>
@@ -75,7 +75,10 @@
 
     import Item from "./Item.vue";
 
-    import Menu from "vue-material-design-icons/Menu.vue";
+    import {useBreakpoints, breakpointsElement} from "@vueuse/core";
+    const verticalLayout = useBreakpoints(breakpointsElement).smallerOrEqual("sm");
+
+    import ChartLineVariant from "vue-material-design-icons/ChartLineVariant.vue";
     import Plus from "vue-material-design-icons/Plus.vue";
     import Magnify from "vue-material-design-icons/Magnify.vue";
 

ui/src/components/demo/Assets.vue

@@ -0,0 +1,34 @@
+<template>
+    <TopNavBar :title="routeInfo.title" />
+    <Layout
+        :title="t(`demos.assets.title`)"
+        :image="{
+            source: img,
+            alt: t(`demos.assets.title`)
+        }"
+        :video="{
+            //TODO: replace with ASSET video
+            source: 'https://www.youtube.com/embed/jMZ9Cs3xxpo',
+        }"
+    >
+        <template #message>
+            {{ $t(`demos.assets.message`) }}
+        </template>
+    </Layout>
+</template>
+
+<script setup lang="ts">
+    import {computed} from "vue";
+    import {useI18n} from "vue-i18n";
+    import img from "../../assets/demo/assets.png";
+    import useRouteContext from "../../composables/useRouteContext";
+    
+    import Layout from "./Layout.vue";
+    import TopNavBar from "../../components/layout/TopNavBar.vue";
+
+    const {t} = useI18n();
+
+    const routeInfo = computed(() => ({title: t("demos.assets.header")}));
+
+    useRouteContext(routeInfo);
+</script>

ui/src/components/demo/Layout.vue

@@ -105,7 +105,7 @@
             position: relative;
             background: $base-gray-200;
             padding: .125rem 0.5rem;
-            border-radius: $border-radius;
+            border-radius: 1rem;
             display: inline-block;
             z-index: 2;
             margin: 0 auto;
@@ -175,6 +175,7 @@
             line-height: 16px;
             font-size: 11px;
             text-align: left;
+            color: var(--ks-content-secondary);
         }
 
         .video-container {
@@ -261,7 +262,7 @@
             }
 
             p {
-                font-size: 14px;
+                font-size: 1rem;
                 line-height: 22px;
             }
         }

ui/src/components/dependencies/Dependencies.vue

@@ -42,6 +42,7 @@
                 :elements="getElements()"
                 @select="selectNode"
                 :selected="selectedNodeID"
+                :subtype="SUBTYPE"
             />
         </el-splitter-panel>
     </el-splitter>
@@ -54,7 +55,7 @@
     import Empty from "../layout/empty/Empty.vue";
 
     import {useDependencies} from "./composables/useDependencies";
-    import {FLOW, EXECUTION, NAMESPACE} from "./utils/types";
+    import {FLOW, EXECUTION, NAMESPACE, ASSET} from "./utils/types";
 
     const PANEL = {size: "70%", min: "30%", max: "80%"};
 
@@ -66,13 +67,27 @@
     import SelectionRemove from "vue-material-design-icons/SelectionRemove.vue";
     import FitToScreenOutline from "vue-material-design-icons/FitToScreenOutline.vue";
 
-    const SUBTYPE = route.name === "flows/update" ? FLOW : route.name === "namespaces/update" ? NAMESPACE : EXECUTION;
+    const props = defineProps<{
+        fetchAssetDependencies?: () => Promise<{
+            data: any[];
+            count: number;
+        }>;
+    }>();
+
+    const SUBTYPE = route.name === "flows/update" ? FLOW : route.name === "namespaces/update" ? NAMESPACE : route.name === "assets/update" ? ASSET : EXECUTION;
 
     const container = ref(null);
-    const initialNodeID: string = SUBTYPE === FLOW || SUBTYPE === NAMESPACE ? String(route.params.id) : String(route.params.flowId);
+    const initialNodeID: string = SUBTYPE === FLOW || SUBTYPE === NAMESPACE || SUBTYPE === ASSET ? String(route.params.id || route.params.assetId) : String(route.params.flowId);
     const TESTING = false; // When true, bypasses API data fetching and uses mock/test data.
 
-    const {getElements, isLoading, isRendering, selectedNodeID, selectNode, handlers} = useDependencies(container, SUBTYPE, initialNodeID, route.params, TESTING);
+    const {
+        getElements,
+        isLoading,
+        isRendering,
+        selectedNodeID,
+        selectNode,
+        handlers,
+    } = useDependencies(container, SUBTYPE, initialNodeID, route.params, TESTING, props.fetchAssetDependencies);
 </script>
 
 <style scoped lang="scss">
@@ -95,7 +110,7 @@
 
         & .controls {
             position: absolute;
-            bottom: 10px;
+            bottom: 16px;
             left: 10px;
             display: flex;
             flex-direction: column;

ui/src/components/dependencies/components/Link.vue

@@ -9,26 +9,32 @@
 <script setup lang="ts">
     import {computed} from "vue";
 
-    import {FLOW, EXECUTION, NAMESPACE, type Node} from "../utils/types";
+    import {FLOW, EXECUTION, NAMESPACE, ASSET, type Node} from "../utils/types";
 
     const props = defineProps<{
         node: Node;
-        subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE;
+        subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE | typeof ASSET;
     }>();
 
     const to = computed(() => {
         const base = {namespace: props.node.namespace};
 
-        if ("id" in props.node.metadata && props.node.metadata.id)
+        if (props.subtype === ASSET) {
+            return {
+                name: "assets/update",
+                params: {...base, assetId: props.node.flow},
+            };
+        } else if ("id" in props.node.metadata && props.node.metadata.id) {
             return {
                 name: "executions/update",
                 params: {...base, flowId: props.node.flow, id: props.node.metadata.id},
             };
-        else
+        } else {
             return {
                 name: "flows/update",
                 params: {...base, id: props.node.flow},
             };
+        }
     });
 </script>
 

ui/src/components/dependencies/components/Table.vue

@@ -2,7 +2,7 @@
     <section id="input">
         <el-input
             v-model="search"
-            :placeholder="$t('dependency.search.placeholder')"
+            :placeholder="$t(props.subtype === ASSET ? 'dependency.search.asset_placeholder' : 'dependency.search.placeholder')"
             clearable
         />
     </section>
@@ -38,10 +38,13 @@
                             size="small"
                         />
                         <RouterLink
-                            v-if="[FLOW, NAMESPACE].includes(row.data.metadata.subtype)"
+                            v-if="[FLOW, NAMESPACE, ASSET].includes(row.data.metadata.subtype)"
                             :to="{
-                                name: 'flows/update',
-                                params: {namespace: row.data.namespace, id: row.data.flow}}"
+                                name: row.data.metadata.subtype === ASSET ? 'assets/update' : 'flows/update',
+                                params: row.data.metadata.subtype === ASSET 
+                                    ? {namespace: row.data.namespace, assetId: row.data.flow}
+                                    : {namespace: row.data.namespace, id: row.data.flow}
+                            }"
                         >
                             <el-icon :size="16">
                                 <OpenInNew />
@@ -64,12 +67,13 @@
 
     import OpenInNew from "vue-material-design-icons/OpenInNew.vue";
 
-    import {NODE, FLOW, EXECUTION, NAMESPACE, type Node} from "../utils/types";
+    import {NODE, FLOW, EXECUTION, NAMESPACE, ASSET, type Node} from "../utils/types";
 
     const emits = defineEmits<{ (e: "select", id: Node["id"]): void }>();
     const props = defineProps<{
         elements: cytoscape.ElementDefinition[];
         selected: Node["id"] | undefined;
+        subtype?: typeof FLOW | typeof EXECUTION | typeof NAMESPACE | typeof ASSET;
     }>();
 
     const focusSelectedRow = () => {
@@ -177,6 +181,10 @@ section#row {
     & section#right {
         flex-shrink: 0;
         margin-left: 0.5rem;
+
+        :deep(a:hover .el-icon) {
+            color: var(--ks-content-link-hover);
+        }
     }
 }
 </style>

ui/src/components/dependencies/composables/useDependencies.ts

@@ -163,7 +163,7 @@ const setExecutionEdgeColors = throttle(
  * @param classes - An array of class names to remove from all elements.
  *                  Defaults to [`selected`, `faded`, `hovered`, `executions`].
  */
-export function clearClasses(cy: cytoscape.Core, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE, classes: string[] = [SELECTED, FADED, HOVERED, EXECUTIONS]): void {
+export function clearClasses(cy: cytoscape.Core, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE | typeof ASSET, classes: string[] = [SELECTED, FADED, HOVERED, EXECUTIONS]): void {
     cy.elements().removeClass(classes.join(" "));
     if (subtype === EXECUTION) cy.edges().style(edgeColors());
 }
@@ -197,7 +197,7 @@ export function fit(cy: cytoscape.Core, padding: number = 50): void {
  * @param subtype - Determines how connected elements are highlighted (`FLOW`, `EXECUTION` or `NAMESPACE`).
  * @param id - Optional explicit ID to assign to the ref (defaults to the node’s own ID).
  */
-function selectHandler(cy: cytoscape.Core, node: cytoscape.NodeSingular, selected: Ref<Node["id"] | undefined>, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE, id?: Node["id"]): void {
+function selectHandler(cy: cytoscape.Core, node: cytoscape.NodeSingular, selected: Ref<Node["id"] | undefined>, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE | typeof ASSET, id?: Node["id"]): void {
     // Clear all existing classes
     clearClasses(cy, subtype);
 
@@ -263,7 +263,17 @@ function hoverHandler(cy: cytoscape.Core): void {
  * @returns An object with element getters, loading state, rendering state, selected node ID,
  *          selection helpers, and control handlers.
  */
-export function useDependencies(container: Ref<HTMLElement | null>, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE = FLOW, initialNodeID: string, params: RouteParams, isTesting = false) {
+export function useDependencies(
+    container: Ref<HTMLElement | null>,
+    subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE | typeof ASSET = FLOW,
+    initialNodeID: string,
+    params: RouteParams,
+    isTesting = false,
+    fetchAssetDependencies?: () => Promise<{
+        data: Element[];
+        count: number;
+    }>
+) {
     const coreStore = useCoreStore();
     const flowStore = useFlowStore();
     const executionsStore = useExecutionsStore();
@@ -301,7 +311,13 @@ export function useDependencies(container: Ref<HTMLElement | null>, subtype: typ
         }
     };
 
-    const elements = ref<{ data: cytoscape.ElementDefinition[]; count: number; }>({data: [], count: 0});
+    const elements = ref<{
+        data: cytoscape.ElementDefinition[];
+        count: number;
+    }>({
+        data: [],
+        count: 0,
+    });
     onMounted(async () => {
         if (isTesting) {
             if (!container.value) {
@@ -313,13 +329,32 @@ export function useDependencies(container: Ref<HTMLElement | null>, subtype: typ
             isLoading.value = false;
         } else {
             try {
-                if (subtype === NAMESPACE) {
-                    const {data} = await namespacesStore.loadDependencies({namespace: params.id as string});
+                if (fetchAssetDependencies) {
+                    const result = await fetchAssetDependencies();
+                    elements.value = {
+                        data: result.data,
+                        count: result.count
+                    };
+                    isLoading.value = false;
+                } else if (subtype === NAMESPACE) {
+                    const {data} = await namespacesStore.loadDependencies({
+                        namespace: params.id as string,
+                    });
                     const nodes = data.nodes ?? [];
-                    elements.value = {data: transformResponse(data, NAMESPACE), count: new Set(nodes.map((r: { uid: string }) => r.uid)).size};
+                    elements.value = {
+                        data: transformResponse(data, NAMESPACE),
+                        count: new Set(nodes.map((r: { uid: string }) => r.uid)).size,
+                    };
                     isLoading.value = false;
                 } else {
-                    const result = await flowStore.loadDependencies({id: (subtype === FLOW ? params.id : params.flowId) as string, namespace: params.namespace as string, subtype}, false);
+                    const result = await flowStore.loadDependencies(
+                        {
+                            id: (subtype === FLOW ? params.id : params.flowId) as string,
+                            namespace: params.namespace as string,
+                            subtype,
+                        },
+                        false
+                    );
                     elements.value = {data: result.data ?? [], count: result.count};
                     isLoading.value = false;
                 }
@@ -448,8 +483,16 @@ export function useDependencies(container: Ref<HTMLElement | null>, subtype: typ
         selectedNodeID,
         selectNode,
         handlers: {
-            zoomIn: () => cy.zoom({level: cy.zoom() + 0.1, renderedPosition: cy.getElementById(selectedNodeID.value!).renderedPosition()}),
-            zoomOut: () => cy.zoom({level: cy.zoom() - 0.1, renderedPosition: cy.getElementById(selectedNodeID.value!).renderedPosition()}),
+            zoomIn: () =>
+                cy.zoom({
+                    level: cy.zoom() + 0.1,
+                    renderedPosition: cy.getElementById(selectedNodeID.value!).renderedPosition(),
+                }),
+            zoomOut: () =>
+                cy.zoom({
+                    level: cy.zoom() - 0.1,
+                    renderedPosition: cy.getElementById(selectedNodeID.value!).renderedPosition(),
+                }),
             clearSelection: () => {
                 clearClasses(cy, subtype);
                 selectedNodeID.value = undefined;
@@ -468,9 +511,23 @@ export function useDependencies(container: Ref<HTMLElement | null>, subtype: typ
  * @param subtype - The node subtype, either `FLOW`, `EXECUTION`, or `NAMESPACE`.
  * @returns An array of cytoscape elements with correctly typed nodes and edges.
  */
-export function transformResponse(response: {nodes: { uid: string; namespace: string; id: string }[]; edges: { source: string; target: string }[]; }, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE): Element[] {
-    const nodes: Node[] = response.nodes.map((node) => ({id: node.uid, type: NODE, flow: node.id, namespace: node.namespace, metadata: {subtype}}));
-    const edges: Edge[] = response.edges.map((edge) => ({id: uuid(), type: EDGE, source: edge.source, target: edge.target}));
+export function transformResponse(response: {nodes: { uid: string; namespace: string; id: string }[]; edges: { source: string; target: string }[];}, subtype: typeof FLOW | typeof EXECUTION | typeof NAMESPACE): Element[] {
+    const nodes: Node[] = response.nodes.map((node) => ({
+        id: node.uid,
+        type: NODE,
+        flow: node.id,
+        namespace: node.namespace,
+        metadata: {subtype},
+    }));
+    const edges: Edge[] = response.edges.map((edge) => ({
+        id: uuid(),
+        type: EDGE,
+        source: edge.source,
+        target: edge.target,
+    }));
 
-    return [...nodes.map((node) => ({data: node}) as Element), ...edges.map((edge) => ({data: edge}) as Element)];
+    return [
+        ...nodes.map((node) => ({data: node}) as Element),
+        ...edges.map((edge) => ({data: edge}) as Element),
+    ];
 }

ui/src/components/dependencies/utils/types.ts

@@ -4,6 +4,7 @@ export const EDGE = "EDGE" as const;
 export const FLOW = "FLOW" as const;
 export const EXECUTION = "EXECUTION" as const;
 export const NAMESPACE = "NAMESPACE" as const;
+export const ASSET = "ASSET" as const;
 
 type Flow = {
     subtype: typeof FLOW;
@@ -19,12 +20,16 @@ type Namespace = {
     subtype: typeof NAMESPACE;
 };
 
+type Asset = {
+    subtype: typeof ASSET;
+};
+
 export type Node = {
     id: string;
     type: "NODE";
     flow: string;
     namespace: string;
-    metadata: Flow | Execution | Namespace;
+    metadata: Flow | Execution | Namespace | Asset;
 };
 
 export type Edge = {

ui/src/components/filter/components/layout/FilterChip.vue

@@ -152,6 +152,8 @@
             font-size: 12px;
             color: var(--ks-content-primary);
             white-space: nowrap;
+            display: flex;
+            align-items: center;
         }
         .value {
             font-weight: 700;

ui/src/components/filter/components/layout/FilterEditPopper.vue

@@ -86,7 +86,7 @@
     );
 
     const isKVPairFilter = computed(() =>
-        props.filterKey?.valueType === "key-value" || (props.filterKey?.key === "labels" && KV_COMPARATORS.includes(state.selectedComparator))
+        props.filterKey?.valueType === "key-value"
     );
 
     const valueComponent = computed(() => {

ui/src/components/filter/composables/useFilters.ts

@@ -14,7 +14,6 @@ import {
     COMPARATOR_LABELS,
     Comparators,
     TEXT_COMPARATORS,
-    KV_COMPARATORS
 } from "../utils/filterTypes";
 import {usePreAppliedFilters} from "./usePreAppliedFilters";
 import {useDefaultFilter} from "./useDefaultFilter";
@@ -67,11 +66,11 @@ export function useFilters(
     };
 
     const clearLegacyParams = (query: Record<string, any>) => {
-        configuration.keys?.forEach(({key}) => {
+        configuration.keys?.forEach(({key, valueType}) => {
             delete query[key];
-            if (key === "details") {
+            if (valueType === "key-value") {
                 Object.keys(query).forEach(queryKey => {
-                    if (queryKey.startsWith("details.")) delete query[queryKey];
+                    if (queryKey.startsWith(`${key}.`)) delete query[queryKey];
                 });
             }
         });
@@ -85,10 +84,10 @@ export function useFilters(
      */
     const buildLegacyQuery = (query: Record<string, any>) => {
         getUniqueFilters(appliedFilters.value.filter(isValidFilter)).forEach(filter => {
-            if (filter.key === "details") {
+            if (configuration.keys?.find(k => k.key === filter.key)?.valueType === "key-value") {
                 (filter.value as string[]).forEach(item => {
                     const [k, v] = item.split(":");
-                    query[`details.${k}`] = v;
+                    query[`${filter.key}.${k}`] = v;
                 });
             } else if (Array.isArray(filter.value)) {
                 filter.value.forEach(item =>
@@ -108,8 +107,6 @@ export function useFilters(
         const query = {...route.query};
         clearFilterQueryParams(query);
 
-        delete query.page;
-
         if (legacyQuery) {
             clearLegacyParams(query);
             buildLegacyQuery(query);
@@ -119,6 +116,15 @@ export function useFilters(
         }
 
         updateSearchQuery(query);
+
+        if (
+            (appliedFilters.value.some(f => Array.isArray(f.value) && f.value.length > 0)
+            || searchQuery.value.trim())
+            && parseInt(String(query.page ?? "1")) > 1
+        ) {
+            delete query.page;
+        }
+
         router.push({query});
     };
 
@@ -145,14 +151,13 @@ export function useFilters(
         value: string | string[]
     ): AppliedFilter => {
         const comparator = (config?.comparators?.[0] as Comparators) ?? Comparators.EQUALS;
-        const valueLabel = Array.isArray(value)
-            ? key === "details" && value.length > 1
-                ? `${value[0]} +${value.length - 1}`
+        return createAppliedFilter(key, config, comparator, value, 
+            config?.valueType === "key-value" && Array.isArray(value)
+                ? value.length > 1 ? `${value[0]} +${value.length - 1}` : value[0] ?? ""
                 : Array.isArray(value)
                     ? value.join(", ")
-                    : value[0]
-            : (value as string);
-        return createAppliedFilter(key, config, comparator, value, valueLabel, "EQUALS");
+                    : value as string
+        , "EQUALS");
     };
 
     const createTimeRangeFilter = (
@@ -161,14 +166,13 @@ export function useFilters(
         endDate: Date,
         comparator = Comparators.EQUALS
     ): AppliedFilter => {
-        const valueLabel = `${startDate.toLocaleDateString()} - ${endDate.toLocaleDateString()}`;
         return {
             ...createAppliedFilter(
                 "timeRange",
                 config,
                 comparator,
                 {startDate, endDate},
-                valueLabel,
+                `${startDate.toLocaleDateString()} - ${endDate.toLocaleDateString()}`,
                 keyOfComparator(comparator)
             ),
             comparatorLabel: "Is Between"
@@ -181,34 +185,36 @@ export function useFilters(
      */
     const parseLegacyFilters = (): AppliedFilter[] => {
         const filtersMap = new Map<string, AppliedFilter>();
-        const details: string[] = [];
+        const keyValueFilters: Record<string, string[]> = {};
 
         Object.entries(route.query).forEach(([key, value]) => {
             if (["q", "search", "filters[q][EQUALS]"].includes(key)) return;
 
-            if (key.startsWith("details.")) {
-                details.push(`${key.split(".")[1]}:${value}`);
+            const kvConfig = configuration.keys?.find(k => key.startsWith(`${k.key}.`) && k.valueType === "key-value");
+            if (kvConfig) {
+                if (!keyValueFilters[kvConfig.key]) keyValueFilters[kvConfig.key] = [];
+                keyValueFilters[kvConfig.key].push(`${key.split(".")[1]}:${value}`);
                 return;
             }
 
             const config = configuration.keys?.find(k => k.key === key);
             if (!config) return;
 
-            const processedValue = Array.isArray(value)
-                ? (value as string[]).filter(v => v !== null)
-                : config?.valueType === "multi-select"
-                    ? ((value as string) ?? "").split(",")
-                    : ((value as string) ?? "");
-
-            filtersMap.set(key, createFilter(key, config, processedValue));
+            filtersMap.set(key, createFilter(key, config, 
+                Array.isArray(value)
+                    ? (value as string[]).filter(v => v !== null)
+                    : config?.valueType === "multi-select"
+                        ? ((value as string) ?? "").split(",")
+                        : ((value as string) ?? "")
+            ));
         });
 
-        if (details.length > 0) {
-            const config = configuration.keys?.find(k => k.key === "details");
+        Object.entries(keyValueFilters).forEach(([key, values]) => {
+            const config = configuration.keys?.find(k => k.key === key);
             if (config) {
-                filtersMap.set("details", createFilter("details", config, details));
+                filtersMap.set(key, createFilter(key, config, values));
             }
-        }
+        });
 
         if (route.query.startDate && route.query.endDate) {
             const timeRangeConfig = configuration.keys?.find(k => k.key === "timeRange");
@@ -227,13 +233,10 @@ export function useFilters(
         return Array.from(filtersMap.values());
     };
 
-    const isKVFilter = (field: string, comparator: Comparators) =>
-        field === "details" || (field === "labels" && KV_COMPARATORS.includes(comparator));
-
-    const processFieldValue = (config: any, params: any[], field: string, comparator: Comparators) => {
+    const processFieldValue = (config: any, params: any[], _field: string, comparator: Comparators) => {
         const isTextOp = TEXT_COMPARATORS.includes(comparator);
 
-        if (isKVFilter(field, comparator)) {
+        if (config?.valueType === "key-value") {
             const combinedValue = params.map(p => p?.value as string);
             return {
                 value: combinedValue,
@@ -253,10 +256,9 @@ export function useFilters(
             };
         }
 
-        const param = params[0];
-        let value = Array.isArray(param?.value)
-            ? param.value[0]
-            : (param?.value as string);
+        let value = Array.isArray(params[0]?.value)
+            ? params[0].value[0]
+            : (params[0]?.value as string);
 
         if (config?.valueType === "date" && typeof value === "string") {
             value = new Date(value);

ui/src/components/filter/configurations/executionFilter.ts

@@ -127,7 +127,7 @@ export const useExecutionFilter = (): ComputedRef<FilterConfiguration> => {
                     label: t("filter.labels_execution.label"),
                     description: t("filter.labels_execution.description"),
                     comparators: [Comparators.EQUALS, Comparators.NOT_EQUALS],
-                    valueType: "text",
+                    valueType: "key-value",
                 },
                 {
                     key: "triggerExecutionId",

ui/src/components/filter/configurations/flowExecutionFilter.ts

@@ -74,7 +74,7 @@ export const useFlowExecutionFilter = (): ComputedRef<FilterConfiguration> => {
                     label: t("filter.labels_execution.label"),
                     description: t("filter.labels_execution.description"),
                     comparators: [Comparators.EQUALS, Comparators.NOT_EQUALS],
-                    valueType: "text",
+                    valueType: "key-value",
                 },
                 {
                     key: "triggerExecutionId",

ui/src/components/filter/configurations/flowFilter.ts

@@ -67,7 +67,7 @@ export const useFlowFilter = (): ComputedRef<FilterConfiguration> => {
                     label: t("filter.labels_flow.label"),
                     description: t("filter.labels_flow.description"),
                     comparators: [Comparators.EQUALS, Comparators.NOT_EQUALS],
-                    valueType: "text",
+                    valueType: "key-value",
                 },
             ]
         };

ui/src/components/filter/utils/helpers.ts

@@ -25,7 +25,7 @@ export const decodeSearchParams = (query: LocationQuery) =>
 
             const [, field, operation, subKey] = match;
 
-            if (field === "labels" && subKey) {
+            if (subKey) {
                 return {
                     field,
                     value: `${subKey}:${decodeURIComponentSafely(value)}`,
@@ -57,30 +57,19 @@ export const encodeFiltersToQuery = (filters: Filter[], keyOfComparator: (compar
                     query[`filters[${key}][${comparatorKey}]`] = value?.toString() ?? "";
                 }
                 return query;
-            case "labels":
-                if (Array.isArray(value)) {
-                    value.forEach((label: string) => {
-                        const [k, v] = label.split(":", 2);
-                        if (k && v) query[`filters[labels][${comparatorKey}][${k}]`] = v;
-                    });
-                } else if (typeof value === "string") {
-                    const [k, v] = value.split(":", 2);
-                    if (k && v) {
-                        query[`filters[labels][${comparatorKey}][${k}]`] = v;
-                    } else {
-                        query[`filters[${key}][${comparatorKey}]`] = value;
-                    }
-                }
-                return query;
             default: {
-                const processedValue = Array.isArray(value)
-                    ? value.join(",")
-                    : typeof value === "object" && "startDate" in value
-                        ? `${value.startDate.toISOString()},${value.endDate.toISOString()}`
+                if (Array.isArray(value) && value.some(v => typeof v === "string" && v.includes(":"))) {
+                    value.forEach((item: string) => {
+                        const [k, v] = item.split(":", 2);
+                        if (k && v) query[`filters[${key}][${comparatorKey}][${k}]`] = v;
+                    });
+                } else {
+                    query[`filters[${key}][${comparatorKey}]`] = Array.isArray(value)
+                        ? value.join(",")
                         : value instanceof Date
                             ? value.toISOString()
-                            : value;
-                query[`filters[${key}][${comparatorKey}]`] = processedValue?.toString() ?? "";
+                            : value?.toString() ?? "";
+                }
                 return query;
             }
         }

ui/src/components/layout/Labels.vue

@@ -26,8 +26,16 @@
     }
 
     const props = withDefaults(
-        defineProps<{ labels: Label[]; readOnly?: boolean }>(),
-        {labels: () => [], readOnly: false},
+        defineProps<{
+            labels?: Label[];
+            readOnly?: boolean;
+            filterType?: "labels" | "metadata";
+        }>(),
+        {
+            labels: () => [],
+            readOnly: false,
+            filterType: "labels",
+        },
     );
 
     import {decodeSearchParams} from "../../components/filter/utils/helpers";
@@ -48,7 +56,7 @@
     };
 
     const updateLabel = (label: Label) => {
-        const getKey = (key: string) => `filters[labels][EQUALS][${key}]`;
+        const getKey = (key: string) => `filters[${props.filterType}][EQUALS][${key}]`;
 
         if (isChecked(label)) {
             const replacementQuery = {...route.query};

ui/src/components/layout/TopNavBar.vue

@@ -33,6 +33,11 @@
                             @click="onStarClick"
                         />
                     </h1>
+                    <div class="description">
+                        <slot name="description">
+                            {{ longDescription }}
+                        </slot>
+                    </div>
                 </div>
             </div>
         </div>
@@ -77,15 +82,20 @@
     const props = defineProps<{
         title: string;
         description?: string;
-        breadcrumb?: { label: string; link?: RouterLinkTo; disabled?: boolean }[];
+        longDescription?: string;
+        breadcrumb?: {
+            label: string;
+            link?: RouterLinkTo;
+            disabled?: boolean;
+        }[];
         beta?: boolean;
     }>();
 
-    const logsStore = useLogsStore();
-    const bookmarksStore = useBookmarksStore();
-    const flowStore = useFlowStore();
     const route = useRoute();
+    const logsStore = useLogsStore();
+    const flowStore = useFlowStore();
     const layoutStore = useLayoutStore();
+    const bookmarksStore = useBookmarksStore();
 
 
     const shouldDisplayDeleteButton = computed(() => {
@@ -182,6 +192,12 @@
             align-items: center;
         }
 
+        .description {
+            font-size: 0.875rem;
+            margin-top: -0.5rem;
+            color: var(--ks-content-secondary);
+        }
+
         .icon {
             border: none;
             color: var(--ks-content-tertiary);

ui/src/components/layout/empty/images.ts

@@ -8,6 +8,7 @@ import plugins from "./assets/visuals/plugins.png";
 import triggers from "./assets/visuals/triggers.png";
 import versionPlugin from "./assets/visuals/versionPlugin.png";
 import panels from "./assets/visuals/panels.png";
+import assets from "./assets/visuals/assets.png";
 
 export const images: Record<string, string> = {
     announcements,
@@ -18,8 +19,10 @@ export const images: Record<string, string> = {
     "dependencies.FLOW": dependencies,
     "dependencies.EXECUTION": dependencies,
     "dependencies.NAMESPACE": dependencies,
+    "dependencies.ASSET": dependencies,
     plugins,
     triggers,
     versionPlugin,
     panels,
+    assets
 };

ui/src/override/components/useLeftMenu.ts

@@ -15,7 +15,7 @@ import ContentCopy from "vue-material-design-icons/ContentCopy.vue";
 import PlayOutline from "vue-material-design-icons/PlayOutline.vue";
 import FileDocumentOutline from "vue-material-design-icons/FileDocumentOutline.vue";
 import FlaskOutline from "vue-material-design-icons/FlaskOutline.vue";
-// import PackageVariantClosed from "vue-material-design-icons/PackageVariantClosed.vue";
+import PackageVariantClosed from "vue-material-design-icons/PackageVariantClosed.vue";
 import FolderOpenOutline from "vue-material-design-icons/FolderOpenOutline.vue";
 import PuzzleOutline from "vue-material-design-icons/PuzzleOutline.vue";
 import ShapePlusOutline from "vue-material-design-icons/ShapePlusOutline.vue";
@@ -145,8 +145,19 @@ export function useLeftMenu() {
                     locked: true,
                 },
             },
-            // TODO: To add Assets entry here in future release
-            // Uncomment PackageVariantClosed on line 25 and use as the icon
+            {
+                title: t("demos.assets.label"),
+                routes: routeStartWith("assets"),
+                href: {
+                    name: "assets/list"
+                },
+                icon: {
+                    element: PackageVariantClosed,
+                },
+                attributes: {
+                    locked: true,
+                },
+            },
             {
                 title: t("namespaces"),
                 routes: routeStartWith("namespaces"),

ui/src/routes/routes.js

@@ -7,6 +7,7 @@ import DemoAuditLogs from "../components/demo/AuditLogs.vue"
 import DemoInstance from "../components/demo/Instance.vue"
 import DemoApps from "../components/demo/Apps.vue"
 import DemoTests from "../components/demo/Tests.vue"
+import DemoAssets from "../components/demo/Assets.vue"
 import {applyDefaultFilters} from "../components/filter/composables/useDefaultFilter";
 
 export default [
@@ -123,6 +124,7 @@ export default [
     //Demo Pages
     {name: "apps/list", path: "/:tenant?/apps", component: DemoApps},
     {name: "tests/list", path: "/:tenant?/tests", component: DemoTests},
+    {name: "assets/list", path: "/:tenant?/assets", component: DemoAssets},
     {name: "admin/iam", path: "/:tenant?/admin/iam", component: DemoIAM},
     {name: "admin/tenants/list", path: "/:tenant?/admin/tenants", component: DemoTenants},
     {name: "admin/auditlogs/list", path: "/:tenant?/admin/auditlogs", component: DemoAuditLogs},

ui/src/translations/de.json

@@ -314,6 +314,12 @@
         "message": "Apps in der Kestra Enterprise Edition ermöglichen es Ihnen, benutzerdefinierte UIs zu erstellen, die mit Kestra-Workflows außerhalb der Plattform interagieren. Diese Funktion erlaubt es Ihnen, Ihre Workflows als Backend für benutzerdefinierte Anwendungen zu nutzen, sodass nicht-technische Benutzer Daten über Formulare einreichen oder pausierte Workflows, die auf eine Genehmigung warten, fortsetzen können.",
         "title": "Erstellen Sie benutzerdefinierte Apps mit Kestra"
       },
+      "assets": {
+        "header": "Assets-Metadaten und Observability",
+        "label": "Assets",
+        "message": "Assets verbinden Observability-, Lineage- und Ownership-Metadaten, damit Plattformteams schneller Probleme beheben und mit Vertrauen bereitstellen können.",
+        "title": "Bringen Sie jedes Dataset, jeden Service und jede Abhängigkeit in Sicht."
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition protokolliert jede Aktivität mit robusten, unveränderlichen Aufzeichnungen, was es einfach macht, Änderungen nachzuverfolgen, die Einhaltung von Vorschriften zu gewährleisten und Probleme zu beheben.",
         "title": "Änderungen mit Audit Logs nachverfolgen"
@@ -389,6 +395,7 @@
         "zoom_out": "Rauszoomen"
       },
       "search": {
+        "asset_placeholder": "Suche nach Asset, flow oder namespace...",
         "no_results": "Keine Ergebnisse gefunden für {term}",
         "placeholder": "Suche nach flow oder namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Beginnen Sie mit dem Erstellen von Apps, um mit Kestra von der Außenwelt zu interagieren.",
         "title": "Sie haben noch keine Apps!"
       },
+      "assets": {
+        "content": "Fügen Sie Assets hinzu, um Ihre Daten-Assets, Dienste und Infrastruktur zu verfolgen und zu verwalten.",
+        "title": "Sie haben noch keine Assets!"
+      },
       "concurrency_executions": {
         "content": "Erfahren Sie mehr über <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Ausführungen</a></strong> in unserer Dokumentation.",
         "title": "Keine laufenden Ausführungen für diesen Flow."
@@ -463,6 +474,10 @@
         "title": "Für diesen Flow sind keine Grenzen festgelegt."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Dieses Asset hat keine Upstream- oder Downstream-Abhängigkeiten mit flows oder anderen Assets.",
+          "title": "Derzeit gibt es keine Abhängigkeiten."
+        },
         "EXECUTION": {
           "content": "Erfahren Sie mehr über <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">Ausführungsabhängigkeiten</a> in unserer Dokumentation.",
           "title": "Derzeit gibt es keine Abhängigkeiten."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Keine Logs für die ausgewählten Filter gefunden. <br> Bitte versuchen Sie, Ihre Filter anzupassen, den Zeitraum zu ändern oder überprüfen Sie, ob der flow kürzlich ausgeführt wurde.",
     "no_namespaces": "Keine Namespaces entsprechen den Suchkriterien.",
     "no_results": {
+      "assets": "Keine Assets gefunden",
       "executions": "Keine Ausführungen gefunden",
       "flows": "Keine Flows gefunden",
       "kv_pairs": "Keine Key-Value-Paare gefunden",

ui/src/translations/en.json

@@ -896,6 +896,12 @@
         "title": "Ensure Reliability with Every Change",
         "message": "Verify the logic of your flows in isolation, detect regressions early, and maintain confidence in your automations as they change and grow."
       },
+      "assets": {
+        "label": "Assets",
+        "header": "Assets Metadata and Observability",
+        "title": "Bring every dataset, service, and dependency into view.",
+        "message": "Assets connect observability, lineage, and ownership metadata so platform teams can troubleshoot faster and deploy with confidence."
+      },
       "IAM": {
         "title": "Manage Users through IAM with SSO, SCIM and RBAC",
         "message": "Kestra Enterprise Edition has built-in IAM capabilities with single sign-on (SSO), SCIM directory sync, and role-based access control (RBAC), integrating with multiple identity providers and letting you assign fine-grained permissions for users and service accounts."
@@ -1370,6 +1376,10 @@
         "title": "You have no Tests yet!",
         "content": "Add tests to validate quality and avoid regressions in your flows."
       },
+      "assets": {
+        "title": "You have no Assets yet!",
+        "content": "Add assets to track and manage your data assets, services, and infrastructure."
+      },
       "concurrency_executions": {
         "title": "No ongoing Executions for this Flow.",
         "content": "Read more about <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong> in our documentation."
@@ -1390,6 +1400,10 @@
         "NAMESPACE": {
           "title": "There are currently no dependencies.",
           "content": "Read more about <a href=\"https://kestra.io/docs/ui/namespaces#dependencies\" target=\"_blank\">Namespace Dependencies</a> in our documentation."
+        },
+        "ASSET": {
+          "title": "There are currently no dependencies.",
+          "content": "This asset has no upstream or downstream dependencies with flows or other assets."
         }
       },
       "plugins": {
@@ -1430,7 +1444,8 @@
       "flows": "No Flows Found",
       "kv_pairs": "No Key-Value pairs Found",
       "secrets": "No Secrets Found",
-      "templates": "No Templates Found"
+      "templates": "No Templates Found",
+      "assets": "No Assets Found"
     },
     "duplicate-pair": "{label} \"{key}\" is duplicated, first key ignored.",
     "dashboards": {
@@ -1575,6 +1590,7 @@
       },
       "search": {
         "placeholder": "Search by flow or namespace...",
+        "asset_placeholder": "Search by asset or flow or namespace...",
         "no_results": "No results found for {term}"
       }
     },

ui/src/translations/es.json

@@ -314,6 +314,12 @@
         "message": "Las aplicaciones en Kestra Enterprise Edition te permiten construir interfaces de usuario personalizadas que interactúan con los workflows de Kestra desde fuera de la plataforma. Esta función te permite usar tus workflows como un backend para aplicaciones personalizadas, permitiendo a usuarios no técnicos enviar datos a través de formularios o reanudar workflows PAUSED que esperan aprobación.",
         "title": "Crea aplicaciones personalizadas con Kestra"
       },
+      "assets": {
+        "header": "Metadatos de Activos y Observabilidad",
+        "label": "Activos",
+        "message": "Los assets conectan la observabilidad, el linaje y los metadatos de propiedad para que los equipos de la plataforma puedan solucionar problemas más rápido y desplegar con confianza.",
+        "title": "Lleva cada dataset, servicio y dependencia a la vista."
+      },
       "audit-logs": {
         "message": "La edición Enterprise de Kestra registra cada actividad con registros robustos e inmutables, lo que facilita el seguimiento de cambios, el mantenimiento de la conformidad y la resolución de problemas.",
         "title": "Rastrear Cambios con Audit Logs"
@@ -389,6 +395,7 @@
         "zoom_out": "Alejar"
       },
       "search": {
+        "asset_placeholder": "Buscar por asset, flow o namespace...",
         "no_results": "No se encontraron resultados para {term}",
         "placeholder": "Buscar por flow o namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Comienza a crear Apps para interactuar con Kestra desde el mundo exterior.",
         "title": "¡Aún no tienes aplicaciones!"
       },
+      "assets": {
+        "content": "Agrega activos para rastrear y gestionar tus activos de datos, servicios e infraestructura.",
+        "title": "¡Aún no tienes Assets!"
+      },
       "concurrency_executions": {
         "content": "Lee más sobre <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Ejecuciones</a></strong> en nuestra documentación.",
         "title": "No hay Ejecuciones en curso para este Flow."
@@ -463,6 +474,10 @@
         "title": "No se han establecido límites para este Flow."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Este recurso no tiene dependencias ascendentes o descendentes con flows u otros recursos.",
+          "title": "Actualmente no hay dependencias."
+        },
         "EXECUTION": {
           "content": "Lea más sobre <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">Dependencias de Ejecución</a> en nuestra documentación.",
           "title": "Actualmente no hay dependencias."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "No se encontraron logs para los filtros seleccionados. <br> Por favor, intente ajustar sus filtros, cambiar el rango de tiempo o verifique si el flow se ha ejecutado recientemente.",
     "no_namespaces": "Ningún namespace coincide con los criterios de búsqueda.",
     "no_results": {
+      "assets": "No se encontraron activos",
       "executions": "No se encontraron ejecuciones",
       "flows": "No se encontraron Flows",
       "kv_pairs": "No se encontraron pares Key-Value",

ui/src/translations/fr.json

@@ -314,6 +314,12 @@
         "message": "Les applications dans Kestra Enterprise Edition vous permettent de créer des interfaces utilisateur personnalisées qui interagissent avec les workflows Kestra depuis l'extérieur de la plateforme. Cette fonctionnalité vous permet d'utiliser vos workflows comme un backend pour des applications personnalisées, permettant aux utilisateurs non techniques de soumettre des données via des formulaires ou de reprendre des workflows PAUSED en attente d'approbation.",
         "title": "Créez des applications personnalisées avec Kestra"
       },
+      "assets": {
+        "header": "Métadonnées des actifs et Observabilité",
+        "label": "Actifs",
+        "message": "Les assets connectent les métadonnées d'observabilité, de lignée et de propriété afin que les équipes de la plateforme puissent résoudre les problèmes plus rapidement et déployer en toute confiance.",
+        "title": "Amenez chaque dataset, service et dépendance en vue."
+      },
       "audit-logs": {
         "message": "L'édition Kestra Enterprise enregistre chaque activité avec des enregistrements robustes et immuables, ce qui facilite le suivi des modifications, le maintien de la conformité et la résolution des problèmes.",
         "title": "Suivre les modifications avec les Audit Logs"
@@ -389,6 +395,7 @@
         "zoom_out": "Dézoomer"
       },
       "search": {
+        "asset_placeholder": "Rechercher par asset, flow ou namespace...",
         "no_results": "Aucun résultat trouvé pour {term}",
         "placeholder": "Rechercher par flow ou namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Commencez à créer des Apps pour interagir avec Kestra depuis le monde extérieur.",
         "title": "Vous n'avez pas encore d'applications !"
       },
+      "assets": {
+        "content": "Ajoutez des ressources pour suivre et gérer vos ressources de données, services et infrastructures.",
+        "title": "Vous n'avez pas encore d'Assets !"
+      },
       "concurrency_executions": {
         "content": "En savoir plus sur les <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Exécutions</a></strong> dans notre documentation.",
         "title": "Aucune exécution en cours pour ce flow."
@@ -463,6 +474,10 @@
         "title": "Aucune limite n'est définie pour ce flow."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Cet actif n'a pas de dépendances amont ou aval avec des flows ou d'autres actifs.",
+          "title": "Il n'y a actuellement aucune dépendance."
+        },
         "EXECUTION": {
           "content": "En savoir plus sur les <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">dépendances d'exécution</a> dans notre documentation.",
           "title": "Il n'y a actuellement aucune dépendance."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Aucun log trouvé pour les filtres sélectionnés. <br> Veuillez essayer d'ajuster vos filtres, de changer la plage de temps, ou vérifier si le flow a été exécuté récemment.",
     "no_namespaces": "Zéro espace de noms correspond aux critères de recherche.",
     "no_results": {
+      "assets": "Aucun actif trouvé",
       "executions": "Aucune exécution trouvée",
       "flows": "Aucun flow trouvé",
       "kv_pairs": "Aucune paire clé-valeur trouvée",

ui/src/translations/hi.json

@@ -314,6 +314,12 @@
         "message": "Kestra एंटरप्राइज एडिशन में ऐप्स आपको कस्टम UI बनाने की अनुमति देते हैं जो प्लेटफ़ॉर्म के बाहर से Kestra वर्कफ़्लोज़ के साथ इंटरैक्ट करते हैं। यह फीचर आपको अपने वर्कफ़्लोज़ को कस्टम एप्लिकेशनों के लिए बैकएंड के रूप में उपयोग करने देता है, जिससे गैर-तकनीकी उपयोगकर्ता फॉर्म के माध्यम से डेटा सबमिट कर सकते हैं या अप्रूवल की प्रतीक्षा कर रहे PAUSED वर्कफ़्लोज़ को फिर से शुरू कर सकते हैं।",
         "title": "Kestra के साथ कस्टम ऐप्स बनाएं"
       },
+      "assets": {
+        "header": "संपत्तियों का मेटाडेटा और अवलोकनीयता",
+        "label": "संपत्तियाँ",
+        "message": "एसेट्स अवलोकन, वंशावली, और स्वामित्व मेटाडेटा को जोड़ते हैं ताकि प्लेटफ़ॉर्म टीमें तेजी से समस्या निवारण कर सकें और आत्मविश्वास के साथ परिनियोजन कर सकें।",
+        "title": "हर dataset, सेवा, और निर्भरता को दृश्य में लाएं।"
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition हर गतिविधि को मजबूत, अपरिवर्तनीय रिकॉर्ड के साथ लॉग करता है, जिससे परिवर्तनों को ट्रैक करना, अनुपालन बनाए रखना और समस्याओं का समाधान करना आसान हो जाता है।",
         "title": "परिवर्तनों को ट्रैक करें Audit Logs के साथ"
@@ -389,6 +395,7 @@
         "zoom_out": "ज़ूम आउट"
       },
       "search": {
+        "asset_placeholder": "एसेट, flow या namespace द्वारा खोजें...",
         "no_results": "{term} के लिए कोई परिणाम नहीं मिला",
         "placeholder": "flow या namespace द्वारा खोजें..."
       }
@@ -454,6 +461,10 @@
         "content": "बाहरी दुनिया से Kestra के साथ इंटरैक्ट करने के लिए Apps बनाना शुरू करें।",
         "title": "आपके पास अभी तक कोई ऐप्स नहीं हैं!"
       },
+      "assets": {
+        "content": "अपने डेटा एसेट्स, सेवाओं और इन्फ्रास्ट्रक्चर को ट्रैक और प्रबंधित करने के लिए एसेट्स जोड़ें।",
+        "title": "आपके पास अभी तक कोई Assets नहीं हैं!"
+      },
       "concurrency_executions": {
         "content": "हमारे दस्तावेज़ में <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong> के बारे में अधिक पढ़ें।",
         "title": "इस Flow के लिए कोई चल रही Executions नहीं हैं।"
@@ -463,6 +474,10 @@
         "title": "इस Flow के लिए कोई सीमाएँ निर्धारित नहीं हैं।"
       },
       "dependencies": {
+        "ASSET": {
+          "content": "इस संपत्ति का flows या अन्य संपत्तियों के साथ कोई upstream या downstream निर्भरता नहीं है।",
+          "title": "वर्तमान में कोई निर्भरता नहीं है।"
+        },
         "EXECUTION": {
           "content": "हमारे दस्तावेज़ में <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">Execution Dependencies</a> के बारे में अधिक पढ़ें।",
           "title": "वर्तमान में कोई निर्भरता नहीं है।"
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "चयनित फ़िल्टर के लिए कोई logs नहीं मिले। <br> कृपया अपने फ़िल्टर समायोजित करें, समय सीमा बदलें, या जांचें कि flow ने हाल ही में निष्पादन किया है या नहीं।",
     "no_namespaces": "कोई भी namespaces खोज मानदंड से मेल नहीं खाते।",
     "no_results": {
+      "assets": "कोई Assets नहीं मिले",
       "executions": "कोई Executions नहीं मिला",
       "flows": "कोई Flows नहीं मिला",
       "kv_pairs": "कोई Key-Value जोड़े नहीं मिले",

ui/src/translations/it.json

@@ -314,6 +314,12 @@
         "message": "Le app nella Kestra Enterprise Edition ti consentono di creare interfacce utente personalizzate che interagiscono con i workflow di Kestra al di fuori della piattaforma. Questa funzionalità ti permette di utilizzare i tuoi workflow come backend per applicazioni personalizzate, consentendo agli utenti non tecnici di inviare dati tramite moduli o riprendere workflow PAUSED in attesa di approvazione.",
         "title": "Crea App personalizzate con Kestra"
       },
+      "assets": {
+        "header": "Metadati e Osservabilità degli Asset",
+        "label": "Risorse",
+        "message": "Le risorse collegano la osservabilità, la lineage e i metadati di proprietà in modo che i team della piattaforma possano risolvere i problemi più rapidamente e distribuire con fiducia.",
+        "title": "Porta ogni dataset, servizio e dipendenza in vista."
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition registra ogni attività con record robusti e immutabili, rendendo facile tracciare le modifiche, mantenere la conformità e risolvere i problemi.",
         "title": "Traccia le modifiche con i Log di Audit"
@@ -389,6 +395,7 @@
         "zoom_out": "Riduci zoom"
       },
       "search": {
+        "asset_placeholder": "Cerca per asset, flow o namespace...",
         "no_results": "Nessun risultato trovato per {term}",
         "placeholder": "Cerca per flow o namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Inizia a creare App per interagire con Kestra dal mondo esterno.",
         "title": "Non hai ancora app!"
       },
+      "assets": {
+        "content": "Aggiungi asset per tracciare e gestire i tuoi asset di dati, servizi e infrastruttura.",
+        "title": "Non hai ancora Asset!"
+      },
       "concurrency_executions": {
         "content": "Leggi di più sulle <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Esecuzioni</a></strong> nella nostra documentazione.",
         "title": "Nessuna Esecuzione in corso per questo Flow."
@@ -463,6 +474,10 @@
         "title": "Nessun limite è impostato per questo Flow."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Questo asset non ha dipendenze upstream o downstream con flow o altri asset.",
+          "title": "Attualmente non ci sono dipendenze."
+        },
         "EXECUTION": {
           "content": "Per saperne di più sulle <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">Dipendenze di Esecuzione</a> nella nostra documentazione.",
           "title": "Attualmente non ci sono dipendenze."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Nessun log trovato per i filtri selezionati. <br> Prova a modificare i filtri, cambiare l'intervallo di tempo o verifica se il flow è stato eseguito di recente.",
     "no_namespaces": "Nessun namespace corrisponde ai criteri di ricerca.",
     "no_results": {
+      "assets": "Nessun asset trovato",
       "executions": "Nessuna Esecuzione Trovata",
       "flows": "Nessun Flow Trovato",
       "kv_pairs": "Nessuna coppia Key-Value trovata",

ui/src/translations/ja.json

@@ -314,6 +314,12 @@
         "message": "Kestra Enterprise Editionのアプリは、プラットフォーム外からKestraのworkflowと連携するカスタムUIを構築することを可能にします。この機能により、workflowをカスタムアプリケーションのバックエンドとして使用でき、技術的でないユーザーがフォームを通じてデータを送信したり、承認待ちのPAUSED状態のworkflowを再開したりすることができます。",
         "title": "Kestraでカスタムアプリを構築する"
       },
+      "assets": {
+        "header": "アセットメタデータとオブザーバビリティ",
+        "label": "アセット",
+        "message": "アセットは、観測性、系譜、および所有権のメタデータを接続し、プラットフォームチームが迅速にトラブルシューティングを行い、自信を持ってデプロイできるようにします。",
+        "title": "すべてのデータセット、サービス、および依存関係を表示します。"
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Editionは、すべてのアクティビティを堅牢で不変の記録としてログに記録し、変更の追跡、コンプライアンスの維持、問題のトラブルシューティングを容易にします。",
         "title": "監査Logで変更を追跡"
@@ -389,6 +395,7 @@
         "zoom_out": "ズームアウト"
       },
       "search": {
+        "asset_placeholder": "アセット、flow、またはnamespaceで検索...",
         "no_results": "{term} に対する結果が見つかりませんでした",
         "placeholder": "flow または namespace で検索..."
       }
@@ -454,6 +461,10 @@
         "content": "外部からKestraと連携するアプリを作成し始めましょう。",
         "title": "まだアプリがありません！"
       },
+      "assets": {
+        "content": "データ資産、サービス、インフラストラクチャを追跡および管理するためのアセットを追加します。",
+        "title": "まだアセットがありません！"
+      },
       "concurrency_executions": {
         "content": "<strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong> についての詳細は、ドキュメントをご覧ください。",
         "title": "このFlowには進行中の実行はありません。"
@@ -463,6 +474,10 @@
         "title": "このFlowには制限が設定されていません。"
       },
       "dependencies": {
+        "ASSET": {
+          "content": "このアセットは、flowや他のアセットとの上流または下流の依存関係がありません。",
+          "title": "現在、依存関係はありません。"
+        },
         "EXECUTION": {
           "content": "<a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">実行の依存関係</a>についての詳細は、ドキュメントをご覧ください。",
           "title": "現在、依存関係はありません。"
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "選択したフィルターに対してログが見つかりませんでした。<br> フィルターを調整するか、時間範囲を変更するか、flowが最近実行されたかを確認してください。",
     "no_namespaces": "検索条件に一致するnamespaceはゼロです。",
     "no_results": {
+      "assets": "アセットが見つかりません",
       "executions": "実行が見つかりません",
       "flows": "フローが見つかりません",
       "kv_pairs": "キー-バリュー ペアが見つかりません",

ui/src/translations/ko.json

@@ -314,6 +314,12 @@
         "message": "Kestra Enterprise Edition의 앱은 플랫폼 외부에서 Kestra 워크플로와 상호작용하는 맞춤형 UI를 구축할 수 있게 해줍니다. 이 기능을 통해 워크플로를 맞춤형 애플리케이션의 백엔드로 사용할 수 있으며, 비기술적 사용자도 양식을 통해 데이터를 제출하거나 승인을 기다리는 PAUSED 워크플로를 재개할 수 있습니다.",
         "title": "Kestra로 맞춤형 앱 만들기"
       },
+      "assets": {
+        "header": "자산 메타데이터 및 관측 가능성",
+        "label": "자산",
+        "message": "자산은 관측 가능성, 계보, 소유권 메타데이터를 연결하여 플랫폼 팀이 더 빠르게 문제를 해결하고 자신 있게 배포할 수 있도록 합니다.",
+        "title": "모든 데이터셋, 서비스 및 종속성을 한눈에 확인하세요."
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition은 모든 활동을 강력하고 변경 불가능한 기록으로 로그하여 변경 사항을 추적하고, 규정을 준수하며, 문제를 해결하는 것을 쉽게 만듭니다.",
         "title": "변경 사항 추적 - 감사 Log"
@@ -389,6 +395,7 @@
         "zoom_out": "축소"
       },
       "search": {
+        "asset_placeholder": "자산, flow 또는 namespace로 검색...",
         "no_results": "{term}에 대한 결과가 없습니다.",
         "placeholder": "flow 또는 namespace로 검색..."
       }
@@ -454,6 +461,10 @@
         "content": "Kestra와 외부 세계를 상호작용하는 앱을 만들기 시작하세요.",
         "title": "앱이 아직 없습니다!"
       },
+      "assets": {
+        "content": "데이터 자산, 서비스 및 인프라를 추적하고 관리하기 위해 자산을 추가하세요.",
+        "title": "아직 자산이 없습니다!"
+      },
       "concurrency_executions": {
         "content": "자세한 내용은 <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong>에 대한 문서를 참조하세요.",
         "title": "이 Flow에 대한 진행 중인 실행이 없습니다."
@@ -463,6 +474,10 @@
         "title": "이 Flow에 대한 제한이 설정되지 않았습니다."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "이 자산은 flow 또는 다른 자산과 상위 또는 하위 종속성이 없습니다.",
+          "title": "현재 종속성이 없습니다."
+        },
         "EXECUTION": {
           "content": "<a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">실행 종속성</a>에 대한 자세한 내용은 문서를 참조하세요.",
           "title": "현재 종속성이 없습니다."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "선택한 필터에 대한 로그가 없습니다. <br> 필터를 조정하거나, 시간 범위를 변경하거나, flow가 최근에 실행되었는지 확인해 주세요.",
     "no_namespaces": "검색 기준에 맞는 namespace가 없습니다.",
     "no_results": {
+      "assets": "자산을 찾을 수 없음",
       "executions": "실행을 찾을 수 없음",
       "flows": "flow를 찾을 수 없음",
       "kv_pairs": "Key-Value 쌍을 찾을 수 없음",

ui/src/translations/pl.json

@@ -314,6 +314,12 @@
         "message": "Aplikacje w Kestra Enterprise Edition umożliwiają tworzenie niestandardowych interfejsów użytkownika, które współdziałają z workflowami Kestra spoza platformy. Ta funkcja pozwala używać workflowów jako backendu dla niestandardowych aplikacji, umożliwiając użytkownikom nietechnicznym przesyłanie danych za pomocą formularzy lub wznawianie PAUSED workflowów oczekujących na zatwierdzenie.",
         "title": "Twórz własne aplikacje z Kestra"
       },
+      "assets": {
+        "header": "Metadane zasobów i obserwowalność",
+        "label": "Zasoby",
+        "message": "Zasoby łączą metadane dotyczące obserwowalności, pochodzenia i własności, aby zespoły platformowe mogły szybciej rozwiązywać problemy i wdrażać z pewnością.",
+        "title": "Przenieś każdy zestaw danych, usługę i zależność do widoku."
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition rejestruje każdą aktywność za pomocą solidnych, niezmiennych zapisów, co ułatwia śledzenie zmian, utrzymanie zgodności oraz rozwiązywanie problemów.",
         "title": "Śledź zmiany za pomocą Audit Logs"
@@ -389,6 +395,7 @@
         "zoom_out": "Oddal"
       },
       "search": {
+        "asset_placeholder": "Szukaj według asset, flow lub namespace...",
         "no_results": "Nie znaleziono wyników dla {term}",
         "placeholder": "Szukaj według flow lub namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Rozpocznij tworzenie aplikacji do interakcji z Kestra ze świata zewnętrznego.",
         "title": "Nie masz jeszcze aplikacji!"
       },
+      "assets": {
+        "content": "Dodaj zasoby, aby śledzić i zarządzać swoimi zasobami danych, usługami i infrastrukturą.",
+        "title": "Nie masz jeszcze zasobów!"
+      },
       "concurrency_executions": {
         "content": "Przeczytaj więcej o <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong> w naszej dokumentacji.",
         "title": "Brak trwających Executions dla tego flow."
@@ -463,6 +474,10 @@
         "title": "Dla tego flow nie ustawiono żadnych limitów."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Ten zasób nie ma zależności w górę ani w dół z flow lub innymi zasobami.",
+          "title": "Obecnie brak zależności."
+        },
         "EXECUTION": {
           "content": "Przeczytaj więcej o <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">zależnościach wykonania</a> w naszej dokumentacji.",
           "title": "Obecnie brak zależności."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Nie znaleziono logów dla wybranych filtrów. <br> Spróbuj dostosować filtry, zmienić zakres czasu lub sprawdź, czy flow został ostatnio wykonany.",
     "no_namespaces": "Zero namespaces spełnia kryteria wyszukiwania.",
     "no_results": {
+      "assets": "Nie znaleziono zasobów",
       "executions": "Nie znaleziono wykonania",
       "flows": "Nie znaleziono flowów",
       "kv_pairs": "Nie znaleziono par klucz-wartość",

ui/src/translations/pt.json

@@ -314,6 +314,12 @@
         "message": "As aplicações na Kestra Enterprise Edition permitem que você construa interfaces personalizadas que interagem com os workflows do Kestra fora da plataforma. Este recurso permite usar seus workflows como um backend para aplicações personalizadas, possibilitando que usuários não técnicos enviem dados através de formulários ou retomem workflows PAUSED aguardando aprovação.",
         "title": "Crie Apps personalizadas com Kestra"
       },
+      "assets": {
+        "header": "Metadados e Observabilidade de Assets",
+        "label": "Ativos",
+        "message": "Os assets conectam observabilidade, linhagem e metadados de propriedade para que as equipes da plataforma possam solucionar problemas mais rapidamente e implantar com confiança.",
+        "title": "Traga todos os datasets, serviços e dependências para a visualização."
+      },
       "audit-logs": {
         "message": "A Kestra Enterprise Edition registra todas as atividades com registros robustos e imutáveis, facilitando o acompanhamento de alterações, a manutenção da conformidade e a resolução de problemas.",
         "title": "Acompanhe Alterações com Audit Logs"
@@ -389,6 +395,7 @@
         "zoom_out": "Reduzir zoom"
       },
       "search": {
+        "asset_placeholder": "Pesquisar por asset, flow ou namespace...",
         "no_results": "Nenhum resultado encontrado para {term}",
         "placeholder": "Pesquisar por flow ou namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Comece a criar Apps para interagir com o Kestra a partir do mundo externo.",
         "title": "Você ainda não tem apps!"
       },
+      "assets": {
+        "content": "Adicione assets para rastrear e gerenciar seus data assets, serviços e infraestrutura.",
+        "title": "Você ainda não tem Assets!"
+      },
       "concurrency_executions": {
         "content": "Leia mais sobre <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Execuções</a></strong> em nossa documentação.",
         "title": "Nenhuma Execução em andamento para este Flow."
@@ -463,6 +474,10 @@
         "title": "Nenhum limite está definido para este Flow."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Este ativo não possui dependências upstream ou downstream com flows ou outros ativos.",
+          "title": "Atualmente, não há dependências."
+        },
         "EXECUTION": {
           "content": "Leia mais sobre <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">Dependências de Execução</a> na nossa documentação.",
           "title": "Atualmente, não há dependências."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Nenhum log encontrado para os filtros selecionados. <br> Por favor, tente ajustar seus filtros, alterar o intervalo de tempo ou verifique se o flow foi executado recentemente.",
     "no_namespaces": "Nenhum namespace corresponde aos critérios de busca.",
     "no_results": {
+      "assets": "Nenhum Asset Encontrado",
       "executions": "Nenhuma Execução Encontrada",
       "flows": "Nenhum Flow Encontrado",
       "kv_pairs": "Nenhum par de Key-Value encontrado",

ui/src/translations/pt_BR.json

@@ -314,6 +314,12 @@
         "message": "As aplicações na Kestra Enterprise Edition permitem que você construa interfaces personalizadas que interagem com os workflows do Kestra fora da plataforma. Este recurso permite usar seus workflows como um backend para aplicações personalizadas, possibilitando que usuários não técnicos enviem dados através de formulários ou retomem workflows PAUSED aguardando aprovação.",
         "title": "Crie Apps personalizadas com Kestra"
       },
+      "assets": {
+        "header": "Metadados e Observabilidade de Assets",
+        "label": "Ativos",
+        "message": "Ativos conectam metadados de observabilidade, linhagem e propriedade para que as equipes da plataforma possam solucionar problemas mais rapidamente e implantar com confiança.",
+        "title": "Traga todos os datasets, serviços e dependências para a visualização."
+      },
       "audit-logs": {
         "message": "A Kestra Enterprise Edition registra todas as atividades com registros robustos e imutáveis, facilitando o acompanhamento de alterações, a manutenção da conformidade e a resolução de problemas.",
         "title": "Acompanhe Alterações com Audit Logs"
@@ -389,6 +395,7 @@
         "zoom_out": "Reduzir zoom"
       },
       "search": {
+        "asset_placeholder": "Pesquisar por asset, flow ou namespace...",
         "no_results": "Nenhum resultado encontrado para {term}",
         "placeholder": "Pesquisar por flow ou namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Comece a criar Apps para interagir com o Kestra a partir do mundo externo.",
         "title": "Você ainda não tem apps!"
       },
+      "assets": {
+        "content": "Adicione assets para rastrear e gerenciar seus assets de dados, serviços e infraestrutura.",
+        "title": "Você ainda não tem Assets!"
+      },
       "concurrency_executions": {
         "content": "Leia mais sobre <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Execuções</a></strong> em nossa documentação.",
         "title": "Nenhuma Execução em andamento para este Flow."
@@ -463,6 +474,10 @@
         "title": "Nenhum limite está definido para este Flow."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Este ativo não possui dependências upstream ou downstream com flows ou outros ativos.",
+          "title": "Atualmente, não há dependências."
+        },
         "EXECUTION": {
           "content": "Leia mais sobre <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">Dependências de Execução</a> na nossa documentação.",
           "title": "Atualmente, não há dependências."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Nenhum log encontrado para os filtros selecionados. <br> Tente ajustar seus filtros, alterar o intervalo de tempo ou verificar se o flow foi executado recentemente.",
     "no_namespaces": "Nenhum namespace corresponde aos critérios de busca.",
     "no_results": {
+      "assets": "Nenhum Asset Encontrado",
       "executions": "Nenhuma Execução Encontrada",
       "flows": "Nenhum Flow Encontrado",
       "kv_pairs": "Nenhum par de Key-Value encontrado",

ui/src/translations/ru.json

@@ -314,6 +314,12 @@
         "message": "Приложения в Kestra Enterprise Edition позволяют создавать пользовательские интерфейсы, которые взаимодействуют с Kestra workflows вне платформы. Эта функция позволяет использовать ваши workflows в качестве backend для пользовательских приложений, что дает возможность нетехническим пользователям отправлять данные через формы или возобновлять PAUSED workflows, ожидающие одобрения.",
         "title": "Создавайте пользовательские приложения с Kestra"
       },
+      "assets": {
+        "header": "Метаданные и наблюдаемость активов",
+        "label": "Активы",
+        "message": "Активы связывают метаданные наблюдаемости, происхождения и владения, чтобы команды платформы могли быстрее устранять неполадки и развертывать с уверенностью.",
+        "title": "Приведите каждый набор данных, сервис и зависимость в поле зрения."
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition записывает каждую активность с надежными, неизменяемыми записями, что упрощает отслеживание изменений, поддержание соответствия и устранение неполадок.",
         "title": "Отслеживание изменений с помощью Audit Logs"
@@ -389,6 +395,7 @@
         "zoom_out": "Уменьшить масштаб"
       },
       "search": {
+        "asset_placeholder": "Поиск по asset, flow или namespace...",
         "no_results": "Результаты для {term} не найдены",
         "placeholder": "Поиск по flow или namespace..."
       }
@@ -454,6 +461,10 @@
         "content": "Начните создавать приложения для взаимодействия с Kestra из внешнего мира.",
         "title": "У вас пока нет приложений!"
       },
+      "assets": {
+        "content": "Добавьте assets, чтобы отслеживать и управлять вашими данными, сервисами и инфраструктурой.",
+        "title": "У вас еще нет Assets!"
+      },
       "concurrency_executions": {
         "content": "Узнайте больше о <strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong> в нашей документации.",
         "title": "Нет текущих выполнений для этого flow."
@@ -463,6 +474,10 @@
         "title": "Для этого flow не установлены ограничения."
       },
       "dependencies": {
+        "ASSET": {
+          "content": "Этот ресурс не имеет зависимостей вверх или вниз по потоку с flow или другими ресурсами.",
+          "title": "В настоящее время зависимости отсутствуют."
+        },
         "EXECUTION": {
           "content": "Узнайте больше о <a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">зависимостях выполнения</a> в нашей документации.",
           "title": "В настоящее время зависимости отсутствуют."
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "Для выбранных фильтров логи не найдены. <br> Попробуйте изменить фильтры, изменить временной диапазон или проверьте, выполнялся ли flow недавно.",
     "no_namespaces": "Ни один namespace не соответствует критериям поиска.",
     "no_results": {
+      "assets": "Активы не найдены",
       "executions": "Исполнения не найдены",
       "flows": "Потоки не найдены",
       "kv_pairs": "Пары kv store не найдены",

ui/src/translations/zh_CN.json

@@ -314,6 +314,12 @@
         "message": "在 Kestra 企业版中，应用程序允许您构建自定义 UI，以便从平台外部与 Kestra 工作流进行交互。此功能使您可以将工作流用作自定义应用程序的后端，允许非技术用户通过表单提交数据或恢复等待批准的暂停工作流。",
         "title": "使用 Kestra 构建自定义应用程序"
       },
+      "assets": {
+        "header": "资产元数据和可观测性",
+        "label": "资产",
+        "message": "资产连接可观测性、血缘关系和所有权元数据，使平台团队能够更快地排查问题并自信地部署。",
+        "title": "将每个数据集、服务和依赖项展示出来。"
+      },
       "audit-logs": {
         "message": "Kestra Enterprise Edition记录每个活动，提供强大且不可更改的记录，使得跟踪更改、保持合规性和解决问题变得容易。",
         "title": "使用审计日志跟踪更改"
@@ -389,6 +395,7 @@
         "zoom_out": "缩小视图"
       },
       "search": {
+        "asset_placeholder": "按资产、flow或namespace搜索...",
         "no_results": "未找到与 {term} 相关的结果",
         "placeholder": "按flow或namespace搜索..."
       }
@@ -454,6 +461,10 @@
         "content": "开始构建应用程序，以便与外部世界的Kestra进行交互。",
         "title": "您还没有应用程序！"
       },
+      "assets": {
+        "content": "添加资产以跟踪和管理您的数据资产、服务和基础设施。",
+        "title": "您还没有资产！"
+      },
       "concurrency_executions": {
         "content": "在我们的文档中阅读更多关于<strong><a href=\"https://kestra.io/docs/workflow-components/execution\" target=\"_blank\">Executions</a></strong>的信息。",
         "title": "此 Flow 没有正在进行的 Executions。"
@@ -463,6 +474,10 @@
         "title": "此Flow未设置限制。"
       },
       "dependencies": {
+        "ASSET": {
+          "content": "此资产与flow或其他资产没有上游或下游依赖关系。",
+          "title": "当前没有依赖项。"
+        },
         "EXECUTION": {
           "content": "了解更多关于<a href=\"https://kestra.io/docs/ui/executions#dependencies\" target=\"_blank\">执行依赖关系</a>的信息，请参阅我们的文档。",
           "title": "当前没有依赖项。"
@@ -1296,6 +1311,7 @@
     "no_logs_data_description": "未找到所选过滤器的日志。<br>请尝试调整过滤器、更改时间范围，或检查flow最近是否已执行。",
     "no_namespaces": "没有符合搜索条件的命名空间。",
     "no_results": {
+      "assets": "未找到资产",
       "executions": "未找到执行",
       "flows": "未找到Flows",
       "kv_pairs": "未找到Key-Value对",

ui/src/utils/constants.ts

@@ -20,6 +20,8 @@ export const storageKeys = {
     DISPLAY_KV_COLUMNS: "displayKvColumns",
     DISPLAY_SECRETS_COLUMNS: "displaySecretsColumns",
     DISPLAY_TRIGGERS_COLUMNS: "displayTriggersColumns",
+    DISPLAY_ASSETS_COLUMNS: "displayAssetsColumns",
+    DISPLAY_ASSET_EXECUTIONS_COLUMNS: "displayAssetExecutionsColumns",
     SELECTED_TENANT: "selectedTenant",
     EXECUTE_FLOW_BEHAVIOUR: "executeFlowBehaviour",
     SHOW_CHART: "showChart",

webserver/src/test/java/io/kestra/webserver/controllers/api/PluginControllerTest.java

@@ -145,7 +145,7 @@ class PluginControllerTest {
         Map<String, Map<String, Object>> properties = (Map<String, Map<String, Object>>) doc.getSchema().getProperties().get("properties");
 
         assertThat(doc.getMarkdown()).contains("io.kestra.plugin.templates.ExampleTask");
-        assertThat(properties.size()).isEqualTo(18);
+        assertThat(properties.size()).isEqualTo(19);
         assertThat(properties.get("id").size()).isEqualTo(5);
         assertThat(((Map<String, Object>) doc.getSchema().getOutputs().get("properties")).size()).isEqualTo(1);
     }

worker/src/main/java/io/kestra/worker/DefaultWorker.java

@@ -9,6 +9,9 @@ import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.Label;
+import io.kestra.core.models.assets.Asset;
+import io.kestra.core.models.assets.AssetsDeclaration;
+import io.kestra.core.models.assets.AssetsInOut;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.Output;
@@ -955,6 +958,15 @@ public class DefaultWorker implements Worker {
         try {
             Variables variables = variablesService.of(StorageContext.forTask(taskRun), workerTaskCallable.getTaskOutput());
             taskRun = taskRun.withOutputs(variables);
+            if (workerTask.getTask().getAssets() != null) {
+                List<Asset> outputAssets = runContext.assets().outputs();
+                Optional<AssetsDeclaration> renderedAssetsDeclaration = runContext.render(workerTask.getTask().getAssets()).as(AssetsDeclaration.class);
+                renderedAssetsDeclaration.map(AssetsDeclaration::getOutputs).ifPresent(outputAssets::addAll);
+                taskRun = taskRun.withAssets(new AssetsInOut(
+                    renderedAssetsDeclaration.map(AssetsDeclaration::getInputs).orElse(null),
+                    outputAssets
+                ));
+            }
         } catch (Exception e) {
             logger.warn("Unable to save output on taskRun '{}'", taskRun, e);
         }