mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-11-10 21:32:43 +08:00
Refactor: better message - canvas -> agent (#20395)
This commit is contained in:
1 parent
c1a680e905
commit
7cd4718a0b
39 files changed
+154
-130
No files matched your search
@@ -125,7 +125,7 @@ func walkAndClose(v any, seen map[any]bool) {
|
||||
func safeClose(closer interface{ Close() }) {
|
||||
defer func() {
|
||||
if rec := recover(); rec != nil {
|
||||
common.Warn("canvas: Close() panicked", zap.Any("recover", rec))
|
||||
common.Warn("agent: Close() panicked", zap.Any("recover", rec))
|
||||
}
|
||||
}()
|
||||
closer.Close()
|
||||
|
||||
@@ -173,7 +173,7 @@ func Compile(ctx context.Context, c *Canvas, opts ...CompileOption) (*CompiledCa
|
||||
}
|
||||
}
|
||||
if n > 0 {
|
||||
common.Info("canvas: Compile received Canvas with legacy LoopItem/IterationItem/Iteration nodes; this path bypassed dsl.NormalizeForCanvas — the fold step is not applied", zap.Int("n", n))
|
||||
common.Info("agent: Compile received Canvas with legacy LoopItem/IterationItem/Iteration nodes; this path bypassed dsl.NormalizeForCanvas — the fold step is not applied", zap.Int("n", n))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -206,7 +206,7 @@ func Compile(ctx context.Context, c *Canvas, opts ...CompileOption) (*CompiledCa
|
||||
}
|
||||
}
|
||||
if len(bad) > 0 {
|
||||
return nil, fmt.Errorf("canvas: Compile: WithInterruptAfterNonTerminalCpn forbids UserFillUp/legacy-no-op nodes %v (plan §4.2.b): ingestion has no user to fill up and no-op nodes do not report progress, breaking the resume/percent invariant", bad)
|
||||
return nil, fmt.Errorf("agent: Compile: WithInterruptAfterNonTerminalCpn forbids UserFillUp/legacy-no-op nodes %v (plan §4.2.b): ingestion has no user to fill up and no-op nodes do not report progress, breaking the resume/percent invariant", bad)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -220,7 +220,7 @@ func Compile(ctx context.Context, c *Canvas, opts ...CompileOption) (*CompiledCa
|
||||
|
||||
wf, err := BuildWorkflow(ctx, c)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: build workflow: %w", err)
|
||||
return nil, fmt.Errorf("agent: build workflow: %w", err)
|
||||
}
|
||||
|
||||
compileOpts := make([]compose.GraphCompileOption, 0, 4)
|
||||
@@ -249,7 +249,7 @@ func Compile(ctx context.Context, c *Canvas, opts ...CompileOption) (*CompiledCa
|
||||
|
||||
runnable, err := wf.Compile(ctx, compileOpts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: eino compile: %w", err)
|
||||
return nil, fmt.Errorf("agent: eino compile: %w", err)
|
||||
}
|
||||
return &CompiledCanvas{Workflow: runnable, CheckPointID: cfg.CheckPointID}, nil
|
||||
}
|
||||
|
||||
@@ -119,7 +119,7 @@ func TestCompile_NoLogOnCleanCanvas(t *testing.T) {
|
||||
|
||||
got := buf.String()
|
||||
if strings.Contains(got, "LoopItem/IterationItem") {
|
||||
t.Errorf("unexpected legacy-node log on clean canvas: %q", got)
|
||||
t.Errorf("unexpected legacy-node log on clean agent: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -23,11 +23,11 @@ import "fmt"
|
||||
// normalized flat shape (`name`/`params`) that NormalizeForCanvas emits.
|
||||
func DecodeFromDSL(dsl map[string]any) (*Canvas, error) {
|
||||
if len(dsl) == 0 {
|
||||
return nil, fmt.Errorf("canvas: empty DSL")
|
||||
return nil, fmt.Errorf("agent: empty DSL")
|
||||
}
|
||||
rawComps, ok := dsl["components"].(map[string]any)
|
||||
if !ok || len(rawComps) == 0 {
|
||||
return nil, fmt.Errorf("canvas: no components")
|
||||
return nil, fmt.Errorf("agent: no components")
|
||||
}
|
||||
c := &Canvas{
|
||||
Components: make(map[string]CanvasComponent, len(rawComps)),
|
||||
@@ -68,7 +68,7 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) {
|
||||
}
|
||||
name, params, downstream, upstream := decodeComponentFields(comp)
|
||||
if name == "" {
|
||||
return nil, fmt.Errorf("canvas: component %q has empty component_name", cpnID)
|
||||
return nil, fmt.Errorf("agent: component %q has empty component_name", cpnID)
|
||||
}
|
||||
c.Components[cpnID] = CanvasComponent{
|
||||
Obj: CanvasComponentObj{
|
||||
@@ -80,7 +80,7 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) {
|
||||
}
|
||||
}
|
||||
if len(c.Components) == 0 {
|
||||
return nil, fmt.Errorf("canvas: no components")
|
||||
return nil, fmt.Errorf("agent: no components")
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
@@ -152,7 +152,7 @@ func UserFillUpNodeBody(cpnID string, params map[string]any) func(ctx context.Co
|
||||
// returns an interrupt error or panics on engine misuse. Keep
|
||||
// the guard so test runs without a runner surface a clear
|
||||
// message rather than a panic.
|
||||
return nil, fmt.Errorf("canvas: UserFillUp %q: interrupt did not halt execution", cpnID)
|
||||
return nil, fmt.Errorf("agent: UserFillUp %q: interrupt did not halt execution", cpnID)
|
||||
}
|
||||
return body
|
||||
}
|
||||
|
||||
@@ -75,13 +75,13 @@ type loopExpansion struct {
|
||||
// sub-graph, not the outer graph).
|
||||
func buildLoopExpansion(ctx context.Context, c *Canvas, loopID string) (*loopExpansion, error) {
|
||||
if c == nil {
|
||||
return nil, fmt.Errorf("canvas: nil canvas")
|
||||
return nil, fmt.Errorf("agent: nil canvas")
|
||||
}
|
||||
if loopID == "" {
|
||||
return nil, fmt.Errorf("canvas: buildLoopExpansion: empty loopID")
|
||||
return nil, fmt.Errorf("agent: buildLoopExpansion: empty loopID")
|
||||
}
|
||||
if _, ok := c.Components[loopID]; !ok {
|
||||
return nil, fmt.Errorf("canvas: buildLoopExpansion: unknown cpn %q", loopID)
|
||||
return nil, fmt.Errorf("agent: buildLoopExpansion: unknown cpn %q", loopID)
|
||||
}
|
||||
|
||||
loopComp := c.Components[loopID]
|
||||
@@ -90,19 +90,19 @@ func buildLoopExpansion(ctx context.Context, c *Canvas, loopID string) (*loopExp
|
||||
|
||||
initValues, err := resolveInitialVariables(loopComp.Obj.Params)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: loop %q: %w", loopID, err)
|
||||
return nil, fmt.Errorf("agent: loop %q: %w", loopID, err)
|
||||
}
|
||||
|
||||
shouldQuit, err := translateLoopCondition(loopID, loopComp.Obj.Params)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: loop %q: %w", loopID, err)
|
||||
return nil, fmt.Errorf("agent: loop %q: %w", loopID, err)
|
||||
}
|
||||
|
||||
maxIters := readMaxLoopCount(loopComp.Obj.Params)
|
||||
|
||||
sub, err := buildSubWorkflow(ctx, c, members, loopID, initValues)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: loop %q: %w", loopID, err)
|
||||
return nil, fmt.Errorf("agent: loop %q: %w", loopID, err)
|
||||
}
|
||||
|
||||
return &loopExpansion{
|
||||
@@ -222,7 +222,7 @@ func buildSubWorkflow(
|
||||
ref, _ := spec.Value.(string)
|
||||
resolved, err := state.GetVar(ref)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: loop %q init: variable %q ref %q: %w", loopID, k, ref, err)
|
||||
return nil, fmt.Errorf("agent: loop %q init: variable %q ref %q: %w", loopID, k, ref, err)
|
||||
}
|
||||
v = resolved
|
||||
}
|
||||
@@ -248,7 +248,7 @@ func buildSubWorkflow(
|
||||
for cpnID := range members {
|
||||
name := c.Components[cpnID].Obj.ComponentName
|
||||
if name == "" {
|
||||
return nil, fmt.Errorf("canvas: loop %q member %q has empty component_name", loopID, cpnID)
|
||||
return nil, fmt.Errorf("agent: loop %q member %q has empty component_name", loopID, cpnID)
|
||||
}
|
||||
deferToMessage := directMessageDownstream(c, cpnID)
|
||||
nodeOpts := runtime.ComponentExecutionOptions{
|
||||
|
||||
@@ -274,7 +274,7 @@ func TestWireMultiBranches_RegistersTwoChildren(t *testing.T) {
|
||||
func TestWireMultiBranches_NilSafety(t *testing.T) {
|
||||
// nil canvas
|
||||
if got := wireMultiBranches(nil, nil, nil); got != nil {
|
||||
t.Errorf("nil canvas: got %v, want nil", got)
|
||||
t.Errorf("nil agent: got %v, want nil", got)
|
||||
}
|
||||
wf := compose.NewWorkflow[map[string]any, map[string]any]()
|
||||
if got := wireMultiBranches(wf, nil, nil); got != nil {
|
||||
|
||||
@@ -149,10 +149,10 @@ func buildNodeBodyWithOptions(ctx context.Context, cpnID, name string, params ma
|
||||
if factory := resolveComponentFactory(ctx); factory != nil {
|
||||
comp, err := factory(name, params)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: component %q (%s): factory: %w", cpnID, name, err)
|
||||
return nil, fmt.Errorf("agent: component %q (%s): factory: %w", cpnID, name, err)
|
||||
}
|
||||
if comp == nil {
|
||||
return nil, fmt.Errorf("canvas: component %q (%s): factory returned nil component", cpnID, name)
|
||||
return nil, fmt.Errorf("agent: component %q (%s): factory returned nil component", cpnID, name)
|
||||
}
|
||||
// Pass the class name through to the body for structured logging
|
||||
// without the runtime.Component interface needing to expose Name().
|
||||
@@ -164,7 +164,7 @@ func buildNodeBodyWithOptions(ctx context.Context, cpnID, name string, params ma
|
||||
// canvas-only unit tests; production wiring always installs a
|
||||
// factory via component.init().
|
||||
if !isKnownPrimitive(name) {
|
||||
return nil, fmt.Errorf("canvas: component %q has unknown component_name %q (typo? not in isKnownPrimitive, not in legacyNoOpNames)", cpnID, name)
|
||||
return nil, fmt.Errorf("agent: component %q has unknown component_name %q (typo? not in isKnownPrimitive, not in legacyNoOpNames)", cpnID, name)
|
||||
}
|
||||
return placeholderBody(cpnID), nil
|
||||
}
|
||||
@@ -258,24 +258,24 @@ func realComponentBodyWithOptions(cpnID, componentClass string, comp runtime.Com
|
||||
// A user cancel is a normal control path; the authoritative
|
||||
// "Task ... cancelled" line is logged by the service layer, so
|
||||
// keep this at debug to avoid duplicate noise.
|
||||
common.Debug("canvas: component invoke cancelled",
|
||||
common.Debug("agent: component invoke cancelled",
|
||||
zap.String("component_id", cpnID),
|
||||
zap.String("component_class", componentClass))
|
||||
return nil, fmt.Errorf("canvas: component %q invoke: cancelled: %w", cpnID, invokeErr)
|
||||
return nil, fmt.Errorf("agent: component %q invoke: cancelled: %w", cpnID, invokeErr)
|
||||
case errors.Is(invokeErr, context.DeadlineExceeded):
|
||||
common.Error("canvas: component invoke failed", invokeErr,
|
||||
common.Error("agent: component invoke failed", invokeErr,
|
||||
zap.String("component_id", cpnID),
|
||||
zap.String("component_class", componentClass))
|
||||
return nil, fmt.Errorf("canvas: component %q invoke: context deadline exceeded: %w", cpnID, invokeErr)
|
||||
return nil, fmt.Errorf("agent: component %q invoke: context deadline exceeded: %w", cpnID, invokeErr)
|
||||
default:
|
||||
// Surface the failure as a structured log line. The wrapped error
|
||||
// already carries the full cause chain (e.g. deepseek DNS/timeout),
|
||||
// but without this the failure only showed up as a generic
|
||||
// "Task ... failed" line with no clear cause in the logs.
|
||||
common.Error("canvas: component invoke failed", invokeErr,
|
||||
common.Error("agent: component invoke failed", invokeErr,
|
||||
zap.String("component_id", cpnID),
|
||||
zap.String("component_class", componentClass))
|
||||
return nil, fmt.Errorf("canvas: component %q invoke: %w", cpnID, invokeErr)
|
||||
return nil, fmt.Errorf("agent: component %q invoke: %w", cpnID, invokeErr)
|
||||
}
|
||||
}
|
||||
if out == nil {
|
||||
|
||||
@@ -43,14 +43,14 @@ type parallelExpansion struct {
|
||||
|
||||
func buildParallelExpansion(ctx context.Context, c *Canvas, parallelID string) (*parallelExpansion, error) {
|
||||
if c == nil {
|
||||
return nil, fmt.Errorf("canvas: nil canvas")
|
||||
return nil, fmt.Errorf("agent: nil canvas")
|
||||
}
|
||||
if parallelID == "" {
|
||||
return nil, fmt.Errorf("canvas: buildParallelExpansion: empty parallelID")
|
||||
return nil, fmt.Errorf("agent: buildParallelExpansion: empty parallelID")
|
||||
}
|
||||
comp, ok := c.Components[parallelID]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("canvas: buildParallelExpansion: unknown cpn %q", parallelID)
|
||||
return nil, fmt.Errorf("agent: buildParallelExpansion: unknown cpn %q", parallelID)
|
||||
}
|
||||
|
||||
members := collectGroupedMembers(c, parallelID)
|
||||
@@ -60,17 +60,17 @@ func buildParallelExpansion(ctx context.Context, c *Canvas, parallelID string) (
|
||||
|
||||
itemsRef, maxConcurrency, outputRefs, err := readParallelParams(comp.Obj.Params)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: parallel %q: %w", parallelID, err)
|
||||
return nil, fmt.Errorf("agent: parallel %q: %w", parallelID, err)
|
||||
}
|
||||
|
||||
sub, err := buildParallelItemWorkflow(ctx, c, parallelID, members)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: parallel %q: %w", parallelID, err)
|
||||
return nil, fmt.Errorf("agent: parallel %q: %w", parallelID, err)
|
||||
}
|
||||
|
||||
graph, err := buildParallelOuterWorkflow(ctx, parallelID, itemsRef, maxConcurrency, outputRefs, sub)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: parallel %q: %w", parallelID, err)
|
||||
return nil, fmt.Errorf("agent: parallel %q: %w", parallelID, err)
|
||||
}
|
||||
|
||||
return ¶llelExpansion{
|
||||
@@ -161,7 +161,7 @@ func buildParallelItemWorkflow(
|
||||
compose.InvokableLambda(func(ctx context.Context, in map[string]any) (map[string]any, error) {
|
||||
localState, err := GetStateFromContext(ctx)
|
||||
if err != nil || localState == nil {
|
||||
return nil, fmt.Errorf("canvas: parallel %q item collector: no canvas state in context", parallelID)
|
||||
return nil, fmt.Errorf("agent: parallel %q item collector: no canvas state in context", parallelID)
|
||||
}
|
||||
out := map[string]any{
|
||||
"item": localState.Globals["__item__"],
|
||||
@@ -203,11 +203,11 @@ func buildParallelOuterWorkflow(
|
||||
compose.InvokableLambda(func(ctx context.Context, _ map[string]any) ([]map[string]any, error) {
|
||||
state, err := GetStateFromContext(ctx)
|
||||
if err != nil || state == nil {
|
||||
return nil, fmt.Errorf("canvas: parallel %q: no canvas state in context", key)
|
||||
return nil, fmt.Errorf("agent: parallel %q: no canvas state in context", key)
|
||||
}
|
||||
raw, err := state.GetVar(itemsRef)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: parallel %q items_ref %q: %w", key, itemsRef, err)
|
||||
return nil, fmt.Errorf("agent: parallel %q items_ref %q: %w", key, itemsRef, err)
|
||||
}
|
||||
return toParallelItems(raw)
|
||||
}),
|
||||
|
||||
@@ -268,7 +268,7 @@ func (r *Runner) Run(
|
||||
out := make(chan RunEvent, 8)
|
||||
|
||||
if run == nil {
|
||||
pushErr(ctx, out, "canvas: nil RunFunc", sessionID)
|
||||
pushErr(ctx, out, "agent: nil RunFunc", sessionID)
|
||||
close(out)
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -385,10 +385,10 @@ func nodeFinishedNow(ctx context.Context, state *CanvasState, cpnID, componentNa
|
||||
// extracts from context for us (via WithGenLocalState — wired in compile.go).
|
||||
func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string]any, map[string]any], error) {
|
||||
if c == nil {
|
||||
return nil, fmt.Errorf("canvas: nil canvas")
|
||||
return nil, fmt.Errorf("agent: nil canvas")
|
||||
}
|
||||
if len(c.Components) == 0 {
|
||||
return nil, fmt.Errorf("canvas: no components")
|
||||
return nil, fmt.Errorf("agent: no components")
|
||||
}
|
||||
|
||||
// GenLocalState copies the request-initialized *CanvasState when the
|
||||
@@ -485,7 +485,7 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string
|
||||
ctx, wf, cpnID, exp.Sub, exp.ShouldQuit, opts...,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: install loop %q: %w", cpnID, err)
|
||||
return nil, fmt.Errorf("agent: install loop %q: %w", cpnID, err)
|
||||
}
|
||||
macroNodes[cpnID] = node
|
||||
for m := range exp.Members {
|
||||
@@ -550,7 +550,7 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string
|
||||
}
|
||||
name := c.Components[cpnID].Obj.ComponentName
|
||||
if name == "" {
|
||||
return nil, fmt.Errorf("canvas: component %q has empty component_name", cpnID)
|
||||
return nil, fmt.Errorf("agent: component %q has empty component_name", cpnID)
|
||||
}
|
||||
deferToMessage := directMessageDownstream(c, cpnID)
|
||||
nodeOpts := runtime.ComponentExecutionOptions{
|
||||
@@ -637,14 +637,14 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string
|
||||
}
|
||||
wired[e] = struct{}{}
|
||||
if e.cpn == e.up {
|
||||
return nil, fmt.Errorf("canvas: self-edge on %q", e.cpn)
|
||||
return nil, fmt.Errorf("agent: self-edge on %q", e.cpn)
|
||||
}
|
||||
if resolveNode(e.up) == nil {
|
||||
return nil, fmt.Errorf("canvas: component %q has unknown upstream %q", e.cpn, e.up)
|
||||
return nil, fmt.Errorf("agent: component %q has unknown upstream %q", e.cpn, e.up)
|
||||
}
|
||||
cpnNode := resolveNode(e.cpn)
|
||||
if cpnNode == nil {
|
||||
return nil, fmt.Errorf("canvas: pending edge references unknown cpn %q", e.cpn)
|
||||
return nil, fmt.Errorf("agent: pending edge references unknown cpn %q", e.cpn)
|
||||
}
|
||||
if !first[e.cpn] {
|
||||
cpnNode.AddInput(e.up)
|
||||
@@ -758,7 +758,7 @@ func wireWorkflowTerminals(
|
||||
) error {
|
||||
if len(terminals) == 0 {
|
||||
if fallback == "" {
|
||||
return fmt.Errorf("canvas: end node not set")
|
||||
return fmt.Errorf("agent: end node not set")
|
||||
}
|
||||
terminals = []string{fallback}
|
||||
}
|
||||
|
||||
@@ -157,7 +157,7 @@ func TestBuildWorkflow_MultiTerminalSucceeds(t *testing.T) {
|
||||
|
||||
cc, err := Compile(t.Context(), c)
|
||||
if err != nil {
|
||||
t.Fatalf("Compile multi-terminal canvas: %v", err)
|
||||
t.Fatalf("Compile multi-terminal agent: %v", err)
|
||||
}
|
||||
if cc.Workflow == nil {
|
||||
t.Fatal("nil compiled multi-terminal workflow")
|
||||
@@ -225,7 +225,7 @@ func TestBuildWorkflow_ParallelGroupWithOuterFollowerSucceeds(t *testing.T) {
|
||||
|
||||
cc, err := Compile(t.Context(), c)
|
||||
if err != nil {
|
||||
t.Fatalf("Compile parallel canvas: %v", err)
|
||||
t.Fatalf("Compile parallel agent: %v", err)
|
||||
}
|
||||
if cc.Workflow == nil {
|
||||
t.Fatal("nil compiled parallel workflow")
|
||||
|
||||
@@ -68,7 +68,7 @@ func (c *ToolBackedComponent) GetInputForm() map[string]any { return c.spec.Inpu
|
||||
func (c *ToolBackedComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map[string]any) (map[string]any, error) {
|
||||
argsJSON, err := json.Marshal(inputs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: %s: encode inputs: %w", c.name, err)
|
||||
return nil, fmt.Errorf("agent: %s: encode inputs: %w", c.name, err)
|
||||
}
|
||||
|
||||
raw, invokeErr := c.tool.InvokableRun(ctx, string(argsJSON))
|
||||
@@ -80,9 +80,9 @@ func (c *ToolBackedComponent) Invoke(ctx context.Context, db *gorm.DB, inputs ma
|
||||
}
|
||||
if rawValue, invalid := decoded["_raw"]; invalid {
|
||||
if invokeErr != nil {
|
||||
return nil, fmt.Errorf("canvas: %s: %w", c.name, invokeErr)
|
||||
return nil, fmt.Errorf("agent: %s: %w", c.name, invokeErr)
|
||||
}
|
||||
return nil, fmt.Errorf("canvas: %s: invalid tool result: %v", c.name, rawValue)
|
||||
return nil, fmt.Errorf("agent: %s: invalid tool result: %v", c.name, rawValue)
|
||||
}
|
||||
if existing, _ := decoded["_ERROR"].(string); strings.TrimSpace(existing) != "" {
|
||||
outputs := c.tool.BuildComponentOutputs(decoded)
|
||||
@@ -93,7 +93,7 @@ func (c *ToolBackedComponent) Invoke(ctx context.Context, db *gorm.DB, inputs ma
|
||||
return outputs, nil
|
||||
}
|
||||
if invokeErr != nil {
|
||||
return nil, fmt.Errorf("canvas: %s: %w", c.name, invokeErr)
|
||||
return nil, fmt.Errorf("agent: %s: %w", c.name, invokeErr)
|
||||
}
|
||||
|
||||
if builder, ok := c.tool.(agenttool.ReferenceBuilder); ok {
|
||||
|
||||
@@ -228,7 +228,7 @@ func (c *retrievalComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map
|
||||
argsJSON, _ := json.Marshal(merged)
|
||||
out, err := c.inner.InvokableRun(ctx, string(argsJSON))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("canvas: Retrieval: %w", err)
|
||||
return nil, fmt.Errorf("agent: Retrieval: %w", err)
|
||||
}
|
||||
common.Debug("agent retrieval component: output",
|
||||
zap.String("tool_output", out),
|
||||
@@ -557,7 +557,7 @@ func (c *codeExecComponent) Invoke(ctx context.Context, db *gorm.DB, inputs map[
|
||||
attachCodeExecArtifacts(ctx, decoded)
|
||||
|
||||
if err != nil {
|
||||
return decoded, fmt.Errorf("canvas: CodeExec: %w", err)
|
||||
return decoded, fmt.Errorf("agent: CodeExec: %w", err)
|
||||
}
|
||||
return decoded, nil
|
||||
}
|
||||
|
||||
@@ -464,11 +464,11 @@ func ResetAgentMessageEmission(ctx context.Context) {
|
||||
func GetStateFromContext(ctx context.Context) (*CanvasState, error) {
|
||||
v := ctx.Value(stateCtxKey{})
|
||||
if v == nil {
|
||||
return nil, fmt.Errorf("canvas: no state in context")
|
||||
return nil, fmt.Errorf("agent: no state in context")
|
||||
}
|
||||
s, ok := v.(*CanvasState)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("canvas: state type mismatch: have %T, want *CanvasState", v)
|
||||
return nil, fmt.Errorf("agent: state type mismatch: have %T, want *CanvasState", v)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
@@ -926,7 +926,7 @@ func getVarLocked(s *CanvasState, ref string) (any, error) {
|
||||
}
|
||||
return dotTraverse(outputs, tail), nil
|
||||
default:
|
||||
return nil, fmt.Errorf("canvas: invalid variable reference %q", ref)
|
||||
return nil, fmt.Errorf("agent: invalid variable reference %q", ref)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -95,7 +95,7 @@ func ResolveTemplate(s string, state *CanvasState) (string, error) {
|
||||
v, err := state.GetVar(ref)
|
||||
if err != nil {
|
||||
if firstErr == nil {
|
||||
firstErr = fmt.Errorf("canvas: resolve %q: %w", ref, err)
|
||||
firstErr = fmt.Errorf("agent: resolve %q: %w", ref, err)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ func applyUserCanvasTagFilter(ctx context.Context, db *gorm.DB, query *gorm.DB,
|
||||
return query.Where(tagQuery)
|
||||
}
|
||||
|
||||
var ErrUserCanvasNotFound = errors.New("user_canvas: not found or access denied")
|
||||
var ErrUserCanvasNotFound = errors.New("agent: not found or access denied")
|
||||
|
||||
// UserCanvasDAO user canvas data access object
|
||||
type UserCanvasDAO struct{}
|
||||
|
||||
@@ -56,7 +56,7 @@ func TestUserCanvasDAOUpdateDSL(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: originalDSL,
|
||||
}); err != nil {
|
||||
t.Fatalf("failed to create canvas: %v", err)
|
||||
t.Fatalf("failed to create agent: %v", err)
|
||||
}
|
||||
|
||||
newDSL := entity.JSONMap{
|
||||
@@ -76,7 +76,7 @@ func TestUserCanvasDAOUpdateDSL(t *testing.T) {
|
||||
|
||||
canvas, err := dao.GetByID(ctx, db, "canvas-1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get canvas: %v", err)
|
||||
t.Fatalf("failed to get agent: %v", err)
|
||||
}
|
||||
graph, ok := canvas.DSL["graph"].(map[string]interface{})
|
||||
if !ok {
|
||||
@@ -111,7 +111,7 @@ func TestUserCanvasDAOUpdateDSLNoMatch(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: originalDSL,
|
||||
}); err != nil {
|
||||
t.Fatalf("failed to create canvas: %v", err)
|
||||
t.Fatalf("failed to create agent: %v", err)
|
||||
}
|
||||
|
||||
rows, err := dao.UpdateDSL(ctx, db, "missing-canvas", entity.JSONMap{"path": []interface{}{"new"}})
|
||||
@@ -124,7 +124,7 @@ func TestUserCanvasDAOUpdateDSLNoMatch(t *testing.T) {
|
||||
|
||||
canvas, err := dao.GetByID(ctx, db, "canvas-1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get canvas: %v", err)
|
||||
t.Fatalf("failed to get agent: %v", err)
|
||||
}
|
||||
path, ok := canvas.DSL["path"].([]interface{})
|
||||
if !ok {
|
||||
@@ -220,7 +220,7 @@ func TestUserCanvasDAOOwnerAndCategoryFilters(t *testing.T) {
|
||||
}
|
||||
for i := range canvases {
|
||||
if err := db.WithContext(ctx).Create(&canvases[i]).Error; err != nil {
|
||||
t.Fatalf("failed to create canvas: %v", err)
|
||||
t.Fatalf("failed to create agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -256,7 +256,7 @@ func TestUserCanvasDAOOwnerAndCategoryFilters(t *testing.T) {
|
||||
for _, c := range categories {
|
||||
catByID[c.ID] = c.Count
|
||||
}
|
||||
// agent_canvas: c1 (own) + c3 (team); c4 hidden. dataflow_canvas: c2.
|
||||
// agent_agent: c1 (own) + c3 (team); c4 hidden. dataflow_agent: c2.
|
||||
if catByID["agent_canvas"] != 2 {
|
||||
t.Fatalf("agent_canvas count = %d, want 2", catByID["agent_canvas"])
|
||||
}
|
||||
@@ -286,7 +286,7 @@ func TestUserCanvasDAOKeywordSearchIncludesTags(t *testing.T) {
|
||||
}
|
||||
for i := range canvases {
|
||||
if err := db.Create(&canvases[i]).Error; err != nil {
|
||||
t.Fatalf("create canvas: %v", err)
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -323,7 +323,7 @@ func TestUserCanvasDAOListByTenantIDsCategoryUnion(t *testing.T) {
|
||||
}
|
||||
for i := range canvases {
|
||||
if err := db.Create(&canvases[i]).Error; err != nil {
|
||||
t.Fatalf("create canvas: %v", err)
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -365,7 +365,7 @@ func TestUserCanvasDAOOrderByTags(t *testing.T) {
|
||||
}
|
||||
for i := range canvases {
|
||||
if err := db.Create(&canvases[i]).Error; err != nil {
|
||||
t.Fatalf("create canvas: %v", err)
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1223,7 +1223,7 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
// DataFlow canvas: run a synchronous, side-effect-free debug (dry-run)
|
||||
// DataFlow agent: run a synchronous, side-effect-free debug (dry-run)
|
||||
// and return the parsed chunks inline — reusing this existing
|
||||
// chat/completions endpoint instead of a dedicated dataflow/debug route.
|
||||
// This mirrors the Python agent_api.py:1569 DataFlow branch, which is
|
||||
@@ -1305,6 +1305,9 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) {
|
||||
emitted := false
|
||||
doneSent := false
|
||||
for ev := range events {
|
||||
if ev.Type == "error" {
|
||||
ev = convertAgentRunErrorToMessage(ev)
|
||||
}
|
||||
emitted = true
|
||||
if ev.Type == "done" {
|
||||
doneSent = true
|
||||
@@ -1367,8 +1370,9 @@ func (h *AgentHandler) AgentChatCompletions(c *gin.Context) {
|
||||
for ev := range events {
|
||||
hasEvents = true
|
||||
if ev.Type == "error" {
|
||||
common.ResponseWithCodeData(c, common.CodeServerError, false, agentRunEventMessage(ev, "Agent run failed."))
|
||||
return
|
||||
ev = convertAgentRunErrorToMessage(ev)
|
||||
copy := ev
|
||||
finalAns = ©
|
||||
}
|
||||
var evData map[string]any
|
||||
if err := json.Unmarshal([]byte(ev.Data), &evData); err == nil {
|
||||
|
||||
@@ -505,7 +505,7 @@ func TestRunCanvasPipelineDebug_ErrorStillExposesMessageID(t *testing.T) {
|
||||
}
|
||||
|
||||
// TestRunCanvasPipelineDebug_WiresMessageIDAndLog asserts the full write-side wiring
|
||||
// of runCanvasPipelineDebug without a live Redis or real canvas: a fake executor
|
||||
// of runCanvasPipelineDebug without a live Redis or real agent: a fake executor
|
||||
// emits component progress, the injected DebugLogSink flushes the
|
||||
// [{component_id, trace}] array (with the END marker last) to the captured
|
||||
// store under the key "{canvasID}-{messageID}-logs", and the returned result
|
||||
|
||||
@@ -37,6 +37,15 @@ func agentRunEventMessage(ev canvas.RunEvent, fallback string) string {
|
||||
return fallback
|
||||
}
|
||||
|
||||
func convertAgentRunErrorToMessage(ev canvas.RunEvent) canvas.RunEvent {
|
||||
data, _ := json.Marshal(map[string]string{
|
||||
"content": agentRunEventMessage(ev, "Agent run failed."),
|
||||
})
|
||||
ev.Type = "message"
|
||||
ev.Data = string(data)
|
||||
return ev
|
||||
}
|
||||
|
||||
func agentWaitingForUser(ev canvas.RunEvent) canvas.WaitingForUserEvent {
|
||||
var waiting canvas.WaitingForUserEvent
|
||||
if err := json.Unmarshal([]byte(ev.Data), &waiting); err != nil {
|
||||
|
||||
@@ -762,6 +762,7 @@ func TestAgentChatCompletions_StreamSetsContentType(t *testing.T) {
|
||||
|
||||
runner := &stubChatRunner{events: []canvas.RunEvent{
|
||||
{Type: "message", MessageID: "msg-1", SessionID: "sess-1", Data: `{"content":"hi back","reference":[]}`},
|
||||
{Type: "error", MessageID: "msg-1", SessionID: "sess-1", Data: `{"message":"rerank model is unavailable","kind":"user"}`},
|
||||
{Type: "done", Data: ""},
|
||||
}}
|
||||
h := &AgentHandler{chatRunner: runner}
|
||||
@@ -775,9 +776,13 @@ func TestAgentChatCompletions_StreamSetsContentType(t *testing.T) {
|
||||
!strings.Contains(body, `"message_id":"msg-1"`) ||
|
||||
!strings.Contains(body, `"task_id":"sess-1"`) ||
|
||||
!strings.Contains(body, `"session_id":"sess-1"`) ||
|
||||
!strings.Contains(body, `"content":"hi back"`) {
|
||||
!strings.Contains(body, `"content":"hi back"`) ||
|
||||
!strings.Contains(body, `"content":"rerank model is unavailable"`) {
|
||||
t.Errorf("body should contain flat agent event with content, got %q", body)
|
||||
}
|
||||
if strings.Contains(body, `"code":500`) {
|
||||
t.Errorf("runner error should be rendered as conversation content, got %q", body)
|
||||
}
|
||||
if !strings.HasSuffix(body, "data:[DONE]\n\n") {
|
||||
t.Errorf("body should end with [DONE] terminator, got %q", body)
|
||||
}
|
||||
@@ -898,7 +903,7 @@ func TestAgentChatCompletions_DefaultBranchNonStreaming(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAgentChatCompletions_NonStreamingReturnsRunnerError(t *testing.T) {
|
||||
func TestAgentChatCompletions_NonStreamingReturnsRunnerErrorAsMessage(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
@@ -915,14 +920,19 @@ func TestAgentChatCompletions_NonStreamingReturnsRunnerError(t *testing.T) {
|
||||
(&AgentHandler{chatRunner: runner}).AgentChatCompletions(c)
|
||||
|
||||
var response struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
Code int `json:"code"`
|
||||
Data struct {
|
||||
Event string `json:"event"`
|
||||
Data struct {
|
||||
Content string `json:"content"`
|
||||
} `json:"data"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
|
||||
t.Fatalf("decode response: %v", err)
|
||||
}
|
||||
if response.Code != int(common.CodeServerError) || response.Message != "Can't find variable: 'Agent:Deleted@content'" {
|
||||
t.Fatalf("response = %+v, want runner error", response)
|
||||
if response.Code != int(common.CodeSuccess) || response.Data.Event != "message" || response.Data.Data.Content != "Can't find variable: 'Agent:Deleted@content'" {
|
||||
t.Fatalf("response = %+v, want runner error as conversation message", response)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -59,7 +59,7 @@ func newWebhookTraceTestHandler(t *testing.T) (*AgentHandler, *goredis.Client) {
|
||||
dao.DB = db
|
||||
t.Cleanup(func() { dao.DB = originalDB })
|
||||
if err := db.Create(&entity.UserCanvas{ID: "c1", UserID: "u1", Title: sptr("Test")}).Error; err != nil {
|
||||
t.Fatalf("create canvas: %v", err)
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
|
||||
mr, err := miniredis.Run()
|
||||
|
||||
@@ -218,7 +218,7 @@ func (p *Pipeline) Run(ctx context.Context, inputs map[string]any, overrideParam
|
||||
}
|
||||
compiled, err := canvas.Compile(compileCtx, p.canvas, compileOpts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("pipeline: Run: compile canvas: %w", err)
|
||||
return nil, fmt.Errorf("pipeline: Run: compile agent: %w", err)
|
||||
}
|
||||
|
||||
// Record the component count as the authoritative denominator for
|
||||
|
||||
@@ -27,8 +27,6 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
"ragflow/internal/common"
|
||||
"ragflow/internal/dao"
|
||||
"ragflow/internal/deepdoc/parser/pdf"
|
||||
@@ -37,6 +35,9 @@ import (
|
||||
"ragflow/internal/ingestion/component"
|
||||
"ragflow/internal/ingestion/pipeline"
|
||||
"ragflow/internal/storage"
|
||||
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
)
|
||||
|
||||
// TestExecute_DebugViaEntry_HonorsPagesCap_Integration is the end-to-end
|
||||
@@ -141,7 +142,7 @@ func TestExecute_DebugViaEntry_HonorsPagesCap_Integration(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: canvasDSL,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("create user canvas: %v", err)
|
||||
t.Fatalf("create user agent: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = realDB.Where("id = ?", canvasID).Delete(&entity.UserCanvas{}).Error })
|
||||
|
||||
|
||||
@@ -392,7 +392,7 @@ func TestRecordPipelineLog_TerminalWithoutDSLResolvesCanvasTitle(t *testing.T) {
|
||||
Title: strPtr("My Pipeline"),
|
||||
Avatar: strPtr("a.png"),
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed canvas: %v", err)
|
||||
t.Fatalf("seed agent: %v", err)
|
||||
}
|
||||
if err := dao.DB.AutoMigrate(&entity.Knowledgebase{}); err != nil {
|
||||
t.Fatalf("migrate knowledgebase: %v", err)
|
||||
|
||||
@@ -85,7 +85,7 @@ func TestPipelineExecutor_Run_RealCanvasDSL_UsesGeneralPipeline(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: templateDSL,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("create user canvas: %v", err)
|
||||
t.Fatalf("create user agent: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
cleanUpCtx := context.Background()
|
||||
@@ -210,7 +210,7 @@ func TestPipelineExecutor_Run_RealPDF_ProducesIndexedChunks(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: templateDSL,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("create user canvas: %v", err)
|
||||
t.Fatalf("create user agent: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_ = realDB.Where("id = ?", canvasID).Delete(&entity.UserCanvas{}).Error
|
||||
|
||||
@@ -34,7 +34,7 @@ func TestPipelineExecutor_DefaultLoadDSL_UsesUserCanvas(t *testing.T) {
|
||||
dslMap := entity.JSONMap{"dsl": map[string]any{"graph": map[string]any{"nodes": []any{}, "edges": []any{}}}}
|
||||
title := "title 1"
|
||||
if err := dao.NewUserCanvasDAO().Create(ctx, dao.DB, &entity.UserCanvas{Title: &title, ID: "canvas-1", UserID: "u1", Permission: "me", CanvasCategory: "agent_canvas", DSL: dslMap}); err != nil {
|
||||
t.Fatalf("create user canvas: %v", err)
|
||||
t.Fatalf("create user agent: %v", err)
|
||||
}
|
||||
|
||||
taskCtx := makeTaskCtx()
|
||||
|
||||
@@ -2932,9 +2932,9 @@ func canvasInvokeError(err error) error {
|
||||
msg := err.Error()
|
||||
if strings.Contains(msg, "[GraphRunError] no tasks to execute") &&
|
||||
strings.Contains(msg, "last completed nodes: [Switch:") {
|
||||
return errors.New("canvas invoke: Switch routing stopped because no connected branch matched the condition; check the Switch branches")
|
||||
return errors.New("agent invoke: Switch routing stopped because no connected branch matched the condition; check the Switch branches")
|
||||
}
|
||||
return fmt.Errorf("canvas invoke: %w", err)
|
||||
return fmt.Errorf("agent invoke: %w", err)
|
||||
}
|
||||
|
||||
// markRunSucceeded records the run as completed successfully via
|
||||
|
||||
@@ -147,7 +147,7 @@ func TestRunAgentInitializationFailureReleasesActiveSession(t *testing.T) {
|
||||
t.Fatalf("migrate agent tables: %v", err)
|
||||
}
|
||||
if err := testDB.Create(&entity.UserCanvas{ID: "agent-1", UserID: "user-a"}).Error; err != nil {
|
||||
t.Fatalf("create canvas: %v", err)
|
||||
t.Fatalf("create agent: %v", err)
|
||||
}
|
||||
tracker, mr := newAgentCancelTracker(t)
|
||||
svc := NewAgentServiceWithOptions(nil, nil, tracker)
|
||||
|
||||
@@ -17,14 +17,14 @@ import (
|
||||
|
||||
func TestDeferredAgentStreamFailureText(t *testing.T) {
|
||||
deferred := &runtime.DeferredStreamError{Text: "**ERROR**: [GraphRunError] model unavailable"}
|
||||
if got := deferredAgentStreamFailureText(fmt.Errorf("canvas invoke: component %q invoke: %w", "Message:x", deferred)); got != deferred.Text {
|
||||
if got := deferredAgentStreamFailureText(fmt.Errorf("agent invoke: component %q invoke: %w", "Message:x", deferred)); got != deferred.Text {
|
||||
t.Errorf("deferredAgentStreamFailureText = %q, want the Agent _ERROR text", got)
|
||||
}
|
||||
embedded := &runtime.DeferredStreamError{Text: "**ERROR**: [GraphRunError] consume deferred Agent stream: model unavailable"}
|
||||
if got := deferredAgentStreamFailureText(fmt.Errorf("canvas invoke: %w", embedded)); got != embedded.Text {
|
||||
if got := deferredAgentStreamFailureText(fmt.Errorf("agent invoke: %w", embedded)); got != embedded.Text {
|
||||
t.Errorf("marker text embedded in the _ERROR value truncated the failure text to %q", got)
|
||||
}
|
||||
if got := deferredAgentStreamFailureText(errors.New("canvas invoke: Message: consume deferred Agent stream: unrelated")); got != "" {
|
||||
if got := deferredAgentStreamFailureText(errors.New("agent invoke: Message: consume deferred Agent stream: unrelated")); got != "" {
|
||||
t.Errorf("plain error containing the marker text returned %q, want empty", got)
|
||||
}
|
||||
if got := deferredAgentStreamFailureText(fmt.Errorf("invoke: %w", &runtime.DeferredStreamError{Err: errors.New("stream dropped")})); got != "stream dropped" {
|
||||
@@ -33,7 +33,7 @@ func TestDeferredAgentStreamFailureText(t *testing.T) {
|
||||
if got := deferredAgentStreamFailureText(fmt.Errorf("invoke: %w", &runtime.DeferredStreamError{Err: context.Canceled})); got != "" {
|
||||
t.Errorf("cancellation inside the deferred stream returned %q, want empty", got)
|
||||
}
|
||||
if got := deferredAgentStreamFailureText(errors.New("canvas invoke: compile error")); got != "" {
|
||||
if got := deferredAgentStreamFailureText(errors.New("agent invoke: compile error")); got != "" {
|
||||
t.Errorf("non-deferred error returned %q, want empty", got)
|
||||
}
|
||||
if got := deferredAgentStreamFailureText(context.DeadlineExceeded); got != "" {
|
||||
@@ -43,7 +43,7 @@ func TestDeferredAgentStreamFailureText(t *testing.T) {
|
||||
|
||||
func TestCanvasInvokeErrorForUnmatchedSwitch(t *testing.T) {
|
||||
got := canvasInvokeError(errors.New("[GraphRunError] no tasks to execute, last completed nodes: [Switch:Route]"))
|
||||
if got.Error() != "canvas invoke: Switch routing stopped because no connected branch matched the condition; check the Switch branches" {
|
||||
if got.Error() != "agent invoke: Switch routing stopped because no connected branch matched the condition; check the Switch branches" {
|
||||
t.Fatalf("got %q", got)
|
||||
}
|
||||
|
||||
|
||||
@@ -1809,7 +1809,7 @@ func TestRunAgent_RealCanvas_InvokeFails(t *testing.T) {
|
||||
if len(errs) == 0 {
|
||||
t.Fatal("expected error event from Invoke of DSL with bad component query ref")
|
||||
}
|
||||
if !strings.Contains(errs[0].Message, "canvas invoke:") {
|
||||
if !strings.Contains(errs[0].Message, "agent invoke:") {
|
||||
t.Errorf("error message %q does not mention invoke context", errs[0].Message)
|
||||
}
|
||||
if strings.Contains(errs[0].Message, "agent storage error") {
|
||||
|
||||
@@ -1269,7 +1269,7 @@ func TestUpdateAgentTagsServiceSuccess(t *testing.T) {
|
||||
|
||||
canvasInstance, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get canvas: %v", err)
|
||||
t.Fatalf("failed to get agent: %v", err)
|
||||
}
|
||||
if canvasInstance.Tags != "alpha,beta,with comma" {
|
||||
t.Fatalf("expected normalized tags, got %q", canvasInstance.Tags)
|
||||
@@ -1295,7 +1295,7 @@ func TestUpdateAgentTagsServiceInvalidPayload(t *testing.T) {
|
||||
|
||||
canvasInstance, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get canvas: %v", err)
|
||||
t.Fatalf("failed to get agent: %v", err)
|
||||
}
|
||||
if canvasInstance.Tags != "" {
|
||||
t.Fatalf("expected tags to remain unchanged, got %q", canvasInstance.Tags)
|
||||
@@ -1321,7 +1321,7 @@ func TestUpdateAgentTagsServiceNoPermission(t *testing.T) {
|
||||
|
||||
canvasInstance, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get canvas: %v", err)
|
||||
t.Fatalf("failed to get agent: %v", err)
|
||||
}
|
||||
if canvasInstance.Tags != "" {
|
||||
t.Fatalf("expected tags to remain unchanged, got %q", canvasInstance.Tags)
|
||||
@@ -1457,7 +1457,7 @@ func TestResetAgentServiceClearsPerRunState(t *testing.T) {
|
||||
DSL: initialDSL,
|
||||
}
|
||||
if err := dao.DB.WithContext(ctx).Create(row).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
got, err := NewAgentService().ResetAgent(ctx, "user-1", "canvas-1")
|
||||
@@ -1506,7 +1506,7 @@ func TestResetAgentServiceClearsPerRunState(t *testing.T) {
|
||||
// DB row was updated in place; release flipped back to false.
|
||||
persisted, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to reload canvas: %v", err)
|
||||
t.Fatalf("failed to reload agent: %v", err)
|
||||
}
|
||||
if persisted.Release {
|
||||
t.Errorf("Release = true after reset, want false")
|
||||
@@ -1558,7 +1558,7 @@ func TestUpdateAgentSettingsPreservesDSL(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: originalDSL,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
err := NewAgentService().UpdateAgent(ctx, "user-1", "canvas-settings", map[string]interface{}{
|
||||
@@ -1570,7 +1570,7 @@ func TestUpdateAgentSettingsPreservesDSL(t *testing.T) {
|
||||
|
||||
persisted, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-settings")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to reload canvas: %v", err)
|
||||
t.Fatalf("failed to reload agent: %v", err)
|
||||
}
|
||||
if persisted.Description == nil || *persisted.Description != "new description" {
|
||||
t.Fatalf("Description = %v, want new description", persisted.Description)
|
||||
@@ -1595,7 +1595,7 @@ func TestUpdateAgentAllowsExistingTitleForSameCanvas(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
err := NewAgentService().UpdateAgent(ctx, "user-1", "canvas-same-title", map[string]interface{}{
|
||||
@@ -1678,7 +1678,7 @@ func TestUpdateAgentRejectsDuplicateTitleInDestinationCategory(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed source canvas: %v", err)
|
||||
t.Fatalf("failed to seed source agent: %v", err)
|
||||
}
|
||||
if err := dao.DB.WithContext(ctx).Create(&entity.UserCanvas{
|
||||
ID: "canvas-destination-duplicate",
|
||||
@@ -1687,7 +1687,7 @@ func TestUpdateAgentRejectsDuplicateTitleInDestinationCategory(t *testing.T) {
|
||||
CanvasCategory: "dataflow_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed duplicate canvas: %v", err)
|
||||
t.Fatalf("failed to seed duplicate agent: %v", err)
|
||||
}
|
||||
|
||||
err := NewAgentService().UpdateAgent(ctx, "user-1", "canvas-source-category", map[string]interface{}{
|
||||
@@ -1709,7 +1709,7 @@ func TestUpdateAgentRejectsCategoryOnlyDuplicateTitleInDestinationCategory(t *te
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed source canvas: %v", err)
|
||||
t.Fatalf("failed to seed source agent: %v", err)
|
||||
}
|
||||
if err := dao.DB.Create(&entity.UserCanvas{
|
||||
ID: "canvas-category-only-duplicate",
|
||||
@@ -1718,7 +1718,7 @@ func TestUpdateAgentRejectsCategoryOnlyDuplicateTitleInDestinationCategory(t *te
|
||||
CanvasCategory: "dataflow_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed duplicate canvas: %v", err)
|
||||
t.Fatalf("failed to seed duplicate agent: %v", err)
|
||||
}
|
||||
|
||||
ctx := t.Context()
|
||||
@@ -1741,7 +1741,7 @@ func TestUpdateAgentPersistsDSLAsJSONMap(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
err := NewAgentService().UpdateAgent(ctx, "user-1", "canvas-dsl-update", map[string]interface{}{
|
||||
@@ -1763,7 +1763,7 @@ func TestUpdateAgentPersistsDSLAsJSONMap(t *testing.T) {
|
||||
|
||||
persisted, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-dsl-update")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to reload canvas: %v", err)
|
||||
t.Fatalf("failed to reload agent: %v", err)
|
||||
}
|
||||
if _, ok := persisted.DSL["graph"]; !ok {
|
||||
t.Fatalf("DSL graph was not persisted: %#v", persisted.DSL)
|
||||
@@ -1784,7 +1784,7 @@ func TestUpdateAgentDSLCreatesAndReplacesDraftVersion(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
patch := map[string]interface{}{
|
||||
@@ -1844,7 +1844,7 @@ func TestUpdateAgentReleaseTrueMarksCanvasAndVersionReleased(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: entity.JSONMap{},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
ctx := t.Context()
|
||||
@@ -1868,7 +1868,7 @@ func TestUpdateAgentReleaseTrueMarksCanvasAndVersionReleased(t *testing.T) {
|
||||
}
|
||||
persisted, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-update-release")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to reload canvas: %v", err)
|
||||
t.Fatalf("failed to reload agent: %v", err)
|
||||
}
|
||||
if !persisted.Release {
|
||||
t.Fatal("canvas release flag was not set after publish update")
|
||||
@@ -1909,7 +1909,7 @@ func TestUpdateAgentWithoutReleaseKeepsCanvasUnreleased(t *testing.T) {
|
||||
DSL: entity.JSONMap{},
|
||||
Release: true,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
// Python parity: update_agent coerces a missing release to False and
|
||||
@@ -1933,7 +1933,7 @@ func TestUpdateAgentWithoutReleaseKeepsCanvasUnreleased(t *testing.T) {
|
||||
}
|
||||
persisted, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-update-draft")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to reload canvas: %v", err)
|
||||
t.Fatalf("failed to reload agent: %v", err)
|
||||
}
|
||||
if persisted.Release {
|
||||
t.Fatal("canvas release flag should be reset to false when release is absent")
|
||||
@@ -2053,7 +2053,7 @@ func TestListAgents_MergesCompilationTemplateGroups(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
BaseModel: entity.BaseModel{UpdateTime: &canvasUpdate},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
// The caller's own group (must appear), updated before the canvas.
|
||||
createAgentSessionTestCompilationGroup(t, "group-own", "user-1", groupUpdate)
|
||||
@@ -2195,7 +2195,7 @@ func TestPublishAgentUpdatesCanvasAndReleasedVersion(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: initialDSL,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
|
||||
description := "published description"
|
||||
@@ -2234,7 +2234,7 @@ func TestPublishAgentUpdatesCanvasAndReleasedVersion(t *testing.T) {
|
||||
|
||||
persisted, err := dao.NewUserCanvasDAO().GetByID(ctx, dao.DB, "canvas-publish")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to reload canvas: %v", err)
|
||||
t.Fatalf("failed to reload agent: %v", err)
|
||||
}
|
||||
if !persisted.Release {
|
||||
t.Fatal("publish did not mark canvas released")
|
||||
@@ -2289,7 +2289,7 @@ func TestUpdateAgentDSLDoesNotOverwriteLatestReleasedVersion(t *testing.T) {
|
||||
CanvasCategory: "agent_canvas",
|
||||
DSL: dsl,
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to seed canvas: %v", err)
|
||||
t.Fatalf("failed to seed agent: %v", err)
|
||||
}
|
||||
releasedAt := time.Now().Add(-time.Minute)
|
||||
if err := dao.DB.WithContext(ctx).Create(&entity.UserCanvasVersion{
|
||||
@@ -2502,7 +2502,7 @@ func createTeamSharedAgentTestFixtures(t *testing.T, canvasOwner, teammate, sess
|
||||
CanvasCategory: "agent_canvas",
|
||||
Permission: string(entity.TenantPermissionTeam),
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("failed to create canvas: %v", err)
|
||||
t.Fatalf("failed to create agent: %v", err)
|
||||
}
|
||||
if err := dao.DB.Create(&entity.UserTenant{
|
||||
ID: "ut-" + teammate + "-" + canvasOwner,
|
||||
|
||||
@@ -158,7 +158,7 @@ func TestBotService_AgentbotInputs_CrossTenantDenied(t *testing.T) {
|
||||
Title: sptr("Victim Agent"),
|
||||
CanvasCategory: "agent_canvas",
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed victim canvas: %v", err)
|
||||
t.Fatalf("seed victim agent: %v", err)
|
||||
}
|
||||
|
||||
// Seed tenant-B (the attacker's tenant).
|
||||
@@ -233,7 +233,7 @@ func TestBotService_AgentbotInputsReadsBeginParams(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed canvas: %v", err)
|
||||
t.Fatalf("seed agent: %v", err)
|
||||
}
|
||||
|
||||
svc := NewBotService(nil)
|
||||
|
||||
@@ -60,7 +60,7 @@ import (
|
||||
func decodeCanvasFromDSL(dsl map[string]any) (*canvas.Canvas, error) {
|
||||
c, err := canvas.DecodeFromDSL(dsl)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decode canvas: %w: %w", err, ErrAgentStorageError)
|
||||
return nil, fmt.Errorf("decode agent: %w: %w", err, ErrAgentStorageError)
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
@@ -973,7 +973,7 @@ func insertDatasetUpdateCanvas(t *testing.T, id, userID string) {
|
||||
DSL: entity.JSONMap{},
|
||||
}
|
||||
if err := dao.DB.Create(canvas).Error; err != nil {
|
||||
t.Fatalf("insert test canvas: %v", err)
|
||||
t.Fatalf("insert test agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1070,7 +1070,7 @@ func seedDatasetUpdateCanvas(t *testing.T, id, userID string, dslJSON []byte) {
|
||||
DSL: entity.JSONMap(dslMap),
|
||||
}
|
||||
if err := dao.DB.Create(canvas).Error; err != nil {
|
||||
t.Fatalf("seed canvas: %v", err)
|
||||
t.Fatalf("seed agent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3770,7 +3770,7 @@ func TestGetDocumentArtifact_AuthGate(t *testing.T) {
|
||||
Title: sptr("Agent"),
|
||||
CanvasCategory: "agent_canvas",
|
||||
}).Error; err != nil {
|
||||
t.Fatalf("seed canvas: %v", err)
|
||||
t.Fatalf("seed agent: %v", err)
|
||||
}
|
||||
// Seed an API4Conversation whose message references the filename.
|
||||
if err := dao.NewAPI4ConversationDAO().Create(t.Context(), db, &entity.API4Conversation{
|
||||
|
||||
@@ -65,7 +65,7 @@ func main() {
|
||||
flag.Parse()
|
||||
|
||||
if *goldenFlag != "" && *writeGolden {
|
||||
fmt.Fprintln(os.Stderr, "migrate-canvas: -golden and -write-golden are mutually exclusive")
|
||||
fmt.Fprintln(os.Stderr, "migrate-agent: -golden and -write-golden are mutually exclusive")
|
||||
os.Exit(2)
|
||||
}
|
||||
|
||||
@@ -77,7 +77,7 @@ func main() {
|
||||
|
||||
args := flag.Args()
|
||||
if len(args) == 0 {
|
||||
fmt.Fprintln(os.Stderr, "migrate-canvas: expected <file.json> (or -walk <dir>)")
|
||||
fmt.Fprintln(os.Stderr, "migrate-agent: expected <file.json> (or -walk <dir>)")
|
||||
flag.CommandLine.Usage()
|
||||
os.Exit(2)
|
||||
}
|
||||
@@ -144,7 +144,7 @@ func runOne(path, goldenPath string, writeGolden bool) error {
|
||||
func runWalk(dir, goldenDir string, writeGolden bool) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "migrate-canvas: walk: %v\n", err)
|
||||
fmt.Fprintf(os.Stderr, "migrate-agent: walk: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
var files []string
|
||||
|
||||
Reference in new issue
Block a user