ragflow/internal/harness/core/tool_invoke.go

package core

import (
	"context"
	"fmt"
	"sync"
	"time"

	"ragflow/internal/harness/core/schema"
)

// ToolInvocationContext captures the full context of a single tool invocation.
// It replaces the separate endpoint function signatures in middleware chains
// with a single unified object, making it easier to implement cross-cutting
// concerns like timeout, retry, and approval.
type ToolInvocationContext struct {
	// Name is the tool name being called (e.g., "get_weather").
	Name string
	// CallID is the unique identifier for this invocation from the LLM.
	CallID string
	// Arguments is the structured tool argument.
	Arguments *schema.ToolArgument
	// Result holds the tool result after successful execution (may be set by middleware).
	Result *schema.ToolResult
	// Timeout is the per-invocation timeout. Zero means no timeout.
	Timeout time.Duration
	// RetryConfig configures retry for this invocation. Nil means no retry.
	RetryConfig *ToolRetryConfig
	// Fallback is an optional fallback tool function to call if the primary fails.
	Fallback func(ctx context.Context, args *schema.ToolArgument) (*schema.ToolResult, error)

	// internal
	err     error
	skipped bool
	mu      sync.Mutex
}

// ToolRetryConfig configures retry behavior for a single tool invocation.
type ToolRetryConfig struct {
	MaxAttempts int
	Backoff     time.Duration
	IsRetryable func(err error) bool
}

// InvokeTool is the standard tool invocation function signature using the unified context.
type InvokeTool func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error)

// ToolInvokeMiddleware wraps a tool invocation with cross-cutting behavior.
// It receives the next handler in the chain and the invocation context.
type ToolInvokeMiddleware func(next InvokeTool) InvokeTool

// ---- ToolWrapper: timeout + retry + fallback ----

// NewTimeoutToolMiddleware creates a ToolInvokeMiddleware that enforces a timeout.
// If the tool invocation exceeds the duration, the context is cancelled.
func NewTimeoutToolMiddleware(timeout time.Duration) ToolInvokeMiddleware {
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			d := timeout
			if ictx.Timeout > 0 {
				d = ictx.Timeout
			}
			if d <= 0 {
				return next(ctx, ictx)
			}
			ctx, cancel := context.WithTimeout(ctx, d)
			defer cancel()
			return next(ctx, ictx)
		}
	}
}

// NewRetryToolMiddleware creates a ToolInvokeMiddleware that retries on failure.
func NewRetryToolMiddleware(cfg *ToolRetryConfig) ToolInvokeMiddleware {
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			rc := cfg
			if ictx.RetryConfig != nil {
				rc = ictx.RetryConfig
			}
			if rc == nil || rc.MaxAttempts <= 0 {
				return next(ctx, ictx)
			}
			backoff := rc.Backoff
			if backoff <= 0 {
				backoff = 100 * time.Millisecond
			}
			var lastErr error
			for attempt := 0; attempt <= rc.MaxAttempts; attempt++ {
				result, err := next(ctx, ictx)
				if err == nil {
					return result, nil
				}
				lastErr = err
				if rc.IsRetryable != nil && !rc.IsRetryable(err) {
					return nil, err
				}
				if attempt < rc.MaxAttempts {
					select {
					case <-ctx.Done():
						return nil, ctx.Err()
					case <-time.After(backoff):
					}
					backoff *= 2
				}
			}
			return nil, fmt.Errorf("tool retry exhausted after %d attempts: %w", rc.MaxAttempts, lastErr)
		}
	}
}

// NewFallbackToolMiddleware creates a ToolInvokeMiddleware that falls back to a secondary function.
func NewFallbackToolMiddleware(fallback func(ctx context.Context, args *schema.ToolArgument) (*schema.ToolResult, error)) ToolInvokeMiddleware {
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			result, err := next(ctx, ictx)
			if err == nil {
				return result, nil
			}
			fb := fallback
			if ictx.Fallback != nil {
				fb = ictx.Fallback
			}
			if fb == nil {
				return nil, err
			}
			return fb(ctx, ictx.Arguments)
		}
	}
}

// ---- Tool wrapper chain builder ----

// ToolWrapperChain builds a composed tool invocation handler from middleware and a final tool function.
func ToolWrapperChain(toolFn InvokeTool, middlewares ...ToolInvokeMiddleware) InvokeTool {
	chained := toolFn
	for i := len(middlewares) - 1; i >= 0; i-- {
		chained = middlewares[i](chained)
	}
	return chained
}

// ---- Approval mechanism ----

// ApprovalRequest is returned when a tool requires human approval before execution.
type ApprovalRequest struct {
	ToolName    string
	CallID      string
	Arguments   *schema.ToolArgument
	Description string
	// ApproveChan receives the approval decision. Send true to approve, false to reject.
	ApproveChan chan bool
}

// ApprovalMiddleware creates a ToolInvokeMiddleware that requests human approval before
// tool invocation. If approval is denied or times out, the tool is skipped.
// The getApproval callback is called for every tool invocation to produce an approval request.
func ApprovalMiddleware(getApproval func(ctx context.Context, ictx *ToolInvocationContext) (*ApprovalRequest, error)) ToolInvokeMiddleware {
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			req, err := getApproval(ctx, ictx)
			if err != nil {
				return nil, fmt.Errorf("approval setup error: %w", err)
			}
			if req == nil {
				return next(ctx, ictx)
			}

			select {
			case approved := <-req.ApproveChan:
				if !approved {
					return &schema.ToolResult{
						Name:    ictx.Name,
						Content: fmt.Sprintf("Tool '%s' execution rejected by user", ictx.Name),
						Error:   "rejected",
					}, nil
				}
				return next(ctx, ictx)
			case <-ctx.Done():
				return nil, ctx.Err()
			}
		}
	}
}

// AutoApprovalMiddleware creates an approval middleware that auto-approves all tools.
// Useful for testing or when no human-in-the-loop is needed.
func AutoApprovalMiddleware() ToolInvokeMiddleware {
	return ApprovalMiddleware(func(ctx context.Context, ictx *ToolInvocationContext) (*ApprovalRequest, error) {
		return nil, nil // nil = auto-approve
	})
}

// ---- Wrapping existing Tool into ToolInvokeMiddleware chain ----

// ToolToInvokeFn converts a standard Tool into an InvokeTool function.
func ToolToInvokeFn(tool Tool) InvokeTool {
	return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
		result, err := tool.Invoke(ctx, ictx.Arguments.Arguments)
		if err != nil {
			// Preserve tool interrupts so ToolsNode can handle them.
			if _, ok := IsToolInterrupt(err); ok {
				return nil, err
			}
			return &schema.ToolResult{Name: ictx.Name, Error: err.Error(), ToolCallID: ictx.CallID}, nil
		}
		return &schema.ToolResult{Name: ictx.Name, Content: result, ToolCallID: ictx.CallID}, nil
	}
}

// EnhancedToolToInvokeFn converts an EnhancedTool into an InvokeTool function.
func EnhancedToolToInvokeFn(tool EnhancedTool) InvokeTool {
	return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
		return tool.EnhancedInvoke(ctx, ictx.Arguments)
	}
}

// ---- Built-in middlewares: event sending and cancel monitoring ----

// NewEventSenderToolMiddleware creates a ToolInvokeMiddleware that emits tool
// result events to the agent's event stream after tool execution.
func NewEventSenderToolMiddleware[M MessageType]() ToolInvokeMiddleware {
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			result, err := next(ctx, ictx)
			if err != nil {
				return nil, err
			}
			ec := getReActExecCtx[M](ctx)
			if ec != nil && ec.generator != nil && result != nil {
				content := result.Content
				if content == "" {
					content = result.Error
				}
				var msg M
				var zero M
				switch any(zero).(type) {
				case *schema.AgenticMessage:
					msg = any(&schema.AgenticMessage{
						Role:    schema.AgenticRoleUser,
						Content: content,
						ContentBlocks: []schema.ContentBlock{
							{Type: "tool_result", ToolResult: &schema.ToolResult{
								ToolCallID: ictx.CallID, Content: content,
							}},
						},
					}).(M)
				default:
					msg = any(schema.ToolMessage(content, ictx.CallID)).(M)
				}
				ev := typedEventFromMessage(msg, nil, schema.RoleTool, ictx.Name)
				ec.send(ev)
			}
			return result, nil
		}
	}
}

// NewCancelToolMiddleware creates a ToolInvokeMiddleware that checks the cancel
// context before tool execution. If immediate cancel is requested, it returns
// ErrStreamCanceled immediately.
func NewCancelToolMiddleware() ToolInvokeMiddleware {
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			cc := getCancelContext(ctx)
			if cc != nil && cc.isImmediate() {
				return nil, ErrStreamCanceled
			}
			return next(ctx, ictx)
		}
	}
}

// ---- Rate limiting ----

// rateLimiter implements a simple per-tool token bucket.
type rateLimiter struct {
	mu     sync.Mutex
	tokens map[string]*tokenBucket
}

type tokenBucket struct {
	capacity int
	tokens   float64
	rate     float64 // tokens per nanosecond
	last     time.Time
}

func (rl *rateLimiter) allow(name string) bool {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	b, ok := rl.tokens[name]
	if !ok {
		return true // first use, always allow
	}
	now := time.Now()
	elapsed := now.Sub(b.last)
	b.tokens += elapsed.Seconds() * b.rate
	if b.tokens > float64(b.capacity) {
		b.tokens = float64(b.capacity)
	}
	b.last = now
	if b.tokens >= 1 {
		b.tokens--
		return true
	}
	return false
}

func (rl *rateLimiter) init(name string, rate_ float64, burst int) {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	rl.tokens[name] = &tokenBucket{
		capacity: burst,
		tokens:   float64(burst),
		rate:     rate_,
		last:     time.Now(),
	}
}

// NewRateLimitToolMiddleware creates a ToolInvokeMiddleware that limits the
// invocation rate per tool name using a per-token token bucket.
// rate is the number of invocations per second, burst is the maximum burst size.
//
// Example: NewRateLimitToolMiddleware(10, 5) allows up to 10 req/s with burst of 5.
func NewRateLimitToolMiddleware(rate float64, burst int) ToolInvokeMiddleware {
	rl := &rateLimiter{tokens: make(map[string]*tokenBucket)}
	return func(next InvokeTool) InvokeTool {
		return func(ctx context.Context, ictx *ToolInvocationContext) (*schema.ToolResult, error) {
			rl.initOnce(ictx.Name, rate, burst)
			if !rl.allow(ictx.Name) {
				return nil, fmt.Errorf("rate limit exceeded for tool '%s'", ictx.Name)
			}
			return next(ctx, ictx)
		}
	}
}

func (rl *rateLimiter) initOnce(name string, rate float64, burst int) {
	rl.mu.Lock()
	defer rl.mu.Unlock()
	if _, ok := rl.tokens[name]; ok {
		return
	}
	rl.tokens[name] = &tokenBucket{
		capacity: burst,
		tokens:   float64(burst),
		rate:     rate,
		last:     time.Now(),
	}
}