yao/agent/robot/api/execution.go
Max 3fee0e3fc3 Enhance execution listing and interaction with pagination and streaming support
- Refactor execution listing to support pagination with `Page` and `PageSize` options, replacing previous `Limit` and `Offset` parameters.
- Introduce `ExcludeStatuses` in execution queries to filter out specific execution statuses.
- Implement streaming interaction methods in the manager, allowing real-time responses from the host agent during interactions.
- Update API endpoints to accommodate new query parameters and enhance interaction capabilities with streaming support.
- Modify tests to ensure coverage for new pagination and streaming functionalities.
2026-02-27 16:38:30 +08:00

188 lines
4.6 KiB
Go

package api
import (
"context"
"fmt"
"sync"
"github.com/yaoapp/yao/agent/robot/store"
"github.com/yaoapp/yao/agent/robot/types"
)
// executionStore singleton
var (
execStore *store.ExecutionStore
execStoreOnce sync.Once
)
// getExecutionStore returns the singleton execution store
func getExecutionStore() *store.ExecutionStore {
execStoreOnce.Do(func() {
execStore = store.NewExecutionStore()
})
return execStore
}
// ResetExecutionStore resets the singleton for testing purposes
// This should only be called in tests
func ResetExecutionStore() {
execStoreOnce = sync.Once{}
execStore = nil
}
// ==================== Execution Query API ====================
// These functions query and manage execution history
// GetExecution returns a specific execution by ID
func GetExecution(ctx *types.Context, execID string) (*types.Execution, error) {
if execID == "" {
return nil, fmt.Errorf("execution_id is required")
}
// Try to get from execution store
record, err := getExecutionStore().Get(context.Background(), execID)
if err != nil {
return nil, fmt.Errorf("failed to get execution: %w", err)
}
if record == nil {
return nil, fmt.Errorf("execution not found: %s", execID)
}
return record.ToExecution(), nil
}
// ListExecutions returns execution history for a robot
func ListExecutions(ctx *types.Context, memberID string, query *ExecutionQuery) (*ExecutionResult, error) {
if memberID == "" {
return nil, fmt.Errorf("member_id is required")
}
if query == nil {
query = &ExecutionQuery{}
}
query.applyDefaults()
opts := &store.ListOptions{
MemberID: memberID,
Page: query.Page,
PageSize: query.PageSize,
OrderBy: "start_time desc",
}
if query.Status != "" {
opts.Status = query.Status
}
if len(query.ExcludeStatuses) > 0 {
opts.ExcludeStatuses = query.ExcludeStatuses
}
if query.Trigger != "" {
opts.TriggerType = query.Trigger
}
result, err := getExecutionStore().List(context.Background(), opts)
if err != nil {
return nil, fmt.Errorf("failed to list executions: %w", err)
}
executions := make([]*types.Execution, 0, len(result.Data))
for _, record := range result.Data {
executions = append(executions, record.ToExecution())
}
return &ExecutionResult{
Data: executions,
Total: result.Total,
Page: result.Page,
PageSize: result.PageSize,
}, nil
}
// ==================== Execution Control API ====================
// These functions control running executions
// PauseExecution pauses a running execution
func PauseExecution(ctx *types.Context, execID string) error {
if execID == "" {
return fmt.Errorf("execution_id is required")
}
mgr, err := getManager()
if err != nil {
return err
}
if err := mgr.PauseExecution(ctx, execID); err != nil {
return err
}
// Update database status to paused
return getExecutionStore().UpdateStatus(context.Background(), execID, types.ExecPaused, "")
}
// ResumeExecution resumes a paused execution
func ResumeExecution(ctx *types.Context, execID string) error {
if execID == "" {
return fmt.Errorf("execution_id is required")
}
mgr, err := getManager()
if err != nil {
return err
}
if err := mgr.ResumeExecution(ctx, execID); err != nil {
return err
}
// Update database status back to running
return getExecutionStore().UpdateStatus(context.Background(), execID, types.ExecRunning, "")
}
// StopExecution stops a running execution
func StopExecution(ctx *types.Context, execID string) error {
if execID == "" {
return fmt.Errorf("execution_id is required")
}
mgr, err := getManager()
if err != nil {
return err
}
if err := mgr.StopExecution(ctx, execID); err != nil {
return err
}
// Update database status to cancelled
return getExecutionStore().UpdateStatus(context.Background(), execID, types.ExecCancelled, "User cancelled")
}
// ==================== Execution Status API ====================
// GetExecutionStatus returns the current status of an execution
// This combines stored data with runtime state
func GetExecutionStatus(ctx *types.Context, execID string) (*types.Execution, error) {
if execID == "" {
return nil, fmt.Errorf("execution_id is required")
}
// Get from store first
exec, err := GetExecution(ctx, execID)
if err != nil {
return nil, err
}
// If manager is running, check for runtime state
mgr, mgrErr := getManager()
if mgrErr == nil {
// Check if execution is being tracked (running)
ctrlExec, ctrlErr := mgr.GetExecutionStatus(execID)
if ctrlErr == nil && ctrlExec != nil {
// Update with runtime state
exec.Status = ctrlExec.Status
exec.Phase = ctrlExec.Phase
}
}
return exec, nil
}