Add stopped state support for pipeline status and stop/cancel AI agents

openmetadata-service/src/main/java/org/openmetadata/service/clients/pipeline/MeteredPipelineServiceClient.java

@@ -141,6 +141,13 @@ public PipelineServiceClientResponse killIngestion(IngestionPipeline ingestionPi
         KILL, () -> this.decoratedClient.killIngestion(ingestionPipeline));
   }
 
+  @Override
+  public PipelineServiceClientResponse killIngestionRun(
+      IngestionPipeline ingestionPipeline, String runId) {
+    return this.respondWithMetering(
+        KILL, () -> this.decoratedClient.killIngestionRun(ingestionPipeline, runId));
+  }
+
   @Override
   public String getPlatform() {
     return this.decoratedClient.getPlatform();

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DataInsightSystemChartRepository.java

@@ -549,6 +549,7 @@ private AppRunRecord convertPipelineStatusToAppRun(App app, PipelineStatus pipel
               case SUCCESS -> AppRunRecord.Status.SUCCESS;
               case FAILED, PARTIAL_SUCCESS -> AppRunRecord.Status.FAILED;
               case RUNNING -> AppRunRecord.Status.RUNNING;
+              case STOPPED -> AppRunRecord.Status.STOPPED;
             })
         .withConfig(pipelineStatus.getConfig());
   }

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java

@@ -702,13 +702,17 @@ public PipelineStatus getLatestPipelineStatus(IngestionPipeline ingestionPipelin
   }
 
   public PipelineStatus getPipelineStatus(String ingestionPipelineFQN, UUID pipelineStatusRunId) {
+    return getPipelineStatus(ingestionPipelineFQN, pipelineStatusRunId.toString());
+  }
+
+  public PipelineStatus getPipelineStatus(String ingestionPipelineFQN, String runId) {
     IngestionPipeline ingestionPipeline = findByName(ingestionPipelineFQN, Include.NON_DELETED);
     return JsonUtils.readValue(
         daoCollection
             .entityExtensionTimeSeriesDao()
             .getExtensionByKey(
                 RUN_ID_EXTENSION_KEY,
-                pipelineStatusRunId.toString(),
+                runId,
                 ingestionPipeline.getFullyQualifiedName(),
                 PIPELINE_STATUS_EXTENSION),
         PipelineStatus.class);

openmetadata-service/src/main/java/org/openmetadata/service/resources/apps/AppResource.java

@@ -59,6 +59,7 @@
 import org.openmetadata.schema.entity.services.ingestionPipelines.IngestionPipeline;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineServiceClientResponse;
 import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatus;
+import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatusType;
 import org.openmetadata.schema.services.connections.metadata.OpenMetadataConnection;
 import org.openmetadata.schema.type.EntityHistory;
 import org.openmetadata.schema.type.EntityReference;
@@ -374,8 +375,13 @@ protected static AppRunRecord convertPipelineStatus(App app, PipelineStatus pipe
               case SUCCESS -> AppRunRecord.Status.SUCCESS;
               case FAILED, PARTIAL_SUCCESS -> AppRunRecord.Status.FAILED;
               case RUNNING -> AppRunRecord.Status.RUNNING;
+              case STOPPED -> AppRunRecord.Status.STOPPED;
             })
-        .withConfig(pipelineStatus.getConfig());
+        .withConfig(pipelineStatus.getConfig())
+        .withProperties(
+            pipelineStatus.getRunId() != null
+                ? Map.of("pipelineRunId", pipelineStatus.getRunId())
+                : null);
   }
 
   private ResultList<AppRunRecord> sortRunsByStartTime(ResultList<AppRunRecord> runs) {
@@ -1227,7 +1233,12 @@ public Response stopApplicationRun(
       @Context SecurityContext securityContext,
       @Parameter(description = "Name of the App", schema = @Schema(type = "string"))
           @PathParam("name")
-          String name) {
+          String name,
+      @Parameter(
+              description = "Pipeline run ID (Argo workflow UID) to stop a specific run",
+              schema = @Schema(type = "string"))
+          @QueryParam("runId")
+          String runId) {
     EntityUtil.Fields fields = getFields(String.format("%s,bot,pipelines", FIELD_OWNERS));
     App app = repository.getByName(uriInfo, name, fields);
     OperationContext operationContext = new OperationContext(entityType, MetadataOperation.TRIGGER);
@@ -1243,15 +1254,83 @@ public Response stopApplicationRun(
       } else {
         if (!app.getPipelines().isEmpty()) {
           IngestionPipeline ingestionPipeline = getIngestionPipeline(uriInfo, securityContext, app);
-          PipelineServiceClientResponse response =
-              pipelineServiceClient.killIngestion(ingestionPipeline);
-          return Response.status(response.getCode()).entity(response).build();
+          if (runId != null && !runId.isBlank()) {
+            markPipelineStatusAsStopped(uriInfo, ingestionPipeline, runId);
+            PipelineServiceClientResponse response =
+                pipelineServiceClient.killIngestionRun(ingestionPipeline, runId);
+            return Response.status(response.getCode()).entity(response).build();
+          } else {
+            markLatestPipelineStatusAsStopped(uriInfo, ingestionPipeline);
+            PipelineServiceClientResponse response =
+                pipelineServiceClient.killIngestion(ingestionPipeline);
+            return Response.status(response.getCode()).entity(response).build();
+          }
         }
       }
     }
     throw new BadRequestException("Application does not support Interrupts.");
   }
 
+  private void markPipelineStatusAsStopped(
+      UriInfo uriInfo, IngestionPipeline ingestionPipeline, String runId) {
+    try {
+      IngestionPipelineRepository ingestionPipelineRepository =
+          (IngestionPipelineRepository) Entity.getEntityRepository(Entity.INGESTION_PIPELINE);
+      PipelineStatus status =
+          ingestionPipelineRepository.getPipelineStatus(
+              ingestionPipeline.getFullyQualifiedName(), runId);
+      if (status == null) {
+        LOG.warn("Pipeline status not found for run {}, skipping DB update", runId);
+        return;
+      }
+      if (!isTerminalState(status.getPipelineState())) {
+        status.setPipelineState(PipelineStatusType.STOPPED);
+        status.setEndDate(System.currentTimeMillis());
+        ingestionPipelineRepository.addPipelineStatus(
+            uriInfo, ingestionPipeline.getFullyQualifiedName(), status);
+      }
+    } catch (RuntimeException e) {
+      LOG.warn(
+          "Failed to mark pipeline run {} as stopped, continuing with kill: {}",
+          runId,
+          e.getMessage());
+    }
+  }
+
+  private void markLatestPipelineStatusAsStopped(
+      UriInfo uriInfo, IngestionPipeline ingestionPipeline) {
+    try {
+      IngestionPipelineRepository ingestionPipelineRepository =
+          (IngestionPipelineRepository) Entity.getEntityRepository(Entity.INGESTION_PIPELINE);
+      // Limit to 20: covers any burst of concurrent runs while avoiding unbounded DB scans.
+      // This path is only hit when no specific runId is provided (fallback kill-all).
+      ResultList<PipelineStatus> recentStatuses =
+          ingestionPipelineRepository.listPipelineStatus(
+              ingestionPipeline.getFullyQualifiedName(), null, null, 20);
+      long now = System.currentTimeMillis();
+      for (PipelineStatus status : recentStatuses.getData()) {
+        if (!isTerminalState(status.getPipelineState())) {
+          status.setPipelineState(PipelineStatusType.STOPPED);
+          status.setEndDate(now);
+          ingestionPipelineRepository.addPipelineStatus(
+              uriInfo, ingestionPipeline.getFullyQualifiedName(), status);
+        }
+      }
+    } catch (RuntimeException e) {
+      LOG.warn("Failed to mark pipeline runs as stopped, continuing with kill: {}", e.getMessage());
+    }
+  }
+
+  private static boolean isTerminalState(PipelineStatusType state) {
+    if (state == null) {
+      return false;
+    }
+    return state == PipelineStatusType.SUCCESS
+        || state == PipelineStatusType.FAILED
+        || state == PipelineStatusType.STOPPED
+        || state == PipelineStatusType.PARTIAL_SUCCESS;
+  }
+
   @POST
   @Path("/deploy/{name}")
   @Operation(

openmetadata-spec/src/main/java/org/openmetadata/sdk/PipelineServiceClientInterface.java

@@ -130,5 +130,11 @@ default PipelineServiceClientResponse runPipeline(
   /* Get the all last run logs of a deployed pipeline */
   PipelineServiceClientResponse killIngestion(IngestionPipeline ingestionPipeline);
 
+  /* Stop a specific run of a deployed pipeline identified by its run ID (Argo workflow UID) */
+  default PipelineServiceClientResponse killIngestionRun(
+      IngestionPipeline ingestionPipeline, String runId) {
+    return killIngestion(ingestionPipeline);
+  }
+
   String getPlatform();
 }

openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/ingestionPipeline.json

@@ -26,7 +26,7 @@
           "description": "Pipeline status denotes if its failed or succeeded.",
           "type": "string",
           "javaType": "org.openmetadata.schema.entity.services.ingestionPipelines.PipelineStatusType",
-          "enum": ["queued","success","failed","running","partialSuccess"]
+          "enum": ["queued","success","failed","running","partialSuccess","stopped"]
         },
         "startDate": {
           "description": "startDate of the pipeline run for this particular execution.",

openmetadata-ui/src/main/resources/ui/src/components/Modals/StopScheduleRun/StopScheduleRunModal.interface.ts

@@ -15,6 +15,7 @@ export interface StopScheduleRunModalProps {
   appName: string;
   isModalOpen: boolean;
   displayName: string;
+  runId?: string;
   onClose: () => void;
   onStopWorkflowsUpdate?: () => void;
 }

openmetadata-ui/src/main/resources/ui/src/components/Modals/StopScheduleRun/StopScheduleRunModal.tsx

@@ -23,6 +23,7 @@ const StopScheduleModal: FC<StopScheduleRunModalProps> = ({
   appName,
   isModalOpen,
   displayName,
+  runId,
   onClose,
   onStopWorkflowsUpdate,
 }) => {
@@ -32,7 +33,7 @@ const StopScheduleModal: FC<StopScheduleRunModalProps> = ({
   const handleConfirm = async () => {
     setIsLoading(true);
     try {
-      const { status } = await stopApp(appName);
+      const { status } = await stopApp(appName, runId);
       if (status === 200) {
         showSuccessToast(
           t('message.application-stop', {

openmetadata-ui/src/main/resources/ui/src/components/Settings/Applications/AppRunsHistory/AppRunsHistory.component.tsx

@@ -86,6 +86,9 @@ const AppRunsHistory = forwardRef(
     >([]);
     const [expandedRowKeys, setExpandedRowKeys] = useState<string[]>([]);
     const [isStopModalOpen, setIsStopModalOpen] = useState<boolean>(false);
+    const [selectedRunId, setSelectedRunId] = useState<string | undefined>(
+      undefined
+    );
     const [showConfigModal, setShowConfigModal] = useState<boolean>(false);
     const [appRunRecordConfig, setAppRunRecordConfig] = useState<
       AppRunRecord['config']
@@ -198,16 +201,23 @@ const AppRunsHistory = forwardRef(
               onClick={() => showAppRunConfig(record)}>
               {t('label.config')}
             </Button>
-            {/* For status running or activewitherror and supportsInterrupt is true, show stop button */}
-            {(record.status === Status.Running ||
-              record.status === Status.ActiveError) &&
+            {record.status !== Status.Success &&
+              record.status !== Status.Failed &&
+              record.status !== Status.Stopped &&
+              record.status !== Status.Completed &&
               Boolean(appData?.supportsInterrupt) && (
                 <Button
                   className="m-l-xs p-0"
                   data-testid="stop-button"
                   size="small"
                   type="link"
-                  onClick={() => setIsStopModalOpen(true)}>
+                  onClick={() => {
+                    const rawRunId = record.properties?.pipelineRunId;
+                    setSelectedRunId(
+                      typeof rawRunId === 'string' ? rawRunId : undefined
+                    );
+                    setIsStopModalOpen(true);
+                  }}>
                   {t('label.stop')}
                 </Button>
               )}
@@ -452,8 +462,10 @@ const AppRunsHistory = forwardRef(
             appName={fqn}
             displayName={appData?.displayName ?? ''}
             isModalOpen={isStopModalOpen}
+            runId={selectedRunId}
             onClose={() => {
               setIsStopModalOpen(false);
+              setSelectedRunId(undefined);
             }}
             onStopWorkflowsUpdate={() => {
               fetchAppHistory();

openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/ingestionPipeline.ts

@@ -761,6 +761,7 @@ export enum PipelineState {
     PartialSuccess = "partialSuccess",
     Queued = "queued",
     Running = "running",
+    Stopped = "stopped",
     Success = "success",
 }
 

openmetadata-ui/src/main/resources/ui/src/rest/applicationAPI.ts

@@ -155,8 +155,14 @@ export const restoreApp = async (id: string) => {
   return response.data;
 };
 
-export const stopApp = async (name: string) => {
-  return await APIClient.post(`${BASE_URL}/stop/${getEncodedFqn(name)}`);
+export const stopApp = async (name: string, runId?: string) => {
+  return await APIClient.post(
+    `${BASE_URL}/stop/${getEncodedFqn(name)}`,
+    undefined,
+    {
+      params: runId ? { runId } : undefined,
+    }
+  );
 };
 
 export const getApplicationLogs = (appName: string, after?: string) => {