diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
index 657ad593c8b..c8f8886e6a9 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
@@ -229,8 +229,10 @@ public void appendOutput(OutputAppendEvent event) throws InterpreterRPCException
   @Override
   public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException, TException {
     if (event.getAppId() == null) {
-      listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),
+      runner.updateBuffer(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           InterpreterResult.Type.valueOf(event.getType()), event.getData());
+      // Complete replacements before the interpreter can publish its terminal result.
+      runner.run();
     } else {
       appListener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), event.getIndex(),
           event.getAppId(), InterpreterResult.Type.valueOf(event.getType()), event.getData());
@@ -239,11 +241,15 @@ public void updateOutput(OutputUpdateEvent event) throws InterpreterRPCException
 
   @Override
   public void updateAllOutput(OutputUpdateAllEvent event) throws InterpreterRPCException, TException {
-    listener.onOutputClear(event.getNoteId(), event.getParagraphId());
-    for (int i = 0; i < event.getMsg().size(); i++) {
-      RemoteInterpreterResultMessage msg = event.getMsg().get(i);
-      listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i,
-          InterpreterResult.Type.valueOf(msg.getType()), msg.getData());
+    synchronized (runner) {
+      // Finish earlier output before the clear; keep replacements ahead of the next drain.
+      runner.run();
+      listener.onOutputClear(event.getNoteId(), event.getParagraphId());
+      for (int i = 0; i < event.getMsg().size(); i++) {
+        RemoteInterpreterResultMessage msg = event.getMsg().get(i);
+        listener.onOutputUpdated(event.getNoteId(), event.getParagraphId(), i,
+            InterpreterResult.Type.valueOf(msg.getType()), msg.getData());
+      }
     }
   }
 
@@ -266,6 +272,9 @@ public void updateAppStatus(AppStatusUpdateEvent event) throws InterpreterRPCExc
 
   @Override
   public void checkpointOutput(String noteId, String paragraphId) throws InterpreterRPCException, TException {
+    // Drain replacements before checkpointing.
+    // Keep storage callbacks outside the runner lock to avoid blocking output delivery.
+    runner.run();
     listener.checkpointOutput(noteId, paragraphId);
   }
 
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
index 93dd3282969..8bc063af106 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java
@@ -17,6 +17,7 @@
 
 package org.apache.zeppelin.interpreter.remote;
 
+import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,10 +30,8 @@
 import java.util.concurrent.LinkedBlockingQueue;
 
 /**
- * This thread sends paragraph's append-data
- * periodically, rather than continously, with
- * a period of BUFFER_TIME_MS. It handles append-data
- * for all paragraphs across all notebooks.
+ * Sends paragraph output periodically. Adjacent append events are batched, while update events
+ * share the same queue so that they cannot overtake earlier appends.
  */
 public class AppendOutputRunner implements Runnable {
 
@@ -48,29 +47,36 @@ public AppendOutputRunner(RemoteInterpreterProcessListener listener) {
     this.listener = listener;
   }
 
+  // Serialize scheduled and RPC drains to preserve callback order.
+  // Empty drains must return immediately to RPC callers.
   @Override
-  public void run() {
+  public synchronized void run() {
 
     Map<String, StringBuilder> stringBufferMap = new HashMap<>();
     List<AppendOutputBuffer> list = new LinkedList<>();
 
-    /* "drainTo" method does not wait for any element
-     * to be present in the queue, and thus this loop would
-     * continuosly run (with period of BUFFER_TIME_MS). "take()" method
-     * waits for the queue to become non-empty and then removes
-     * one element from it. Rest elements from queue (if present) are
-     * removed using "drainTo" method. Thus we save on some un-necessary
-     * cpu-cycles.
-     */
-    try {
-      list.add(queue.take());
-    } catch (InterruptedException e) {
-      LOGGER.error("Wait for OutputBuffer queue interrupted: {}", e.getMessage());
+    queue.drainTo(list);
+    if (list.isEmpty()) {
+      return;
     }
     Long processingStartTime = System.currentTimeMillis();
-    queue.drainTo(list);
 
-    for (AppendOutputBuffer buffer: list) {
+    Long sizeProcessed = Long.valueOf(0);
+    for (AppendOutputBuffer buffer : list) {
+      if (buffer instanceof UpdateOutputBuffer) {
+        sizeProcessed += flushAppendBuffers(stringBufferMap);
+        UpdateOutputBuffer update = (UpdateOutputBuffer) buffer;
+        try {
+          listener.onOutputUpdated(update.getNoteId(), update.getParagraphId(), update.getIndex(),
+              update.getType(), update.getData());
+        } catch (RuntimeException e) {
+          // A stale callback must not abort another paragraph's synchronous drain.
+          LOGGER.warn("Failed to update output for note {} paragraph {}",
+              update.getNoteId(), update.getParagraphId(), e);
+        }
+        continue;
+      }
+
       String noteId = buffer.getNoteId();
       String paragraphId = buffer.getParagraphId();
       int index = buffer.getIndex();
@@ -82,6 +88,7 @@ public void run() {
       builder.append(buffer.getData());
       stringBufferMap.put(stringBufferKey, builder);
     }
+    sizeProcessed += flushAppendBuffers(stringBufferMap);
     Long processingTime = System.currentTimeMillis() - processingStartTime;
 
     if (processingTime > SAFE_PROCESSING_TIME) {
@@ -90,15 +97,6 @@ public void run() {
       LOGGER.debug("Processing time for append-output took {} milliseconds", processingTime);
     }
 
-    Long sizeProcessed = Long.valueOf(0);
-    for (Entry<String, StringBuilder> stringBufferMapEntry : stringBufferMap.entrySet()) {
-      String stringBufferKey = stringBufferMapEntry.getKey();
-      StringBuilder buffer = stringBufferMapEntry.getValue();
-      sizeProcessed += buffer.length();
-      String[] keys = stringBufferKey.split(":");
-      listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString());
-    }
-
     if (sizeProcessed > SAFE_PROCESSING_STRING_SIZE) {
       LOGGER.warn("Processing size for buffered append-output is high: {} characters.", sizeProcessed);
     } else {
@@ -106,8 +104,31 @@ public void run() {
     }
   }
 
+  private long flushAppendBuffers(Map<String, StringBuilder> stringBufferMap) {
+    long sizeProcessed = 0;
+    for (Entry<String, StringBuilder> stringBufferMapEntry : stringBufferMap.entrySet()) {
+      String stringBufferKey = stringBufferMapEntry.getKey();
+      StringBuilder buffer = stringBufferMapEntry.getValue();
+      sizeProcessed += buffer.length();
+      try {
+        String[] keys = stringBufferKey.split(":");
+        listener.onOutputAppend(keys[0], keys[1], Integer.parseInt(keys[2]), buffer.toString());
+      } catch (RuntimeException e) {
+        // One stale append must not abort another paragraph's synchronous drain.
+        LOGGER.warn("Failed to append output for {}", stringBufferKey, e);
+      }
+    }
+    stringBufferMap.clear();
+    return sizeProcessed;
+  }
+
   public void appendBuffer(String noteId, String paragraphId, int index, String outputToAppend) {
     queue.offer(new AppendOutputBuffer(noteId, paragraphId, index, outputToAppend));
   }
 
+  /** Enqueues a replacement; callers needing completion must also invoke run(). */
+  public void updateBuffer(String noteId, String paragraphId, int index,
+                           InterpreterResult.Type type, String output) {
+    queue.offer(new UpdateOutputBuffer(noteId, paragraphId, index, type, output));
+  }
 }
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java
new file mode 100644
index 00000000000..15de2d5092b
--- /dev/null
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/UpdateOutputBuffer.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.remote;
+
+import org.apache.zeppelin.interpreter.InterpreterResult;
+
+/**
+ * This element stores the buffered update-data of paragraph's output. It shares the
+ * append-data queue so that an update, which replaces a result, can never be sent
+ * ahead of the appends that preceded it.
+ */
+public class UpdateOutputBuffer extends AppendOutputBuffer {
+
+  private final InterpreterResult.Type type;
+
+  public UpdateOutputBuffer(String noteId, String paragraphId, int index,
+                            InterpreterResult.Type type, String data) {
+    super(noteId, paragraphId, index, data);
+    this.type = type;
+  }
+
+  public InterpreterResult.Type getType() {
+    return type;
+  }
+
+}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
index fcf76a8105d..28fde11fc71 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
@@ -1769,7 +1769,19 @@ public void onOutputAppend(String noteId, String paragraphId, int index, String
         .put("paragraphId", paragraphId)
         .put("index", index)
         .put("data", output);
-    connectionManager.broadcast(noteId, msg);
+    try {
+      getNotebook().processNote(noteId, note -> {
+        if (note == null) {
+          LOGGER.warn("Note {} not found", noteId);
+        } else if (!note.isPersonalizedMode()) {
+          // Streaming events do not identify the user that owns the execution.
+          connectionManager.broadcast(noteId, msg);
+        }
+        return null;
+      });
+    } catch (IOException e) {
+      LOGGER.warn("Fail to call onOutputAppend", e);
+    }
   }
 
   /**
@@ -1796,16 +1808,15 @@ public void onOutputUpdated(String noteId, String paragraphId, int index,
             LOGGER.warn("Note {} not found", noteId);
             return null;
           }
-          Paragraph paragraph = note.getParagraph(paragraphId);
-          paragraph.updateOutputBuffer(index, type, output);
           if (note.isPersonalizedMode()) {
-            String user = note.getParagraph(paragraphId).getUser();
-            if (null != user) {
-              connectionManager.multicastToUser(user, msg);
-            }
-          } else {
-            connectionManager.broadcast(noteId, msg);
+            // Streaming events carry no owner. The shared outputBuffer is what checkpointOutput
+            // saves as the shared result and what other users' paragraphs are cloned from, so
+            // one user's output must not be written there. Personalized clients get their
+            // user-specific terminal snapshot instead.
+            return null;
           }
+          note.getParagraph(paragraphId).updateOutputBuffer(index, type, output);
+          connectionManager.broadcast(noteId, msg);
           return null;
         });
     } catch (IOException e) {
@@ -1828,11 +1839,15 @@ public void onOutputClear(String noteId, String paragraphId) {
           if (note == null) {
             // It is possible the note is removed, but the job is still running
             LOGGER.warn("Note {} doesn't existed, it maybe deleted.", noteId);
-          } else {
-            note.clearParagraphOutput(paragraphId);
-            Paragraph paragraph = note.getParagraph(paragraphId);
-            broadcastParagraph(note, paragraph, MSG_ID_NOT_DEFINED);
+            return null;
           }
+          if (note.isPersonalizedMode()) {
+            // Streaming events carry no owner, so they must not mutate shared paragraph state.
+            return null;
+          }
+          note.clearParagraphOutput(paragraphId);
+          Paragraph paragraph = note.getParagraph(paragraphId);
+          broadcastParagraph(note, paragraph, MSG_ID_NOT_DEFINED);
           return null;
         });
 
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
index ad385c612e7..db73b10a427 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServerTest.java
@@ -20,24 +20,152 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.io.Serializable;
+import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.remote.AppendOutputRunner;
 import org.apache.zeppelin.interpreter.remote.InvokeResourceMethodEventMessage;
 import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
 import org.apache.zeppelin.interpreter.thrift.InterpreterRPCException;
+import org.apache.zeppelin.interpreter.thrift.OutputAppendEvent;
+import org.apache.zeppelin.interpreter.thrift.OutputUpdateAllEvent;
+import org.apache.zeppelin.interpreter.thrift.OutputUpdateEvent;
+import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResultMessage;
 import org.apache.zeppelin.resource.Resource;
 import org.apache.zeppelin.resource.ResourceId;
 import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+
 
 public class RemoteInterpreterEventServerTest {
   
+  @Test
+  void updateOutputCompletesBeforeReturning() throws Exception {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    RemoteInterpreterEventServer server =
+        serverWithRunner(listener, new AppendOutputRunner(listener));
+    try {
+      server.updateOutput(new OutputUpdateEvent("note", "para", 0, "TEXT", "final", null));
+      // A caller may publish terminal status as soon as the RPC returns.
+      verify(listener).onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "final");
+    } finally {
+      server.stop();
+    }
+  }
+
+  @Test
+  void checkpointDrainsPendingOutputBeforeSaving() throws Exception {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    RemoteInterpreterEventServer server =
+        serverWithRunner(listener, new AppendOutputRunner(listener));
+    try {
+      server.appendOutput(new OutputAppendEvent("note", "para", 0, "pending", null));
+      server.checkpointOutput("note", "para");
+      InOrder order = inOrder(listener);
+      order.verify(listener).onOutputAppend("note", "para", 0, "pending");
+      order.verify(listener).checkpointOutput("note", "para");
+    } finally {
+      server.stop();
+    }
+  }
+
+  @Test
+  void updateAllIsAnOrderedClearAndReplacement() throws Exception {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    RemoteInterpreterEventServer server = serverWithRunner(listener, runner);
+    try {
+      runner.appendBuffer("note", "para", 0, "old");
+      server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.singletonList(
+          new RemoteInterpreterResultMessage("HTML", "replacement"))));
+      verify(listener).onOutputUpdated("note", "para", 0,
+          InterpreterResult.Type.HTML, "replacement");
+      runner.appendBuffer("note", "para", 0, "new");
+      runner.run();
+      InOrder order = inOrder(listener);
+      order.verify(listener).onOutputAppend("note", "para", 0, "old");
+      order.verify(listener).onOutputClear("note", "para");
+      order.verify(listener).onOutputUpdated("note", "para", 0,
+          InterpreterResult.Type.HTML, "replacement");
+      order.verify(listener).onOutputAppend("note", "para", 0, "new");
+    } finally {
+      server.stop();
+    }
+  }
+
+  @Test
+  void updateAllWaitsForInFlightAppendAndCompletesBeforeReturning() throws Exception {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    RemoteInterpreterEventServer server = serverWithRunner(listener, runner);
+    CountDownLatch entered = new CountDownLatch(1);
+    CountDownLatch release = new CountDownLatch(1);
+    CountDownLatch updateStarted = new CountDownLatch(1);
+    doAnswer(invocation -> {
+      entered.countDown();
+      assertTrue(release.await(5, TimeUnit.SECONDS));
+      return null;
+    }).when(listener).onOutputAppend("note", "para", 0, "old");
+    ExecutorService executor = Executors.newFixedThreadPool(2);
+    try {
+      runner.appendBuffer("note", "para", 0, "old");
+      Future<?> first = executor.submit(runner);
+      assertTrue(entered.await(5, TimeUnit.SECONDS));
+      Future<?> update = executor.submit(() -> {
+        updateStarted.countDown();
+        server.updateAllOutput(new OutputUpdateAllEvent("note", "para", Collections.singletonList(
+            new RemoteInterpreterResultMessage("HTML", "replacement"))));
+        verify(listener).onOutputUpdated("note", "para", 0,
+            InterpreterResult.Type.HTML, "replacement");
+        return null;
+      });
+      assertTrue(updateStarted.await(5, TimeUnit.SECONDS));
+      assertThrows(TimeoutException.class, () -> update.get(100, TimeUnit.MILLISECONDS));
+      release.countDown();
+      first.get(5, TimeUnit.SECONDS);
+      update.get(5, TimeUnit.SECONDS);
+      InOrder order = inOrder(listener);
+      order.verify(listener).onOutputAppend("note", "para", 0, "old");
+      order.verify(listener).onOutputClear("note", "para");
+      order.verify(listener).onOutputUpdated("note", "para", 0,
+          InterpreterResult.Type.HTML, "replacement");
+    } finally {
+      release.countDown();
+      executor.shutdownNow();
+      server.stop();
+    }
+  }
+
+  private RemoteInterpreterEventServer serverWithRunner(
+      RemoteInterpreterProcessListener listener, AppendOutputRunner runner) throws Exception {
+    InterpreterSettingManager manager = mock(InterpreterSettingManager.class);
+    when(manager.getRemoteInterpreterProcessListener()).thenReturn(listener);
+    RemoteInterpreterEventServer server = new RemoteInterpreterEventServer(
+        mock(ZeppelinConfiguration.class), manager);
+    Field field = RemoteInterpreterEventServer.class.getDeclaredField("runner");
+    field.setAccessible(true);
+    field.set(server, runner);
+    return server;
+  }
+
   @Test
   void invokeMethodThrowsRpcExceptionWhenSerializationFails() throws Exception {
     ZeppelinConfiguration zConf = mock(ZeppelinConfiguration.class);
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
index 1d8d273cb73..2d6e08beba7 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunnerTest.java
@@ -17,28 +17,39 @@
 
 package org.apache.zeppelin.interpreter.remote;
 
+import org.apache.zeppelin.interpreter.InterpreterResult;
 import org.apache.log4j.AppenderSkeleton;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.log4j.spi.LoggingEvent;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.atMost;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -88,6 +99,24 @@ public void testMultipleEventsOfSameParagraph() throws InterruptedException {
     verify(listener, times(1)).onOutputAppend(note1, para1, 0, "data1\ndata2\ndata3\n");
   }
 
+  @Test
+  void testUpdateDoesNotOvertakeQueuedAppend() {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    runner.appendBuffer("note", "para", 0, "before-1\n");
+    runner.appendBuffer("note", "para", 0, "before-2\n");
+    runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n");
+    runner.appendBuffer("note", "para", 0, "after\n");
+
+    runner.run();
+
+    InOrder order = inOrder(listener);
+    order.verify(listener).onOutputAppend("note", "para", 0, "before-1\nbefore-2\n");
+    order.verify(listener).onOutputUpdated(
+        "note", "para", 0, InterpreterResult.Type.TEXT, "replacement\n");
+    order.verify(listener).onOutputAppend("note", "para", 0, "after\n");
+  }
+
   @Test
   void testMultipleEventsOfDifferentParagraphs() throws InterruptedException {
     RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
@@ -145,24 +174,85 @@ void testWarnLoggerForLargeData() throws InterruptedException {
     logger.addAppender(appender);
 
     runner.run();
-    List<LoggingEvent> log;
-
-    int warnLogCounter;
-    LoggingEvent sizeWarnLogEntry = null;
-    do {
-      warnLogCounter = 0;
-      log = appender.getLog();
-      for (LoggingEvent logEntry: log) {
-        if (Level.WARN.equals(logEntry.getLevel())) {
-          sizeWarnLogEntry = logEntry;
-          warnLogCounter += 1;
-        }
-      }
-    } while(warnLogCounter != 2);
+    try {
+      String expected = "Processing size for buffered append-output is high: "
+          + (data.length() * numEvents) + " characters.";
+      assertTrue(appender.getLog().stream().anyMatch(event ->
+          Level.WARN.equals(event.getLevel()) && expected.equals(event.getMessage())));
+    } finally {
+      logger.removeAppender(appender);
+    }
+  }
+
+  @Test
+  void emptyDrainDoesNotBlock() {
+    AppendOutputRunner runner =
+        new AppendOutputRunner(mock(RemoteInterpreterProcessListener.class));
+    assertTimeoutPreemptively(Duration.ofSeconds(1), runner::run);
+  }
+
+  @Test
+  void updateFailureDoesNotDiscardOtherEventsOrLaterDrains() {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    doThrow(new IllegalStateException("removed")).when(listener)
+        .onOutputUpdated("note", "gone", 0, InterpreterResult.Type.TEXT, "bad");
+    runner.appendBuffer("note", "gone", 0, "bad");
+    runner.updateBuffer("note", "gone", 0, InterpreterResult.Type.TEXT, "bad");
+    runner.appendBuffer("note", "present", 0, "good");
+    runner.run();
+    runner.appendBuffer("note", "present", 0, "later");
+    runner.run();
+    verify(listener).onOutputAppend("note", "present", 0, "good");
+    verify(listener).onOutputAppend("note", "present", 0, "later");
+  }
+
+  @Test
+  void appendFailureDoesNotDiscardLaterUpdateOrDrain() {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    doThrow(new IllegalStateException("removed")).when(listener)
+        .onOutputAppend("note", "gone", 0, "bad");
+    runner.appendBuffer("note", "gone", 0, "bad");
+    runner.updateBuffer("note", "present", 0, InterpreterResult.Type.TEXT, "current");
+    runner.run();
+    runner.appendBuffer("note", "present", 0, "later");
+    runner.run();
+    verify(listener).onOutputUpdated("note", "present", 0,
+        InterpreterResult.Type.TEXT, "current");
+    verify(listener).onOutputAppend("note", "present", 0, "later");
+  }
 
-    String loggerString = "Processing size for buffered append-output is high: " +
-        (data.length() * numEvents) + " characters.";
-    assertEquals(loggerString, sizeWarnLogEntry.getMessage());
+  @Test
+  void concurrentDrainCannotOvertakeInFlightCallback() throws Exception {
+    RemoteInterpreterProcessListener listener = mock(RemoteInterpreterProcessListener.class);
+    AppendOutputRunner runner = new AppendOutputRunner(listener);
+    CountDownLatch entered = new CountDownLatch(1);
+    CountDownLatch release = new CountDownLatch(1);
+    doAnswer(invocation -> {
+      entered.countDown();
+      assertTrue(release.await(5, TimeUnit.SECONDS));
+      return null;
+    }).when(listener).onOutputAppend("note", "para", 0, "old");
+    ExecutorService executor = Executors.newFixedThreadPool(2);
+    try {
+      runner.appendBuffer("note", "para", 0, "old");
+      Future<?> first = executor.submit(runner);
+      assertTrue(entered.await(5, TimeUnit.SECONDS));
+      runner.updateBuffer("note", "para", 0, InterpreterResult.Type.TEXT, "new");
+      Future<?> second = executor.submit(runner);
+      assertThrows(TimeoutException.class, () -> second.get(100, TimeUnit.MILLISECONDS));
+      release.countDown();
+      first.get(5, TimeUnit.SECONDS);
+      second.get(5, TimeUnit.SECONDS);
+      InOrder order = inOrder(listener);
+      order.verify(listener).onOutputAppend("note", "para", 0, "old");
+      order.verify(listener).onOutputUpdated("note", "para", 0,
+          InterpreterResult.Type.TEXT, "new");
+    } finally {
+      release.countDown();
+      executor.shutdownNow();
+    }
   }
 
   private class BombardEvents implements Runnable {
@@ -232,4 +322,4 @@ private void loopForCompletingEvents(RemoteInterpreterProcessListener listener,
       }
     }
   }
-}
\ No newline at end of file
+}
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java
new file mode 100644
index 00000000000..cdc4d5c7bf3
--- /dev/null
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerStreamingScopeTest.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.socket;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.List;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterResultMessage;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.notebook.Notebook;
+import org.apache.zeppelin.notebook.Paragraph;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class NotebookServerStreamingScopeTest {
+
+  private NotebookServer server;
+  private Note note;
+  private ConnectionManager connections;
+
+  @BeforeEach
+  void setUp() throws Exception {
+    ZeppelinConfiguration conf = mock(ZeppelinConfiguration.class);
+    when(conf.getBoolean(
+        ZeppelinConfiguration.ConfVars.ZEPPELIN_WEBSOCKET_PARAGRAPH_STATUS_PROGRESS))
+        .thenReturn(true);
+    note = new Note();
+    note.setId("note");
+    note.getParagraphs().add(new Paragraph("para", note, null));
+    Notebook notebook = mock(Notebook.class);
+    doAnswer(invocation -> {
+      Notebook.NoteProcessor<?> processor = invocation.getArgument(1);
+      return processor.process(note);
+    }).when(notebook).processNote(eq("note"), any());
+    connections = mock(ConnectionManager.class);
+    server = new NotebookServer();
+    server.setZeppelinConfiguration(conf);
+    server.setNotebook(() -> notebook);
+    server.setConnectionManager(connections);
+  }
+
+  @Test
+  void sharedNoteReceivesIncrementalOutput() {
+    server.onOutputAppend("note", "para", 0, "append");
+    server.onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "update");
+
+    verify(connections, times(2)).broadcast(eq("note"), any());
+  }
+
+  @Test
+  void personalizedNoteDoesNotReceiveUnownedIncrementalOutput() {
+    note.setPersonalizedMode(true);
+
+    server.onOutputAppend("note", "para", 0, "private append");
+    server.onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "private update");
+
+    verify(connections, never()).broadcast(eq("note"), any());
+    verify(connections, never()).multicastToUser(any(), any());
+  }
+
+  @Test
+  void personalizedCheckpointDoesNotExposeUnownedOutputToOtherUsers() {
+    note.setPersonalizedMode(true);
+
+    server.onOutputUpdated("note", "para", 0, InterpreterResult.Type.TEXT, "private update");
+    server.checkpointOutput("note", "para");
+
+    InterpreterResult otherUserResult =
+        note.getParagraph("para").getUserParagraph("other").getReturn();
+    List<InterpreterResultMessage> otherUserMessages =
+        otherUserResult == null ? Collections.emptyList() : otherUserResult.message();
+    assertTrue(otherUserMessages.stream().noneMatch(m -> m.getData().contains("private update")),
+        "another user's paragraph sees the checkpointed output: " + otherUserMessages);
+  }
+
+  @Test
+  void personalizedUnownedClearPreservesSharedOutputForFutureUsers() {
+    note.setPersonalizedMode(true);
+    Paragraph sharedParagraph = note.getParagraph("para");
+    sharedParagraph.setResult(
+        new InterpreterResult(InterpreterResult.Code.SUCCESS, "shared result"));
+    sharedParagraph.updateOutputBuffer(0, InterpreterResult.Type.TEXT, "buffered result");
+    sharedParagraph.getUserParagraph("existing");
+
+    server.onOutputClear("note", "para");
+
+    InterpreterResult futureUserResult =
+        sharedParagraph.getUserParagraph("future").getReturn();
+    assertNotNull(futureUserResult,
+        "unowned clear removed the shared output inherited by a future user");
+    assertEquals(1, futureUserResult.message().size());
+    assertEquals("shared result", futureUserResult.message().get(0).getData());
+
+    sharedParagraph.checkpointOutput();
+    InterpreterResult checkpointedFutureUserResult =
+        sharedParagraph.getUserParagraph("checkpointed-future").getReturn();
+    assertNotNull(checkpointedFutureUserResult,
+        "unowned clear removed the shared output buffer used by checkpoint");
+    assertEquals(1, checkpointedFutureUserResult.message().size());
+    assertEquals("buffered result", checkpointedFutureUserResult.message().get(0).getData());
+    verify(connections, never()).broadcast(eq("note"), any());
+    verify(connections, never()).multicastToUser(any(), any());
+  }
+}
diff --git a/zeppelin-web-angular/e2e/models/published-paragraph-page.ts b/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
index e63350da022..8916fef0f7a 100644
--- a/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
+++ b/zeppelin-web-angular/e2e/models/published-paragraph-page.ts
@@ -16,6 +16,7 @@ import { BasePage } from './base-page';
 
 export class PublishedParagraphPage extends BasePage {
   readonly confirmationModal: Locator;
+  readonly textOutput: Locator;
   readonly angularRenderer: Locator;
   readonly reactWidget: Locator;
   readonly reactWidgetOrEmptyState: Locator;
@@ -30,6 +31,7 @@ export class PublishedParagraphPage extends BasePage {
     // The result count is 0 in both modes, so dynamic-forms is the discriminator: it renders only in Angular mode.
     this.angularRenderer = page.locator('zeppelin-notebook-paragraph-dynamic-forms');
     this.reactWidget = page.locator('[data-testid="react-published-paragraph"]');
+    this.textOutput = page.locator('zeppelin-publish-paragraph pre');
     // Without paragraph data the remote mounts an <Empty>, so tests that only assert "React took over" accept either.
     this.reactWidgetOrEmptyState = this.reactWidget.or(page.locator('.ant-alert'));
   }
diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
index 9ee808f9078..7da8c388176 100644
--- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
+++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.json
@@ -401,6 +401,86 @@
       ],
       "verificationEvidence": []
     },
+    {
+      "id": "NB-PARITY-022",
+      "name": "Streaming interpreter output accumulates while a paragraph is running",
+      "area": "result",
+      "preconditions": [
+        "A disposable notebook has a shell paragraph that prints several chunks separated by delays.",
+        "The server streams paragraph output (zeppelin.websocket.paragraph_status_progress.enable is true)."
+      ],
+      "action": "Run the paragraph and observe the result panel before and after the paragraph finishes.",
+      "observableOutcomes": [
+        {
+          "id": "NB-PARITY-022-OUTCOME-001",
+          "description": "The first output chunk is visible while the paragraph status is still RUNNING."
+        },
+        {
+          "id": "NB-PARITY-022-OUTCOME-002",
+          "description": "Later chunks are appended after the earlier chunks instead of replacing them while the paragraph remains RUNNING."
+        },
+        {
+          "id": "NB-PARITY-022-OUTCOME-003",
+          "description": "The FINISHED result contains every chunk exactly once, in emission order."
+        }
+      ],
+      "interpreter": "sh",
+      "roleExpectations": {
+        "owner": "allow",
+        "writer": "allow",
+        "reader": "deny",
+        "runner": "allow"
+      },
+      "roleVerification": {
+        "owner": "unverified",
+        "writer": "unverified",
+        "reader": "unverified",
+        "runner": "unverified"
+      },
+      "coverage": {
+        "status": "covered",
+        "tests": [
+          {
+            "path": "zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts",
+            "tag": "@NB-PARITY-022"
+          }
+        ],
+        "issues": [],
+        "uncoveredOutcomes": []
+      },
+      "implementationEvidence": [
+        {
+          "path": "zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts",
+          "symbol": "ParagraphOutputState"
+        },
+        {
+          "path": "zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts",
+          "symbol": "ParagraphBase.onParagraphAppendOutput"
+        },
+        {
+          "path": "zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts",
+          "symbol": "ParagraphBase.onParagraphUpdateOutput"
+        },
+        {
+          "path": "zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java",
+          "symbol": "AppendOutputRunner"
+        }
+      ],
+      "verificationEvidence": [
+        {
+          "path": "zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts",
+          "symbol": "ParagraphOutputState"
+        },
+        {
+          "path": "zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json",
+          "symbol": "enabled"
+        },
+        {
+          "path": "zeppelin-web-angular/e2e/models/notebook-paragraph-page.ts",
+          "symbol": "NotebookParagraphPage"
+        }
+      ]
+    },
     {
       "id": "NB-PARITY-050",
       "name": "Notebook editor persists the latest text after typing stops",
diff --git a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
index 3da42b6e890..e90d35f5b8e 100644
--- a/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
+++ b/zeppelin-web-angular/e2e/scenarios/notebook-parity.md
@@ -32,6 +32,7 @@ Coverage note: `covered` mechanically means this registry points to a matching e
 | NB-PARITY-010 | editor | History inline completion can be dismissed without losing editor focus | covered | not-applicable | zeppelin-web-angular/e2e/tests/notebook/inline-completion.spec.ts<br>@NB-PARITY-010 |  |
 | NB-PARITY-011 | editor | The second Escape after inline completion dismissal blurs the editor | covered | not-applicable | zeppelin-web-angular/e2e/tests/notebook/inline-completion.spec.ts<br>@NB-PARITY-011 |  |
 | NB-PARITY-021 | result | Text and table result displays preserve output semantics after paragraph execution | partial | owner: allow<br>writer: allow<br>reader: deny<br>runner: allow | zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts<br>@NB-PARITY-021 | ZEPPELIN-6514, ZEPPELIN-6516 |
+| NB-PARITY-022 | result | Streaming interpreter output accumulates while a paragraph is running | covered | owner: allow<br>writer: allow<br>reader: deny<br>runner: allow | zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts<br>@NB-PARITY-022 |  |
 | NB-PARITY-050 | persistence | Notebook editor persists the latest text after typing stops | gap | owner: allow<br>writer: allow<br>reader: deny<br>runner: not-applicable |  | ZEPPELIN-6661 |
 | NB-PARITY-051 | persistence | Notebook editor does not lose an edit made while a prior save is in flight | gap | owner: allow<br>writer: allow<br>reader: deny<br>runner: not-applicable |  | ZEPPELIN-6661 |
 | NB-PARITY-060 | theme | Notebook honors host theme selection | gap | not-applicable |  | ZEPPELIN-6640 |
@@ -135,6 +136,18 @@ Coverage note: `covered` mechanically means this registry points to a matching e
 - Verification evidence: not-applicable
 - Uncovered outcomes: NB-PARITY-021-OUTCOME-002: The UI offers every display mode that the Angular notebook exposes for the returned result type. NB-PARITY-021-OUTCOME-003: Visualization control changes preserve the field mapping from result columns to configured dimensions or measures. NB-PARITY-021-OUTCOME-004: The paragraph's persisted config reflects the resulting configuration object after a visualization option changes. NB-PARITY-021-OUTCOME-005: Text and table results expose accessible table output row by row so migrated React rendering can be compared without relying on screenshots alone.
 
+### NB-PARITY-022 Streaming interpreter output accumulates while a paragraph is running
+
+- Area: result
+- Coverage: covered
+- Interpreter: sh
+- Role verification: owner: unverified; writer: unverified; reader: unverified; runner: unverified
+- Preconditions: A disposable notebook has a shell paragraph that prints several chunks separated by delays. The server streams paragraph output (zeppelin.websocket.paragraph_status_progress.enable is true).
+- Action: Run the paragraph and observe the result panel before and after the paragraph finishes.
+- Observable outcomes: NB-PARITY-022-OUTCOME-001: The first output chunk is visible while the paragraph status is still RUNNING. NB-PARITY-022-OUTCOME-002: Later chunks are appended after the earlier chunks instead of replacing them while the paragraph remains RUNNING. NB-PARITY-022-OUTCOME-003: The FINISHED result contains every chunk exactly once, in emission order.
+- Implementation evidence: zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts (ParagraphOutputState); zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts (ParagraphBase.onParagraphAppendOutput); zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts (ParagraphBase.onParagraphUpdateOutput); zeppelin-server/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java (AppendOutputRunner)
+- Verification evidence: zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts (ParagraphOutputState); zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json (enabled); zeppelin-web-angular/e2e/models/notebook-paragraph-page.ts (NotebookParagraphPage)
+
 ### NB-PARITY-050 Notebook editor persists the latest text after typing stops
 
 - Area: persistence
diff --git a/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
index 8a937dc1a16..9211fd5e743 100644
--- a/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
+++ b/zeppelin-web-angular/e2e/tests/notebook/paragraph/paragraph-functionality.spec.ts
@@ -15,10 +15,11 @@ import { NotebookParagraphPage } from 'e2e/models/notebook-paragraph-page';
 import { NotebookKeyboardPage } from 'e2e/models/notebook-keyboard-page';
 import {
   addPageAnnotationBeforeEach,
+  createTestNotebook,
   performLoginIfRequired,
-  waitForZeppelinReady,
   PAGES,
-  createTestNotebook
+  setParagraphText,
+  waitForZeppelinReady
 } from '../../../utils';
 
 test.describe('Notebook Paragraph Functionality', () => {
@@ -94,6 +95,37 @@ test.describe('Notebook Paragraph Functionality', () => {
     await expect(paragraphPage.resultDisplay).not.toBeEmpty();
   });
 
+  test(
+    'should accumulate interpreter output while the paragraph is running',
+    { tag: '@NB-PARITY-022' },
+    async ({ page }) => {
+      await test.step('Given a shell paragraph that emits three delayed output chunks', async () => {
+        await setParagraphText(
+          page,
+          testNotebook.noteId,
+          testNotebook.paragraphId,
+          '%sh\necho first; sleep 3; echo second; sleep 5; echo third'
+        );
+        await page.reload();
+        await expect(paragraphPage.paragraphContainer).toBeVisible({ timeout: 30000 });
+      });
+
+      await test.step('When the paragraph runs', async () => {
+        await paragraphPage.runParagraph();
+      });
+
+      await test.step('Then output accumulates before the paragraph finishes', async () => {
+        await expect(paragraphPage.resultDisplay).toContainText('first', { timeout: 30000 });
+        await expect(paragraphPage.status).toHaveText('RUNNING');
+        await expect(paragraphPage.resultDisplay).toContainText(/first\s+second/, { timeout: 10000 });
+        await expect(paragraphPage.status).toHaveText('RUNNING');
+        await expect(paragraphPage.resultDisplay).toContainText(/first\s+second\s+third/, { timeout: 10000 });
+        await expect(paragraphPage.status).toHaveText('FINISHED');
+        await expect(paragraphPage.resultDisplay).toHaveText('first\nsecond\nthird\n');
+      });
+    }
+  );
+
   test('should display dynamic forms', async ({ page }) => {
     test.skip(!!process.env.CI, 'Dynamic form tests require a Spark interpreter — skipped on CI');
 
diff --git a/zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts b/zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts
new file mode 100644
index 00000000000..208a8721424
--- /dev/null
+++ b/zeppelin-web-angular/e2e/tests/notebook/published/published-streaming.spec.ts
@@ -0,0 +1,120 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { createServer } from 'node:http';
+
+import { expect, test } from '@playwright/test';
+import { NotebookParagraphPage } from '../../../models/notebook-paragraph-page';
+import { PublishedParagraphPage } from '../../../models/published-paragraph-page';
+import {
+  addPageAnnotationBeforeEach,
+  createTestNotebook,
+  PAGES,
+  setParagraphText,
+  waitForZeppelinReady
+} from '../../../utils';
+
+test.describe('Published paragraph streaming', () => {
+  // JUSTIFIED: shared owner and notebook state must remain within one worker.
+  test.describe.configure({ mode: 'default' });
+  addPageAnnotationBeforeEach(PAGES.WORKSPACE.PUBLISHED_PARAGRAPH);
+
+  let owner: NotebookParagraphPage;
+  let notebook: { noteId: string; paragraphId: string };
+
+  test.beforeEach(async ({ page }) => {
+    owner = new NotebookParagraphPage(page);
+    await page.goto('/#/');
+    await waitForZeppelinReady(page);
+    notebook = await createTestNotebook(page);
+  });
+
+  for (const react of [false, true]) {
+    test(`accumulates live output and preserves the exact terminal snapshot (${react ? 'React' : 'Angular'})`, async ({
+      page,
+      context
+    }) => {
+      const viewer = await context.newPage();
+      const published = new PublishedParagraphPage(viewer);
+
+      // The local shell waits for the viewer before emitting the next chunk.
+      const release: Array<() => void> = [];
+      const gates = [0, 1].map(index => new Promise<void>(resolve => (release[index] = resolve)));
+      const outputGate = createServer((request, response) => {
+        const gate = gates[Number(request.url?.slice(1))];
+        if (!gate) {
+          response.writeHead(404).end();
+          return;
+        }
+        void gate.then(() => response.end());
+      });
+      await new Promise<void>((resolve, reject) => {
+        outputGate.once('error', reject);
+        outputGate.listen(0, '127.0.0.1', resolve);
+      });
+
+      try {
+        const address = outputGate.address();
+        if (!address || typeof address === 'string') {
+          throw new Error('Failed to bind output gate');
+        }
+        const waitForViewer = `curl --noproxy '*' --fail --silent --show-error --max-time 180 http://127.0.0.1:${address.port}`;
+
+        await test.step('Given a published viewer of a shell paragraph with controlled output', async () => {
+          await setParagraphText(
+            page,
+            notebook.noteId,
+            notebook.paragraphId,
+            `%sh\nset -e\necho first\n${waitForViewer}/0\necho second\n${waitForViewer}/1\necho third`
+          );
+          await page.goto(`/#/notebook/${notebook.noteId}`);
+          await expect(owner.paragraphContainer).toBeVisible();
+          await viewer.goto(`/#/notebook/${notebook.noteId}/paragraph/${notebook.paragraphId}?react=${react}`);
+          await waitForZeppelinReady(viewer);
+          await expect(published.confirmationModal).toBeVisible();
+          await published.cancelButton.click();
+        });
+
+        await test.step('When the owner runs the paragraph, published output accumulates while RUNNING', async () => {
+          await owner.runParagraph();
+          await expect(published.textOutput).toHaveText(/^first\n$/);
+          await expect(published.textOutput).toBeVisible();
+          await expect(owner.status).toHaveText('RUNNING');
+          release[0]();
+          await expect(published.textOutput).toHaveText(/^first\nsecond\n$/);
+          await expect(published.textOutput).toBeVisible();
+          await expect(owner.status).toHaveText('RUNNING');
+          release[1]();
+        });
+
+        await test.step('Then the terminal snapshot contains every chunk exactly once', async () => {
+          await expect(owner.status).toHaveText('FINISHED');
+          await expect(published.textOutput).toHaveText(/^first\nsecond\nthird\n$/);
+          await expect(published.reactWidget).toHaveCount(react ? 1 : 0);
+        });
+
+        await test.step('And a new run replaces the previous stream', async () => {
+          await setParagraphText(page, notebook.noteId, notebook.paragraphId, '%sh\necho rerun');
+          await owner.runParagraph();
+          await expect(owner.status).toHaveText('FINISHED');
+          await expect(published.textOutput).toHaveText(/^rerun\n$/);
+        });
+      } finally {
+        await new Promise<void>((resolve, reject) => {
+          outputGate.close(error => (error ? reject(error) : resolve()));
+          outputGate.closeAllConnections();
+        });
+        await viewer.close();
+      }
+    });
+  }
+});
diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts
new file mode 100644
index 00000000000..ec7797ecba1
--- /dev/null
+++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.spec.ts
@@ -0,0 +1,38 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { expect, expectTypeOf, it } from 'vitest';
+
+import { MessageReceiveDataTypeMap } from './message-data-type-map.interface';
+import { OP } from './message-operator.interface';
+import { DatasetType, ParagraphAppendOutput, ParagraphUpdateOutput } from './message-paragraph.interface';
+
+it('declares the asymmetric paragraph output payloads sent by the server', () => {
+  const append: MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT] = {
+    noteId: 'note',
+    paragraphId: 'paragraph',
+    index: 0,
+    data: 'chunk'
+  };
+  const update: MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT] = {
+    ...append,
+    type: DatasetType.TEXT
+  };
+  expect(append).not.toHaveProperty('type');
+  expect(update.type).toBe(DatasetType.TEXT);
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]>().toEqualTypeOf<ParagraphAppendOutput>();
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]>().not.toHaveProperty('type');
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]>().toEqualTypeOf<ParagraphUpdateOutput>();
+  expectTypeOf<MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]>()
+    .toHaveProperty('type')
+    .toEqualTypeOf<DatasetType>();
+});
diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts
index f86dbdb2b12..6c023a09231 100644
--- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts
+++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-data-type-map.interface.ts
@@ -67,12 +67,14 @@ import {
   CopyParagraph,
   InsertParagraph,
   MoveParagraph,
+  ParagraphAppendOutput,
   ParagraphClearAllOutput,
   ParagraphClearOutput,
   ParagraphExecutedBySpell,
   ParagraphRemove,
   ParagraphRemoved,
   ParagraphStatus,
+  ParagraphUpdateOutput,
   ParasInfo,
   PatchParagraphReceived,
   PatchParagraphSend,
@@ -108,6 +110,8 @@ export interface MessageReceiveDataTypeMap {
   [OP.IMPORT_NOTE]: ImportNoteReceived;
   [OP.SAVE_NOTE_FORMS]: SaveNoteFormsSend;
   [OP.PARAGRAPH]: UpdateParagraph;
+  [OP.PARAGRAPH_APPEND_OUTPUT]: ParagraphAppendOutput;
+  [OP.PARAGRAPH_UPDATE_OUTPUT]: ParagraphUpdateOutput;
   [OP.PATCH_PARAGRAPH]: PatchParagraphSend;
   [OP.PARAGRAPH_REMOVED]: ParagraphRemoved;
   [OP.EDITOR_SETTING]: EditorSettingReceived;
diff --git a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts
index f75cd1f5f31..f9e01351e2e 100644
--- a/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts
+++ b/zeppelin-web-angular/projects/zeppelin-sdk/src/interfaces/message-paragraph.interface.ts
@@ -97,6 +97,17 @@ export class ParagraphIResultsMsgItem {
   data = '';
 }
 
+export interface ParagraphAppendOutput {
+  noteId: string;
+  paragraphId: string;
+  index: number;
+  data: string;
+}
+
+export interface ParagraphUpdateOutput extends ParagraphAppendOutput {
+  type: DatasetType;
+}
+
 export interface ParasInfo {
   id: string;
   infos: RuntimeInfos;
diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts
new file mode 100644
index 00000000000..34d5cc44183
--- /dev/null
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.spec.ts
@@ -0,0 +1,247 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ChangeDetectorRef } from '@angular/core';
+import { DatasetType, Message, ParagraphItem } from '@zeppelin/sdk';
+import { EMPTY } from 'rxjs';
+import { describe, expect, it, vi } from 'vitest';
+
+import { AngularContextManager } from './angular-context-manager';
+import { ParagraphBase } from './paragraph-base';
+
+class TestParagraph extends ParagraphBase {
+  changeColWidth = vi.fn();
+  updateParagraphResult = vi.fn();
+  protected currentNoteId = 'note';
+
+  hydrate(snapshot: ParagraphItem) {
+    this.setParagraphSnapshot(snapshot);
+  }
+
+  constructor(paragraph: ParagraphItem) {
+    super(
+      { receive: () => EMPTY } as unknown as Message,
+      { isParagraphRunning: item => item.status === 'RUNNING', isEntireNoteRunning: () => false },
+      {} as AngularContextManager,
+      { markForCheck: vi.fn() } as unknown as ChangeDetectorRef
+    );
+    this.paragraph = paragraph;
+  }
+}
+
+const paragraph = (id: string, status = 'RUNNING', dateStarted = '2026-01-01T00:00:00Z'): ParagraphItem => ({
+  id,
+  status,
+  dateStarted,
+  text: '',
+  user: 'anonymous',
+  dateUpdated: '',
+  dateCreated: '',
+  config: {},
+  settings: { params: {}, forms: {} },
+  apps: [],
+  progressUpdateIntervalMs: 500,
+  jobName: '',
+  aborted: false,
+  lineNumbers: false,
+  fontSize: 9
+});
+
+const beginOutput = (component: TestParagraph) =>
+  component.onParagraphUpdateOutput({
+    noteId: 'note',
+    paragraphId: 'A',
+    index: 0,
+    type: DatasetType.TEXT,
+    data: 'first\n'
+  });
+
+const appendOutput = (component: TestParagraph) =>
+  component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index: 0, data: 'second\n' });
+
+describe('ParagraphBase streaming state isolation', () => {
+  it.each(['append', 'update'])('ignores %s from another note with the same paragraph ID', kind => {
+    const component = new TestParagraph(paragraph('A'));
+    beginOutput(component);
+    if (kind === 'append') {
+      component.onParagraphAppendOutput({ noteId: 'other-note', paragraphId: 'A', index: 0, data: 'foreign' });
+    } else {
+      component.onParagraphUpdateOutput({
+        noteId: 'other-note',
+        paragraphId: 'A',
+        index: 0,
+        type: DatasetType.HTML,
+        data: 'foreign'
+      });
+    }
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\n' }]);
+    component.ngOnDestroy();
+  });
+
+  it.each(['FINISHED', 'ERROR', 'ABORT', 'RUNNING', 'PENDING'])(
+    'keeps accumulating output when another paragraph becomes %s',
+    status => {
+      const component = new TestParagraph(paragraph('A'));
+      beginOutput(component);
+
+      component.paragraphData({ paragraph: paragraph('B', status, '2026-01-01T00:00:01Z') });
+      appendOutput(component);
+
+      expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]);
+      expect(component.paragraph?.id).toBe('A');
+      component.ngOnDestroy();
+    }
+  );
+
+  it('ignores late output after its own terminal snapshot', () => {
+    const component = new TestParagraph(paragraph('A'));
+    beginOutput(component);
+    const finished = { ...paragraph('A', 'FINISHED'), results: { msg: [{ type: DatasetType.TEXT, data: 'final\n' }] } };
+
+    component.paragraphData({ paragraph: finished });
+    appendOutput(component);
+
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'final\n' }]);
+    component.ngOnDestroy();
+  });
+
+  it('accepts fresh output when its own paragraph starts a new run', () => {
+    const component = new TestParagraph(paragraph('A'));
+    beginOutput(component);
+    component.paragraphData({ paragraph: paragraph('A', 'FINISHED') });
+    component.paragraphData({ paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') });
+
+    beginOutput(component);
+    appendOutput(component);
+
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]);
+    component.ngOnDestroy();
+  });
+});
+
+describe('ParagraphBase streaming boundaries', () => {
+  it('preserves a saved revision during live output and lifecycle messages', () => {
+    const component = new TestParagraph(paragraph('A', 'FINISHED'));
+    const saved = [{ type: DatasetType.TEXT, data: 'saved revision' }];
+    component.paragraph!.results = { msg: saved };
+    component.results = saved;
+    component.revisionView = true;
+
+    component.paragraphData({ paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') });
+    beginOutput(component);
+    appendOutput(component);
+    component.paragraphData({ paragraph: paragraph('A') });
+
+    expect(component.results).toEqual(saved);
+    expect(component.paragraph?.status).toBe('FINISHED');
+    component.revisionView = false;
+    beginOutput(component);
+    expect(component.results).toEqual(saved);
+    component.ngOnDestroy();
+  });
+
+  it('does not refresh an unchanged paragraph for an omitted run timestamp', () => {
+    const component = new TestParagraph(paragraph('A'));
+    const update = vi.spyOn(component, 'updateParagraph');
+    const statusOnly = paragraph('A');
+    delete statusOnly.dateStarted;
+
+    component.paragraphData({ paragraph: statusOnly });
+    component.paragraphData({ paragraph: statusOnly });
+
+    expect(update).not.toHaveBeenCalled();
+    expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:00Z');
+    component.paragraphData({ paragraph: paragraph('A', 'RUNNING', '2026-01-01T00:00:01Z') });
+    expect(update).toHaveBeenCalledOnce();
+    expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:01Z');
+    component.ngOnDestroy();
+  });
+
+  it('publishes the final result when terminal status arrives before its snapshot', () => {
+    const component = new TestParagraph(paragraph('A'));
+    beginOutput(component);
+    component.onParagraphStatus({ id: 'A', status: 'FINISHED' });
+    const finished = paragraph('A', 'FINISHED');
+    delete finished.dateStarted;
+    finished.results = { msg: [{ type: DatasetType.TEXT, data: 'first\nfinal\n' }] };
+
+    component.paragraphData({ paragraph: finished });
+
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nfinal\n' }]);
+    expect(component.updateParagraphResult).toHaveBeenCalledWith(0, expect.anything(), {
+      type: DatasetType.TEXT,
+      data: 'first\nfinal\n'
+    });
+    component.ngOnDestroy();
+  });
+
+  it('preserves streamed types across a status snapshot without a run timestamp', () => {
+    const component = new TestParagraph(paragraph('A', 'PENDING'));
+    beginOutput(component);
+    const running = paragraph('A');
+    delete running.dateStarted;
+    component.paragraphData({ paragraph: running });
+    component.paragraphData({ paragraph: paragraph('A') });
+    appendOutput(component);
+
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]);
+    expect(component.paragraph?.dateStarted).toBe('2026-01-01T00:00:00Z');
+    component.ngOnDestroy();
+  });
+
+  it('preserves streaming output across ordinary RUNNING and unrelated paragraph snapshots', () => {
+    const component = new TestParagraph(paragraph('A'));
+    beginOutput(component);
+    component.paragraphData({ paragraph: paragraph('A') });
+    component.paragraphData({ paragraph: paragraph('B') });
+    appendOutput(component);
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]);
+    component.ngOnDestroy();
+  });
+
+  it('buffers sparse result slots without rendering holes or changing server indexes', () => {
+    const component = new TestParagraph(paragraph('A'));
+    component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index: 2, data: ' tail' });
+    component.setResults(component.paragraph!);
+    component.onParagraphUpdateOutput({
+      noteId: 'note',
+      paragraphId: 'A',
+      index: 2,
+      type: DatasetType.TEXT,
+      data: 'third'
+    });
+    expect(component.results).toEqual([]);
+    expect(component.updateParagraphResult).not.toHaveBeenCalled();
+    component.setResults(component.paragraph!);
+    beginOutput(component);
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'first\n' }]);
+    component.onParagraphUpdateOutput({
+      noteId: 'note',
+      paragraphId: 'A',
+      index: 1,
+      type: DatasetType.TABLE,
+      data: 'second'
+    });
+    expect(component.results).toEqual([
+      { type: DatasetType.TEXT, data: 'first\n' },
+      { type: DatasetType.TABLE, data: 'second' },
+      { type: DatasetType.TEXT, data: 'third' }
+    ]);
+    expect(component.updateParagraphResult).toHaveBeenCalledWith(2, expect.anything(), {
+      type: DatasetType.TEXT,
+      data: 'third'
+    });
+    component.onParagraphAppendOutput({ noteId: 'note', paragraphId: 'A', index: 2, data: ' end' });
+    expect(component.results[2].data).toBe('third end');
+    component.ngOnDestroy();
+  });
+});
diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
index 4e7c0e0fde8..b9e514abd6e 100644
--- a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-base.ts
@@ -26,12 +26,13 @@ import {
   ParagraphIResultsMsgItem
 } from '@zeppelin/sdk';
 
-import * as DiffMatchPatch from 'diff-match-patch';
+import { diff_match_patch as DiffMatchPatch } from 'diff-match-patch';
 import { isEmpty, isEqual } from 'lodash';
 
 import { MessageListener, MessageListenersManager } from '../message-listener/message-listener';
 import { AngularContextManager } from './angular-context-manager';
 import { NoteStatus } from './note-status';
+import { ParagraphOutputState } from './paragraph-output-state';
 
 export const ParagraphStatus = {
   READY: 'READY',
@@ -42,6 +43,9 @@ export const ParagraphStatus = {
   ERROR: 'ERROR'
 };
 
+const isTerminalParagraphStatus = (status?: string): boolean =>
+  status === ParagraphStatus.FINISHED || status === ParagraphStatus.ABORT || status === ParagraphStatus.ERROR;
+
 export abstract class ParagraphBase extends MessageListenersManager {
   paragraph?: ParagraphItem;
   dirtyText?: string;
@@ -58,6 +62,7 @@ export abstract class ParagraphBase extends MessageListenersManager {
     params: {},
     forms: {}
   };
+  private readonly outputState = new ParagraphOutputState();
 
   constructor(
     public messageService: Message,
@@ -68,6 +73,8 @@ export abstract class ParagraphBase extends MessageListenersManager {
     super(messageService);
   }
 
+  protected abstract get currentNoteId(): string | null | undefined;
+
   abstract changeColWidth(needCommit: boolean, updateResult?: boolean): void;
 
   @MessageListener(OP.PROGRESS)
@@ -114,16 +121,54 @@ export abstract class ParagraphBase extends MessageListenersManager {
     }
   }
 
+  @MessageListener(OP.PARAGRAPH_APPEND_OUTPUT)
+  onParagraphAppendOutput(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_APPEND_OUTPUT]) {
+    if (this.revisionView || data.noteId !== this.currentNoteId || data.paragraphId !== this.paragraph?.id) {
+      return;
+    }
+    this.initializeOutputState();
+    const result = this.outputState.append(data.index, data.data);
+    if (result) {
+      this.applyStreamingResult(data.index);
+    }
+  }
+
+  @MessageListener(OP.PARAGRAPH_UPDATE_OUTPUT)
+  onParagraphUpdateOutput(data: MessageReceiveDataTypeMap[OP.PARAGRAPH_UPDATE_OUTPUT]) {
+    if (this.revisionView || data.noteId !== this.currentNoteId || data.paragraphId !== this.paragraph?.id) {
+      return;
+    }
+    this.initializeOutputState();
+    const result = this.outputState.update(data.index, data.type, data.data);
+    if (result) {
+      this.applyStreamingResult(data.index);
+    }
+  }
+
   @MessageListener(OP.PARAGRAPH)
   paragraphData(data: MessageReceiveDataTypeMap[OP.PARAGRAPH]) {
     const oldPara = this.paragraph;
     if (!oldPara) {
-      throw new Error('paragraph is not defined');
+      return;
     }
     const newPara = data.paragraph;
+    if (this.revisionView || newPara.id !== oldPara.id) {
+      return;
+    }
     if (!newPara.results) {
       newPara.results = {};
     }
+    const oldRunActive = oldPara.status === ParagraphStatus.PENDING || oldPara.status === ParagraphStatus.RUNNING;
+    const newRunActive = newPara.status === ParagraphStatus.PENDING || newPara.status === ParagraphStatus.RUNNING;
+    const runChanged =
+      newPara.dateStarted != null && oldPara.dateStarted != null && newPara.dateStarted !== oldPara.dateStarted;
+    if (newRunActive && (!oldRunActive || runChanged)) {
+      this.outputState.reset();
+    }
+    // Close the stream before publishing the terminal snapshot.
+    if (isTerminalParagraphStatus(newPara.status)) {
+      this.outputState.finish(newPara.results?.msg);
+    }
     if (this.isUpdateRequired(oldPara, newPara)) {
       this.updateParagraph(oldPara, newPara, () => {
         if (newPara.results && newPara.results.msg) {
@@ -186,6 +231,32 @@ export abstract class ParagraphBase extends MessageListenersManager {
     }
   }
 
+  private initializeOutputState(): void {
+    if (!this.outputState.isInitialized) {
+      this.outputState.reset(this.results, isTerminalParagraphStatus(this.paragraph?.status));
+    }
+  }
+
+  private applyStreamingResult(index: number): void {
+    if (!this.paragraph) {
+      return;
+    }
+    const previousLength = this.results.length;
+    const results = this.outputState.snapshot();
+    if (!this.paragraph.results) {
+      this.paragraph.results = {};
+    }
+    this.paragraph.results.msg = results;
+    this.results = results;
+    results.forEach((visibleResult, visibleIndex) => {
+      if (visibleIndex === index || visibleIndex >= previousLength) {
+        const config = this.paragraph!.config.results?.[visibleIndex] ?? { graph: new GraphConfig() };
+        this.updateParagraphResult(visibleIndex, config, visibleResult);
+      }
+    });
+    this.cdr.markForCheck();
+  }
+
   updateParagraph(oldPara: ParagraphItem, newPara: ParagraphItem, updateCallback: () => void) {
     // 1. can't update on revision view
     if (!this.revisionView) {
@@ -221,12 +292,13 @@ export abstract class ParagraphBase extends MessageListenersManager {
       (newPara.dateCreated !== oldPara.dateCreated ||
         newPara.text !== oldPara.text ||
         newPara.dateFinished !== oldPara.dateFinished ||
-        newPara.dateStarted !== oldPara.dateStarted ||
+        (newPara.dateStarted != null && newPara.dateStarted !== oldPara.dateStarted) ||
         newPara.dateUpdated !== oldPara.dateUpdated ||
         newPara.status !== oldPara.status ||
         newPara.jobName !== oldPara.jobName ||
         newPara.title !== oldPara.title ||
         isEmpty(newPara.results) !== isEmpty(oldPara.results) ||
+        (isTerminalParagraphStatus(newPara.status) && !isEqual(newPara.results?.msg, oldPara.results?.msg)) ||
         newPara.errorMessage !== oldPara.errorMessage ||
         !isEqual(newPara.settings, oldPara.settings) ||
         !isEqual(newPara.config, oldPara.config) ||
@@ -270,7 +342,10 @@ export abstract class ParagraphBase extends MessageListenersManager {
     this.paragraph.dateUpdated = newPara.dateUpdated;
     this.paragraph.dateCreated = newPara.dateCreated;
     this.paragraph.dateFinished = newPara.dateFinished;
-    this.paragraph.dateStarted = newPara.dateStarted;
+    // Status-only snapshots can omit the start time of the current run.
+    if (newPara.dateStarted != null) {
+      this.paragraph.dateStarted = newPara.dateStarted;
+    }
     this.paragraph.errorMessage = newPara.errorMessage;
     this.paragraph.jobName = newPara.jobName;
     this.paragraph.title = newPara.title;
@@ -361,4 +436,15 @@ export abstract class ParagraphBase extends MessageListenersManager {
     }
     this.messageService.cancelParagraph(this.paragraph.id);
   }
+  protected setParagraphSnapshot(paragraph: ParagraphItem | undefined): void {
+    this.paragraph = paragraph;
+    this.results = [];
+    this.configs = {};
+    if (paragraph) {
+      this.setResults(paragraph);
+    }
+    const terminal = isTerminalParagraphStatus(paragraph?.status);
+    this.outputState.reset(this.results, terminal);
+    this.cdr.markForCheck();
+  }
 }
diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts
new file mode 100644
index 00000000000..e9ce623d4db
--- /dev/null
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.spec.ts
@@ -0,0 +1,151 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { DatasetType, ParagraphIResultsMsgItem } from '@zeppelin/sdk';
+import { describe, expect, it } from 'vitest';
+
+import capture from './paragraph-output-stream.capture.json';
+import { ParagraphOutputState } from './paragraph-output-state';
+
+interface CapturedEvent {
+  op: string;
+  data: {
+    index?: number;
+    type?: string;
+    data?: string;
+    paragraph?: {
+      status: string;
+      results?: {
+        msg?: Array<{ type: string; data: string }>;
+      };
+    };
+  };
+}
+
+const capturedType = (type: string): DatasetType => {
+  expect(Object.values(DatasetType)).toContain(type);
+  return type as DatasetType;
+};
+
+const capturedResults = (results: Array<{ type: string; data: string }>): ParagraphIResultsMsgItem[] =>
+  results.map(result => ({ type: capturedType(result.type), data: result.data }));
+
+const replay = (state: ParagraphOutputState, events: CapturedEvent[]): string[] => {
+  const rendered: string[] = [];
+  for (const event of events) {
+    if (event.op === 'PARAGRAPH_UPDATE_OUTPUT') {
+      const result = state.update(event.data.index!, capturedType(event.data.type!), event.data.data!);
+      if (result) {
+        rendered.push(result.data);
+      }
+    } else if (event.op === 'PARAGRAPH_APPEND_OUTPUT') {
+      const result = state.append(event.data.index!, event.data.data!);
+      if (result) {
+        rendered.push(result.data);
+      }
+    } else if (event.op === 'PARAGRAPH') {
+      state.finish(capturedResults(event.data.paragraph?.results?.msg ?? []));
+    }
+  }
+  return rendered;
+};
+
+describe('ParagraphOutputState', () => {
+  it('replays the captured callback and WebSocket order without dropping output', () => {
+    const state = new ParagraphOutputState();
+    state.reset();
+    const events = capture.enabled.events as CapturedEvent[];
+
+    expect(capture.schemaVersion).toBe(1);
+    expect(events.slice(0, -1).map(({ op, data }) => ({ op, data }))).toEqual(capture.callbackOrder.events);
+    expect(replay(state, events)).toEqual(['', 'first\n', 'first\nsecond\n', 'first\nsecond\nthird\n']);
+    expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\nthird\n' }]);
+  });
+
+  it('accumulates coalesced APPEND chunks at their result index', () => {
+    const state = new ParagraphOutputState();
+    const appends = capture.enabled.events.filter(event => event.op === 'PARAGRAPH_APPEND_OUTPUT');
+    state.reset([{ type: DatasetType.TEXT, data: '' }]);
+
+    state.append(0, appends[0].data.data! + appends[1].data.data!);
+    const result = state.append(0, appends[2].data.data!);
+
+    expect(result).toEqual({ type: DatasetType.TEXT, data: 'first\nsecond\nthird\n' });
+    expect(state.snapshot()).toEqual([result]);
+  });
+
+  it('holds APPEND chunks until a typed UPDATE can render them', () => {
+    const state = new ParagraphOutputState();
+    const update = capture.enabled.events.find(event => event.op === 'PARAGRAPH_UPDATE_OUTPUT')!;
+    const appends = capture.enabled.events.filter(event => event.op === 'PARAGRAPH_APPEND_OUTPUT');
+    state.reset();
+
+    expect(state.append(0, appends[0].data.data!)).toBeUndefined();
+    expect(state.append(0, appends[1].data.data!)).toBeUndefined();
+
+    expect(state.update(0, capturedType(update.data.type!), update.data.data!)).toEqual({
+      type: DatasetType.TEXT,
+      data: 'first\nsecond\n'
+    });
+  });
+
+  it('discards pending APPEND chunks when a non-empty UPDATE replaces the result', () => {
+    const state = new ParagraphOutputState();
+    state.reset();
+
+    expect(state.append(0, 'stale\n')).toBeUndefined();
+    expect(state.update(0, DatasetType.TEXT, 'replacement\n')).toEqual({
+      type: DatasetType.TEXT,
+      data: 'replacement\n'
+    });
+    expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: 'replacement\n' }]);
+    expect(state.update(0, DatasetType.TEXT, '')).toEqual({
+      type: DatasetType.TEXT,
+      data: ''
+    });
+    expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: '' }]);
+  });
+
+  it('uses the terminal snapshot after UPDATE overtakes a queued APPEND', () => {
+    const state = new ParagraphOutputState();
+    state.reset([{ type: DatasetType.TEXT, data: 'stale\n' }]);
+
+    state.update(0, DatasetType.TEXT, 'replacement\n');
+    state.append(0, 'queued-before-update\n');
+    state.finish([{ type: DatasetType.TEXT, data: 'replacement\n' }]);
+
+    expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: 'replacement\n' }]);
+  });
+
+  it('ignores an APPEND observed after the terminal PARAGRAPH', () => {
+    const state = new ParagraphOutputState();
+    const events = capture.enabled.events as CapturedEvent[];
+    const terminal = events.find(event => event.op === 'PARAGRAPH')!;
+    const finalAppend = [...events].reverse().find(event => event.op === 'PARAGRAPH_APPEND_OUTPUT')!;
+    state.reset([{ type: DatasetType.TEXT, data: 'first\nsecond\n' }]);
+
+    replay(state, [terminal, finalAppend]);
+
+    expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\nthird\n' }]);
+  });
+
+  it('falls back to the terminal snapshot when streaming messages are disabled', () => {
+    const state = new ParagraphOutputState();
+    const events = capture.disabled.events as CapturedEvent[];
+    state.reset();
+
+    expect(capture.disabled.configuration['zeppelin.websocket.paragraph_status_progress.enable']).toBe(false);
+    expect(events.map(event => event.op)).toEqual(['PARAGRAPH']);
+    expect(() => replay(state, events)).not.toThrow();
+    expect(state.snapshot()).toEqual([{ type: DatasetType.TEXT, data: 'first\nsecond\nthird\n' }]);
+  });
+});
diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts
new file mode 100644
index 00000000000..93934ededca
--- /dev/null
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-state.ts
@@ -0,0 +1,79 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { DatasetType, ParagraphIResultsMsgItem } from '@zeppelin/sdk';
+
+export class ParagraphOutputState {
+  private results: ParagraphIResultsMsgItem[] = [];
+  private readonly pendingAppends = new Map<number, string>();
+  private initialized = false;
+  private terminal = false;
+
+  get isInitialized(): boolean {
+    return this.initialized;
+  }
+
+  reset(results: ParagraphIResultsMsgItem[] = [], terminal = false): void {
+    this.results = results.map(result => ({ ...result }));
+    this.pendingAppends.clear();
+    this.initialized = true;
+    this.terminal = terminal;
+  }
+
+  finish(results: ParagraphIResultsMsgItem[] = []): void {
+    this.reset(results, true);
+  }
+
+  update(index: number, type: DatasetType, data: string): ParagraphIResultsMsgItem | undefined {
+    if (this.terminal) {
+      return undefined;
+    }
+
+    // Non-empty UPDATE data replaces stale appends; an empty type declaration adopts them.
+    const result = {
+      type,
+      data: data === '' ? (this.pendingAppends.get(index) ?? '') : data
+    };
+    this.pendingAppends.delete(index);
+    this.results[index] = result;
+    return result;
+  }
+
+  append(index: number, data: string): ParagraphIResultsMsgItem | undefined {
+    if (this.terminal) {
+      return undefined;
+    }
+
+    const current = this.results[index];
+    if (!current) {
+      this.pendingAppends.set(index, (this.pendingAppends.get(index) ?? '') + data);
+      return undefined;
+    }
+
+    const result = {
+      ...current,
+      data: current.data + data
+    };
+    this.results[index] = result;
+    return result;
+  }
+
+  snapshot(): ParagraphIResultsMsgItem[] {
+    // Preserve server indexes by waiting for missing slots to acquire their types.
+    // Keep later slots buffered until those gaps are filled.
+    let length = 0;
+    while (this.results[length]) {
+      length++;
+    }
+    return this.results.slice(0, length);
+  }
+}
diff --git a/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json
new file mode 100644
index 00000000000..655ca5fbba3
--- /dev/null
+++ b/zeppelin-web-angular/src/app/core/paragraph-base/paragraph-output-stream.capture.json
@@ -0,0 +1,148 @@
+{
+  "_license": "Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0.",
+  "schemaVersion": 1,
+  "capture": {
+    "issue": "ZEPPELIN-6659",
+    "zeppelinVersion": "0.13.0-SNAPSHOT",
+    "capturedAt": "2026-09-05T07:15:34Z",
+    "browser": "Chromium",
+    "interpreter": "sh",
+    "paragraph": "echo first; sleep 3; echo second; sleep 5; echo third",
+    "normalization": "noteId, paragraphId, principal, ticket, message ids, and absolute timestamps are normalized or omitted"
+  },
+  "callbackOrder": {
+    "evidence": "InterpreterResultMessageOutput.java:108-119 emits the initial typed update before flushing the first append callback",
+    "events": [
+      {
+        "op": "PARAGRAPH_UPDATE_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "type": "TEXT",
+          "data": ""
+        }
+      },
+      {
+        "op": "PARAGRAPH_APPEND_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "data": "first\n"
+        }
+      },
+      {
+        "op": "PARAGRAPH_APPEND_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "data": "second\n"
+        }
+      },
+      {
+        "op": "PARAGRAPH_APPEND_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "data": "third\n"
+        }
+      }
+    ]
+  },
+  "enabled": {
+    "configuration": {
+      "zeppelin.websocket.paragraph_status_progress.enable": true
+    },
+    "events": [
+      {
+        "elapsedMs": 0,
+        "op": "PARAGRAPH_UPDATE_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "type": "TEXT",
+          "data": ""
+        }
+      },
+      {
+        "elapsedMs": 107,
+        "op": "PARAGRAPH_APPEND_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "data": "first\n"
+        }
+      },
+      {
+        "elapsedMs": 2984,
+        "op": "PARAGRAPH_APPEND_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "data": "second\n"
+        }
+      },
+      {
+        "elapsedMs": 7990,
+        "op": "PARAGRAPH_APPEND_OUTPUT",
+        "data": {
+          "noteId": "<note-id>",
+          "paragraphId": "<paragraph-id>",
+          "index": 0,
+          "data": "third\n"
+        }
+      },
+      {
+        "elapsedMs": 8027,
+        "op": "PARAGRAPH",
+        "data": {
+          "paragraph": {
+            "id": "<paragraph-id>",
+            "status": "FINISHED",
+            "results": {
+              "code": "SUCCESS",
+              "msg": [
+                {
+                  "type": "TEXT",
+                  "data": "first\nsecond\nthird\n"
+                }
+              ]
+            }
+          }
+        }
+      }
+    ]
+  },
+  "disabled": {
+    "configuration": {
+      "zeppelin.websocket.paragraph_status_progress.enable": false
+    },
+    "events": [
+      {
+        "elapsedMs": 8006,
+        "op": "PARAGRAPH",
+        "data": {
+          "paragraph": {
+            "id": "<paragraph-id>",
+            "status": "FINISHED",
+            "results": {
+              "code": "SUCCESS",
+              "msg": [
+                {
+                  "type": "TEXT",
+                  "data": "first\nsecond\nthird\n"
+                }
+              ]
+            }
+          }
+        }
+      }
+    ]
+  }
+}
diff --git a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
index 17c9a4d5780..c7365dd3c9b 100644
--- a/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
+++ b/zeppelin-web-angular/src/app/pages/workspace/notebook/paragraph/paragraph.component.ts
@@ -102,6 +102,10 @@ export class NotebookParagraphComponent
   @Input() useReactFooter = false;
   reactFooterFailed = false;
 
+  protected get currentNoteId(): string | undefined {
+    return this.note?.id;
+  }
+
   get shouldUseReactFooter(): boolean {
     return this.useReactFooter && !this.reactFooterFailed;
   }
@@ -627,7 +631,7 @@ export class NotebookParagraphComponent
         this.handleKeyEvent(event.action, event.event);
         this.notebookParagraphCodeEditorComponent?.handleKeyEvent(event.action);
       });
-    this.setResults(this.paragraph);
+    this.setParagraphSnapshot(this.paragraph);
     this.originalText = this.paragraph.text;
     this.isEntireNoteRunning = this.noteStatusService.isEntireNoteRunning(this.note);
     this.isParagraphRunning = this.noteStatusService.isParagraphRunning(this.paragraph);
@@ -756,6 +760,9 @@ export class NotebookParagraphComponent
   }
 
   ngOnChanges(changes: SimpleChanges): void {
+    if (changes.paragraph || changes.note) {
+      this.setParagraphSnapshot(this.paragraph);
+    }
     const { index, select, scrolled } = changes;
     if (
       (index && index.currentValue !== index.previousValue && this.select) ||
diff --git a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts
new file mode 100644
index 00000000000..a9c9b75c407
--- /dev/null
+++ b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.spec.ts
@@ -0,0 +1,143 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { QueryList } from '@angular/core';
+import { convertToParamMap } from '@angular/router';
+import { DatasetType, Note, ParagraphItem } from '@zeppelin/sdk';
+import { BehaviorSubject, EMPTY, of } from 'rxjs';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+
+vi.mock('@zeppelin/services', () => ({
+  MessageService: class {},
+  HeliumService: class {},
+  NgZService: class {},
+  NoteStatusService: class {},
+  ReactFeatureService: class {}
+}));
+vi.mock('@zeppelin/core', async () => ({
+  ...(await import('../../../../core/paragraph-base/paragraph-base')),
+  ...(await import('../../../../core/message-listener/message-listener')),
+  publishedSymbol: Symbol('published')
+}));
+vi.mock('../../share/result/result.component', () => ({ NotebookParagraphResultComponent: class {} }));
+
+import { PublishedParagraphComponent } from './paragraph.component';
+
+const snapshot = (noteId: string, status: string, text: string): Note => ({
+  note: {
+    id: noteId,
+    name: noteId,
+    paragraphs: [
+      {
+        id: 'same-paragraph',
+        status,
+        text: '',
+        results: { msg: [{ type: DatasetType.TEXT, data: text }] },
+        config: {},
+        settings: { params: {}, forms: {} }
+      } as ParagraphItem
+    ]
+  } as Note['note']
+});
+
+const components: PublishedParagraphComponent[] = [];
+afterEach(() => components.splice(0).forEach(component => component.ngOnDestroy()));
+
+const setup = (useReact = false) => {
+  const params = new BehaviorSubject({ noteId: 'note-A', paragraphId: 'same-paragraph' });
+  const component = new PublishedParagraphComponent(
+    { receive: () => EMPTY, getNote: vi.fn() } as never,
+    { params, queryParamMap: of(convertToParamMap({})) } as never,
+    {} as never,
+    {} as never,
+    {} as never,
+    { isEnabled: () => useReact } as never,
+    { isParagraphRunning: (p: ParagraphItem) => p.status === 'RUNNING' } as never,
+    {} as never,
+    { markForCheck: vi.fn() } as never
+  );
+  component.notebookParagraphResultComponents = new QueryList();
+  components.push(component);
+  return { component, params };
+};
+
+const append = (component: PublishedParagraphComponent, noteId: string, data: string) =>
+  component.onParagraphAppendOutput({ noteId, paragraphId: 'same-paragraph', index: 0, data });
+
+describe('Published paragraph authoritative snapshots', () => {
+  it('resumes from a same-note refreshed snapshot instead of cached output', () => {
+    const { component } = setup();
+    component.getNote(snapshot('note-A', 'RUNNING', 'old'));
+    append(component, 'note-A', ' tail');
+    component.getNote(snapshot('note-A', 'RUNNING', 'fresh'));
+    append(component, 'note-A', ' next');
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'fresh next' }]);
+  });
+
+  it('unfreezes terminal output when the reused route loads a running paragraph', () => {
+    const { component, params } = setup();
+    component.getNote(snapshot('note-A', 'FINISHED', 'A final'));
+    append(component, 'note-A', ' ignored');
+    params.next({ noteId: 'note-B', paragraphId: 'same-paragraph' });
+    component.getNote(snapshot('note-B', 'RUNNING', 'B first'));
+    append(component, 'note-B', ' second');
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'B first second' }]);
+  });
+
+  it('rejects stale NOTE and stream events while waiting for the new route snapshot', () => {
+    const { component, params } = setup();
+    component.getNote(snapshot('note-A', 'RUNNING', 'A first'));
+    append(component, 'note-A', ' tail');
+    params.next({ noteId: 'note-B', paragraphId: 'same-paragraph' });
+    component.getNote(snapshot('note-A', 'RUNNING', 'stale A'));
+    append(component, 'note-A', ' stale tail');
+    append(component, 'note-B', ' before snapshot');
+    expect(component.results).toEqual([]);
+    component.getNote(snapshot('note-B', 'RUNNING', 'B first'));
+    append(component, 'note-A', ' foreign');
+    append(component, 'note-B', ' second');
+    expect(component.results).toEqual([{ type: DatasetType.TEXT, data: 'B first second' }]);
+  });
+});
+
+describe('Published React streaming output', () => {
+  it.each(['append', 'update'])('refreshes React props after a streaming %s', operation => {
+    const { component } = setup(true);
+    component.getNote(snapshot('note-A', 'RUNNING', 'initial'));
+    const initialProps = component.reactProps;
+    if (operation === 'append') {
+      append(component, 'note-A', ' tail');
+    } else {
+      component.onParagraphUpdateOutput({
+        noteId: 'note-A',
+        paragraphId: 'same-paragraph',
+        index: 0,
+        type: DatasetType.TEXT,
+        data: 'replacement'
+      });
+    }
+    const expected = [{ type: DatasetType.TEXT, data: operation === 'append' ? 'initial tail' : 'replacement' }];
+    expect(component.results).toEqual(expected);
+    expect(component.reactProps.results).toEqual(expected);
+    expect(component.reactProps).not.toBe(initialProps);
+  });
+
+  it('continues updating Angular results after React falls back', () => {
+    const { component } = setup(true);
+    component.reactFailed = true;
+    const updateResult = vi.fn();
+    component.notebookParagraphResultComponents.reset([{ updateResult } as never]);
+    component.getNote(snapshot('note-A', 'RUNNING', 'initial'));
+    append(component, 'note-A', ' tail');
+    expect(updateResult).toHaveBeenCalledWith(expect.anything(), { type: DatasetType.TEXT, data: 'initial tail' });
+  });
+});
diff --git a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
index 94f9af7d943..07d0439710e 100644
--- a/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
+++ b/zeppelin-web-angular/src/app/pages/workspace/published/paragraph/paragraph.component.ts
@@ -58,6 +58,10 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis
   @ViewChildren(NotebookParagraphResultComponent)
   notebookParagraphResultComponents!: QueryList<NotebookParagraphResultComponent>;
 
+  protected get currentNoteId(): string | null {
+    return this.noteId;
+  }
+
   constructor(
     public messageService: MessageService,
     private activatedRoute: ActivatedRoute,
@@ -80,6 +84,7 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis
       }
       this.noteId = params.noteId;
       this.paragraphId = params.paragraphId!;
+      this.setParagraphSnapshot(undefined);
       this.messageService.getNote(params.noteId);
     });
   }
@@ -87,21 +92,19 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis
   @MessageListener(OP.NOTE)
   getNote(data: MessageReceiveDataTypeMap[OP.NOTE]) {
     const note = data.note;
-    if (!isNil(note)) {
-      this.paragraph = note.paragraphs.find(p => p.id === this.paragraphId);
+    if (!isNil(note) && note.id === this.noteId) {
+      this.setParagraphSnapshot(note.paragraphs.find(p => p.id === this.paragraphId));
       if (this.paragraph) {
         if (!this.paragraph.results) {
           this.showRunConfirmationModal();
         }
         if (this.useReact && !this.reactFailed) {
-          this.setResults(this.paragraph);
           this.reactProps = this.buildReactProps(this.paragraph);
           this.isLoading = false;
           this.cdr.markForCheck();
           return;
         }
 
-        this.setResults(this.paragraph);
         this.originalText = this.paragraph.text;
         this.initializeDefault(this.paragraph.config, this.paragraph.settings);
       } else {
@@ -153,8 +156,11 @@ export class PublishedParagraphComponent extends ParagraphBase implements Publis
   }
 
   updateParagraphResult(resultIndex: number, config: ParagraphConfigResult, result: ParagraphIResultsMsgItem): void {
-    // In React mode the Angular result components never render, so this query is empty.
-    // The remote is refreshed from updateParagraphObjectWhenUpdated instead.
+    if (this.useReact && !this.reactFailed && this.paragraph) {
+      this.reactProps = this.buildReactProps(this.paragraph);
+      this.cdr.markForCheck();
+      return;
+    }
     const resultComponent = this.notebookParagraphResultComponents.toArray()[resultIndex];
     if (resultComponent) {
       resultComponent.updateResult(config, result);