diff --git a/src/main/java/org/jenkinsci/plugins/workflow/cps/FlowHead.java b/src/main/java/org/jenkinsci/plugins/workflow/cps/FlowHead.java
index c525c5ca2..3ecff643f 100644
--- a/src/main/java/org/jenkinsci/plugins/workflow/cps/FlowHead.java
+++ b/src/main/java/org/jenkinsci/plugins/workflow/cps/FlowHead.java
@@ -117,6 +117,19 @@ void newStartNode(FlowStartNode n) throws IOException {
             this.head = execution.startNodes.push(n);
         }
         execution.storage.storeNode(head, false);
+
+        CpsThreadGroup c = CpsThreadGroup.current();
+        if (c !=null) {
+            // if the manipulation is from within the program executing thread, then
+            // defer the notification till we get to a safe point.
+            c.notifyNewHead(head);
+        } else {
+            // in recovering from error and such situation, we sometimes need to grow the graph
+            // without running the program.
+            // TODO can CpsThreadGroup.notifyNewHead be used instead to notify both kinds of listeners?
+            execution.notifyListeners(Collections.singletonList(head), true);
+            execution.notifyListeners(Collections.singletonList(head), false);
+        }
     }
 
     /** Could be better described as "append to Flow graph" except for parallel cases. */
diff --git a/src/test/java/org/jenkinsci/plugins/workflow/WorkflowTest.java b/src/test/java/org/jenkinsci/plugins/workflow/WorkflowTest.java
index 290459f86..3a9acd010 100644
--- a/src/test/java/org/jenkinsci/plugins/workflow/WorkflowTest.java
+++ b/src/test/java/org/jenkinsci/plugins/workflow/WorkflowTest.java
@@ -42,8 +42,10 @@
 import hudson.slaves.SlaveComputer;
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
@@ -52,7 +54,11 @@
 import org.jenkinsci.plugins.scriptsecurity.scripts.ScriptApproval;
 import org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition;
 import org.jenkinsci.plugins.workflow.flow.FlowExecution;
+import org.jenkinsci.plugins.workflow.flow.GraphListener;
+import org.jenkinsci.plugins.workflow.graph.FlowNode;
+import org.jenkinsci.plugins.workflow.graph.FlowStartNode;
 import org.jenkinsci.plugins.workflow.job.WorkflowJob;
+import org.jenkinsci.plugins.workflow.job.WorkflowRun;
 import org.jenkinsci.plugins.workflow.steps.AbstractStepDescriptorImpl;
 import org.jenkinsci.plugins.workflow.steps.AbstractStepExecutionImpl;
 import org.jenkinsci.plugins.workflow.steps.AbstractStepImpl;
@@ -262,6 +268,30 @@ public static final class Execution extends AbstractSynchronousNonBlockingStepEx
         }
     }
 
+    @Issue("JENKINS-52189")
+    @Test
+    public void notifyFlowStartNode() {
+        story.then(s->{
+            WorkflowJob j = jenkins().createProject(WorkflowJob.class, "bob");
+            j.setDefinition(new CpsFlowDefinition("echo 'I did a thing'", true));
+            WorkflowRun r = story.j.buildAndAssertSuccess(j);
+            FlowStartNodeListener listener = jenkins().getExtensionList(FlowStartNodeListener.class).get(0);
+            assertTrue(listener.execNames.contains(r.getExecution().toString()));
+        });
+    }
+
+    @TestExtension("notifyFlowStartNode")
+    public static class FlowStartNodeListener implements GraphListener {
+        List<String> execNames = new ArrayList<String>();
+
+        @Override
+        public void onNewHead(FlowNode node) {
+            if (node instanceof FlowStartNode) {
+                execNames.add(node.getExecution().toString());
+            }
+        }
+    }
+
     @Issue("JENKINS-29952")
     @Test public void env() {
         story.addStep(new Statement() {