fix(go-ingestion): write last component output JSON as debug-log END message (#17786)

internal/ingestion/task/debug_log_sink.go

@@ -110,6 +110,11 @@ type DebugLogSink struct {
 	// to the sink via SetResult (the ResultSink capability). It is written onto
 	// the END marker's trace entry in Flush. Empty until SetResult is called.
 	resultDSL json.RawMessage
+	// runOutput is the raw pipeline run output map (output["state"][<id>] is
+	// each component's outputs), handed over together with resultDSL. Flush
+	// uses it to build the END marker's message (the last component's output
+	// JSON) exactly like Python does. Nil until SetResult is called.
+	runOutput map[string]any
 }
 
 // NewDebugLogSink builds a sink that writes to "{canvasID}-{messageID}-logs".
@@ -128,11 +133,11 @@ func NewDebugLogSink(canvasID, messageID string, store DebugLogStore) *DebugLogS
 func (s *DebugLogSink) OnComponentTotal(_ context.Context, _ string, _ int) {}
 
 // SetResult receives the debug-run result DSL (built by BuildDebugResultDSL)
-// and stores it for the END marker. It implements the optional ResultSink
-// capability the executor probes for, so the sink stays decoupled from how the
-// DSL is produced. Safe to call at most once per run; a later call replaces the
-// previous value.
-func (s *DebugLogSink) SetResult(dsl map[string]any) {
+// plus the raw pipeline run output, and stores them for the END marker. It
+// implements the optional ResultSink capability the executor probes for, so
+// the sink stays decoupled from how the DSL is produced. Safe to call at most
+// once per run; a later call replaces the previous value.
+func (s *DebugLogSink) SetResult(dsl map[string]any, output map[string]any) {
 	b, err := json.Marshal(dsl)
 	if err != nil {
 		return
@@ -140,6 +145,7 @@ func (s *DebugLogSink) SetResult(dsl map[string]any) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	s.resultDSL = b
+	s.runOutput = output
 }
 
 // OnComponentProgress maps one component lifecycle event to a trace line and
@@ -223,6 +229,13 @@ func (s *DebugLogSink) Flush(ctx context.Context, finalErr error) {
 	endMsg := "Debug run completed"
 	if finalErr != nil {
 		endMsg = "[ERROR] " + finalErr.Error()
+	} else if msg := endOutputMessage(entries, s.runOutput); msg != "" {
+		// Mirror Python's END marker message: a JSON dump of the last
+		// component's output (rag/flow/pipeline.py:171). The front-end
+		// "Export JSON" button JSON.parses this message
+		// (use-download-output.ts findEndOutput), so a plain-text sentinel
+		// would leave the button permanently disabled.
+		endMsg = msg
 	}
 	entries = append(entries, debugLogEntry{
 		ComponentID: "END",
@@ -253,6 +266,31 @@ func (s *DebugLogSink) Flush(ctx context.Context, finalErr error) {
 	s.store.Set(ctx, key, string(payload), s.ttl)
 }
 
+// endOutputMessage builds the END marker's message for a successful run: the
+// JSON dump of the last executed component's output, mirroring Python's
+// `json.dumps(self.get_component_obj(self.path[-1]).output())`
+// (rag/flow/pipeline.py:171). The last executed component is the sink's final
+// entry (path[-1] always emits its exit progress last). Raw embedding vectors
+// are stripped (deepCopyStrip) so the Redis log stays at Python-scale size.
+// Returns "" when no output is available; the caller then keeps the
+// plain-text fallback and the front-end export button stays disabled (empty
+// output, matching Python's isEmpty check).
+func endOutputMessage(entries []debugLogEntry, runOutput map[string]any) string {
+	if len(entries) == 0 || len(runOutput) == 0 {
+		return ""
+	}
+	lastID := entries[len(entries)-1].ComponentID
+	out, ok := lookupComponentOutput(runOutput, lastID).(map[string]any)
+	if !ok || len(out) == 0 {
+		return ""
+	}
+	b, err := json.Marshal(deepCopyStrip(out))
+	if err != nil {
+		return ""
+	}
+	return string(b)
+}
+
 // truncateRunes returns s truncated to at most max runes, preserving the original
 // bytes when shorter. Mirrors internal/service/agent_sessions.go's truncateRunes.
 func truncateRunes(s string, max int) string {

internal/ingestion/task/debug_log_sink_test.go

@@ -408,7 +408,7 @@ func TestDebugLogSink_RealPipeline_EndMarkerCarriesDSL(t *testing.T) {
 	if err != nil {
 		t.Fatalf("BuildDebugResultDSL: %v", err)
 	}
-	sink.SetResult(resultDSL)
+	sink.SetResult(resultDSL, output)
 	sink.Flush(ctx, nil)
 
 	raw := store.Get(ctx, "c-dsl-m-dsl-logs")
@@ -505,7 +505,7 @@ func TestDebugLogSink_RealPipeline_EndMarkerDSLShowsChunks(t *testing.T) {
 	if err != nil {
 		t.Fatalf("BuildDebugResultDSL: %v", err)
 	}
-	sink.SetResult(resultDSL)
+	sink.SetResult(resultDSL, output)
 	sink.Flush(ctx, nil)
 
 	raw := store.Get(ctx, "c-chunk-m-chunk-logs")
@@ -562,6 +562,58 @@ func TestDebugLogSink_RealPipeline_EndMarkerDSLShowsChunks(t *testing.T) {
 	if _, exists := dParams["outputs"]; exists {
 		t.Errorf("d has no recognized output, outputs must be absent: %#v", dParams)
 	}
+
+	// The END marker MESSAGE must be the JSON dump of the LAST component's
+	// ("d") raw output — the contract the front-end "Export JSON" button
+	// relies on (use-download-output.ts findEndOutput JSON.parses it). A
+	// plain-text sentinel leaves the button permanently disabled.
+	endMsg, _ := endFirst["message"].(string)
+	var endOutput map[string]any
+	if err := json.Unmarshal([]byte(endMsg), &endOutput); err != nil {
+		t.Fatalf("END message must be the last component's output JSON, got %q", endMsg)
+	}
+	if endOutput["ok"] != true {
+		t.Errorf("END message output=%#v want d's raw output {ok:true}", endOutput)
+	}
+}
+
+// TestDebugLogSink_EndMessageFallsBackWithoutRunOutput pins the two cases
+// where the END marker message must NOT be a JSON dump: no run output was
+// handed to the sink (SetResult never called, e.g. the run failed before
+// producing output) and the run ended with an error. In both cases the
+// front-end export button stays disabled — matching Python, whose END message
+// is only the output JSON on success.
+func TestDebugLogSink_EndMessageFallsBackWithoutRunOutput(t *testing.T) {
+	ctx := t.Context()
+
+	// Case 1: successful run but SetResult was never called -> plain text.
+	store := &capturedStore{}
+	sink := NewDebugLogSink("c-nores", "m-nores", store)
+	sink.OnComponentProgress(ctx, pipeline.ProgressEvent{
+		Component: "A", Message: "A Done", Phase: phaseExit,
+	})
+	sink.Flush(ctx, nil)
+	arr := loadArray(t, store.Get(ctx, "c-nores-m-nores-logs"))
+	endFirst, _ := arr[len(arr)-1]["trace"].([]any)[0].(map[string]any)
+	if msg, _ := endFirst["message"].(string); msg != "Debug run completed" {
+		t.Errorf("END message without run output=%q want %q", msg, "Debug run completed")
+	}
+
+	// Case 2: run output IS available but the run failed -> error wins.
+	store2 := &capturedStore{}
+	sink2 := NewDebugLogSink("c-fail", "m-fail", store2)
+	sink2.OnComponentProgress(ctx, pipeline.ProgressEvent{
+		Component: "A", Message: "A Done", Phase: phaseExit,
+	})
+	sink2.SetResult(map[string]any{"components": map[string]any{}},
+		map[string]any{"state": map[string]any{"A": map[string]any{"ok": true}}})
+	sink2.Flush(ctx, errors.New("boom"))
+	arr2 := loadArray(t, store2.Get(ctx, "c-fail-m-fail-logs"))
+	endFirst2, _ := arr2[len(arr2)-1]["trace"].([]any)[0].(map[string]any)
+	msg2, _ := endFirst2["message"].(string)
+	if !strings.HasPrefix(msg2, "[ERROR] ") {
+		t.Errorf("END message on failure must stay [ERROR]-prefixed, got %q", msg2)
+	}
 }
 
 // TestDebugLogSink_ElapsedTimeIsInSeconds locks the unit contract for

internal/ingestion/task/debug_result_dsl.go

@@ -33,11 +33,13 @@ import (
 )
 
 // ResultSink is an OPTIONAL capability a ProgressSink may implement to receive
-// the debug-run result DSL. The pipeline executor probes for it via a type
-// assertion, so the ProgressSink contract stays unchanged and non-debug
-// (DB-backed) sinks simply ignore it — keeping the coupling one-directional.
+// the debug-run result DSL plus the raw pipeline run output (output["state"]
+// [<id>] is each component's outputs map). The pipeline executor probes for it
+// via a type assertion, so the ProgressSink contract stays unchanged and
+// non-debug (DB-backed) sinks simply ignore it — keeping the coupling
+// one-directional.
 type ResultSink interface {
-	SetResult(dsl map[string]any)
+	SetResult(dsl map[string]any, output map[string]any)
 }
 
 // outputFormats is the priority order used to pick a component's payload key,

internal/ingestion/task/pipeline_executor.go

@@ -495,7 +495,7 @@ func (s *PipelineExecutor) runPipelineWithDSL(ctx context.Context, dsl string) (
 	// contract is unchanged, keeping the coupling one-directional.
 	if rs, ok := s.progressSink.(ResultSink); ok {
 		if resultDSL, e := BuildDebugResultDSL(dsl, output); e == nil {
-			rs.SetResult(resultDSL)
+			rs.SetResult(resultDSL, output)
 		}
 	}