From 7ea95f393e5e2a8ee36bf77c331c492a2fb90013 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Mathieu?= <loikeseke@gmail.com>
Date: Mon, 15 Dec 2025 13:28:02 +0100
Subject: [PATCH] feat(execution): add a `system.from` label

Closes https://github.com/kestra-io/kestra-ee/issues/4699
---
 .../java/io/kestra/core/models/Label.java     |  1 +
 .../core/models/triggers/TriggerService.java  |  1 +
 .../trigger/SchedulableExecutionFactory.java  |  1 +
 .../plugin/core/trigger/ScheduleTest.java     | 30 +++++++-------
 .../SchedulerPollingTriggerTest.java          |  2 +
 .../scheduler/SchedulerStreamingTest.java     |  1 +
 ui/src/components/flows/FlowRun.vue           | 16 ++++----
 .../controllers/api/ExecutionController.java  | 39 +++++++++----------
 .../api/ExecutionControllerRunnerTest.java    | 25 +++++++-----
 9 files changed, 65 insertions(+), 51 deletions(-)

diff --git a/core/src/main/java/io/kestra/core/models/Label.java b/core/src/main/java/io/kestra/core/models/Label.java
index d6141cd238..d1ceabb653 100644
--- a/core/src/main/java/io/kestra/core/models/Label.java
+++ b/core/src/main/java/io/kestra/core/models/Label.java
@@ -26,6 +26,7 @@ public record Label(
     public static final String REPLAYED = SYSTEM_PREFIX + "replayed";
     public static final String SIMULATED_EXECUTION = SYSTEM_PREFIX + "simulatedExecution";
     public static final String TEST = SYSTEM_PREFIX + "test";
+    public static final String FROM = SYSTEM_PREFIX + "from";
 
     /**
      * Static helper method for converting a list of labels to a nested map.
diff --git a/core/src/main/java/io/kestra/core/models/triggers/TriggerService.java b/core/src/main/java/io/kestra/core/models/triggers/TriggerService.java
index 0c5a028794..6cea547d8f 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/TriggerService.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/TriggerService.java
@@ -56,6 +56,7 @@ public abstract class TriggerService {
         ConditionContext conditionContext
     ) {
         List<Label> executionLabels = new ArrayList<>(ListUtils.emptyOnNull(trigger.getLabels()));
+        executionLabels.add(new Label(Label.FROM, "trigger"));
         if (executionLabels.stream().noneMatch(label -> Label.CORRELATION_ID.equals(label.key()))) {
             // add a correlation ID if none exist
             executionLabels.add(new Label(Label.CORRELATION_ID, id));
diff --git a/core/src/main/java/io/kestra/plugin/core/trigger/SchedulableExecutionFactory.java b/core/src/main/java/io/kestra/plugin/core/trigger/SchedulableExecutionFactory.java
index 890a867d3a..861e3f4ffa 100644
--- a/core/src/main/java/io/kestra/plugin/core/trigger/SchedulableExecutionFactory.java
+++ b/core/src/main/java/io/kestra/plugin/core/trigger/SchedulableExecutionFactory.java
@@ -50,6 +50,7 @@ final class SchedulableExecutionFactory {
         List<Label> labels = getLabels(trigger, runContext, triggerContext.getBackfill(), conditionContext.getFlow());
 
         List<Label> executionLabels = new ArrayList<>(ListUtils.emptyOnNull(labels));
+        executionLabels.add(new Label(Label.FROM, "trigger"));
         if (executionLabels.stream().noneMatch(label -> Label.CORRELATION_ID.equals(label.key()))) {
             // add a correlation ID if none exist
             executionLabels.add(new Label(Label.CORRELATION_ID, runContext.getTriggerExecutionId()));
diff --git a/core/src/test/java/io/kestra/plugin/core/trigger/ScheduleTest.java b/core/src/test/java/io/kestra/plugin/core/trigger/ScheduleTest.java
index bb7afeeac1..1f37a3c8a7 100644
--- a/core/src/test/java/io/kestra/plugin/core/trigger/ScheduleTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/trigger/ScheduleTest.java
@@ -104,8 +104,9 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-        assertThat(evaluate.get().getLabels()).hasSize(3);
+        assertThat(evaluate.get().getLabels()).hasSize(4);
         assertTrue(evaluate.get().getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
+        assertTrue(evaluate.get().getLabels().stream().anyMatch(label -> label.equals(new Label(Label.FROM, "trigger"))));
         assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
         var vars = evaluate.get().getTrigger().getVariables();
         var inputs = evaluate.get().getInputs();
@@ -138,8 +139,9 @@ class ScheduleTest {
         );
 
         assertThat(evaluate.isPresent()).isTrue();
-        assertThat(evaluate.get().getLabels()).hasSize(3);
+        assertThat(evaluate.get().getLabels()).hasSize(4);
         assertTrue(evaluate.get().getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
+        assertTrue(evaluate.get().getLabels().stream().anyMatch(label -> label.equals(new Label(Label.FROM, "trigger"))));
         assertThat(evaluate.get().getVariables()).containsEntry("custom_var", "VARIABLE VALUE");
         var inputs = evaluate.get().getInputs();
 
@@ -645,14 +647,14 @@ class ScheduleTest {
     private ZonedDateTime dateFromVars(String date, ZonedDateTime expexted) {
         return ZonedDateTime.parse(date).withZoneSameInstant(expexted.getZone());
     }
-    
+
     @Test
     void shouldGetNextExecutionDateWithConditionMatchingFutureDate() throws InternalException {
-        
+
         ZonedDateTime now = ZonedDateTime.now().withZoneSameLocal(ZoneId.of("Europe/Paris"));
         OffsetTime before = now.minusHours(1).toOffsetDateTime().toOffsetTime().withMinute(0).withSecond(0).withNano(0);
         OffsetTime after = now.minusHours(4).toOffsetDateTime().toOffsetTime().withMinute(0).withSecond(0).withNano(0);
-        
+
         Schedule trigger = Schedule.builder()
             .id("schedule").type(Schedule.class.getName())
             .cron("0 * * * *") // every hour
@@ -665,25 +667,25 @@ class ScheduleTest {
                 .build()
             ))
             .build();
-        
+
         TriggerContext triggerContext = triggerContext(now, trigger).toBuilder().build();
-        
+
         ConditionContext conditionContext = ConditionContext.builder()
             .runContext(runContextInitializer.forScheduler((DefaultRunContext) runContextFactory.of(), triggerContext, trigger))
             .build();
-        
+
         Optional<ZonedDateTime> result = trigger.truePreviousNextDateWithCondition(trigger.executionTime(), conditionContext, now, true);
         assertThat(result).isNotEmpty();
     }
-    
+
     @Test
     void shouldGetNextExecutionDateWithConditionMatchingCurrentDate() throws InternalException {
-        
+
         ZonedDateTime now = ZonedDateTime.now().withZoneSameLocal(ZoneId.of("Europe/Paris"));
 
         OffsetTime before = now.plusHours(2).toOffsetDateTime().toOffsetTime().withMinute(0).withSecond(0).withNano(0);
         OffsetTime after = now.minusHours(2).toOffsetDateTime().toOffsetTime().withMinute(0).withSecond(0).withNano(0);
-        
+
         Schedule trigger = Schedule.builder()
             .id("schedule").type(Schedule.class.getName())
             .cron("*/30 * * * * *")
@@ -696,13 +698,13 @@ class ScheduleTest {
                 .build()
             ))
             .build();
-        
+
         TriggerContext triggerContext = triggerContext(now, trigger).toBuilder().build();
-        
+
         ConditionContext conditionContext = ConditionContext.builder()
             .runContext(runContextInitializer.forScheduler((DefaultRunContext) runContextFactory.of(), triggerContext, trigger))
             .build();
-        
+
         Optional<ZonedDateTime> result = trigger.truePreviousNextDateWithCondition(trigger.executionTime(), conditionContext, now, true);
         assertThat(result).isNotEmpty();
     }
diff --git a/scheduler/src/test/java/io/kestra/scheduler/SchedulerPollingTriggerTest.java b/scheduler/src/test/java/io/kestra/scheduler/SchedulerPollingTriggerTest.java
index f5fbdec73d..31b03d3036 100644
--- a/scheduler/src/test/java/io/kestra/scheduler/SchedulerPollingTriggerTest.java
+++ b/scheduler/src/test/java/io/kestra/scheduler/SchedulerPollingTriggerTest.java
@@ -91,6 +91,7 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
             assertThat(queueCount.getCount()).isEqualTo(0L);
             assertThat(last.get()).isNotNull();
             assertTrue(last.get().getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
+            assertTrue(last.get().getLabels().stream().anyMatch(label -> label.equals(new Label(Label.FROM, "trigger"))));
         }
     }
 
@@ -136,6 +137,7 @@ public class SchedulerPollingTriggerTest extends AbstractSchedulerTest {
             assertThat(queueCount.getCount()).isEqualTo(0L);
             assertThat(last.get()).isNotNull();
             assertTrue(last.get().getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
+            assertTrue(last.get().getLabels().stream().anyMatch(label -> label.equals(new Label(Label.FROM, "trigger"))));
 
             // Assert that the trigger is now disabled.
             // It needs to await on assertion as it will be disabled AFTER we receive a success execution.
diff --git a/scheduler/src/test/java/io/kestra/scheduler/SchedulerStreamingTest.java b/scheduler/src/test/java/io/kestra/scheduler/SchedulerStreamingTest.java
index 455250aac0..9d805b10fd 100644
--- a/scheduler/src/test/java/io/kestra/scheduler/SchedulerStreamingTest.java
+++ b/scheduler/src/test/java/io/kestra/scheduler/SchedulerStreamingTest.java
@@ -104,6 +104,7 @@ public class SchedulerStreamingTest extends AbstractSchedulerTest {
                 assertThat(SchedulerStreamingTest.startedEvaluate.get(false), is(1));
                 assertThat(last.getTrigger().getVariables().get("startedEvaluate"), is(1));
                 assertTrue(last.getLabels().stream().anyMatch(label -> label.key().equals(Label.CORRELATION_ID)));
+                assertTrue(last.getLabels().stream().anyMatch(label -> label.equals(new Label(Label.FROM, "trigger"))));
             }
         );
     }
diff --git a/ui/src/components/flows/FlowRun.vue b/ui/src/components/flows/FlowRun.vue
index 66be920771..2c24279adf 100644
--- a/ui/src/components/flows/FlowRun.vue
+++ b/ui/src/components/flows/FlowRun.vue
@@ -10,15 +10,15 @@
             </el-alert>
         </div>
         <el-form labelPosition="top" :model="inputs" ref="form" @submit.prevent="false">
-            <InputsForm 
-                :initialInputs="flow.inputs" 
-                :selectedTrigger="selectedTrigger" 
-                :flow="flow" 
+            <InputsForm
+                :initialInputs="flow.inputs"
+                :selectedTrigger="selectedTrigger"
+                :flow="flow"
                 v-model="inputs"
                 :executeClicked="executeClicked"
                 @confirm="onSubmit($refs.form)"
-                @update:model-value-no-default="values => inputsNoDefaults=values" 
-                @update:checks="values => checks=values" 
+                @update:model-value-no-default="values => inputsNoDefaults=values"
+                @update:checks="values => checks=values"
             />
 
             <el-collapse v-model="collapseName">
@@ -208,7 +208,7 @@
                                     this.executionLabels
                                         .filter(label => label.key && label.value)
                                         .map(label => `${label.key}:${label.value}`)
-                                )],
+                                ), "system.from:ui"],
                                 scheduleDate: this.scheduleDate
                             });
                         } else {
@@ -221,7 +221,7 @@
                                     this.executionLabels
                                         .filter(label => label.key && label.value)
                                         .map(label => `${label.key}:${label.value}`)
-                                )],
+                                ), "system.from:ui"],
                                 scheduleDate: this.$moment(this.scheduleDate).tz(localStorage.getItem(storageKeys.TIMEZONE_STORAGE_KEY) ?? moment.tz.guess()).toISOString(true),
                                 nextStep: true,
                             });
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 6cea36b90c..74446331db 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -38,9 +38,6 @@ import io.kestra.core.utils.ListUtils;
 import io.kestra.core.utils.Logs;
 import io.kestra.plugin.core.flow.Pause;
 import io.kestra.plugin.core.trigger.Webhook;
-import io.kestra.webserver.controllers.api.ExecutionController.ApiValidateExecutionInputsResponse;
-import io.kestra.webserver.controllers.api.ExecutionController.ApiValidateExecutionInputsResponse.ApiCheckFailure;
-import io.kestra.webserver.controllers.api.ExecutionController.ExecutionResponse;
 import io.kestra.webserver.converters.QueryFilterFormat;
 import io.kestra.webserver.responses.BulkErrorResponse;
 import io.kestra.webserver.responses.BulkResponse;
@@ -590,21 +587,16 @@ public class ExecutionController {
             throw new HttpStatusException(HttpStatus.NOT_FOUND, "No execution triggered");
         }
 
-        var result = execution.get();
+        List<Label> labels = new ArrayList<>();
+        labels.add(new Label(Label.FROM, "trigger"));
         if (flow.getLabels() != null) {
-            result = result.withLabels(LabelService.labelsExcludingSystem(flow));
+            labels.addAll(LabelService.labelsExcludingSystem(flow));
+        }
+        if (labels.stream().noneMatch(label -> label.key().equals(CORRELATION_ID))) {
+            labels.add(new Label(CORRELATION_ID, execution.get().getId()));
         }
 
-        List<Label> labels = ListUtils.emptyOnNull(result.getLabels());
-
-        boolean hasCorrelationId = labels.stream()
-            .anyMatch(label -> label.key().equals(CORRELATION_ID));
-
-        if (!hasCorrelationId) {
-            List<Label> newLabels = new ArrayList<>(labels);
-            newLabels.add(new Label(CORRELATION_ID, result.getId()));
-            result = result.withLabels(newLabels);
-        }
+        var result = execution.get().withLabels(labels);
 
         // we check conditions here as it's easier as the execution is created we have the body and headers available for the runContext
         var conditionContext = conditionService.conditionContext(runContextFactory.of(flow, result), flow, result);
@@ -636,7 +628,7 @@ public class ExecutionController {
             }
 
             executionQueue.emit(result);
-            eventPublisher.publishEvent(new CrudEvent<>(result, CrudEventType.CREATE));
+            eventPublisher.publishEvent(CrudEvent.create(result));
 
             if (webhook.getWait()) {
                 var subscriberId = UUID.randomUUID().toString();
@@ -864,15 +856,22 @@ public class ExecutionController {
     }
 
     protected List<Label> parseLabels(List<String> labels) {
-        List<Label> parsedLabels = labels == null ? Collections.emptyList() : RequestUtils.toMap(labels).entrySet().stream()
+        List<Label> parsedLabels = labels == null ? new ArrayList<>() : RequestUtils.toMap(labels).entrySet().stream()
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
-            .toList();
+            .collect(Collectors.toList());
 
-        // check for system labels: none can be passed at execution creation time except system.correlationId
-        Optional<Label> first = parsedLabels.stream().filter(label -> !label.key().equals(CORRELATION_ID) && label.key().startsWith(SYSTEM_PREFIX)).findFirst();
+        // check for system labels: none can be passed at execution creation time except system.correlationId and system.from
+        Optional<Label> first = parsedLabels.stream().filter(label -> !label.key().equals(CORRELATION_ID) && !label.key().equals(Label.FROM) && label.key().startsWith(SYSTEM_PREFIX)).findFirst();
         if (first.isPresent()) {
             throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().key() + "=" + first.get().value());
         }
+
+        // from can be passed by the UI so we only add it if it didn't exist anymore
+        // if we want to be more restrictive, we may want to restrict it to only have the `ui` value
+        if (parsedLabels.stream().noneMatch(l -> l.key().equals(Label.FROM))) {
+            parsedLabels.add(new Label(Label.FROM, "api"));
+        }
+
         return parsedLabels;
     }
 
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java
index d1e3ecaacc..f1076d9e5a 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java
@@ -53,7 +53,6 @@ import org.awaitility.Awaitility;
 import org.hamcrest.Matcher;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
-import org.junitpioneer.jupiter.RetryingTest;
 import reactor.core.publisher.Flux;
 
 import java.io.ByteArrayInputStream;
@@ -170,12 +169,15 @@ class ExecutionControllerRunnerTest {
         assertThat(result.getInputs().get("file").toString()).startsWith("kestra:///io/kestra/tests/inputs/executions/");
         assertThat(result.getInputs().containsKey("bool")).isTrue();
         assertThat(result.getInputs().get("bool")).isNull();
-        assertThat(result.getLabels().size()).isEqualTo(6);
-        assertThat(result.getLabels().getFirst()).isEqualTo(new Label("flow-label-1", "flow-label-1"));
-        assertThat(result.getLabels().get(1)).isEqualTo(new Label("flow-label-2", "flow-label-2"));
-        assertThat(result.getLabels().get(2)).isEqualTo(new Label("a", "label-1"));
-        assertThat(result.getLabels().get(3)).isEqualTo(new Label("b", "label-2"));
-        assertThat(result.getLabels().get(4)).isEqualTo(new Label("url", URL_LABEL_VALUE));
+        assertThat(result.getLabels()).containsExactlyInAnyOrder(
+            new Label("flow-label-1", "flow-label-1"),
+            new Label("flow-label-2", "flow-label-2"),
+            new Label("a", "label-1"),
+            new Label("b", "label-2"),
+            new Label("url", URL_LABEL_VALUE),
+            new Label(Label.CORRELATION_ID, result.getId()),
+            new Label(Label.FROM, "api")
+        );
 
         var notFound = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().exchange(
             HttpRequest
@@ -203,6 +205,7 @@ class ExecutionControllerRunnerTest {
 
         assertThat(execution.getLabels()).containsExactlyInAnyOrder(
             new Label(Label.CORRELATION_ID, execution.getId()),
+            new Label(Label.FROM, "api"),
             new Label("existing", "fromExecution")
         );
     }
@@ -908,8 +911,12 @@ class ExecutionControllerRunnerTest {
         assertThat((Boolean) ((Map<String, Object>) execution.getTrigger().getVariables().get("body")).get("b")).isTrue();
         assertThat(((Map<String, Object>) execution.getTrigger().getVariables().get("parameters")).get("name")).isEqualTo(List.of("john"));
         assertThat(((Map<String, List<String>>) execution.getTrigger().getVariables().get("parameters")).get("age")).containsExactlyInAnyOrder("12", "13");
-        assertThat(execution.getLabels().getFirst()).isEqualTo(new Label("flow-label-1", "flow-label-1"));
-        assertThat(execution.getLabels().get(1)).isEqualTo(new Label("flow-label-2", "flow-label-2"));
+        assertThat(execution.getLabels()).containsExactlyInAnyOrder(
+            new Label(Label.CORRELATION_ID, execution.getId()),
+            new Label(Label.FROM, "trigger"),
+            new Label("flow-label-1", "flow-label-1"),
+            new Label("flow-label-2", "flow-label-2")
+        );
 
         execution = client.toBlocking().retrieve(
             HttpRequest