From 4c2398cb793550bf54e808a2dedad2acdc1c4a66 Mon Sep 17 00:00:00 2001 From: euvre <93761161+euvre@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:11:13 +0800 Subject: [PATCH] fix(go-ingestion): write last component output JSON as debug-log END message (#17786) --- internal/ingestion/task/debug_log_sink.go | 48 ++++++++++++++-- .../ingestion/task/debug_log_sink_test.go | 56 ++++++++++++++++++- internal/ingestion/task/debug_result_dsl.go | 10 ++-- internal/ingestion/task/pipeline_executor.go | 2 +- 4 files changed, 104 insertions(+), 12 deletions(-) diff --git a/internal/ingestion/task/debug_log_sink.go b/internal/ingestion/task/debug_log_sink.go index 51b3a7dc91..0bef558c43 100644 --- a/internal/ingestion/task/debug_log_sink.go +++ b/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"][] 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 { diff --git a/internal/ingestion/task/debug_log_sink_test.go b/internal/ingestion/task/debug_log_sink_test.go index 3e0c306ffa..1bb048ba62 100644 --- a/internal/ingestion/task/debug_log_sink_test.go +++ b/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 diff --git a/internal/ingestion/task/debug_result_dsl.go b/internal/ingestion/task/debug_result_dsl.go index 284bec3f34..e9131e65d8 100644 --- a/internal/ingestion/task/debug_result_dsl.go +++ b/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"] +// [] 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, diff --git a/internal/ingestion/task/pipeline_executor.go b/internal/ingestion/task/pipeline_executor.go index 2d9a3168be..e976efaa22 100644 --- a/internal/ingestion/task/pipeline_executor.go +++ b/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) } }