[SigEvents] Migrate continuous extraction workflow inputs under manual trigger (elastic#270116)

Author: cesco-f

x-pack/platform/plugins/shared/streams/server/lib/workflows/continuous_extraction_workflow.test.ts

@@ -27,21 +27,21 @@ describe('continuous_extraction_workflow.yaml stays in sync with constants', ()
 
   it('uses the correct maxScheduledStreams input', () => {
     assertYamlContains(
-      `name: maxScheduledStreams\n    type: number\n    default: ${MAX_SCHEDULED_STREAMS}`
+      `name: maxScheduledStreams\n        type: number\n        default: ${MAX_SCHEDULED_STREAMS}`
     );
   });
 
   it('uses the correct lookbackHours input', () => {
-    assertYamlContains(`name: lookbackHours\n    type: number\n    default: 24`);
+    assertYamlContains(`name: lookbackHours\n        type: number\n        default: 24`);
   });
 
   it('declares extractionIntervalHours as an optional input without default', () => {
-    assertYamlContains('name: extractionIntervalHours\n    type: number\n    description:');
+    assertYamlContains('name: extractionIntervalHours\n        type: number\n        description:');
     expect(WORKFLOW_YAML).not.toMatch(/name: extractionIntervalHours[\s\S]*?default:/m);
   });
 
   it('declares excludedStreamPatterns as an optional input without default', () => {
-    assertYamlContains('name: excludedStreamPatterns\n    type: string\n    description:');
+    assertYamlContains('name: excludedStreamPatterns\n        type: string\n        description:');
     expect(WORKFLOW_YAML).not.toMatch(/name: excludedStreamPatterns[\s\S]*?default:/m);
   });
 

x-pack/platform/plugins/shared/streams/server/lib/workflows/continuous_extraction_workflow.yaml

@@ -14,6 +14,11 @@
 version: "1"
 name: ".streams-continuous-ki-extraction"
 description: "This workflow is used by the system and should not be modified."
+tags:
+  - observability
+  - streams
+  - knowledge-indicators
+  - continuous-extraction
 settings:
   # Timeout is 1 minute shorter than the schedule to avoid overlapping runs.
   timeout: "9m"
@@ -26,21 +31,22 @@ triggers:
   - type: scheduled
     with:
       every: "10m"
-inputs:
-  - name: maxScheduledStreams
-    type: number
-    default: 5
-    description: "Max streams to schedule per run; remaining eligible streams are skipped."
-  - name: lookbackHours
-    type: number
-    default: 24
-    description: "Hours of data to sample when identifying KIs."
-  - name: extractionIntervalHours
-    type: number
-    description: "Override for the min hours between extractions per stream. When omitted the Advanced Setting is used."
-  - name: excludedStreamPatterns
-    type: string
-    description: "Override for the comma-separated exclude patterns. When omitted the Advanced Setting is used."
+  - type: manual
+    inputs:
+      - name: maxScheduledStreams
+        type: number
+        default: 5
+        description: "Max streams to schedule per run; remaining eligible streams are skipped."
+      - name: lookbackHours
+        type: number
+        default: 24
+        description: "Hours of data to sample when identifying KIs."
+      - name: extractionIntervalHours
+        type: number
+        description: "Override for the min hours between extractions per stream. When omitted the Advanced Setting is used."
+      - name: excludedStreamPatterns
+        type: string
+        description: "Override for the comma-separated exclude patterns. When omitted the Advanced Setting is used."
 steps:
   # Step 1: Fetch eligible streams from the classification endpoint.
   - name: get_eligible
@@ -51,7 +57,6 @@ steps:
         /internal/streams/_extraction/_eligible?maxScheduledStreams={{ inputs.maxScheduledStreams }}&lookbackHours={{ inputs.lookbackHours }}
         {%- if inputs.extractionIntervalHours %}&extractionIntervalHours={{ inputs.extractionIntervalHours }}{% endif -%}
         {%- if inputs.excludedStreamPatterns %}&excludedStreamPatterns={{ inputs.excludedStreamPatterns }}{% endif -%}
-      use_server_info: true
       headers:
         x-elastic-internal-origin: "kibana"
     on-failure:
@@ -77,7 +82,6 @@ steps:
             with:
               method: POST
               path: "/internal/streams/{{ foreach.item.streamName }}/features/_task"
-              use_server_info: true
               headers:
                 x-elastic-internal-origin: "kibana"
               body:
@@ -98,7 +102,6 @@ steps:
                 with:
                   method: GET
                   path: "/internal/streams/{{ foreach.item.streamName }}/features/_status"
-                  use_server_info: true
                   headers:
                     x-elastic-internal-origin: "kibana"
               - name: if_not_started_delay
@@ -126,7 +129,6 @@ steps:
                 with:
                   method: GET
                   path: "/internal/streams/{{ foreach.item.streamName }}/features/_status"
-                  use_server_info: true
                   headers:
                     x-elastic-internal-origin: "kibana"
                 on-failure:

x-pack/platform/plugins/shared/streams/server/routes/internal/sig_events/extraction/eligible_streams_route.ts

@@ -53,13 +53,17 @@ export interface EligibleStreamsResponse {
   };
 }
 
-const NumberFromString = z.string().transform((value) => {
-  const trimmed = value.trim();
-  if (trimmed === '') {
-    return undefined;
-  }
-  return Number(trimmed);
-});
+const NumberFromString = z
+  .string()
+  .optional()
+  .transform((value) => {
+    if (value === undefined) return undefined;
+    const trimmed = value.trim();
+    if (trimmed === '') {
+      return undefined;
+    }
+    return Number(trimmed);
+  });
 
 const eligibleStreamsRoute = createServerRoute({
   endpoint: 'GET /internal/streams/_extraction/_eligible',