diff --git a/internal/agent/canvas/loop_semantics_test.go b/internal/agent/canvas/loop_semantics_test.go index c34d3c13fa..be37ca63ea 100644 --- a/internal/agent/canvas/loop_semantics_test.go +++ b/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 diff --git a/internal/agent/canvas/loop_subgraph.go b/internal/agent/canvas/loop_subgraph.go index 90f04a8f41..30227877af 100644 --- a/internal/agent/canvas/loop_subgraph.go +++ b/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), ) } diff --git a/internal/agent/canvas/node_body.go b/internal/agent/canvas/node_body.go index c8859811ee..2eb93be504 100644 --- a/internal/agent/canvas/node_body.go +++ b/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 } } diff --git a/internal/agent/canvas/scheduler.go b/internal/agent/canvas/scheduler.go index eb8fcecd5f..e5a2df82be 100644 --- a/internal/agent/canvas/scheduler.go +++ b/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)) } diff --git a/internal/agent/component/message.go b/internal/agent/component/message.go index 817efe0608..1b02c3688d 100644 --- a/internal/agent/component/message.go +++ b/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 diff --git a/internal/agent/component/message_test.go b/internal/agent/component/message_test.go index cace320e32..6ff4ac7901 100644 --- a/internal/agent/component/message_test.go +++ b/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") diff --git a/internal/agent/runtime/context.go b/internal/agent/runtime/context.go index af288ce68e..d0c7f04e7d 100644 --- a/internal/agent/runtime/context.go +++ b/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 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 { diff --git a/internal/agent/workflowx/loop.go b/internal/agent/workflowx/loop.go index 923304e91c..90b64da5fd 100644 --- a/internal/agent/workflowx/loop.go +++ b/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 diff --git a/internal/agent/workflowx/loop_integration_test.go b/internal/agent/workflowx/loop_integration_test.go index df4561bc5f..94b4c418d1 100644 --- a/internal/agent/workflowx/loop_integration_test.go +++ b/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) diff --git a/internal/agent/workflowx/loop_options_test.go b/internal/agent/workflowx/loop_options_test.go index 3bfd14ac16..5d145e5d76 100644 --- a/internal/agent/workflowx/loop_options_test.go +++ b/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 diff --git a/internal/agent/workflowx/loop_test.go b/internal/agent/workflowx/loop_test.go index af3caf16a2..1aff248fe5 100644 --- a/internal/agent/workflowx/loop_test.go +++ b/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) } } diff --git a/internal/service/agent.go b/internal/service/agent.go index 111018e8bd..05c68de3e2 100644 --- a/internal/service/agent.go +++ b/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