From 63bdf5c5b1a96303d009279ce4ff7b5c319611d9 Mon Sep 17 00:00:00 2001
From: Yingfeng <yingfeng.zhang@gmail.com>
Date: Tue, 30 Jun 2026 18:06:03 +0800
Subject: [PATCH] Fix harness streaming emit (#16486)

---
 internal/harness/graph/pregel/stream.go | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/internal/harness/graph/pregel/stream.go b/internal/harness/graph/pregel/stream.go
index 626b7cbd7d..f39505625b 100644
--- a/internal/harness/graph/pregel/stream.go
+++ b/internal/harness/graph/pregel/stream.go
@@ -276,6 +276,8 @@ func (sm *StreamManager) EmitDebug(step int, message string, data any) {
 }
 
 // EmitFinal emits final event with complete state.
+// Uses blocking channel send to guarantee delivery — dropping the final event
+// would cause RunSync to return nil state (silent data loss).
 func (sm *StreamManager) EmitFinal(step int, state any) {
 	if !sm.shouldEmit(EventTypeFinal) {
 		return
@@ -285,8 +287,7 @@ func (sm *StreamManager) EmitFinal(step int, state any) {
 	event.Data = map[string]any{
 		"state": state,
 	}
-
-	sm.emit(event)
+	sm.eventCh <- event
 }
 
 // emit sends an event to the channel.