[Streams] Mark ES|QL rule execution errors as user-triggered (#255011)

## 📓 Summary

Closes #255003 

The ES|QL request executor in Streams rules currently throws a plain
`Error` when query execution fails. Task Manager treats these as
framework errors, which can trigger unnecessary retries and negatively
impact task health metrics. This change wraps the thrown error with
`createTaskRunError` using `TaskErrorSource.USER`, correctly attributing
failures (e.g., invalid ES|QL queries written by the user) to user input
rather than the platform.

## 🧪 Testing
- Create a Stream rule with an invalid ES|QL query and verify the error
is reported without triggering framework-level retries.
- Confirm valid ES|QL rules continue to execute and return results as
expected.

---------

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

Author: tonyghiani

x-pack/platform/plugins/shared/streams/moon.yml

@@ -79,6 +79,8 @@ dependsOn:
   - '@kbn/core-http-common'
   - '@kbn/scout'
   - '@kbn/esql-language'
+  - '@kbn/core-elasticsearch-server-mocks'
+  - '@kbn/core-logging-server-mocks'
 tags:
   - plugin
   - prod

x-pack/platform/plugins/shared/streams/server/lib/rules/esql/lib/execute_esql_request.test.ts

@@ -0,0 +1,77 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { elasticsearchServiceMock } from '@kbn/core-elasticsearch-server-mocks';
+import { loggingSystemMock } from '@kbn/core-logging-server-mocks';
+import { getErrorSource, TaskErrorSource } from '@kbn/task-manager-plugin/server/task_running';
+import { executeEsqlRequest } from './execute_esql_request';
+
+describe('executeEsqlRequest', () => {
+  const logger = loggingSystemMock.createLogger();
+  const esClient = elasticsearchServiceMock.createElasticsearchClient();
+
+  const esqlRequest = {
+    query: 'FROM logs-* | WHERE host.name == "test"',
+    filter: { match_all: {} },
+  };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('returns parsed results when the query succeeds', async () => {
+    esClient.esql.query.mockResolvedValueOnce({
+      columns: [
+        { name: '_source', type: 'object' },
+        { name: '_id', type: 'keyword' },
+      ],
+      values: [[{ host: { name: 'test' } }, 'doc-1']],
+    } as never);
+
+    const results = await executeEsqlRequest({ esClient, esqlRequest, logger });
+
+    expect(results).toEqual([{ _id: 'doc-1', _source: { host: { name: 'test' } } }]);
+  });
+
+  it('returns an empty array when _source or _id columns are missing', async () => {
+    esClient.esql.query.mockResolvedValueOnce({
+      columns: [{ name: 'host.name', type: 'keyword' }],
+      values: [['test']],
+    } as never);
+
+    const results = await executeEsqlRequest({ esClient, esqlRequest, logger });
+
+    expect(results).toEqual([]);
+  });
+
+  it('throws a user error when the ES|QL query fails', async () => {
+    esClient.esql.query.mockRejectedValueOnce(
+      new Error('verification_exception: Unknown index [logs.kafka]')
+    );
+
+    try {
+      await executeEsqlRequest({ esClient, esqlRequest, logger });
+      fail('Expected executeEsqlRequest to throw');
+    } catch (e) {
+      const error = e as Error;
+      expect(error).toBeInstanceOf(Error);
+      expect(error.message).toContain('Error executing ES|QL request');
+      expect(error.message).toContain('verification_exception');
+      expect(getErrorSource(error)).toBe(TaskErrorSource.USER);
+    }
+  });
+
+  it('logs the error message at debug level', async () => {
+    esClient.esql.query.mockRejectedValueOnce(new Error('some query error'));
+
+    await expect(executeEsqlRequest({ esClient, esqlRequest, logger })).rejects.toThrow();
+
+    expect(logger.debug).toHaveBeenCalledWith(
+      expect.stringContaining('Error executing ES|QL request: some query error')
+    );
+  });
+});

x-pack/platform/plugins/shared/streams/server/lib/rules/esql/lib/execute_esql_request.ts

@@ -8,6 +8,7 @@
 import type { estypes } from '@elastic/elasticsearch';
 import type { ElasticsearchClient, Logger } from '@kbn/core/server';
 import type { ESQLSearchResponse } from '@kbn/es-types';
+import { createTaskRunError, TaskErrorSource } from '@kbn/task-manager-plugin/server';
 
 type Response = Array<{
   _id: string;
@@ -52,6 +53,6 @@ export const executeEsqlRequest = async ({
       error instanceof Error ? error.message : String(error)
     }`;
     logger.debug(message);
-    throw new Error(message);
+    throw createTaskRunError(new Error(message), TaskErrorSource.USER);
   }
 };

x-pack/platform/plugins/shared/streams/tsconfig.json

@@ -78,5 +78,7 @@
     "@kbn/core-http-common",
     "@kbn/scout",
     "@kbn/esql-language",
+    "@kbn/core-elasticsearch-server-mocks",
+    "@kbn/core-logging-server-mocks",
   ]
 }