Configurable webhook responses

CHANGELOG.md

@@ -17,11 +17,16 @@ and this project adheres to
 
 ### Added
 
+- Allow users to respond back with custom webhook responses via the
+  `webhookResponse` field in the job state.
+  [#3102](https://github.com/OpenFn/lightning/issues/3102)
+
 ### Changed
 
 ### Fixed
 
 ## [2.16.3] - 2026-05-07
+
 ## [2.16.3-pre3] - 2026-05-07
 
 ### Fixed

assets/js/collaborative-editor/adapters/YAMLStateToYDoc.ts

@@ -77,6 +77,10 @@ export class YAMLStateToYDoc {
 
     if (trigger.type === 'webhook') {
       triggerMap.set('webhook_reply', trigger.webhook_reply ?? null);
+      triggerMap.set(
+        'sync_webhook_response_config',
+        trigger.sync_webhook_response_config ?? null
+      );
     }
 
     return triggerMap;

assets/js/collaborative-editor/hooks/useUnsavedChanges.ts

@@ -106,6 +106,8 @@ function transformTrigger(trigger: Trigger) {
       break;
     case 'webhook':
       output.webhook_reply = trigger.webhook_reply ?? 'before_start';
+      output.sync_webhook_response_config =
+        trigger.sync_webhook_response_config ?? null;
       break;
   }
   return output;

assets/js/collaborative-editor/types/session.ts

@@ -71,6 +71,10 @@ export namespace Session {
     cron_expression: string | null;
     has_auth_method: boolean;
     webhook_reply: 'before_start' | 'after_completion' | null;
+    sync_webhook_response_config: {
+      success_code: number | null;
+      error_code: number | null;
+    } | null;
     webhook_auth_methods: Array<{
       id: string;
       name: string;

assets/js/collaborative-editor/types/trigger.ts

@@ -24,6 +24,13 @@ const webhookTriggerSchema = baseTriggerSchema.extend({
     .enum(['before_start', 'after_completion'])
     .nullable()
     .default('before_start'),
+  sync_webhook_response_config: z
+    .object({
+      success_code: z.number().int().nullable().default(null),
+      error_code: z.number().int().nullable().default(null),
+    })
+    .nullable()
+    .default(null),
 });
 
 // Cron trigger schema with professional validation using cron-validator
@@ -49,6 +56,7 @@ const cronTriggerSchema = baseTriggerSchema.extend({
     ),
   cron_cursor_job_id: z.string().uuid().nullable().default(null),
   kafka_configuration: z.null().default(null),
+  sync_webhook_response_config: z.null().default(null),
   webhook_reply: z.null().default(null).catch(null),
 });
 
@@ -103,6 +111,7 @@ const kafkaTriggerSchema = baseTriggerSchema.extend({
   cron_expression: z.null().default(null),
   cron_cursor_job_id: z.null().default(null),
   kafka_configuration: kafkaConfigSchema,
+  sync_webhook_response_config: z.null().default(null),
   webhook_reply: z.null().default(null).catch(null),
 });
 
@@ -137,6 +146,7 @@ export const createDefaultTrigger = (
         cron_cursor_job_id: null,
         kafka_configuration: null,
         webhook_reply: 'before_start' as const,
+        sync_webhook_response_config: null,
       };
 
     case 'cron':

assets/js/yaml/schema/workflow-spec.json

@@ -62,6 +62,14 @@
               "type": ["string", "null"],
               "enum": ["before_start", "after_completion", "custom", null]
             },
+            "webhook_response": {
+              "type": ["object", "null"],
+              "properties": {
+                "success_code": { "type": ["integer", "null"] },
+                "error_code": { "type": ["integer", "null"] }
+              },
+              "additionalProperties": false
+            },
             "pos": {
               "type": "object",
               "properties": {

assets/js/yaml/types.ts

@@ -28,7 +28,11 @@ export type StateWebhookTrigger = {
   id: string;
   enabled: boolean;
   type: 'webhook';
-  webhook_reply: 'before_start' | 'after_completion' | 'custom' | null;
+  webhook_reply: 'before_start' | 'after_completion' | null | undefined;
+  sync_webhook_response_config?: {
+    success_code?: number | null;
+    error_code?: number | null;
+  } | null;
 };
 
 export type StateKafkaTrigger = {
@@ -87,11 +91,17 @@ export type SpecCronTrigger = {
   pos: Position | undefined;
 };
 
+export type WebhookResponseConfig = {
+  success_code?: number | null;
+  error_code?: number | null;
+};
+
 export type SpecWebhookTrigger = {
   id?: string;
   type: 'webhook';
   enabled: boolean;
   webhook_reply: string | null;
+  webhook_response?: WebhookResponseConfig | null;
   pos: Position | undefined;
 };
 

assets/js/yaml/util.ts

@@ -82,6 +82,18 @@ export const convertWorkflowStateToSpec = (
 
     if (trigger.type === 'webhook') {
       triggerDetails.webhook_reply = trigger.webhook_reply ?? null;
+      const config = trigger.sync_webhook_response_config;
+      if (
+        config &&
+        (config.success_code != null || config.error_code != null)
+      ) {
+        triggerDetails.webhook_response = {
+          ...(config.success_code != null && {
+            success_code: config.success_code,
+          }),
+          ...(config.error_code != null && { error_code: config.error_code }),
+        };
+      }
     }
 
     // TODO: handle kafka config
@@ -186,6 +198,9 @@ export const convertWorkflowSpecToState = (
         type: 'webhook',
         enabled,
         webhook_reply: specTrigger.webhook_reply,
+        sync_webhook_response_config: parseWebhookResponseConfig(
+          specTrigger.webhook_response
+        ),
       };
     } else {
       trigger = {
@@ -251,6 +266,13 @@ export const convertWorkflowSpecToState = (
   return workflowState;
 };
 
+function parseWebhookResponseConfig(
+  config: import('./types').WebhookResponseConfig | null | undefined
+): import('./types').WebhookResponseConfig | null {
+  if (!config) return null;
+  return config;
+}
+
 export const extractJobCredentials = (jobs: Workflow.Job[]): JobCredentials => {
   const credentials: JobCredentials = {};
   for (const job of jobs) {

assets/test/collaborative-editor/components/inspector/TriggerForm.test.tsx

@@ -192,9 +192,11 @@ describe('TriggerForm - Response Mode Field', () => {
 
     // Check both options exist (use exact text to avoid "Async" matching "sync")
     expect(
-      screen.getByRole('option', { name: 'Async (default)' })
+      screen.getByRole('option', { name: 'Async (Before Start)' })
+    ).toBeInTheDocument();
+    expect(
+      screen.getByRole('option', { name: 'Sync (After Completion)' })
     ).toBeInTheDocument();
-    expect(screen.getByRole('option', { name: 'Sync' })).toBeInTheDocument();
   });
 
   test('shows Async as default selected value', async () => {
@@ -261,7 +263,7 @@ describe('TriggerForm - Response Mode Field', () => {
     await waitFor(() => {
       expect(
         screen.getByText(
-          /responds with the final output state after the run completes/i
+          /holds the http connection open and responds when the run completes/i
         )
       ).toBeInTheDocument();
     });

lib/lightning/collaboration/workflow_serializer.ex

@@ -228,7 +228,18 @@ defmodule Lightning.Collaboration.WorkflowSerializer do
           "id" => trigger.id,
           "type" => trigger.type |> to_string(),
           "webhook_reply" =>
-            trigger.webhook_reply && to_string(trigger.webhook_reply)
+            trigger.webhook_reply && to_string(trigger.webhook_reply),
+          "sync_webhook_response_config" =>
+            case trigger.sync_webhook_response_config do
+              nil ->
+                nil
+
+              config ->
+                Yex.MapPrelim.from(%{
+                  "success_code" => config.success_code,
+                  "error_code" => config.error_code
+                })
+            end
         })
 
       Yex.Array.push(triggers_array, trigger_map)
@@ -274,9 +285,11 @@ defmodule Lightning.Collaboration.WorkflowSerializer do
     |> Enum.map(fn trigger ->
       trigger
       |> Map.take(
-        ~w(id type enabled cron_expression cron_cursor_job_id webhook_reply kafka_configuration)
+        ~w(id type enabled cron_expression cron_cursor_job_id webhook_reply
+           kafka_configuration sync_webhook_response_config)
       )
       |> normalize_kafka_configuration()
+      |> normalize_sync_webhook_response_config()
     end)
   end
 
@@ -299,6 +312,27 @@ defmodule Lightning.Collaboration.WorkflowSerializer do
 
   defp normalize_kafka_configuration(trigger), do: trigger
 
+  # Y.Doc serialises numbers as floats; convert integer codes back.
+  defp normalize_sync_webhook_response_config(
+         %{"sync_webhook_response_config" => %{} = config} = trigger
+       ) do
+    normalized =
+      config
+      |> normalize_integer_field("success_code")
+      |> normalize_integer_field("error_code")
+
+    Map.put(trigger, "sync_webhook_response_config", normalized)
+  end
+
+  defp normalize_sync_webhook_response_config(trigger), do: trigger
+
+  defp normalize_integer_field(map, key) do
+    case Map.fetch(map, key) do
+      {:ok, value} when is_float(value) -> Map.put(map, key, trunc(value))
+      _ -> map
+    end
+  end
+
   defp extract_positions(positions_map) do
     Yex.Map.to_json(positions_map)
   end

lib/lightning/export_utils.ex

@@ -31,6 +31,7 @@ defmodule Lightning.ExportUtils do
     trigger: [
       :type,
       :webhook_reply,
+      :webhook_response,
       :cron_expression,
       :cron_cursor_job,
       :enabled,
@@ -121,14 +122,37 @@ defmodule Lightning.ExportUtils do
         Map.put(base, :kafka_configuration, kafka_config)
 
       :webhook ->
-        if trigger.webhook_reply do
-          Map.put(base, :webhook_reply, Atom.to_string(trigger.webhook_reply))
-        else
-          base
-        end
+        base
+        |> maybe_put_webhook_reply(trigger.webhook_reply)
+        |> maybe_put_webhook_response(trigger.sync_webhook_response_config)
+    end
+  end
+
+  defp maybe_put_webhook_reply(map, nil), do: map
+
+  defp maybe_put_webhook_reply(map, reply) when is_atom(reply) do
+    Map.put(map, :webhook_reply, Atom.to_string(reply))
+  end
+
+  defp maybe_put_webhook_response(map, %{} = config) do
+    webhook_response =
+      Map.reject(
+        %{
+          success_code: config.success_code,
+          error_code: config.error_code
+        },
+        fn {_k, v} -> is_nil(v) end
+      )
+
+    if map_size(webhook_response) > 0 do
+      Map.put(map, :webhook_response, webhook_response)
+    else
+      map
     end
   end
 
+  defp maybe_put_webhook_response(map, _), do: map
+
   defp edge_to_treenode(%{source_job_id: nil} = edge, triggers, jobs) do
     source_trigger =
       Enum.find(triggers, fn t -> t.id == edge.source_trigger_id end)

lib/lightning/projects/provisioner.ex

@@ -32,6 +32,7 @@ defmodule Lightning.Projects.Provisioner do
   alias Lightning.Workflows.Snapshot
   alias Lightning.Workflows.Trigger
   alias Lightning.Workflows.Triggers.KafkaConfiguration
+  alias Lightning.Workflows.Triggers.SyncWebhookResponseConfig
   alias Lightning.Workflows.Workflow
   alias Lightning.Workflows.WorkflowUsageLimiter
   alias Lightning.WorkflowVersions
@@ -458,13 +459,20 @@ defmodule Lightning.Projects.Provisioner do
   end
 
   defp trigger_changeset(trigger, attrs) do
+    attrs = remap_webhook_response(attrs)
+
     trigger
     |> Trigger.cast_changeset(attrs)
     |> cast_embed(
       :kafka_configuration,
       required: false,
       with: &kafka_config_changeset/2
     )
+    |> cast_embed(
+      :sync_webhook_response_config,
+      required: false,
+      with: &SyncWebhookResponseConfig.changeset/2
+    )
     |> Trigger.validate()
     |> cast(attrs, [:delete])
     |> validate_required([:id])
@@ -473,6 +481,16 @@ defmodule Lightning.Projects.Provisioner do
     |> maybe_mark_for_deletion()
   end
 
+  defp remap_webhook_response(attrs) do
+    case Map.pop(attrs, "webhook_response") do
+      {nil, attrs} ->
+        attrs
+
+      {config, attrs} ->
+        Map.put(attrs, "sync_webhook_response_config", config)
+    end
+  end
+
   defp kafka_config_changeset(kafka_config, attrs) do
     kafka_config
     |> KafkaConfiguration.changeset(attrs)