fix: loop component usage issue (#17028)

### Summary

As title

<img width="3718" height="2029" alt="image"
src="https://github.com/user-attachments/assets/227bcd8b-e5b8-465e-983c-c0c0650a4ea5"
/>

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

internal/agent/canvas/loop_semantics_test.go

@@ -33,7 +33,8 @@ package canvas
 
 import (
 	"context"
-	"errors"
+	"encoding/json"
+	"io"
 	"strings"
 	"testing"
 
@@ -43,7 +44,6 @@ import (
 	// exercise real component invocation.
 	_ "ragflow/internal/agent/component"
 	"ragflow/internal/agent/runtime"
-	"ragflow/internal/agent/workflowx"
 )
 
 // runLoopCanvas is the common harness for the e2e loop tests. It
@@ -153,15 +153,8 @@ func TestLoop_DoWhileCounter(t *testing.T) {
 // counter=5 (5 successful body runs).
 func TestLoop_MaxCount(t *testing.T) {
 	state, err := runLoopCanvas(t, counterLoopDSL(1, 100, 5))
-	// workflowx surfaces a MaxIterationsExceeded error when the cap
-	// is hit. Both the error path AND the partial state must be
-	// observable to the caller — the state writes that succeeded
-	// before the cap should still be present.
-	if err == nil {
-		t.Fatalf("expected ErrLoopMaxIterationsExceeded, got nil")
-	}
-	if !errors.Is(err, workflowx.ErrLoopMaxIterationsExceeded) {
-		t.Fatalf("want ErrLoopMaxIterationsExceeded, got: %v", err)
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
 	}
 	v, err := state.GetVar("loop@counter")
 	if err != nil {
@@ -176,6 +169,141 @@ func TestLoop_MaxCount(t *testing.T) {
 	}
 }
 
+// TestLoop_StreamEmitsEveryIteration pins the canvas Loop streaming mode.
+// Python emits Message output from inside a loop on every iteration; the Go
+// canvas must therefore install workflowx loops in every_iteration mode rather
+// than buffering and exposing only the final body result.
+func TestLoop_StreamEmitsEveryIteration(t *testing.T) {
+	cc, err := Compile(context.Background(), counterLoopDSL(1, 3, 50))
+	if err != nil {
+		t.Fatalf("Compile: %v", err)
+	}
+	state := NewCanvasState("run-loop-stream", "task-loop-stream")
+	ctx := withState(context.Background(), state)
+	sr, err := cc.Workflow.Stream(ctx, map[string]any{"query": "go"})
+	if err != nil {
+		t.Fatalf("Stream: %v", err)
+	}
+	defer sr.Close()
+
+	var chunks []map[string]any
+	for {
+		chunk, recvErr := sr.Recv()
+		if recvErr != nil {
+			if recvErr == io.EOF {
+				break
+			}
+			t.Fatalf("Recv: %v", recvErr)
+		}
+		chunks = append(chunks, chunk)
+	}
+	if len(chunks) != 3 {
+		t.Fatalf("stream chunks: got %d (%v), want 3", len(chunks), chunks)
+	}
+	v, err := state.GetVar("loop@counter")
+	if err != nil {
+		t.Fatalf("GetVar: %v", err)
+	}
+	if got, ok := v.(float64); !ok || got != 3 {
+		t.Fatalf("counter after stream: got %T %v, want float64(3)", v, v)
+	}
+}
+
+func TestLoop_EmitsLifecycleEventsForMacroAndBody(t *testing.T) {
+	cc, err := Compile(context.Background(), counterLoopDSL(1, 3, 50))
+	if err != nil {
+		t.Fatalf("Compile: %v", err)
+	}
+	state := NewCanvasState("run-loop-events", "task-loop-events")
+	events := make(chan RunEvent, 32)
+	ctx := withState(context.Background(), state)
+	ctx = WithRunMeta(ctx, &RunMeta{Events: events})
+
+	if _, err := cc.Workflow.Invoke(ctx, map[string]any{"query": "go"}); err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+
+	started, finished := collectLifecycleEvents(t, events)
+	if started["loop"] != 1 {
+		t.Fatalf("loop node_started: got %d, want 1; all started: %v", started["loop"], started)
+	}
+	if finished["loop"] != 1 {
+		t.Fatalf("loop node_finished: got %d, want 1; all finished: %v", finished["loop"], finished)
+	}
+	if started["bump"] != 3 {
+		t.Fatalf("bump node_started: got %d, want 3; all started: %v", started["bump"], started)
+	}
+	if finished["bump"] != 3 {
+		t.Fatalf("bump node_finished: got %d, want 3; all finished: %v", finished["bump"], finished)
+	}
+}
+
+func TestLoop_MessageEmitsEveryIteration(t *testing.T) {
+	dsl := counterLoopDSL(1, 3, 50)
+	bump := dsl.Components["bump"]
+	bump.Downstream = []string{"msg"}
+	dsl.Components["bump"] = bump
+	dsl.Components["msg"] = CanvasComponent{
+		Obj: CanvasComponentObj{
+			ComponentName: "Message",
+			Params: map[string]any{
+				"content": []any{"tick"},
+			},
+		},
+		Upstream: []string{"bump"},
+	}
+
+	cc, err := Compile(context.Background(), dsl)
+	if err != nil {
+		t.Fatalf("Compile: %v", err)
+	}
+	state := NewCanvasState("run-loop-message", "task-loop-message")
+	ctx := withState(context.Background(), state)
+	var emitted []string
+	ctx = runtime.WithCanvasMessageEmitter(ctx, func(content string) {
+		emitted = append(emitted, content)
+	})
+
+	if _, err := cc.Workflow.Invoke(ctx, map[string]any{"query": "go"}); err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if len(emitted) != 3 {
+		t.Fatalf("emitted messages: got %d %#v, want 3", len(emitted), emitted)
+	}
+	for i, msg := range emitted {
+		if msg != "tick" {
+			t.Fatalf("emitted[%d] = %q, want tick; all = %#v", i, msg, emitted)
+		}
+	}
+}
+
+func collectLifecycleEvents(t *testing.T, events <-chan RunEvent) (map[string]int, map[string]int) {
+	t.Helper()
+	started := map[string]int{}
+	finished := map[string]int{}
+	for {
+		select {
+		case ev := <-events:
+			if ev.Type != "node_started" && ev.Type != "node_finished" {
+				continue
+			}
+			var payload struct {
+				ComponentID string `json:"component_id"`
+			}
+			if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {
+				t.Fatalf("decode %s payload %q: %v", ev.Type, ev.Data, err)
+			}
+			if ev.Type == "node_started" {
+				started[payload.ComponentID]++
+			} else {
+				finished[payload.ComponentID]++
+			}
+		default:
+			return started, finished
+		}
+	}
+}
+
 // TestLoop_FactoryErrorSurfaces proves that a factory rejection of a
 // loop body member produces a cpn-scoped error from BuildWorkflow
 // (not a silent placeholder fallback or an opaque error from the

internal/agent/canvas/loop_subgraph.go

@@ -246,7 +246,7 @@ func buildSubWorkflow(
 			return nil, err
 		}
 		nodes[cpnID] = sub.AddLambdaNode(cpnID,
-			compose.InvokableLambda[map[string]any, map[string]any](withStateBracket(body)),
+			compose.InvokableLambda[map[string]any, map[string]any](withStateBracket(cpnID, name, body)),
 			compose.WithNodeName(cpnID),
 		)
 	}

internal/agent/canvas/node_body.go

@@ -300,10 +300,12 @@ func placeholderBody(cpnID string) nodeBodyFn {
 // the body directly), the wrapper degrades to a plain invocation:
 // the body still runs, its output is still tagged with __cpn_id__,
 // but no state snapshot is injected and no result is persisted.
-func withStateBracket(body nodeBodyFn) nodeBodyFn {
+func withStateBracket(cpnID, componentName string, body nodeBodyFn) nodeBodyFn {
 	return func(ctx context.Context, in map[string]any) (map[string]any, error) {
+		originalIn := in
 		state, _, _ := runtime.GetStateFromContext[*runtime.CanvasState](ctx)
 		if state != nil {
+			nodeStartedAt(ctx, state, cpnID, componentName, componentName, originalIn)
 			if in == nil {
 				in = map[string]any{}
 			}
@@ -317,21 +319,30 @@ func withStateBracket(body nodeBodyFn) nodeBodyFn {
 		}
 		out, err := body(ctx, in)
 		if err != nil {
+			if state != nil {
+				nodeFinishedNow(ctx, state, cpnID, componentName, componentName, err)
+			}
 			return nil, err
 		}
-		if state == nil || out == nil {
+		if state == nil {
 			return out, nil
 		}
-		cpnID, _ := out["__cpn_id__"].(string)
-		if cpnID == "" {
+		if out == nil {
+			nodeFinishedNow(ctx, state, cpnID, componentName, componentName, nil)
+			return out, nil
+		}
+		outputCpnID, _ := out["__cpn_id__"].(string)
+		if outputCpnID == "" {
+			nodeFinishedNow(ctx, state, cpnID, componentName, componentName, nil)
 			return out, nil
 		}
 		for k, v := range out {
 			if k == "__cpn_id__" || k == "state" || k == "__legacy_noop__" {
 				continue
 			}
-			state.SetVar(cpnID, k, v)
+			state.SetVar(outputCpnID, k, v)
 		}
+		nodeFinishedNow(ctx, state, cpnID, componentName, componentName, nil)
 		return out, nil
 	}
 }

internal/agent/canvas/scheduler.go

@@ -463,6 +463,18 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string
 				return nil, err
 			}
 			var opts []workflowx.LoopOption
+			opts = append(opts, workflowx.WithLoopStream(workflowx.LoopStreamEveryIteration))
+			opts = append(opts, workflowx.WithLoopLifecycleHooks(
+				func(ctx context.Context, input any) {
+					state, _, _ := runtime.GetStateFromContext[*CanvasState](ctx)
+					in, _ := input.(map[string]any)
+					nodeStartedAt(ctx, state, cpnID, comp.Obj.ComponentName, comp.Obj.ComponentName, in)
+				},
+				func(ctx context.Context, loopErr error) {
+					state, _, _ := runtime.GetStateFromContext[*CanvasState](ctx)
+					nodeFinishedNow(ctx, state, cpnID, comp.Obj.ComponentName, comp.Obj.ComponentName, loopErr)
+				},
+			))
 			if exp.MaxIters > 0 {
 				opts = append(opts, workflowx.WithLoopMaxIterations(exp.MaxIters))
 			}

internal/agent/component/message.go

@@ -216,6 +216,11 @@ func (m *MessageComponent) Invoke(ctx context.Context, inputs map[string]any) (m
 			Text:   resolved,
 		})
 	}
+	if rendered != "" {
+		if !runtime.EmitCanvasMessage(ctx, rendered) {
+			runtime.EmitAgentMessage(ctx, rendered, "")
+		}
+	}
 
 	// Python's Message output schema always contains downloads, including an
 	// empty list. Keeping the key is also important for the full terminal

internal/agent/component/message_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	"ragflow/internal/agent/canvas"
+	"ragflow/internal/agent/runtime"
 )
 
 // TestMessage_ResolveTemplate asserts the canonical {{sys.x}} substitution
@@ -130,6 +131,68 @@ func TestMessage_RuntimeContentInput(t *testing.T) {
 	}
 }
 
+func TestMessage_EmitsAgentMessage(t *testing.T) {
+	c, _ := NewMessageComponent(nil)
+	state := canvas.NewCanvasState("run-emit", "task-emit")
+	ctx := withStateForTest(context.Background(), state)
+	var emitted []string
+	ctx = runtime.WithAgentMessageEmitter(ctx, func(contentDelta, thinkingDelta string) {
+		if contentDelta != "" {
+			emitted = append(emitted, contentDelta)
+		}
+		if thinkingDelta != "" {
+			t.Fatalf("unexpected thinking delta: %q", thinkingDelta)
+		}
+	})
+
+	out, err := c.Invoke(ctx, map[string]any{"content": "visible message"})
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if got, _ := out["content"].(string); got != "visible message" {
+		t.Fatalf("content: got %q, want visible message", got)
+	}
+	if len(emitted) != 1 || emitted[0] != "visible message" {
+		t.Fatalf("emitted = %#v, want [visible message]", emitted)
+	}
+	if !runtime.AgentMessageEventsEmitted(ctx) {
+		t.Fatal("AgentMessageEventsEmitted = false, want true")
+	}
+}
+
+func TestMessage_EmitsDirectCanvasMessage(t *testing.T) {
+	c, _ := NewMessageComponent(nil)
+	state := canvas.NewCanvasState("run-direct", "task-direct")
+	ctx := withStateForTest(context.Background(), state)
+	var direct []string
+	var agent []string
+	ctx = runtime.WithAgentMessageEmitter(ctx, func(contentDelta, thinkingDelta string) {
+		if contentDelta != "" {
+			agent = append(agent, contentDelta)
+		}
+	})
+	ctx = runtime.WithCanvasMessageEmitter(ctx, func(content string) {
+		direct = append(direct, content)
+	})
+
+	out, err := c.Invoke(ctx, map[string]any{"content": "direct message"})
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if got, _ := out["content"].(string); got != "direct message" {
+		t.Fatalf("content: got %q, want direct message", got)
+	}
+	if len(direct) != 1 || direct[0] != "direct message" {
+		t.Fatalf("direct = %#v, want [direct message]", direct)
+	}
+	if len(agent) != 0 {
+		t.Fatalf("agent fallback should not run when direct emitter exists: %#v", agent)
+	}
+	if !runtime.AgentMessageEventsEmitted(ctx) {
+		t.Fatal("AgentMessageEventsEmitted = false, want true")
+	}
+}
+
 func TestMessage_FormalizedContentFallback(t *testing.T) {
 	c, _ := NewMessageComponent(nil)
 	state := canvas.NewCanvasState("run-5", "task-5")

internal/agent/runtime/context.go

@@ -37,11 +37,13 @@ import (
 // distinctly and break ctx.Value lookups).
 type stateCtxKey struct{}
 type agentMessageEmitterCtxKey struct{}
+type canvasMessageEmitterCtxKey struct{}
 
 // AgentMessageEmitter emits visible assistant deltas for an Agent component.
 // The service layer owns the actual SSE envelope; runtime keeps the callback
 // shape free of canvas/service imports.
 type AgentMessageEmitter func(contentDelta, thinkingDelta string)
+type CanvasMessageEmitter func(content string)
 
 type agentMessageEmitterState struct {
 	emit     AgentMessageEmitter
@@ -84,6 +86,16 @@ func WithAgentMessageEmitterControl(ctx context.Context, emit AgentMessageEmitte
 	})
 }
 
+// WithCanvasMessageEmitter attaches the direct Message-component output
+// callback. Unlike AgentMessageEmitter, this path does not parse <think> tags
+// or buffer chunks; Message nodes are already resolved visible content.
+func WithCanvasMessageEmitter(ctx context.Context, emit CanvasMessageEmitter) context.Context {
+	if emit == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, canvasMessageEmitterCtxKey{}, emit)
+}
+
 // HasAgentMessageEmitter reports whether the service layer installed an
 // Agent message stream callback on ctx.
 func HasAgentMessageEmitter(ctx context.Context) bool {
@@ -105,6 +117,23 @@ func EmitAgentMessage(ctx context.Context, contentDelta, thinkingDelta string) b
 	return true
 }
 
+// EmitCanvasMessage emits already-rendered Message-component content directly.
+// It also marks the shared message stream as emitted so the service layer does
+// not re-emit the final state content after workflow completion.
+func EmitCanvasMessage(ctx context.Context, content string) bool {
+	emit, ok := ctx.Value(canvasMessageEmitterCtxKey{}).(CanvasMessageEmitter)
+	if !ok || emit == nil {
+		return false
+	}
+	emit(content)
+	if content != "" {
+		if state, ok := ctx.Value(agentMessageEmitterCtxKey{}).(*agentMessageEmitterState); ok && state != nil {
+			state.emitted = true
+		}
+	}
+	return true
+}
+
 // AgentMessageEventsEmitted reports whether the invocation-scoped Agent
 // message emitter has emitted any deltas during the current run.
 func AgentMessageEventsEmitted(ctx context.Context) bool {

internal/agent/workflowx/loop.go

@@ -81,7 +81,7 @@ const (
 	LoopStreamEveryIteration LoopStreamMode = "every_iteration"
 )
 
-// defaultMaxIterations caps the loop when the caller does not
+// defaultMaxIterations is a safety guard used only when the caller does not
 // configure an explicit limit. The cap is intentionally generous to
 // accommodate real workflows (deep research, iterative refinement) but
 // is finite so a bug in the quit condition cannot spin forever.
@@ -114,9 +114,9 @@ type LoopCondition[T any] func(ctx context.Context, iteration int, prev, next T)
 
 // Sentinel errors. Tests use errors.Is to assert these.
 var (
-	// ErrLoopMaxIterationsExceeded is returned when the configured
-	// (or default) iteration cap is reached without shouldQuit
-	// returning true. The cap exists purely as a safety net.
+	// ErrLoopMaxIterationsExceeded is returned when the default safety
+	// cap is reached without shouldQuit returning true. Explicit loop
+	// limits are normal termination, matching the canvas Python runtime.
 	ErrLoopMaxIterationsExceeded = errors.New("workflowx: loop max iterations exceeded")
 
 	// ErrLoopSubGraphInterrupted is wrapped around an interrupt error
@@ -142,22 +142,26 @@ type LoopOption func(*loopOptions)
 
 type loopOptions struct {
 	maxIterations       int
+	maxIterationsSet    bool
 	compileOpts         []compose.GraphCompileOption
 	runOpts             []compose.Option
 	streamMode          LoopStreamMode
 	checkpointBuilder   func(nodeKey string, iteration int) string
 	enableSubCheckpoint bool
+	onStart             func(context.Context, any)
+	onFinish            func(context.Context, error)
 }
 
-// WithLoopMaxIterations caps the loop at n iterations. The cap is
-// checked AFTER each completed iteration. A value of 0 keeps the
-// default cap in effect. A value of 1 is legal and yields the
-// do-while contract: the sub-workflow executes at least once and the
-// loop exits immediately (subject to shouldQuit).
+// WithLoopMaxIterations caps the loop at n iterations. The cap is checked
+// AFTER each completed iteration. A value of 0 keeps the default safety cap in
+// effect. A positive value is normal loop termination, not an error; this
+// matches the canvas Python runtime's maximum_loop_count semantics. A value of
+// 1 is legal and yields the single-iteration do-while case.
 func WithLoopMaxIterations(n int) LoopOption {
 	return func(o *loopOptions) {
-		if n >= 0 {
+		if n > 0 {
 			o.maxIterations = n
+			o.maxIterationsSet = true
 		}
 	}
 }
@@ -171,6 +175,17 @@ func WithLoopCompileOptions(opts ...compose.GraphCompileOption) LoopOption {
 	}
 }
 
+// WithLoopLifecycleHooks installs callbacks around the outer loop node's
+// execution. Canvas uses this to emit node lifecycle events for the Loop macro
+// without relying on eino state post-processing, which is unsafe for streamed
+// multi-iteration output.
+func WithLoopLifecycleHooks(onStart func(context.Context, any), onFinish func(context.Context, error)) LoopOption {
+	return func(o *loopOptions) {
+		o.onStart = onStart
+		o.onFinish = onFinish
+	}
+}
+
 // WithLoopRunOptions appends run options to every nested sub-workflow
 // Invoke / Stream call. Use this to forward run-level options such as
 // per-iteration callbacks or extra callbacks.
@@ -286,14 +301,7 @@ type loopInterruptState struct {
 // failures are returned as an error and the outer workflow is not
 // modified, so the caller does not need to roll back any state on
 // failure.
-func AddLoopNode[T any](
-	ctx context.Context,
-	wf *compose.Workflow[T, T],
-	key string,
-	sub *compose.Workflow[T, T],
-	shouldQuit LoopCondition[T],
-	opts ...LoopOption,
-) (*compose.WorkflowNode, error) {
+func AddLoopNode[T any](ctx context.Context, wf *compose.Workflow[T, T], key string, sub *compose.Workflow[T, T], shouldQuit LoopCondition[T], opts ...LoopOption) (*compose.WorkflowNode, error) {
 	if wf == nil {
 		return nil, errors.New("workflowx: outer workflow is nil")
 	}
@@ -323,10 +331,27 @@ func AddLoopNode[T any](
 	// struct{} options at run time.
 	lambda, err := compose.AnyLambda[T, T, struct{}](
 		func(ctx context.Context, input T, _ ...struct{}) (T, error) {
-			return runLoopInvoke(ctx, key, compiled, input, shouldQuit, options)
+			if options.onStart != nil {
+				options.onStart(ctx, input)
+			}
+			out, err := runLoopInvoke(ctx, key, compiled, input, shouldQuit, options)
+			if options.onFinish != nil {
+				options.onFinish(ctx, err)
+			}
+			return out, err
 		},
 		func(ctx context.Context, input T, _ ...struct{}) (*schema.StreamReader[T], error) {
-			return runLoopStream(ctx, key, compiled, input, shouldQuit, options)
+			if options.onStart != nil {
+				options.onStart(ctx, input)
+			}
+			reader, err := runLoopStream(ctx, key, compiled, input, shouldQuit, options)
+			if err != nil {
+				if options.onFinish != nil {
+					options.onFinish(ctx, err)
+				}
+				return nil, err
+			}
+			return wrapLoopStreamFinish(ctx, reader, options.onFinish), nil
 		},
 		nil,
 		nil,
@@ -400,14 +425,7 @@ func encodeState(s loopSnapshot) ([]byte, error) {
 // runLoopInvoke executes the loop on the invoke path. It is the
 // body of the loop lambda's Invoke handler. See the package doc and
 // plan §"Invoke path" for the documented state machine.
-func runLoopInvoke[T any](
-	ctx context.Context,
-	nodeKey string,
-	sub compose.Runnable[T, T],
-	input T,
-	shouldQuit LoopCondition[T],
-	options *loopOptions,
-) (T, error) {
+func runLoopInvoke[T any](ctx context.Context, nodeKey string, sub compose.Runnable[T, T], input T, shouldQuit LoopCondition[T], options *loopOptions) (T, error) {
 	var zero T
 
 	snap, err := loadLoopSnapshot[T](ctx, options.streamMode)
@@ -487,10 +505,15 @@ func runLoopInvoke[T any](
 			return next, nil
 		}
 
-		// Cap enforcement. The check uses iteration, not a
-		// pre-decrement, so WithLoopMaxIterations(1) is the
-		// single-iteration do-while case.
+		// Cap enforcement. The check uses iteration, not a pre-decrement,
+		// so WithLoopMaxIterations(1) is the single-iteration do-while
+		// case. Explicit caps are normal termination; the default cap is
+		// the runaway-loop safety net and remains an error.
 		if iteration >= options.maxIterations {
+			bridgeState.delete(subCheckID)
+			if options.maxIterationsSet {
+				return next, nil
+			}
 			return zero, fmt.Errorf("%w: %d", ErrLoopMaxIterationsExceeded, options.maxIterations)
 		}
 
@@ -692,6 +715,10 @@ func runLoopStream[T any](
 				return
 			}
 			if iteration >= options.maxIterations {
+				bridgeState.delete(subCheckID)
+				if options.maxIterationsSet {
+					return
+				}
 				pipeErrCh <- fmt.Errorf("%w: %d", ErrLoopMaxIterationsExceeded, options.maxIterations)
 				return
 			}
@@ -738,6 +765,37 @@ func runLoopStream[T any](
 	return outReader, nil
 }
 
+func wrapLoopStreamFinish[T any](ctx context.Context, reader *schema.StreamReader[T], onFinish func(context.Context, error)) *schema.StreamReader[T] {
+	if onFinish == nil || reader == nil {
+		return reader
+	}
+	outReader, outWriter := schema.Pipe[T](16)
+	go func() {
+		var zero T
+		var finishErr error
+		defer func() {
+			reader.Close()
+			outWriter.Close()
+			onFinish(ctx, finishErr)
+		}()
+		for {
+			v, err := reader.Recv()
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					return
+				}
+				finishErr = err
+				outWriter.Send(zero, err)
+				return
+			}
+			if outWriter.Send(v, nil) {
+				return
+			}
+		}
+	}()
+	return outReader
+}
+
 // sendIterations writes the supplied iteration readers to w. For
 // LoopStreamEveryIteration every reader is forwarded in order; for
 // LoopStreamFinalOnly only the last reader is forwarded (or none

internal/agent/workflowx/loop_integration_test.go

@@ -297,12 +297,11 @@ func TestIntegration_LoopStatePersistedOnInterrupt(t *testing.T) {
 	}
 }
 
-// TestIntegration_MaxIterationsExceeded_OnInvokePath asserts
-// that a sustained (non-converging) loop run surfaces
-// ErrLoopMaxIterationsExceeded through the outer invoke. This
-// uses a non-interrupting sub-workflow so the loop actually
-// reaches the cap.
-func TestIntegration_MaxIterationsExceeded_OnInvokePath(t *testing.T) {
+// TestIntegration_ExplicitMaxIterationsStops_OnInvokePath asserts that a
+// sustained non-converging loop returns normally when it reaches an explicit
+// cap. This uses a non-interrupting sub-workflow so the loop actually reaches
+// the cap.
+func TestIntegration_ExplicitMaxIterationsStops_OnInvokePath(t *testing.T) {
 	var subCalls atomic.Int64
 	subStore := newInMemoryStore()
 	sub := counterSub(t, &subCalls)
@@ -328,9 +327,12 @@ func TestIntegration_MaxIterationsExceeded_OnInvokePath(t *testing.T) {
 		t.Fatalf("compile: %v", err)
 	}
 
-	_, err = compiled.Invoke(context.Background(), 0)
-	if !errors.Is(err, ErrLoopMaxIterationsExceeded) {
-		t.Fatalf("got %v, want ErrLoopMaxIterationsExceeded", err)
+	out, err := compiled.Invoke(context.Background(), 0)
+	if err != nil {
+		t.Fatalf("Invoke: %v", err)
+	}
+	if out != 3 {
+		t.Errorf("output: got %d, want 3", out)
 	}
 	if got := subCalls.Load(); got != 3 {
 		t.Errorf("sub invocations: got %d, want 3", got)

internal/agent/workflowx/loop_options_test.go

@@ -83,6 +83,9 @@ func TestOptions_DefaultMaxIterations(t *testing.T) {
 	if opts.maxIterations <= 0 {
 		t.Errorf("default max iterations: got %d, want > 0", opts.maxIterations)
 	}
+	if opts.maxIterationsSet {
+		t.Error("default max iterations should be a safety cap, not an explicit user cap")
+	}
 }
 
 // TestOptions_WithLoopMaxIterations_ZeroKeepsDefault asserts that
@@ -93,6 +96,9 @@ func TestOptions_WithLoopMaxIterations_ZeroKeepsDefault(t *testing.T) {
 	if opts.maxIterations <= 0 {
 		t.Errorf("explicit zero: got %d, want > 0 (default)", opts.maxIterations)
 	}
+	if opts.maxIterationsSet {
+		t.Error("explicit zero should keep the default safety cap")
+	}
 }
 
 // TestOptions_WithLoopMaxIterations_NegativeKeepsDefault asserts
@@ -103,6 +109,9 @@ func TestOptions_WithLoopMaxIterations_NegativeKeepsDefault(t *testing.T) {
 	if opts.maxIterations <= 0 {
 		t.Errorf("negative: got %d, want > 0 (default)", opts.maxIterations)
 	}
+	if opts.maxIterationsSet {
+		t.Error("negative max iterations should keep the default safety cap")
+	}
 }
 
 // TestOptions_WithLoopMaxIterations_Positive asserts the positive
@@ -112,6 +121,9 @@ func TestOptions_WithLoopMaxIterations_Positive(t *testing.T) {
 	if opts.maxIterations != 42 {
 		t.Errorf("got %d, want 42", opts.maxIterations)
 	}
+	if !opts.maxIterationsSet {
+		t.Error("positive max iterations should be marked as an explicit user cap")
+	}
 }
 
 // TestOptions_CheckpointBuilder_Default is non-empty. The default

internal/agent/workflowx/loop_test.go

@@ -160,9 +160,10 @@ func TestLoop_DoWhileContract(t *testing.T) {
 	}
 }
 
-// TestLoop_MaxIterationsExceeded asserts that exceeding the
-// configured cap returns ErrLoopMaxIterationsExceeded.
-func TestLoop_MaxIterationsExceeded(t *testing.T) {
+// TestLoop_ExplicitMaxIterationsStops asserts that reaching an
+// explicitly configured cap returns the final iteration output. Canvas
+// maximum_loop_count uses this as normal termination.
+func TestLoop_ExplicitMaxIterationsStops(t *testing.T) {
 	shouldQuit := func(_ context.Context, _ int, _, _ int) (bool, error) {
 		return false, nil
 	}
@@ -180,9 +181,33 @@ func TestLoop_MaxIterationsExceeded(t *testing.T) {
 	if err != nil {
 		t.Fatalf("compile: %v", err)
 	}
-	_, err = compiled.Invoke(context.Background(), 0)
+	out, err := compiled.Invoke(context.Background(), 0)
+	if err != nil {
+		t.Fatalf("invoke: %v", err)
+	}
+	if out != 3 {
+		t.Fatalf("output: got %d, want 3", out)
+	}
+}
+
+// TestLoop_DefaultSafetyMaxIterationsExceeded asserts that the non-explicit
+// safety cap remains an error. The test lowers the internal cap to keep the
+// case small while preserving the "not user configured" option state.
+func TestLoop_DefaultSafetyMaxIterationsExceeded(t *testing.T) {
+	shouldQuit := func(_ context.Context, _ int, _, _ int) (bool, error) {
+		return false, nil
+	}
+	compiledSub, err := buildSubIncrement(t).Compile(context.Background())
+	if err != nil {
+		t.Fatalf("compile sub: %v", err)
+	}
+	options := getLoopOptions(nil)
+	options.maxIterations = 3
+	options.enableSubCheckpoint = false
+
+	_, err = runLoopInvoke(context.Background(), "loop", compiledSub, 0, shouldQuit, options)
 	if !errors.Is(err, ErrLoopMaxIterationsExceeded) {
-		t.Fatalf("invoke: got %v, want ErrLoopMaxIterationsExceeded", err)
+		t.Fatalf("runLoopInvoke: got %v, want ErrLoopMaxIterationsExceeded", err)
 	}
 }
 

internal/service/agent.go

@@ -1185,6 +1185,9 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv
 		})
 		agentMessageEmit, agentMessageFinalize, agentMessageReset := makeAgentMessageDeltaEmitterWithFinalizer(emit)
 		ctx2 = runtime.WithAgentMessageEmitterControl(ctx2, agentMessageEmit, agentMessageFinalize, agentMessageReset)
+		ctx2 = runtime.WithCanvasMessageEmitter(ctx2, func(content string) {
+			emitAgentMessageEvent(emit, canvas.MessageEvent{Content: content})
+		})
 
 		// Seed initial env/sys values from the Canvas DSL globals.
 		// Python's self.globals dict stores "sys.*" and "env.*" under