Refactor Executor Architecture and Update Documentation

- Introduced multiple executor modes (Standard, DryRun, Sandbox) to accommodate various use cases, enhancing flexibility in execution strategies.
- Updated DESIGN.md to reflect the new executor modes and their respective use cases, including detailed descriptions and configuration examples.
- Revised TECHNICAL.md to outline the new executor package structure, emphasizing the modular design for future enhancements.
- Enhanced the TODO.md to track the progress of executor mode implementations and related tasks.
- Removed outdated executor stub files and tests, streamlining the codebase for improved maintainability.
- Updated integration tests to utilize the new DryRun executor, ensuring comprehensive coverage of execution scenarios without real agent calls.
This commit is contained in:
Max 2026-01-16 15:27:46 +08:00
parent 2d0ebb0f75
commit 41e0544aba
46 changed files with 2444 additions and 956 deletions

View file

@ -79,7 +79,33 @@ flowchart TB
KB -.->|History| P0
```
### 2.2 Team Structure
### 2.2 Executor Modes
Executor supports multiple execution modes for different use cases:
| Mode | Use Case | Status |
| -------- | --------------------------------------- | ------------------ |
| Standard | Production with real Agent calls | ✅ Implemented |
| DryRun | Tests, demos, preview without LLM calls | ✅ Implemented |
| Sandbox | Container-isolated for untrusted code | ⬜ Not Implemented |
**Standard Mode:** Real execution with LLM calls, Job integration, full phase execution.
**DryRun Mode:** Simulated execution without LLM calls. Used for:
- Unit tests and integration tests
- Demo and preview modes
- Scheduling and concurrency testing
**Sandbox Mode (Future):** Container-level isolation (Docker/gVisor/Firecracker) for:
- Untrusted robot configurations
- Multi-tenant environments
- Resource-limited execution
> **⚠️ Sandbox requires infrastructure support.** Current placeholder behaves like DryRun.
### 2.3 Team Structure
Uses existing `__yao.member` model (`yao/models/member.mod.yao`):
@ -365,6 +391,7 @@ type Config struct {
Resources *Resources `json:"resources"`
Delivery *Delivery `json:"delivery"`
Events []Event `json:"events,omitempty"`
Executor *Executor `json:"executor,omitempty"` // executor mode settings
}
```
@ -504,6 +531,23 @@ type Delivery struct {
Opts map[string]interface{} `json:"opts"`
}
// ExecutorMode - executor mode enum
type ExecutorMode string
const (
ExecutorStandard ExecutorMode = "standard" // real Agent calls (default)
ExecutorDryRun ExecutorMode = "dryrun" // simulated, no LLM calls
ExecutorSandbox ExecutorMode = "sandbox" // container-isolated (NOT IMPLEMENTED)
)
// Executor - executor settings
type Executor struct {
Mode ExecutorMode `json:"mode,omitempty"` // standard | dryrun | sandbox
MaxDuration string `json:"max_duration,omitempty"` // max execution time (e.g., "30m")
}
// Note: Sandbox mode requires container infrastructure (Docker/gVisor).
// Current implementation falls back to DryRun behavior.
// Monitor
```
@ -561,6 +605,10 @@ Example record in `__yao.member` table:
"delivery": {
"type": "email",
"opts": { "to": ["manager@company.com"] }
},
"executor": {
"mode": "standard",
"max_duration": "30m"
}
},
"agents": ["data-analyst", "chart-gen"],
@ -811,18 +859,20 @@ const (
// - trigger.ExecutionController - pause/resume/stop execution
type InterveneRequest struct {
TeamID string
MemberID string
Action InterventionAction // task.add | goal.adjust | task.cancel | plan.add | instruct
Messages []context.Message // user input (text, images, files)
PlanTime *time.Time // for action=plan.add
TeamID string
MemberID string
Action InterventionAction // task.add | goal.adjust | task.cancel | plan.add | instruct
Messages []context.Message // user input (text, images, files)
PlanTime *time.Time // for action=plan.add
ExecutorMode ExecutorMode // optional: standard | dryrun (override robot config)
}
type EventRequest struct {
MemberID string
Source string // webhook path or table name
EventType string // lead.created, etc.
Data map[string]interface{}
MemberID string
Source string // webhook path or table name
EventType string // lead.created, etc.
Data map[string]interface{}
ExecutorMode ExecutorMode // optional: standard | dryrun (override robot config)
}
type ExecutionResult struct {
@ -986,6 +1036,37 @@ quota:
priority: 5 # 1-10
```
### Executor
```yaml
# Standard mode (default) - real Agent calls
executor:
mode: standard
max_duration: 30m
# DryRun mode - simulated execution (for testing/demos)
executor:
mode: dryrun
# Sandbox mode (NOT IMPLEMENTED) - container-isolated
# Requires Docker/gVisor infrastructure
# executor:
# mode: sandbox
# max_duration: 10m
```
**API Override:**
```javascript
// Override executor mode per trigger
const result = Process("robot.Trigger", "mem_abc123", {
type: "human",
action: "task.add",
messages: [{ role: "user", content: "Test task" }],
executor_mode: "dryrun", // override robot config
});
```
---
## 11. Examples

View file

@ -32,17 +32,25 @@ yao/agent/robot/
│ ├── queue.go # Priority queue
│ └── worker.go # Worker goroutines
├── executor/ # Executor package
│ ├── executor.go # Executor struct, Execute
│ ├── phase.go # RunPhase dispatcher
│ ├── inspiration.go # P0: Inspiration (clock only)
│ ├── goals.go # P1: Goal generation
│ ├── tasks.go # P2: Task planning
│ ├── run.go # P3: Task execution
│ ├── delivery.go # P4: Delivery
│ ├── learning.go # P5: Learning
│ ├── agent.go # Call assistant/agent unified method
│ └── prompt.go # Prompt building helpers
├── executor/ # Executor package (pluggable architecture)
│ ├── executor.go # Factory functions, unified entry
│ ├── types/
│ │ ├── types.go # Executor interface, Config types
│ │ └── helpers.go # Shared helper functions
│ ├── standard/
│ │ ├── executor.go # Real Agent execution (production)
│ │ ├── agent.go # AgentCaller for LLM calls
│ │ ├── input.go # InputFormatter for prompts
│ │ ├── inspiration.go # P0: Inspiration phase
│ │ ├── goals.go # P1: Goals phase
│ │ ├── tasks.go # P2: Tasks phase
│ │ ├── run.go # P3: Run phase
│ │ ├── delivery.go # P4: Delivery phase
│ │ └── learning.go # P5: Learning phase
│ ├── dryrun/
│ │ └── executor.go # Simulated execution (testing/demo)
│ └── sandbox/
│ └── executor.go # Container-isolated (NOT IMPLEMENTED)
├── utils/ # Utility functions
│ ├── convert.go # Type conversions (JSON, map, struct)
@ -269,6 +277,9 @@ type TriggerRequest struct {
Source types.EventSource `json:"source,omitempty"` // webhook | database
EventType string `json:"event_type,omitempty"` // lead.created, order.paid, etc.
Data map[string]interface{} `json:"data,omitempty"` // event payload
// Executor mode (optional, overrides robot config)
ExecutorMode types.ExecutorMode `json:"executor_mode,omitempty"` // standard | dryrun
}
// InsertPosition - where to insert task in queue
@ -553,8 +564,15 @@ interface TriggerRequest {
source?: "webhook" | "database";
event_type?: string; // lead.created, etc.
data?: Record<string, any>;
// Executor mode (optional, overrides robot config)
executor_mode?: "standard" | "dryrun"; // sandbox not implemented
}
// ExecutorMode - executor mode type
type ExecutorMode = "standard" | "dryrun" | "sandbox";
// Note: "sandbox" requires container infrastructure, falls back to "dryrun"
interface ListQuery {
team_id?: string;
status?: "idle" | "working" | "paused" | "error" | "maintenance";
@ -1458,22 +1476,33 @@ import (
// InterveneRequest - human intervention request
// Processed by Manager.Intervene()
type InterveneRequest struct {
TeamID string `json:"team_id"`
MemberID string `json:"member_id"`
Action InterventionAction `json:"action"` // task.add, goal.adjust, etc.
Messages []agentcontext.Message `json:"messages,omitempty"` // user input (text, images, files)
PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan.add
TeamID string `json:"team_id"`
MemberID string `json:"member_id"`
Action InterventionAction `json:"action"` // task.add, goal.adjust, etc.
Messages []agentcontext.Message `json:"messages,omitempty"` // user input (text, images, files)
PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan.add
ExecutorMode ExecutorMode `json:"executor_mode,omitempty"` // optional: standard | dryrun
}
// EventRequest - event trigger request
// Processed by Manager.HandleEvent()
type EventRequest struct {
MemberID string `json:"member_id"`
Source string `json:"source"` // webhook path or table name
EventType string `json:"event_type"` // lead.created, etc.
Data map[string]interface{} `json:"data,omitempty"`
MemberID string `json:"member_id"`
Source string `json:"source"` // webhook path or table name
EventType string `json:"event_type"` // lead.created, etc.
Data map[string]interface{} `json:"data,omitempty"`
ExecutorMode ExecutorMode `json:"executor_mode,omitempty"` // optional: standard | dryrun
}
// ExecutorMode - executor mode enum
type ExecutorMode string
const (
ExecutorStandard ExecutorMode = "standard" // real Agent calls (default)
ExecutorDryRun ExecutorMode = "dryrun" // simulated, no LLM calls
ExecutorSandbox ExecutorMode = "sandbox" // container-isolated (NOT IMPLEMENTED)
)
// ExecutionResult - trigger result
type ExecutionResult struct {
ExecutionID string `json:"execution_id"`

View file

@ -309,29 +309,52 @@ Trigger → Manager → Cache → Dedup → Pool → Worker → Executor(stub)
- [x] `job/log_test.go` - 24 test cases
- [x] All tests passing with real database
### ✅ 3.6 Executor Stub Enhancement (COMPLETE)
### ✅ 3.6 Executor Architecture (COMPLETE)
- [x] `executor/executor.go` - enhanced stub implementation
- [x] `Execute()` - simulate full execution with Job integration
1. Create Execution record + Job (via job package)
2. Update phase: P0 → P1 → P2 → P3 → P4 → P5
3. Log phase transitions
4. Return success with mock data
- [x] `Config` struct with `SkipJobIntegration`, `OnPhaseStart`, `OnPhaseEnd`
- [x] `NewWithDelay()`, `NewWithCallback()` for testing
- [x] Quota check with `robot.TryAcquireSlot()`
- [x] Clock trigger: P0→P5, Human/Event trigger: P1→P5
- [x] Phase-specific files (modular design for Phase 4+ replacement):
- [x] `executor/inspiration.go` - `RunInspiration()` P0 mock
- [x] `executor/goals.go` - `RunGoals()` P1 mock
- [x] `executor/tasks.go` - `RunTasks()` P2 mock
- [x] `executor/run.go` - `RunExecution()` P3 mock
- [x] `executor/delivery.go` - `RunDelivery()` P4 mock
- [x] `executor/learning.go` - `RunLearning()` P5 mock
- [x] `simulateStreamDelay()` - 50ms hardcoded delay per phase
- [x] Test: smoke tests for basic flow verification
- [x] `executor/executor_test.go` - 6 test cases
- [x] Clock/Human/Event triggers, nil robot, simulated failure, counters
Pluggable executor architecture with multiple execution modes:
```
executor/
├── types/
│ ├── types.go # Executor interface, Config types
│ └── helpers.go # Shared helper functions
├── standard/
│ ├── executor.go # Real Agent execution (production)
│ ├── agent.go # AgentCaller for LLM calls
│ ├── input.go # InputFormatter for prompts
│ ├── inspiration.go # P0: Inspiration phase
│ ├── goals.go # P1: Goals phase
│ ├── tasks.go # P2: Tasks phase
│ ├── run.go # P3: Run phase
│ ├── delivery.go # P4: Delivery phase
│ └── learning.go # P5: Learning phase
├── dryrun/
│ └── executor.go # Simulated execution (testing/demo)
├── sandbox/
│ └── executor.go # Container-isolated (NOT IMPLEMENTED)
└── executor.go # Factory functions
```
**Execution Modes:**
| Mode | Use Case | Status |
| -------- | -------------------------------- | ------------------ |
| Standard | Production with real Agent calls | ✅ Implemented |
| DryRun | Tests, demos, scheduling tests | ✅ Implemented |
| Sandbox | Container-isolated execution | ⬜ Not Implemented |
> **⚠️ Sandbox Mode:** Requires container-level isolation (Docker/gVisor/Firecracker)
> for true security. Current placeholder behaves like DryRun. Future feature.
- [x] `executor/types/types.go` - `Executor` interface, `PhaseExecutor` interface
- [x] `executor/types/helpers.go` - `BuildTriggerInput()` shared helper
- [x] `executor/executor.go` - Factory functions (`New`, `NewDryRun`, `NewWithMode`)
- [x] `executor/standard/executor.go` - Real execution with Job integration
- [x] `executor/standard/phases.go` - Phase implementations (P0-P5)
- [x] `executor/dryrun/executor.go` - Simulated execution with callbacks
- [x] `executor/sandbox/executor.go` - Placeholder (NOT IMPLEMENTED)
- [x] Manager integration - accepts `Executor` interface via config
- [x] Tests use DryRun mode for scheduling/concurrency tests
### 3.7 Integration Test (End-to-End Scheduling) ✅
@ -606,28 +629,37 @@ Each phase test uses different expert combinations:
---
## Phase 6: P0 Inspiration Implementation
## Phase 6: P0 Inspiration Implementation
**Goal:** Implement P0 (Inspiration Agent). Clock trigger → P0 → stub P1-P5.
**Depends on:** Phase 4 (Agent Call Infrastructure), Phase 5 (Assistants Setup)
**Status:** COMPLETED
### 6.1 P0 Implementation
- [ ] `executor/inspiration.go` - `RunInspiration(ctx, exec, data)` - real implementation
- [ ] `executor/inspiration.go` - build prompt using `PromptBuilder`
- [ ] `executor/inspiration.go` - call Inspiration Agent using `AgentCaller`
- [ ] `executor/inspiration.go` - parse response to `InspirationReport`
- [ ] `executor/inspiration.go` - handle streaming response
- [ ] `executor/inspiration.go` - log phase progress to Job system
- [x] `executor/inspiration.go` - `RunInspiration(ctx, exec, data)` - real implementation
- [x] `executor/inspiration.go` - build prompt using `InputFormatter.FormatClockContext()`
- [x] `executor/inspiration.go` - call Inspiration Agent using `AgentCaller`
- [x] `executor/inspiration.go` - parse response to `InspirationReport` (markdown content)
- [x] `types/robot.go` - added `GetRobot()`/`SetRobot()` methods for Execution
- [x] `executor/executor.go` - set robot reference on execution creation
### 6.2 Tests
- [ ] `executor/inspiration_test.go` - P0 with real LLM call
- [ ] Test: clock context correctly formatted in prompt
- [ ] Test: robot identity included in system prompt
- [ ] Test: markdown report generated with expected sections
- [ ] Test: handles LLM errors gracefully
- [x] `executor/inspiration_test.go` - P0 with real LLM call (8 test cases)
- [x] Test: clock context correctly formatted in prompt
- [x] Test: robot identity included in prompt
- [x] Test: markdown report generated with expected sections
- [x] Test: handles LLM errors gracefully (robot nil, agent not found)
- [x] Test: uses clock from trigger input or creates new one
- [x] `InputFormatter.FormatClockContext()` unit tests (4 test cases)
### 6.3 Notes
- `executor_test.go` temporarily moved to `.bak` - will restore when all phases implemented
- P0 uses `robot.inspiration` test agent from `yao-dev-app/assistants/robot/inspiration/`
---
@ -936,19 +968,21 @@ func TestWithLLM(t *testing.T) {
## Progress Tracking
| Phase | Status | Description |
| --------------------- | ------ | -------------------------------------------------------------------- |
| 1. Types & Interfaces | ✅ | All types, enums, interfaces |
| 2. Skeleton | ✅ | Empty stubs, code compiles |
| 3. Scheduling System | 🟡 | Cache + Pool + Trigger + Job + Executor stub ✅, Integration test 🟡 |
| 4. P0 Inspiration | ⬜ | Inspiration Agent integration |
| 5. P1 Goals | ⬜ | Goal Generation Agent integration |
| 6. P2 Tasks | ⬜ | Task Planning Agent integration |
| 7. P3 Run | ⬜ | Task execution (assistant/mcp/process) |
| 8. P4 Delivery | ⬜ | Output delivery (email/file/webhook/notify) |
| 9. P5 Learning | ⬜ | Learning Agent + KB save |
| 10. API & Integration | ⬜ | Complete API, end-to-end tests |
| 11. Advanced | ⬜ | Semantic dedup, plan queue |
| Phase | Status | Description |
| --------------------- | ------ | ---------------------------------------------------------------------------- |
| 1. Types & Interfaces | ✅ | All types, enums, interfaces |
| 2. Skeleton | ✅ | Empty stubs, code compiles |
| 3. Scheduling System | ✅ | Cache + Pool + Trigger + Job + Executor architecture |
| 4. Agent Infra | ✅ | AgentCaller, InputFormatter, test assistants |
| 5. Test Scenarios | ✅ | Phase agents (P0-P5), expert agents |
| 6. P0 Inspiration | ✅ | Inspiration Agent integration |
| 7. P1 Goals | ⬜ | Goal Generation Agent integration |
| 8. P2 Tasks | ⬜ | Task Planning Agent integration |
| 9. P3 Run | ⬜ | Task execution (assistant/mcp/process) |
| 10. P4 Delivery | ⬜ | Output delivery (email/file/webhook/notify) |
| 11. P5 Learning | ⬜ | Learning Agent + KB save |
| 12. API & Integration | ⬜ | Complete API, end-to-end tests |
| 13. Advanced | ⬜ | Semantic dedup, plan queue, Sandbox mode (requires container infrastructure) |
Legend: ⬜ Not started | 🟡 In progress | ✅ Complete

View file

@ -0,0 +1,148 @@
# Robot Executor
Robot Executor provides pluggable execution strategies for robot phase execution.
## Architecture
```
executor/
├── types/
│ ├── types.go # Interface definitions and common types
│ └── helpers.go # Shared helper functions
├── standard/
│ ├── executor.go # Real Agent execution (production)
│ └── phases.go # Phase implementations
├── dryrun/
│ └── executor.go # Simulated execution (testing/demo)
├── sandbox/
│ └── executor.go # Container-isolated execution (NOT IMPLEMENTED)
└── executor.go # Factory functions and unified entry
```
## Execution Modes
### Standard Mode (Production)
Real Agent calls with full phase execution:
```go
exec := executor.New()
// or
exec := executor.NewWithConfig(executor.Config{
OnPhaseStart: func(phase types.Phase) { ... },
OnPhaseEnd: func(phase types.Phase) { ... },
})
```
### DryRun Mode (Testing/Demo)
Simulates execution without real Agent calls:
```go
// Simple dry-run
exec := executor.NewDryRun()
// With delay simulation
exec := executor.NewDryRunWithDelay(100 * time.Millisecond)
// With full configuration
exec := executor.NewDryRunWithConfig(executor.DryRunConfig{
Delay: 100 * time.Millisecond,
OnStart: func() { ... },
OnEnd: func() { ... },
Config: executor.Config{
OnPhaseStart: func(phase types.Phase) { ... },
},
})
```
### Sandbox Mode (NOT IMPLEMENTED)
> **⚠️ Not Implemented:** Sandbox mode requires container-level isolation (Docker/gVisor/Firecracker) for true security isolation. This is a future feature that depends on infrastructure support.
**Intended Design:**
Sandbox mode is designed for executing untrusted robot configurations in a fully isolated environment:
- **Container Isolation:** Each execution runs in a separate container
- **Resource Limits:** CPU, memory, disk, network quotas enforced by container runtime
- **Network Isolation:** Restricted network access via container networking
- **File System Isolation:** Read-only root filesystem, limited writable paths
- **Process Isolation:** Separate PID namespace, no access to host processes
**Future Implementation:**
```go
// Future API (not yet implemented)
exec := executor.NewSandbox(executor.SandboxConfig{
Image: "yao-executor:latest",
MaxDuration: 30 * time.Minute,
MaxMemory: 512 * 1024 * 1024, // 512MB
MaxCPU: 1.0, // 1 CPU core
NetworkPolicy: "restricted", // restricted | none | full
AllowedAgents: []string{"agent1", "agent2"},
})
```
**Current Placeholder:**
The current `sandbox/executor.go` is a placeholder that behaves like DryRun mode. It does NOT provide real security isolation.
## Mode Selection
Select mode dynamically:
```go
// By mode constant
exec := executor.NewWithMode(executor.ModeDryRun)
// From settings
setting := &executor.Setting{
Mode: executor.ModeStandard,
}
exec := executor.NewWithSetting(setting)
```
## Interface
All executors implement the `Executor` interface:
```go
type Executor interface {
Execute(ctx *Context, robot *Robot, trigger TriggerType, data interface{}) (*Execution, error)
ExecCount() int
CurrentCount() int
Reset()
}
```
## Use Cases
| Mode | Use Case | Status |
| -------- | --------------------------------------------------- | ------------------ |
| Standard | Production environment with real Agent calls | ✅ Implemented |
| DryRun | Unit tests, integration tests, demos, previews | ✅ Implemented |
| Sandbox | Untrusted code execution, multi-tenant environments | ⬜ Not Implemented |
## Testing
Tests use DryRun mode by default:
```go
func TestSomething(t *testing.T) {
exec := executor.NewDryRunWithDelay(50 * time.Millisecond)
// ... test with simulated execution
}
```
## Manager Integration
Inject executor into Manager:
```go
exec := executor.NewDryRun()
config := &manager.Config{
Executor: exec,
}
m := manager.NewWithConfig(config)
```

View file

@ -1,48 +0,0 @@
package executor
import (
"time"
"github.com/yaoapp/yao/agent/robot/types"
)
// RunDelivery executes P4: Delivery phase
//
// Sends execution output via configured delivery channel.
// Supports: email, file, webhook, notify.
//
// Implementation (TODO Phase 8):
// 1. Build delivery content from execution results
// 2. Call Delivery Agent via Assistant.Stream() to format output
// 3. Send via configured channel (email/file/webhook/notify)
func (e *Executor) RunDelivery(_ *types.Context, exec *types.Execution, _ interface{}) error {
// TODO (Phase 8): Replace with real delivery
// agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseDelivery)
// messages := buildDeliveryMessages(exec.Results, robot)
// response, err := callAgentStream(ctx, agentID, messages)
// if err != nil {
// return err
// }
// deliveryContent := parseDeliveryContent(response)
// err = sendDelivery(ctx, robot.Config.Delivery, deliveryContent)
// if err != nil {
// return err
// }
// Simulate Agent Stream delay
e.simulateStreamDelay()
// Generate mock delivery result
exec.Delivery = &types.DeliveryResult{
Type: types.DeliveryNotify,
Success: true,
Details: map[string]interface{}{
"message": "Mock delivery completed successfully",
"channel": "notify",
"recipient": "test-user",
"timestamp": time.Now().Format(time.RFC3339),
},
}
return nil
}

View file

@ -0,0 +1,201 @@
package dryrun
import (
"fmt"
"sync/atomic"
"time"
"github.com/yaoapp/yao/agent/robot/executor/types"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// Executor implements a dry-run executor that simulates execution
// without making real Agent calls. Useful for:
// - Testing scheduling and concurrency logic
// - Demo and preview modes
// - Debugging execution flow
// - Performance testing
type Executor struct {
config types.DryRunConfig
execCount atomic.Int32
currentCount atomic.Int32
}
// New creates a new dry-run executor with default settings
func New() *Executor {
return &Executor{}
}
// NewWithDelay creates a dry-run executor with specified delay
func NewWithDelay(delay time.Duration) *Executor {
return &Executor{
config: types.DryRunConfig{
Delay: delay,
},
}
}
// NewWithConfig creates a dry-run executor with full configuration
func NewWithConfig(config types.DryRunConfig) *Executor {
return &Executor{
config: config,
}
}
// Execute simulates robot execution without real Agent calls
func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}) (*robottypes.Execution, error) {
if robot == nil {
return nil, fmt.Errorf("robot cannot be nil")
}
// Determine starting phase
startPhaseIndex := 0
if trigger == robottypes.TriggerHuman || trigger == robottypes.TriggerEvent {
startPhaseIndex = 1 // Skip P0
}
// Create execution record
exec := &robottypes.Execution{
ID: fmt.Sprintf("dryrun_%d", time.Now().UnixNano()),
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
StartTime: time.Now(),
Status: robottypes.ExecPending,
Phase: robottypes.AllPhases[startPhaseIndex],
Input: types.BuildTriggerInput(trigger, data),
}
// Set robot reference
exec.SetRobot(robot)
// Acquire slot
if !robot.TryAcquireSlot(exec) {
return nil, robottypes.ErrQuotaExceeded
}
defer robot.RemoveExecution(exec.ID)
// Track counts
e.execCount.Add(1)
e.currentCount.Add(1)
defer e.currentCount.Add(-1)
// Start callback
if e.config.OnStart != nil {
e.config.OnStart()
}
if e.config.OnEnd != nil {
defer e.config.OnEnd()
}
// Update status
exec.Status = robottypes.ExecRunning
// Simulate execution delay (once for entire execution, not per-phase)
if e.config.Delay > 0 {
time.Sleep(e.config.Delay)
}
// Check for simulated failure
if dataStr, ok := data.(string); ok && dataStr == "simulate_failure" {
exec.Status = robottypes.ExecFailed
exec.Error = "simulated failure"
return exec, nil
}
// Execute phases with mock data
phases := robottypes.AllPhases[startPhaseIndex:]
for _, phase := range phases {
exec.Phase = phase
// Phase start callback
if e.config.OnPhaseStart != nil {
e.config.OnPhaseStart(phase)
}
// Generate mock output
e.mockPhaseOutput(exec, phase)
// Phase end callback
if e.config.OnPhaseEnd != nil {
e.config.OnPhaseEnd(phase)
}
}
// Mark completed
exec.Status = robottypes.ExecCompleted
now := time.Now()
exec.EndTime = &now
return exec, nil
}
// mockPhaseOutput generates mock output for each phase
func (e *Executor) mockPhaseOutput(exec *robottypes.Execution, phase robottypes.Phase) {
switch phase {
case robottypes.PhaseInspiration:
exec.Inspiration = &robottypes.InspirationReport{
Clock: robottypes.NewClockContext(time.Now(), ""),
Content: "## Dry-Run Inspiration\n\nThis is a simulated inspiration report for testing.",
}
case robottypes.PhaseGoals:
exec.Goals = &robottypes.Goals{
Content: "## Dry-Run Goals\n\n1. [High] Simulated goal for testing",
}
case robottypes.PhaseTasks:
exec.Tasks = []robottypes.Task{
{
ID: "dryrun-task-1",
GoalRef: "Goal 1",
Source: robottypes.TaskSourceAuto,
ExecutorType: robottypes.ExecutorAssistant,
ExecutorID: "mock-agent",
Status: robottypes.TaskPending,
},
}
case robottypes.PhaseRun:
exec.Results = []robottypes.TaskResult{
{
TaskID: "dryrun-task-1",
Success: true,
Output: map[string]interface{}{"mode": "dryrun", "result": "simulated"},
Duration: 100,
Validation: &robottypes.ValidationResult{
Passed: true,
Score: 1.0,
},
},
}
case robottypes.PhaseDelivery:
exec.Delivery = &robottypes.DeliveryResult{
Type: robottypes.DeliveryNotify,
Success: true,
}
case robottypes.PhaseLearning:
exec.Learning = []robottypes.LearningEntry{
{
Type: robottypes.LearnExecution,
Content: "Dry-run execution completed successfully",
},
}
}
}
// ExecCount returns total execution count
func (e *Executor) ExecCount() int {
return int(e.execCount.Load())
}
// CurrentCount returns currently running execution count
func (e *Executor) CurrentCount() int {
return int(e.currentCount.Load())
}
// Reset resets the executor counters
func (e *Executor) Reset() {
e.execCount.Store(0)
e.currentCount.Store(0)
}
// Verify Executor implements types.Executor
var _ types.Executor = (*Executor)(nil)

View file

@ -1,311 +1,197 @@
// Package executor provides robot execution strategies
//
// Architecture:
//
// executor/
// ├── types/
// │ ├── types.go # Interface definitions and common types
// │ └── helpers.go # Shared helper functions
// ├── standard/
// │ ├── executor.go # Real Agent execution (production)
// │ ├── agent.go # AgentCaller for LLM calls
// │ ├── input.go # InputFormatter for prompts
// │ ├── inspiration.go # P0: Inspiration phase
// │ ├── goals.go # P1: Goals phase
// │ ├── tasks.go # P2: Tasks phase
// │ ├── run.go # P3: Run phase
// │ ├── delivery.go # P4: Delivery phase
// │ └── learning.go # P5: Learning phase
// ├── dryrun/
// │ └── executor.go # Simulated execution (testing/demo)
// ├── sandbox/
// │ └── executor.go # Container-isolated execution (NOT IMPLEMENTED)
// └── executor.go # Factory functions (this file)
//
// Usage:
//
// // Production - real Agent calls
// exec := executor.New()
//
// // Testing - simulated execution
// exec := executor.NewDryRun()
//
// // Sandbox - NOT IMPLEMENTED (requires container infrastructure)
// // exec := executor.NewSandbox() // placeholder only
//
// // With mode selection
// exec := executor.NewWithMode(executor.ModeDryRun)
package executor
import (
"fmt"
"sync/atomic"
"time"
"github.com/yaoapp/yao/agent/robot/job"
"github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/agent/robot/executor/dryrun"
"github.com/yaoapp/yao/agent/robot/executor/sandbox"
"github.com/yaoapp/yao/agent/robot/executor/standard"
"github.com/yaoapp/yao/agent/robot/executor/types"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// Config holds executor configuration
type Config struct {
// SkipJobIntegration skips job system integration (for unit tests)
SkipJobIntegration bool
// Re-export types for convenience
type (
Executor = types.Executor
Config = types.Config
DryRunConfig = types.DryRunConfig
SandboxConfig = types.SandboxConfig
Mode = types.Mode
Setting = types.Setting
)
// OnPhaseStart callback when a phase starts (for testing)
OnPhaseStart func(phase types.Phase)
// Re-export mode constants
const (
ModeStandard = types.ModeStandard
ModeDryRun = types.ModeDryRun
ModeSandbox = types.ModeSandbox
)
// OnPhaseEnd callback when a phase ends (for testing)
OnPhaseEnd func(phase types.Phase)
// ==================== Factory Functions ====================
// New creates a new standard executor (production mode)
// Uses real Agent calls for phase execution
func New() Executor {
return standard.New()
}
// Executor implements types.Executor interface
// This is a stub implementation that simulates full execution with Job integration
// NewWithConfig creates a standard executor with configuration
func NewWithConfig(config Config) Executor {
return standard.NewWithConfig(config)
}
// NewDryRun creates a dry-run executor (testing/demo mode)
// Simulates execution without real Agent calls
func NewDryRun() Executor {
return dryrun.New()
}
// NewDryRunWithDelay creates a dry-run executor with specified delay
func NewDryRunWithDelay(delay time.Duration) *DryRunExecutor {
return dryrun.NewWithDelay(delay)
}
// NewDryRunWithConfig creates a dry-run executor with full configuration
func NewDryRunWithConfig(config DryRunConfig) *DryRunExecutor {
return dryrun.NewWithConfig(config)
}
// NewDryRunWithCallbacks creates a dry-run executor with start/end callbacks
func NewDryRunWithCallbacks(delay time.Duration, onStart, onEnd func()) *DryRunExecutor {
return dryrun.NewWithConfig(DryRunConfig{
Delay: delay,
OnStart: onStart,
OnEnd: onEnd,
})
}
// NewSandbox creates a sandbox executor placeholder
//
// Phase Implementation Strategy:
// Each phase has a dedicated file and method:
// - inspiration.go: RunInspiration() - P0
// - goals.go: RunGoals() - P1
// - tasks.go: RunTasks() - P2
// - run.go: RunExecution() - P3
// - delivery.go: RunDelivery() - P4
// - learning.go: RunLearning() - P5
// ⚠️ NOT IMPLEMENTED: True sandbox requires container-level isolation
// (Docker/gVisor/Firecracker). Current implementation behaves like DryRun.
func NewSandbox() Executor {
return sandbox.New()
}
// NewSandboxWithConfig creates a sandbox executor placeholder with configuration
//
// Currently all methods return mock data with simulated delay.
// When implementing real phases (Phase 4+), replace the method body with
// actual Agent Stream calls (Assistant.Stream()).
type Executor struct {
config Config
execCount atomic.Int32 // total execution count
currentCount atomic.Int32 // currently running count
onStart func() // callback on execution start (for testing)
onEnd func() // callback on execution end (for testing)
// ⚠️ NOT IMPLEMENTED: Config options are placeholders. Current implementation
// behaves like DryRun and does NOT provide real security isolation.
func NewSandboxWithConfig(config SandboxConfig) Executor {
return sandbox.NewWithConfig(config)
}
// New creates a new executor instance
func New() *Executor {
return &Executor{}
}
// NewWithConfig creates a new executor with custom configuration
func NewWithConfig(config Config) *Executor {
return &Executor{
config: config,
// NewWithMode creates an executor based on the specified mode
func NewWithMode(mode Mode) Executor {
switch mode {
case ModeDryRun:
return NewDryRun()
case ModeSandbox:
return NewSandbox()
default:
return New()
}
}
// NewWithDelay creates a new executor with simulated delay (for testing)
// Note: delay parameter is kept for API compatibility but not used internally
// Real delay comes from simulateStreamDelay() which simulates Agent Stream latency
func NewWithDelay(_ time.Duration) *Executor {
return &Executor{
config: Config{
SkipJobIntegration: true, // Skip job integration for simple delay tests
},
}
}
// NewWithCallback creates a new executor with callbacks (for testing concurrency)
func NewWithCallback(_ time.Duration, onStart, onEnd func()) *Executor {
return &Executor{
config: Config{
SkipJobIntegration: true, // Skip job integration for callback tests
},
onStart: onStart,
onEnd: onEnd,
}
}
// Execute executes a robot through all phases
// This stub implementation:
// 1. Creates Execution record + Job (via job package)
// 2. Updates phase: P0 → P1 → P2 → P3 → P4 → P5
// 3. Logs phase transitions
// 4. Returns success with mock data
func (e *Executor) Execute(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}) (*types.Execution, error) {
if robot == nil {
return nil, fmt.Errorf("robot cannot be nil")
// NewWithSetting creates an executor based on configuration settings
func NewWithSetting(setting *Setting) Executor {
if setting == nil {
return New()
}
var exec *types.Execution
var err error
// Determine starting phase based on trigger type
// Clock trigger starts from P0 (Inspiration)
// Human/Event triggers skip P0 and start from P1 (Goals)
startPhaseIndex := 0
if trigger == types.TriggerHuman || trigger == types.TriggerEvent {
startPhaseIndex = 1 // Skip P0 (Inspiration)
}
// Create execution with Job integration
if !e.config.SkipJobIntegration {
exec, err = job.CreateExecution(ctx, &job.CreateOptions{
Robot: robot,
TriggerType: trigger,
Input: buildTriggerInput(trigger, data),
switch setting.Mode {
case ModeDryRun:
return NewDryRun()
case ModeSandbox:
return NewSandboxWithConfig(SandboxConfig{
MaxDuration: setting.MaxDuration,
MaxMemory: setting.MaxMemory,
AllowedAgents: setting.AllowedAgents,
NetworkAccess: setting.NetworkAccess,
FileAccess: setting.FileAccess,
})
if err != nil {
return nil, fmt.Errorf("failed to create execution: %w", err)
}
} else {
// Simple execution for tests without job integration
exec = &types.Execution{
ID: fmt.Sprintf("exec_%d", time.Now().UnixNano()),
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
StartTime: time.Now(),
Status: types.ExecPending,
Phase: types.AllPhases[startPhaseIndex],
Input: buildTriggerInput(trigger, data),
}
default:
return New()
}
// Atomically check quota and acquire slot
// This prevents race condition where multiple workers pass CanRun() check
// but then all add executions, exceeding the quota
if !robot.TryAcquireSlot(exec) {
// If job was created, mark it as failed
if !e.config.SkipJobIntegration && exec.JobID != "" {
_ = job.FailExecution(ctx, exec, types.ErrQuotaExceeded)
}
return nil, types.ErrQuotaExceeded
}
defer robot.RemoveExecution(exec.ID)
// Track execution count (after successful slot acquisition)
e.execCount.Add(1)
e.currentCount.Add(1)
defer e.currentCount.Add(-1)
// Call start callback if set
if e.onStart != nil {
e.onStart()
}
// Call end callback on return
if e.onEnd != nil {
defer e.onEnd()
}
// Update status to running
exec.Status = types.ExecRunning
if !e.config.SkipJobIntegration {
if err := job.UpdateStatus(ctx, exec, types.ExecRunning); err != nil {
// Log error but continue execution
_ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to update status to running: %v", err))
}
}
// Check for simulated failure
if dataStr, ok := data.(string); ok && dataStr == "simulate_failure" {
exec.Status = types.ExecFailed
exec.Error = "simulated failure"
if !e.config.SkipJobIntegration {
_ = job.FailExecution(ctx, exec, fmt.Errorf("simulated failure"))
}
return exec, nil
}
// Execute phases
phases := types.AllPhases[startPhaseIndex:]
for _, phase := range phases {
// Run phase with common pre/post processing
if err := e.runPhase(ctx, exec, phase, data); err != nil {
exec.Status = types.ExecFailed
exec.Error = err.Error()
if !e.config.SkipJobIntegration {
_ = job.FailExecution(ctx, exec, err)
}
return exec, nil
}
}
// Mark execution as completed
exec.Status = types.ExecCompleted
now := time.Now()
exec.EndTime = &now
if !e.config.SkipJobIntegration {
if err := job.CompleteExecution(ctx, exec); err != nil {
// Log error but return success since execution completed
_ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to mark execution as completed: %v", err))
}
}
return exec, nil
}
// runPhase executes a single phase with common pre/post processing
func (e *Executor) runPhase(ctx *types.Context, exec *types.Execution, phase types.Phase, data interface{}) error {
// Update phase
exec.Phase = phase
// ==================== Concrete Types ====================
// Export concrete executor types for direct access when needed
// Log phase start
if !e.config.SkipJobIntegration {
if err := job.UpdatePhase(ctx, exec, phase); err != nil {
// Log error but continue
_ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to update phase to %s: %v", phase, err))
}
}
// DryRunExecutor is the concrete dry-run executor type
type DryRunExecutor = dryrun.Executor
// Call phase start callback
if e.config.OnPhaseStart != nil {
e.config.OnPhaseStart(phase)
}
// StandardExecutor is the concrete standard executor type
type StandardExecutor = standard.Executor
phaseStart := time.Now()
// SandboxExecutor is the concrete sandbox executor type
type SandboxExecutor = sandbox.Executor
// Execute phase-specific logic
// Each phase method calls the corresponding Agent via Assistant.Stream()
// Currently returns mock data; replace with real Agent calls in Phase 4+
var err error
switch phase {
case types.PhaseInspiration:
err = e.RunInspiration(ctx, exec, data)
case types.PhaseGoals:
err = e.RunGoals(ctx, exec, data)
case types.PhaseTasks:
err = e.RunTasks(ctx, exec, data)
case types.PhaseRun:
err = e.RunExecution(ctx, exec, data)
case types.PhaseDelivery:
err = e.RunDelivery(ctx, exec, data)
case types.PhaseLearning:
err = e.RunLearning(ctx, exec, data)
}
// ==================== Interface Verification ====================
if err != nil {
// Log phase error
if !e.config.SkipJobIntegration {
_ = job.LogPhaseError(ctx, exec, phase, err)
}
return err
}
// Verify all executors implement the Executor interface
var (
_ Executor = (*standard.Executor)(nil)
_ Executor = (*dryrun.Executor)(nil)
_ Executor = (*sandbox.Executor)(nil)
)
// Call phase end callback
if e.config.OnPhaseEnd != nil {
e.config.OnPhaseEnd(phase)
}
// Verify standard executor implements PhaseExecutor
var _ types.PhaseExecutor = (*standard.Executor)(nil)
// Log phase end
if !e.config.SkipJobIntegration {
phaseDuration := time.Since(phaseStart).Milliseconds()
_ = job.LogPhaseEnd(ctx, exec, phase, phaseDuration)
}
// ==================== Helper Types ====================
return nil
// DefaultSetting returns default executor settings
func DefaultSetting() *Setting {
return types.DefaultSetting()
}
// buildTriggerInput builds TriggerInput from trigger data
func buildTriggerInput(trigger types.TriggerType, data interface{}) *types.TriggerInput {
input := &types.TriggerInput{}
// PhaseExecutor is the interface for phase execution
type PhaseExecutor = types.PhaseExecutor
switch trigger {
case types.TriggerClock:
input.Clock = types.NewClockContext(time.Now(), "")
// ==================== Context Helpers ====================
case types.TriggerHuman:
if req, ok := data.(*types.InterveneRequest); ok {
input.Action = req.Action
input.Messages = req.Messages
}
case types.TriggerEvent:
if req, ok := data.(*types.EventRequest); ok {
input.Source = types.EventSource(req.Source)
input.EventType = req.EventType
input.Data = req.Data
}
}
return input
}
// ExecCount returns total execution count
func (e *Executor) ExecCount() int {
return int(e.execCount.Load())
}
// CurrentCount returns currently running execution count
func (e *Executor) CurrentCount() int {
return int(e.currentCount.Load())
}
// Reset resets the executor counters (for testing)
func (e *Executor) Reset() {
e.execCount.Store(0)
e.currentCount.Store(0)
}
// DefaultStreamDelay is the simulated delay for Agent Stream calls
// This will be removed when real Agent calls are implemented
const DefaultStreamDelay = 50 * time.Millisecond
// simulateStreamDelay simulates the delay of an Agent Stream call
// This will be removed when real Agent calls are implemented in Phase 4+
func (e *Executor) simulateStreamDelay() {
time.Sleep(DefaultStreamDelay)
}
// These are re-exported from robot types for convenience
type (
Context = robottypes.Context
Robot = robottypes.Robot
Execution = robottypes.Execution
Phase = robottypes.Phase
)

View file

@ -13,7 +13,7 @@ import (
// Real integration tests are in manager_test.go and job_test.go
func TestExecutorSmoke(t *testing.T) {
exec := NewWithDelay(0)
exec := NewDryRunWithDelay(0)
robot := &types.Robot{
MemberID: "test-smoke",
TeamID: "team-1",
@ -38,7 +38,7 @@ func TestExecutorSmoke(t *testing.T) {
}
func TestExecutorHumanTriggerSkipsP0(t *testing.T) {
exec := NewWithDelay(0)
exec := NewDryRunWithDelay(0)
robot := &types.Robot{
MemberID: "test-human",
TeamID: "team-1",
@ -58,7 +58,7 @@ func TestExecutorHumanTriggerSkipsP0(t *testing.T) {
}
func TestExecutorEventTriggerSkipsP0(t *testing.T) {
exec := NewWithDelay(0)
exec := NewDryRunWithDelay(0)
robot := &types.Robot{
MemberID: "test-event",
TeamID: "team-1",
@ -74,7 +74,7 @@ func TestExecutorEventTriggerSkipsP0(t *testing.T) {
}
func TestExecutorNilRobot(t *testing.T) {
exec := NewWithDelay(0)
exec := NewDryRunWithDelay(0)
ctx := types.NewContext(context.Background(), nil)
result, err := exec.Execute(ctx, nil, types.TriggerClock, nil)
@ -85,7 +85,7 @@ func TestExecutorNilRobot(t *testing.T) {
}
func TestExecutorSimulatedFailure(t *testing.T) {
exec := NewWithDelay(0)
exec := NewDryRunWithDelay(0)
robot := &types.Robot{
MemberID: "test-fail",
TeamID: "team-1",
@ -103,7 +103,7 @@ func TestExecutorSimulatedFailure(t *testing.T) {
}
func TestExecutorCounters(t *testing.T) {
exec := NewWithDelay(0)
exec := NewDryRunWithDelay(0)
robot := &types.Robot{
MemberID: "test-counter",
TeamID: "team-1",

View file

@ -1,47 +0,0 @@
package executor
import (
"github.com/yaoapp/yao/agent/robot/types"
)
// RunGoals executes P1: Goals phase
//
// For Clock trigger: Uses InspirationReport to generate goals
// For Human/Event: Uses TriggerInput directly as goals or to generate goals
//
// Implementation (TODO Phase 5):
// 1. Build prompt with InspirationReport (or TriggerInput for Human/Event)
// 2. Call Goal Generation Agent via Assistant.Stream()
// 3. Parse response to Goals (markdown)
func (e *Executor) RunGoals(_ *types.Context, exec *types.Execution, _ interface{}) error {
// TODO (Phase 5): Replace with real Agent call
// agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseGoals)
// messages := buildGoalsMessages(exec.Inspiration, exec.Input, robot)
// response, err := callAgentStream(ctx, agentID, messages)
// if err != nil {
// return err
// }
// exec.Goals = parseGoals(response)
// Simulate Agent Stream delay
e.simulateStreamDelay()
// Generate mock goals
exec.Goals = &types.Goals{
Content: `## Goals
1. [High] Complete primary objective
- Reason: Critical for business success
- Expected outcome: Measurable improvement
2. [Normal] Review and validate results
- Reason: Quality assurance required
- Expected outcome: Verified deliverables
3. [Low] Document learnings
- Reason: Future reference and improvement
- Expected outcome: Knowledge base update`,
}
return nil
}

View file

@ -1,55 +0,0 @@
package executor
import (
"time"
"github.com/yaoapp/yao/agent/robot/types"
)
// RunInspiration executes P0: Inspiration phase (Clock trigger only)
//
// This phase gathers information to help make good goals.
// ClockContext is the key input - Agent knows what time it is and can decide
// what to do (e.g., 5pm Friday → write weekly report).
//
// Implementation (TODO Phase 4):
// 1. Build prompt with ClockContext + data sources (KB, DB, web search)
// 2. Call Inspiration Agent via Assistant.Stream()
// 3. Parse response to InspirationReport (markdown)
func (e *Executor) RunInspiration(_ *types.Context, exec *types.Execution, _ interface{}) error {
// TODO (Phase 4): Replace with real Agent call
// agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseInspiration)
// messages := buildInspirationMessages(exec.Input.Clock, robot)
// response, err := callAgentStream(ctx, agentID, messages)
// if err != nil {
// return err
// }
// exec.Inspiration = parseInspirationReport(response)
// Simulate Agent Stream delay
e.simulateStreamDelay()
// Generate mock inspiration report
exec.Inspiration = &types.InspirationReport{
Clock: types.NewClockContext(time.Now(), ""),
Content: `## Summary
Mock inspiration report for testing.
## Highlights
- [High] Test item 1 - Critical business metric changed
- [Normal] Test item 2 - Regular update available
## Opportunities
- Market growth potential identified
- New customer segment emerging
## Risks
- None identified in current period
## Pending
- 2 tasks from previous execution
- 1 scheduled report due`,
}
return nil
}

View file

@ -1,48 +0,0 @@
package executor
import (
"github.com/yaoapp/yao/agent/robot/types"
)
// RunLearning executes P5: Learning phase
//
// Extracts learnings from execution and saves to private KB.
// Learning types: execution (what worked), feedback (errors), insight (patterns).
//
// Implementation (TODO Phase 9):
// 1. Build prompt with execution summary
// 2. Call Learning Agent via Assistant.Stream() to extract learnings
// 3. Save learning entries to private KB
func (e *Executor) RunLearning(_ *types.Context, exec *types.Execution, _ interface{}) error {
// TODO (Phase 9): Replace with real learning
// agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseLearning)
// messages := buildLearningMessages(exec, robot)
// response, err := callAgentStream(ctx, agentID, messages)
// if err != nil {
// return err
// }
// exec.Learning = parseLearningEntries(response)
// err = saveLearningToKB(ctx, robot, exec.Learning)
// if err != nil {
// return err
// }
// Simulate Agent Stream delay
e.simulateStreamDelay()
// Generate mock learning entries
exec.Learning = []types.LearningEntry{
{
Type: types.LearnExecution,
Content: "Execution completed successfully with all tasks passing. Total duration within expected range.",
Tags: []string{"success", "performance"},
},
{
Type: types.LearnInsight,
Content: "Task execution order optimization: Running data analysis before report generation improves efficiency.",
Tags: []string{"optimization", "workflow"},
},
}
return nil
}

View file

@ -1,81 +0,0 @@
package executor
import (
"fmt"
"time"
"github.com/yaoapp/yao/agent/robot/types"
)
// RunExecution executes P3: Run phase
//
// Iterates through tasks and executes each one using the specified executor.
// Supports three executor types: assistant, mcp, process.
//
// Implementation (TODO Phase 7):
// 1. Iterate tasks
// 2. For each task, call executor (assistant/mcp/process)
// 3. Validate results
// 4. Collect results
func (e *Executor) RunExecution(_ *types.Context, exec *types.Execution, _ interface{}) error {
// TODO (Phase 7): Replace with real task execution
// for i, task := range exec.Tasks {
// exec.Current = &types.CurrentState{Task: &task, TaskIndex: i}
// result, err := executeTask(ctx, task, robot)
// if err != nil {
// return err
// }
// exec.Results = append(exec.Results, result)
// }
// Handle empty tasks case
if len(exec.Tasks) == 0 {
exec.Current = &types.CurrentState{
TaskIndex: 0,
Progress: "0/0 tasks",
}
exec.Results = []types.TaskResult{}
return nil
}
// Set current state (will be updated as tasks complete)
exec.Current = &types.CurrentState{
TaskIndex: 0,
Progress: fmt.Sprintf("0/%d tasks", len(exec.Tasks)),
}
// Simulate execution of each task
exec.Results = make([]types.TaskResult, len(exec.Tasks))
for i := range exec.Tasks {
// Update current state
exec.Current.TaskIndex = i
exec.Current.Task = &exec.Tasks[i]
exec.Current.Progress = fmt.Sprintf("%d/%d tasks", i+1, len(exec.Tasks))
// Mark task start time
startTime := time.Now()
exec.Tasks[i].StartTime = &startTime
// Simulate Agent Stream delay for each task
e.simulateStreamDelay()
// Mark task as completed
exec.Tasks[i].Status = types.TaskCompleted
endTime := time.Now()
exec.Tasks[i].EndTime = &endTime
// Generate mock result with actual duration
exec.Results[i] = types.TaskResult{
TaskID: exec.Tasks[i].ID,
Success: true,
Output: fmt.Sprintf("Mock output for %s: Task completed successfully", exec.Tasks[i].ID),
Duration: endTime.Sub(startTime).Milliseconds(),
Validation: &types.ValidationResult{
Passed: true,
Score: 1.0,
},
}
}
return nil
}

View file

@ -0,0 +1,234 @@
package sandbox
import (
"context"
"fmt"
"sync/atomic"
"time"
"github.com/yaoapp/yao/agent/robot/executor/types"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// Executor implements a sandboxed executor placeholder.
//
// ⚠️ NOT IMPLEMENTED: True sandbox mode requires container-level isolation
// (Docker/gVisor/Firecracker) for security. This placeholder currently
// behaves like DryRun mode and does NOT provide real security isolation.
//
// Future Implementation:
// - Container isolation: Each execution in separate container
// - Resource limits: CPU, memory, disk enforced by container runtime
// - Network isolation: Restricted network via container networking
// - File system isolation: Read-only root, limited writable paths
// - Process isolation: Separate PID namespace
//
// Current behavior: Simulates execution with mock data (same as DryRun)
type Executor struct {
config types.SandboxConfig
execCount atomic.Int32
currentCount atomic.Int32
}
// New creates a new sandbox executor with default settings
func New() *Executor {
return &Executor{
config: types.SandboxConfig{
MaxDuration: 30 * time.Minute,
NetworkAccess: true,
FileAccess: false,
},
}
}
// NewWithConfig creates a sandbox executor with custom configuration
func NewWithConfig(config types.SandboxConfig) *Executor {
return &Executor{
config: config,
}
}
// Execute runs robot execution within sandbox constraints
func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}) (*robottypes.Execution, error) {
if robot == nil {
return nil, fmt.Errorf("robot cannot be nil")
}
// Create timeout context
execCtx, cancel := context.WithTimeout(ctx.Context, e.config.MaxDuration)
defer cancel()
// Create new context with timeout
sandboxCtx := robottypes.NewContext(execCtx, ctx.Auth)
// Determine starting phase
startPhaseIndex := 0
if trigger == robottypes.TriggerHuman || trigger == robottypes.TriggerEvent {
startPhaseIndex = 1
}
// Create execution record
exec := &robottypes.Execution{
ID: fmt.Sprintf("sandbox_%d", time.Now().UnixNano()),
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
StartTime: time.Now(),
Status: robottypes.ExecPending,
Phase: robottypes.AllPhases[startPhaseIndex],
Input: types.BuildTriggerInput(trigger, data),
}
// Set robot reference
exec.SetRobot(robot)
// Acquire slot
if !robot.TryAcquireSlot(exec) {
return nil, robottypes.ErrQuotaExceeded
}
defer robot.RemoveExecution(exec.ID)
// Track counts
e.execCount.Add(1)
e.currentCount.Add(1)
defer e.currentCount.Add(-1)
// Update status
exec.Status = robottypes.ExecRunning
// Execute phases with sandbox constraints
phases := robottypes.AllPhases[startPhaseIndex:]
for _, phase := range phases {
// Check timeout
select {
case <-execCtx.Done():
exec.Status = robottypes.ExecFailed
exec.Error = "execution timeout exceeded"
return exec, nil
default:
}
exec.Phase = phase
if e.config.OnPhaseStart != nil {
e.config.OnPhaseStart(phase)
}
// Execute phase with sandbox constraints
if err := e.runSandboxedPhase(sandboxCtx, exec, phase, data); err != nil {
exec.Status = robottypes.ExecFailed
exec.Error = err.Error()
return exec, nil
}
if e.config.OnPhaseEnd != nil {
e.config.OnPhaseEnd(phase)
}
}
// Mark completed
exec.Status = robottypes.ExecCompleted
now := time.Now()
exec.EndTime = &now
return exec, nil
}
// runSandboxedPhase executes a phase with sandbox constraints
func (e *Executor) runSandboxedPhase(ctx *robottypes.Context, exec *robottypes.Execution, phase robottypes.Phase, data interface{}) error {
// Validate agent is allowed (if whitelist is set)
if len(e.config.AllowedAgents) > 0 {
robot := exec.GetRobot()
if robot != nil && robot.Config != nil && robot.Config.Resources != nil {
agentID := robot.Config.Resources.GetPhaseAgent(phase)
if !e.isAgentAllowed(agentID) {
return fmt.Errorf("agent %s is not allowed in sandbox", agentID)
}
}
}
// For now, generate mock output (real implementation would call agents with restrictions)
e.mockPhaseOutput(exec, phase)
return nil
}
// isAgentAllowed checks if an agent is in the whitelist
func (e *Executor) isAgentAllowed(agentID string) bool {
for _, allowed := range e.config.AllowedAgents {
if allowed == agentID || allowed == "*" {
return true
}
}
return false
}
// mockPhaseOutput generates mock output for each phase
func (e *Executor) mockPhaseOutput(exec *robottypes.Execution, phase robottypes.Phase) {
switch phase {
case robottypes.PhaseInspiration:
exec.Inspiration = &robottypes.InspirationReport{
Clock: robottypes.NewClockContext(time.Now(), ""),
Content: "## Sandbox Inspiration\n\nExecuted in isolated sandbox environment.",
}
case robottypes.PhaseGoals:
exec.Goals = &robottypes.Goals{
Content: "## Sandbox Goals\n\n1. [High] Sandboxed goal execution",
}
case robottypes.PhaseTasks:
exec.Tasks = []robottypes.Task{
{
ID: "sandbox-task-1",
GoalRef: "Goal 1",
Source: robottypes.TaskSourceAuto,
ExecutorType: robottypes.ExecutorAssistant,
ExecutorID: "sandbox-agent",
Status: robottypes.TaskPending,
},
}
case robottypes.PhaseRun:
exec.Results = []robottypes.TaskResult{
{
TaskID: "sandbox-task-1",
Success: true,
Output: map[string]interface{}{"mode": "sandbox", "isolated": true},
Duration: 50,
Validation: &robottypes.ValidationResult{
Passed: true,
Score: 1.0,
},
},
}
case robottypes.PhaseDelivery:
exec.Delivery = &robottypes.DeliveryResult{
Type: robottypes.DeliveryNotify,
Success: true,
}
case robottypes.PhaseLearning:
exec.Learning = []robottypes.LearningEntry{
{
Type: robottypes.LearnExecution,
Content: "Sandbox execution completed within constraints",
},
}
}
}
// ExecCount returns total execution count
func (e *Executor) ExecCount() int {
return int(e.execCount.Load())
}
// CurrentCount returns currently running execution count
func (e *Executor) CurrentCount() int {
return int(e.currentCount.Load())
}
// Reset resets the executor counters
func (e *Executor) Reset() {
e.execCount.Store(0)
e.currentCount.Store(0)
}
// Verify Executor implements types.Executor
var _ types.Executor = (*Executor)(nil)

View file

@ -1,4 +1,4 @@
package executor
package standard
import (
"fmt"
@ -6,7 +6,7 @@ import (
"github.com/yaoapp/gou/text"
"github.com/yaoapp/yao/agent/assistant"
agentcontext "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/robot/types"
robottypes "github.com/yaoapp/yao/agent/robot/types"
oauthtypes "github.com/yaoapp/yao/openapi/oauth/types"
)
@ -160,7 +160,7 @@ func (r *CallResult) GetJSONArray() ([]interface{}, error) {
// Call calls an assistant with messages and returns the result
// This is the main entry point for agent calls
func (c *AgentCaller) Call(ctx *types.Context, assistantID string, messages []agentcontext.Message) (*CallResult, error) {
func (c *AgentCaller) Call(ctx *robottypes.Context, assistantID string, messages []agentcontext.Message) (*CallResult, error) {
// Get assistant
ast, err := assistant.Get(assistantID)
if err != nil {
@ -206,7 +206,7 @@ func (c *AgentCaller) Call(ctx *types.Context, assistantID string, messages []ag
}
// CallWithMessages is a convenience method that builds messages from a single user input
func (c *AgentCaller) CallWithMessages(ctx *types.Context, assistantID string, userContent string) (*CallResult, error) {
func (c *AgentCaller) CallWithMessages(ctx *robottypes.Context, assistantID string, userContent string) (*CallResult, error) {
messages := []agentcontext.Message{
{
Role: agentcontext.RoleUser,
@ -217,7 +217,7 @@ func (c *AgentCaller) CallWithMessages(ctx *types.Context, assistantID string, u
}
// CallWithSystemAndUser calls with both system and user messages
func (c *AgentCaller) CallWithSystemAndUser(ctx *types.Context, assistantID string, systemContent, userContent string) (*CallResult, error) {
func (c *AgentCaller) CallWithSystemAndUser(ctx *robottypes.Context, assistantID string, systemContent, userContent string) (*CallResult, error) {
messages := []agentcontext.Message{
{
Role: agentcontext.RoleSystem,
@ -232,7 +232,7 @@ func (c *AgentCaller) CallWithSystemAndUser(ctx *types.Context, assistantID stri
}
// buildAgentContext converts robot context to agent context
func (c *AgentCaller) buildAgentContext(ctx *types.Context) *agentcontext.Context {
func (c *AgentCaller) buildAgentContext(ctx *robottypes.Context) *agentcontext.Context {
// Build authorized info for agent context
var authorized *oauthtypes.AuthorizedInfo
if ctx.Auth != nil {
@ -337,7 +337,7 @@ func (c *Conversation) WithHistory(messages []agentcontext.Message) *Conversatio
// Turn executes a single turn in the conversation
// userInput: the user's message for this turn
// Returns the turn result with agent response
func (c *Conversation) Turn(ctx *types.Context, userInput string) (*TurnResult, error) {
func (c *Conversation) Turn(ctx *robottypes.Context, userInput string) (*TurnResult, error) {
// Check max turns
turnNum := c.TurnCount() + 1
if c.maxTurns > 0 && turnNum > c.maxTurns {
@ -431,7 +431,7 @@ func (c *Conversation) Reset() {
// checkFn: called after each turn, returns (done, error)
// Returns all turn results
func (c *Conversation) RunUntil(
ctx *types.Context,
ctx *robottypes.Context,
inputFn func(turn int, lastResult *CallResult) (string, error),
checkFn func(turn int, result *CallResult) (done bool, err error),
) ([]*TurnResult, error) {

View file

@ -1,4 +1,4 @@
package executor_test
package standard_test
import (
"context"
@ -6,7 +6,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/yaoapp/yao/agent/robot/executor"
"github.com/yaoapp/yao/agent/robot/executor/standard"
"github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/agent/testutils"
oauthtypes "github.com/yaoapp/yao/openapi/oauth/types"
@ -32,7 +32,7 @@ func TestAgentCallerSingleCall(t *testing.T) {
testutils.Prepare(t)
defer testutils.Clean(t)
caller := executor.NewAgentCaller()
caller := standard.NewAgentCaller()
ctx := types.NewContext(context.Background(), testAuth())
// Test basic call - verify assistant responds and returns parseable JSON
@ -72,7 +72,7 @@ func TestAgentCallerNextHookData(t *testing.T) {
testutils.Prepare(t)
defer testutils.Clean(t)
caller := executor.NewAgentCaller()
caller := standard.NewAgentCaller()
ctx := types.NewContext(context.Background(), testAuth())
t.Run("next_hook inspiration returns structured data", func(t *testing.T) {
@ -121,7 +121,7 @@ func TestAgentCallerJSONArray(t *testing.T) {
testutils.Prepare(t)
defer testutils.Clean(t)
caller := executor.NewAgentCaller()
caller := standard.NewAgentCaller()
ctx := types.NewContext(context.Background(), testAuth())
t.Run("array_test returns JSON array", func(t *testing.T) {
@ -150,7 +150,7 @@ func TestAgentCallerEmptyResponse(t *testing.T) {
testutils.Prepare(t)
defer testutils.Clean(t)
caller := executor.NewAgentCaller()
caller := standard.NewAgentCaller()
ctx := types.NewContext(context.Background(), testAuth())
t.Run("empty_test falls back to completion content", func(t *testing.T) {
@ -171,7 +171,7 @@ func TestAgentCallerAssistantNotFound(t *testing.T) {
testutils.Prepare(t)
defer testutils.Clean(t)
caller := executor.NewAgentCaller()
caller := standard.NewAgentCaller()
ctx := types.NewContext(context.Background(), testAuth())
t.Run("non-existent assistant returns error", func(t *testing.T) {
@ -191,7 +191,7 @@ func TestAgentCallerWithSystemAndUser(t *testing.T) {
testutils.Prepare(t)
defer testutils.Clean(t)
caller := executor.NewAgentCaller()
caller := standard.NewAgentCaller()
ctx := types.NewContext(context.Background(), testAuth())
t.Run("call with system and user messages", func(t *testing.T) {
@ -223,7 +223,7 @@ func TestConversationMultiTurn(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("multi-turn conversation maintains state", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-1", 10)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-1", 10)
// Turn 1: Start planning
turn1, err := conv.Turn(ctx, "Plan tasks for sending weekly report")
@ -273,7 +273,7 @@ func TestConversationTurnCount(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("turn count increments correctly", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-2", 10)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-2", 10)
assert.Equal(t, 0, conv.TurnCount())
@ -298,7 +298,7 @@ func TestConversationMaxTurns(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("exceeding max turns returns error", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-3", 2)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-3", 2)
_, err := conv.Turn(ctx, "First")
require.NoError(t, err)
@ -324,7 +324,7 @@ func TestConversationMessages(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("messages history is maintained", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-4", 5)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-4", 5)
// Initially empty
assert.Empty(t, conv.Messages())
@ -349,7 +349,7 @@ func TestConversationLastResponse(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("last response returns assistant message", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-5", 5)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-5", 5)
// No response yet
assert.Nil(t, conv.LastResponse())
@ -375,7 +375,7 @@ func TestConversationReset(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("reset clears conversation history", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-6", 5)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-6", 5)
_, err := conv.Turn(ctx, "First message")
require.NoError(t, err)
@ -398,7 +398,7 @@ func TestConversationWithSystemPrompt(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("system prompt is preserved after reset", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-7", 5).
conv := standard.NewConversation("tests.robot-conversation", "test-conv-7", 5).
WithSystemPrompt("You are a task planner.")
msgs := conv.Messages()
@ -428,7 +428,7 @@ func TestConversationSpecialCommands(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("skip command jumps to completed", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-8", 5)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-8", 5)
turn, err := conv.Turn(ctx, "skip")
require.NoError(t, err)
@ -440,7 +440,7 @@ func TestConversationSpecialCommands(t *testing.T) {
})
t.Run("abort command ends conversation", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-9", 5)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-9", 5)
turn, err := conv.Turn(ctx, "abort")
require.NoError(t, err)
@ -452,7 +452,7 @@ func TestConversationSpecialCommands(t *testing.T) {
})
t.Run("reset command resets conversation state", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-10", 5)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-10", 5)
// First do a turn
_, err := conv.Turn(ctx, "Start planning")
@ -479,7 +479,7 @@ func TestConversationRunUntil(t *testing.T) {
ctx := types.NewContext(context.Background(), testAuth())
t.Run("run until completion", func(t *testing.T) {
conv := executor.NewConversation("tests.robot-conversation", "test-conv-11", 10)
conv := standard.NewConversation("tests.robot-conversation", "test-conv-11", 10)
inputs := []string{
"Plan weekly report tasks",
@ -490,7 +490,7 @@ func TestConversationRunUntil(t *testing.T) {
results, err := conv.RunUntil(
ctx,
func(turn int, lastResult *executor.CallResult) (string, error) {
func(turn int, lastResult *standard.CallResult) (string, error) {
if inputIdx < len(inputs) {
input := inputs[inputIdx]
inputIdx++
@ -498,7 +498,7 @@ func TestConversationRunUntil(t *testing.T) {
}
return "confirm", nil
},
func(turn int, result *executor.CallResult) (bool, error) {
func(turn int, result *standard.CallResult) (bool, error) {
data, err := result.GetJSON()
if err != nil {
return false, nil
@ -524,29 +524,29 @@ func TestConversationRunUntil(t *testing.T) {
func TestCallResultGetText(t *testing.T) {
t.Run("returns content when available", func(t *testing.T) {
result := &executor.CallResult{Content: "Hello World"}
result := &standard.CallResult{Content: "Hello World"}
assert.Equal(t, "Hello World", result.GetText())
})
t.Run("returns empty for empty result", func(t *testing.T) {
result := &executor.CallResult{}
result := &standard.CallResult{}
assert.Equal(t, "", result.GetText())
})
}
func TestCallResultIsEmpty(t *testing.T) {
t.Run("empty when no content and no next", func(t *testing.T) {
result := &executor.CallResult{}
result := &standard.CallResult{}
assert.True(t, result.IsEmpty())
})
t.Run("not empty when has content", func(t *testing.T) {
result := &executor.CallResult{Content: "test"}
result := &standard.CallResult{Content: "test"}
assert.False(t, result.IsEmpty())
})
t.Run("not empty when has next", func(t *testing.T) {
result := &executor.CallResult{Next: map[string]interface{}{"key": "value"}}
result := &standard.CallResult{Next: map[string]interface{}{"key": "value"}}
assert.False(t, result.IsEmpty())
})
}
@ -558,7 +558,7 @@ func TestCallResultIsEmpty(t *testing.T) {
func TestExtractCodeBlock(t *testing.T) {
t.Run("extracts JSON code block", func(t *testing.T) {
content := "Here is the result:\n```json\n{\"key\": \"value\"}\n```"
block := executor.ExtractCodeBlock(content)
block := standard.ExtractCodeBlock(content)
require.NotNil(t, block)
assert.Equal(t, "json", block.Type)
@ -567,7 +567,7 @@ func TestExtractCodeBlock(t *testing.T) {
t.Run("returns nil for no code block", func(t *testing.T) {
content := "Just plain text"
block := executor.ExtractCodeBlock(content)
block := standard.ExtractCodeBlock(content)
// gou/text returns text type for plain text
require.NotNil(t, block)
@ -578,7 +578,7 @@ func TestExtractCodeBlock(t *testing.T) {
func TestExtractAllCodeBlocks(t *testing.T) {
t.Run("extracts multiple code blocks", func(t *testing.T) {
content := "```json\n{}\n```\n\n```python\nprint('hello')\n```"
blocks := executor.ExtractAllCodeBlocks(content)
blocks := standard.ExtractAllCodeBlocks(content)
assert.Len(t, blocks, 2)
})

View file

@ -0,0 +1,32 @@
package standard
import (
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// RunDelivery executes P4: Delivery phase
// Delivers results to configured targets
//
// Input:
// - TaskResults (from P3)
// - Delivery config from robot
//
// Output:
// - DeliveryResult with success status
//
// Delivery Types:
// - DeliveryEmail: Send email
// - DeliveryNotify: Send notification
// - DeliveryWebhook: Call webhook
// - DeliveryStore: Store to database
//
// TODO: Implement real delivery
func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
e.simulateStreamDelay()
exec.Delivery = &robottypes.DeliveryResult{
Type: robottypes.DeliveryNotify,
Success: true,
}
return nil
}

View file

@ -0,0 +1,224 @@
package standard
import (
"fmt"
"sync/atomic"
"time"
"github.com/yaoapp/yao/agent/robot/executor/types"
"github.com/yaoapp/yao/agent/robot/job"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// Executor implements the standard executor with real Agent calls
// This is the production executor that:
// - Creates Job records for tracking
// - Calls real Agents via Assistant.Stream()
// - Logs phase transitions and errors
type Executor struct {
config types.Config
execCount atomic.Int32
currentCount atomic.Int32
onStart func()
onEnd func()
}
// New creates a new standard executor
func New() *Executor {
return &Executor{}
}
// NewWithConfig creates a new standard executor with configuration
func NewWithConfig(config types.Config) *Executor {
return &Executor{
config: config,
}
}
// Execute runs a robot through all applicable phases with real Agent calls
func (e *Executor) Execute(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}) (*robottypes.Execution, error) {
if robot == nil {
return nil, fmt.Errorf("robot cannot be nil")
}
var exec *robottypes.Execution
var err error
// Determine starting phase based on trigger type
startPhaseIndex := 0
if trigger == robottypes.TriggerHuman || trigger == robottypes.TriggerEvent {
startPhaseIndex = 1 // Skip P0 (Inspiration)
}
// Create execution with Job integration
if !e.config.SkipJobIntegration {
exec, err = job.CreateExecution(ctx, &job.CreateOptions{
Robot: robot,
TriggerType: trigger,
Input: types.BuildTriggerInput(trigger, data),
})
if err != nil {
return nil, fmt.Errorf("failed to create execution: %w", err)
}
} else {
exec = &robottypes.Execution{
ID: fmt.Sprintf("exec_%d", time.Now().UnixNano()),
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
StartTime: time.Now(),
Status: robottypes.ExecPending,
Phase: robottypes.AllPhases[startPhaseIndex],
Input: types.BuildTriggerInput(trigger, data),
}
}
// Set robot reference for phase methods
exec.SetRobot(robot)
// Acquire execution slot
if !robot.TryAcquireSlot(exec) {
if !e.config.SkipJobIntegration && exec.JobID != "" {
_ = job.FailExecution(ctx, exec, robottypes.ErrQuotaExceeded)
}
return nil, robottypes.ErrQuotaExceeded
}
defer robot.RemoveExecution(exec.ID)
// Track execution count
e.execCount.Add(1)
e.currentCount.Add(1)
defer e.currentCount.Add(-1)
// Callbacks
if e.onStart != nil {
e.onStart()
}
if e.onEnd != nil {
defer e.onEnd()
}
// Update status to running
exec.Status = robottypes.ExecRunning
if !e.config.SkipJobIntegration {
if err := job.UpdateStatus(ctx, exec, robottypes.ExecRunning); err != nil {
_ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to update status to running: %v", err))
}
}
// Check for simulated failure (for testing)
if dataStr, ok := data.(string); ok && dataStr == "simulate_failure" {
exec.Status = robottypes.ExecFailed
exec.Error = "simulated failure"
if !e.config.SkipJobIntegration {
_ = job.FailExecution(ctx, exec, fmt.Errorf("simulated failure"))
}
return exec, nil
}
// Execute phases
phases := robottypes.AllPhases[startPhaseIndex:]
for _, phase := range phases {
if err := e.runPhase(ctx, exec, phase, data); err != nil {
exec.Status = robottypes.ExecFailed
exec.Error = err.Error()
if !e.config.SkipJobIntegration {
_ = job.FailExecution(ctx, exec, err)
}
return exec, nil
}
}
// Mark completed
exec.Status = robottypes.ExecCompleted
now := time.Now()
exec.EndTime = &now
if !e.config.SkipJobIntegration {
if err := job.CompleteExecution(ctx, exec); err != nil {
_ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to mark execution as completed: %v", err))
}
}
return exec, nil
}
// runPhase executes a single phase
func (e *Executor) runPhase(ctx *robottypes.Context, exec *robottypes.Execution, phase robottypes.Phase, data interface{}) error {
exec.Phase = phase
if !e.config.SkipJobIntegration {
if err := job.UpdatePhase(ctx, exec, phase); err != nil {
_ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to update phase to %s: %v", phase, err))
}
}
if e.config.OnPhaseStart != nil {
e.config.OnPhaseStart(phase)
}
phaseStart := time.Now()
// Execute phase-specific logic
var err error
switch phase {
case robottypes.PhaseInspiration:
err = e.RunInspiration(ctx, exec, data)
case robottypes.PhaseGoals:
err = e.RunGoals(ctx, exec, data)
case robottypes.PhaseTasks:
err = e.RunTasks(ctx, exec, data)
case robottypes.PhaseRun:
err = e.RunExecution(ctx, exec, data)
case robottypes.PhaseDelivery:
err = e.RunDelivery(ctx, exec, data)
case robottypes.PhaseLearning:
err = e.RunLearning(ctx, exec, data)
}
if err != nil {
if !e.config.SkipJobIntegration {
_ = job.LogPhaseError(ctx, exec, phase, err)
}
return err
}
if e.config.OnPhaseEnd != nil {
e.config.OnPhaseEnd(phase)
}
if !e.config.SkipJobIntegration {
phaseDuration := time.Since(phaseStart).Milliseconds()
_ = job.LogPhaseEnd(ctx, exec, phase, phaseDuration)
}
return nil
}
// ExecCount returns total execution count
func (e *Executor) ExecCount() int {
return int(e.execCount.Load())
}
// CurrentCount returns currently running execution count
func (e *Executor) CurrentCount() int {
return int(e.currentCount.Load())
}
// Reset resets the executor counters
func (e *Executor) Reset() {
e.execCount.Store(0)
e.currentCount.Store(0)
}
// DefaultStreamDelay is the simulated delay for Agent Stream calls
// This will be removed when real Agent calls are implemented
const DefaultStreamDelay = 50 * time.Millisecond
// simulateStreamDelay simulates the delay of an Agent Stream call
func (e *Executor) simulateStreamDelay() {
time.Sleep(DefaultStreamDelay)
}
// Verify Executor implements types.Executor
var _ types.Executor = (*Executor)(nil)

View file

@ -0,0 +1,25 @@
package standard
import (
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// RunGoals executes P1: Goals phase
// Calls the Goals Agent to plan daily objectives
//
// Input:
// - InspirationReport (from P0) for clock trigger
// - TriggerInput for human/event trigger
//
// Output:
// - Goals with markdown content listing prioritized objectives
//
// TODO: Implement real Agent call
func (e *Executor) RunGoals(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
e.simulateStreamDelay()
exec.Goals = &robottypes.Goals{
Content: "## Today's Goals\n\n1. [High] Review pending tasks\n2. [Medium] Process new requests\n3. [Low] Organize knowledge base",
}
return nil
}

View file

@ -1,4 +1,4 @@
package executor
package standard
import (
"encoding/json"
@ -6,7 +6,7 @@ import (
"strings"
agentcontext "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/robot/types"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// InputFormatter provides methods to format input data for assistant prompts
@ -26,7 +26,7 @@ func NewInputFormatter() *InputFormatter {
// FormatClockContext formats ClockContext as user message content
// Used by P0 (Inspiration) phase
func (f *InputFormatter) FormatClockContext(clock *types.ClockContext, robot *types.Robot) string {
func (f *InputFormatter) FormatClockContext(clock *robottypes.ClockContext, robot *robottypes.Robot) string {
if clock == nil {
return ""
}
@ -82,7 +82,7 @@ func (f *InputFormatter) FormatClockContext(clock *types.ClockContext, robot *ty
// FormatInspirationReport formats InspirationReport as user message content
// Used by P1 (Goals) phase when trigger is Clock
func (f *InputFormatter) FormatInspirationReport(report *types.InspirationReport) string {
func (f *InputFormatter) FormatInspirationReport(report *robottypes.InspirationReport) string {
if report == nil {
return ""
}
@ -127,7 +127,7 @@ func (f *InputFormatter) FormatInspirationReport(report *types.InspirationReport
// FormatTriggerInput formats TriggerInput as user message content
// Used by P1 (Goals) phase when trigger is Human or Event
func (f *InputFormatter) FormatTriggerInput(input *types.TriggerInput) string {
func (f *InputFormatter) FormatTriggerInput(input *robottypes.TriggerInput) string {
if input == nil {
return ""
}
@ -178,7 +178,7 @@ func (f *InputFormatter) FormatTriggerInput(input *types.TriggerInput) string {
// FormatGoals formats Goals as user message content
// Used by P2 (Tasks) phase
func (f *InputFormatter) FormatGoals(goals *types.Goals, robot *types.Robot) string {
func (f *InputFormatter) FormatGoals(goals *robottypes.Goals, robot *robottypes.Robot) string {
if goals == nil {
return ""
}
@ -220,7 +220,7 @@ func (f *InputFormatter) FormatGoals(goals *types.Goals, robot *types.Robot) str
// FormatTasks formats Tasks as user message content
// Used by P3 (Run) phase
func (f *InputFormatter) FormatTasks(tasks []types.Task) string {
func (f *InputFormatter) FormatTasks(tasks []robottypes.Task) string {
if len(tasks) == 0 {
return "No tasks to execute."
}
@ -262,7 +262,7 @@ func (f *InputFormatter) FormatTasks(tasks []types.Task) string {
// FormatTaskResults formats TaskResults as user message content
// Used by P4 (Delivery) and P5 (Learning) phases
func (f *InputFormatter) FormatTaskResults(results []types.TaskResult) string {
func (f *InputFormatter) FormatTaskResults(results []robottypes.TaskResult) string {
if len(results) == 0 {
return "No task results."
}
@ -335,7 +335,7 @@ func (f *InputFormatter) FormatTaskResults(results []types.TaskResult) string {
}
// FormatExecutionSummary formats the entire execution for P5 (Learning) phase
func (f *InputFormatter) FormatExecutionSummary(exec *types.Execution) string {
func (f *InputFormatter) FormatExecutionSummary(exec *robottypes.Execution) string {
if exec == nil {
return ""
}

View file

@ -1,4 +1,4 @@
package executor_test
package standard_test
import (
"testing"
@ -7,7 +7,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
agentcontext "github.com/yaoapp/yao/agent/context"
"github.com/yaoapp/yao/agent/robot/executor"
"github.com/yaoapp/yao/agent/robot/executor/standard"
"github.com/yaoapp/yao/agent/robot/types"
)
@ -16,7 +16,7 @@ import (
// ============================================================================
func TestInputFormatterFormatClockContext(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats clock context with all fields", func(t *testing.T) {
now := time.Date(2024, 1, 15, 9, 30, 0, 0, time.UTC)
@ -90,7 +90,7 @@ func TestInputFormatterFormatClockContext(t *testing.T) {
}
func TestInputFormatterFormatInspirationReport(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats inspiration report with clock", func(t *testing.T) {
now := time.Date(2024, 1, 15, 9, 30, 0, 0, time.UTC)
@ -127,7 +127,7 @@ func TestInputFormatterFormatInspirationReport(t *testing.T) {
}
func TestInputFormatterFormatTriggerInput(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats human intervention", func(t *testing.T) {
input := &types.TriggerInput{
@ -180,7 +180,7 @@ func TestInputFormatterFormatTriggerInput(t *testing.T) {
}
func TestInputFormatterFormatGoals(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats goals with resources", func(t *testing.T) {
goals := &types.Goals{
@ -230,7 +230,7 @@ func TestInputFormatterFormatGoals(t *testing.T) {
}
func TestInputFormatterFormatTasks(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats multiple tasks", func(t *testing.T) {
tasks := []types.Task{
@ -277,7 +277,7 @@ func TestInputFormatterFormatTasks(t *testing.T) {
}
func TestInputFormatterFormatTaskResults(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats task results with summary", func(t *testing.T) {
results := []types.TaskResult{
@ -330,7 +330,7 @@ func TestInputFormatterFormatTaskResults(t *testing.T) {
}
func TestInputFormatterFormatExecutionSummary(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("formats complete execution summary", func(t *testing.T) {
startTime := time.Date(2024, 1, 15, 9, 0, 0, 0, time.UTC)
@ -404,7 +404,7 @@ func TestInputFormatterFormatExecutionSummary(t *testing.T) {
}
func TestInputFormatterBuildMessages(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("builds user message", func(t *testing.T) {
msgs := formatter.BuildMessages("Hello, world!")
@ -416,7 +416,7 @@ func TestInputFormatterBuildMessages(t *testing.T) {
}
func TestInputFormatterBuildMessagesWithSystem(t *testing.T) {
formatter := executor.NewInputFormatter()
formatter := standard.NewInputFormatter()
t.Run("builds system and user messages", func(t *testing.T) {
msgs := formatter.BuildMessagesWithSystem(

View file

@ -0,0 +1,64 @@
package standard
import (
"fmt"
"time"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// RunInspiration executes P0: Inspiration phase
// Calls the Inspiration Agent to generate daily briefing
//
// Input:
// - ClockContext from trigger input or current time
// - Robot identity and resources
//
// Output:
// - InspirationReport with markdown content
func (e *Executor) RunInspiration(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
// Get robot for identity and resources
robot := exec.GetRobot()
if robot == nil {
return fmt.Errorf("robot not found in execution")
}
// Build clock context from trigger input or current time
var clock *robottypes.ClockContext
if exec.Input != nil && exec.Input.Clock != nil {
clock = exec.Input.Clock
} else {
clock = robottypes.NewClockContext(time.Now(), "")
}
// Get agent ID for inspiration phase
agentID := "__yao.inspiration" // default
if robot.Config != nil && robot.Config.Resources != nil {
agentID = robot.Config.Resources.GetPhaseAgent(robottypes.PhaseInspiration)
}
// Build prompt using InputFormatter
formatter := NewInputFormatter()
userContent := formatter.FormatClockContext(clock, robot)
// Call agent
caller := NewAgentCaller()
result, err := caller.CallWithMessages(ctx, agentID, userContent)
if err != nil {
return fmt.Errorf("inspiration agent call failed: %w", err)
}
// Parse response - get markdown content
content := result.GetText()
if content == "" {
return fmt.Errorf("inspiration agent returned empty response")
}
// Build InspirationReport
exec.Inspiration = &robottypes.InspirationReport{
Clock: clock,
Content: content,
}
return nil
}

View file

@ -0,0 +1,353 @@
package standard_test
import (
"context"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/yaoapp/yao/agent/robot/executor/standard"
"github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/agent/testutils"
)
// ============================================================================
// P0 Inspiration Phase Tests
// ============================================================================
func TestRunInspirationBasic(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
ctx := types.NewContext(context.Background(), testAuth())
t.Run("generates inspiration report with clock context", func(t *testing.T) {
// Create robot with inspiration agent configured
robot := createTestRobot(t, "robot.inspiration")
// Create executor and execution
exec := createTestExecution(robot, types.TriggerClock)
// Run inspiration phase
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
require.NoError(t, err)
require.NotNil(t, exec.Inspiration)
assert.NotEmpty(t, exec.Inspiration.Content)
assert.NotNil(t, exec.Inspiration.Clock)
})
t.Run("includes expected markdown sections", func(t *testing.T) {
robot := createTestRobot(t, "robot.inspiration")
exec := createTestExecution(robot, types.TriggerClock)
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
require.NoError(t, err)
content := exec.Inspiration.Content
// Verify expected sections in markdown output
// Note: LLM output is non-deterministic, so we check for likely sections
hasSection := strings.Contains(content, "##") ||
strings.Contains(content, "Summary") ||
strings.Contains(content, "Highlight") ||
strings.Contains(content, "Recommend")
assert.True(t, hasSection, "should contain markdown sections, got: %s", content)
})
}
func TestRunInspirationClockContext(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
ctx := types.NewContext(context.Background(), testAuth())
t.Run("uses clock from trigger input", func(t *testing.T) {
robot := createTestRobot(t, "robot.inspiration")
exec := createTestExecution(robot, types.TriggerClock)
// Set specific clock context
specificTime := time.Date(2024, 12, 31, 17, 0, 0, 0, time.UTC)
exec.Input = &types.TriggerInput{
Clock: types.NewClockContext(specificTime, "UTC"),
}
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
require.NoError(t, err)
require.NotNil(t, exec.Inspiration.Clock)
// Clock should match input
assert.Equal(t, specificTime.Year(), exec.Inspiration.Clock.Year)
assert.Equal(t, int(specificTime.Month()), exec.Inspiration.Clock.Month)
assert.Equal(t, specificTime.Day(), exec.Inspiration.Clock.DayOfMonth)
})
t.Run("creates clock context when not provided", func(t *testing.T) {
robot := createTestRobot(t, "robot.inspiration")
exec := createTestExecution(robot, types.TriggerClock)
exec.Input = nil // No input
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
require.NoError(t, err)
require.NotNil(t, exec.Inspiration.Clock)
// Clock should be current time (approximately)
now := time.Now()
assert.Equal(t, now.Year(), exec.Inspiration.Clock.Year)
})
}
func TestRunInspirationRobotIdentity(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
ctx := types.NewContext(context.Background(), testAuth())
t.Run("robot identity influences output", func(t *testing.T) {
// Create robot with specific identity
robot := &types.Robot{
MemberID: "test-robot-1",
TeamID: "test-team-1",
DisplayName: "Sales Assistant",
Config: &types.Config{
Identity: &types.Identity{
Role: "Sales Assistant",
Duties: []string{"Track sales metrics", "Prepare weekly reports"},
Rules: []string{"Focus on actionable insights"},
},
Resources: &types.Resources{
Phases: map[types.Phase]string{
types.PhaseInspiration: "robot.inspiration",
},
},
},
}
exec := createTestExecution(robot, types.TriggerClock)
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
require.NoError(t, err)
assert.NotEmpty(t, exec.Inspiration.Content)
// The content should be influenced by robot identity
// (exact content varies due to LLM non-determinism)
})
}
func TestRunInspirationErrorHandling(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
ctx := types.NewContext(context.Background(), testAuth())
t.Run("returns error when robot is nil", func(t *testing.T) {
exec := &types.Execution{
ID: "test-exec-1",
TriggerType: types.TriggerClock,
}
// Don't set robot
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
assert.Error(t, err)
assert.Contains(t, err.Error(), "robot not found")
})
t.Run("returns error when agent not found", func(t *testing.T) {
robot := &types.Robot{
MemberID: "test-robot-1",
TeamID: "test-team-1",
Config: &types.Config{
Identity: &types.Identity{Role: "Test"},
Resources: &types.Resources{
Phases: map[types.Phase]string{
types.PhaseInspiration: "non.existent.agent",
},
},
},
}
exec := createTestExecution(robot, types.TriggerClock)
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
// Real AgentCaller returns error for non-existent agent
assert.Error(t, err)
assert.Contains(t, err.Error(), "agent call failed")
})
}
func TestRunInspirationWithDefaultAgent(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
ctx := types.NewContext(context.Background(), testAuth())
t.Run("uses default agent when not configured", func(t *testing.T) {
robot := &types.Robot{
MemberID: "test-robot-1",
TeamID: "test-team-1",
Config: &types.Config{
Identity: &types.Identity{Role: "Test Robot"},
// No Resources configured - should use default __yao.inspiration
},
}
exec := createTestExecution(robot, types.TriggerClock)
e := standard.New()
err := e.RunInspiration(ctx, exec, nil)
// This will fail if __yao.inspiration doesn't exist
// In test environment, we expect it to fail with "agent not found"
// In production, it would use the default agent
if err != nil {
assert.Contains(t, err.Error(), "agent call failed")
}
})
}
// ============================================================================
// InputFormatter Tests for P0
// ============================================================================
func TestInputFormatterClockContext(t *testing.T) {
t.Run("formats clock context correctly", func(t *testing.T) {
formatter := standard.NewInputFormatter()
// Create a specific clock context
clock := types.NewClockContext(
time.Date(2024, 12, 31, 17, 30, 0, 0, time.UTC),
"UTC",
)
robot := &types.Robot{
Config: &types.Config{
Identity: &types.Identity{
Role: "Sales Assistant",
Duties: []string{"Track metrics", "Send reports"},
},
},
}
content := formatter.FormatClockContext(clock, robot)
// Verify time context
assert.Contains(t, content, "Current Time Context")
assert.Contains(t, content, "2024")
assert.Contains(t, content, "12")
assert.Contains(t, content, "31")
assert.Contains(t, content, "Tuesday") // Dec 31, 2024 is Tuesday
// Verify robot identity
assert.Contains(t, content, "Robot Identity")
assert.Contains(t, content, "Sales Assistant")
assert.Contains(t, content, "Track metrics")
})
t.Run("handles nil clock", func(t *testing.T) {
formatter := standard.NewInputFormatter()
content := formatter.FormatClockContext(nil, nil)
assert.Empty(t, content)
})
t.Run("handles nil robot", func(t *testing.T) {
formatter := standard.NewInputFormatter()
clock := types.NewClockContext(time.Now(), "")
content := formatter.FormatClockContext(clock, nil)
// Should have time context but no robot identity
assert.Contains(t, content, "Current Time Context")
assert.NotContains(t, content, "Robot Identity")
})
t.Run("includes time markers", func(t *testing.T) {
formatter := standard.NewInputFormatter()
// Create a weekend + month start clock context
// Jan 1, 2028 is Saturday (weekend + month start)
clock := types.NewClockContext(
time.Date(2028, 1, 1, 10, 0, 0, 0, time.UTC),
"UTC",
)
content := formatter.FormatClockContext(clock, nil)
assert.Contains(t, content, "Weekend")
assert.Contains(t, content, "Month Start")
})
}
// ============================================================================
// Helper Functions
// ============================================================================
// createTestRobot creates a test robot with specified inspiration agent
func createTestRobot(t *testing.T, agentID string) *types.Robot {
t.Helper()
return &types.Robot{
MemberID: "test-robot-1",
TeamID: "test-team-1",
DisplayName: "Test Robot",
Config: &types.Config{
Identity: &types.Identity{
Role: "Test Assistant",
Duties: []string{"Testing"},
},
Resources: &types.Resources{
Phases: map[types.Phase]string{
types.PhaseInspiration: agentID,
},
},
},
}
}
// createTestExecution creates a test execution for a robot
func createTestExecution(robot *types.Robot, trigger types.TriggerType) *types.Execution {
exec := &types.Execution{
ID: "test-exec-1",
MemberID: robot.MemberID,
TeamID: robot.TeamID,
TriggerType: trigger,
StartTime: time.Now(),
Status: types.ExecRunning,
Phase: types.PhaseInspiration,
Input: &types.TriggerInput{
Clock: types.NewClockContext(time.Now(), ""),
},
}
exec.SetRobot(robot)
return exec
}

View file

@ -0,0 +1,32 @@
package standard
import (
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// RunLearning executes P5: Learning phase
// Extracts learnings and saves to knowledge base
//
// Input:
// - Execution summary (all phases)
//
// Output:
// - LearningEntry list with extracted knowledge
//
// Learning Types:
// - LearnExecution: Execution patterns
// - LearnTask: Task-specific insights
// - LearnError: Error patterns for improvement
//
// TODO: Implement real learning extraction
func (e *Executor) RunLearning(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
e.simulateStreamDelay()
exec.Learning = []robottypes.LearningEntry{
{
Type: robottypes.LearnExecution,
Content: "Completed daily tasks successfully",
},
}
return nil
}

View file

@ -0,0 +1,39 @@
package standard
import (
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// RunExecution executes P3: Run phase
// Executes each task using the appropriate executor (Assistant, Process, or Function)
//
// Input:
// - Tasks (from P2)
//
// Output:
// - TaskResult for each task with output and validation
//
// Executor Types:
// - ExecutorAssistant: Call AI assistant
// - ExecutorMCP: Call MCP tool
// - ExecutorProcess: Run Yao process
// - ExecutorFunction: Call JavaScript function
//
// TODO: Implement real task execution
func (e *Executor) RunExecution(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
e.simulateStreamDelay()
exec.Results = []robottypes.TaskResult{
{
TaskID: "task-1",
Success: true,
Output: map[string]interface{}{"status": "completed"},
Duration: 100,
Validation: &robottypes.ValidationResult{
Passed: true,
Score: 0.95,
},
},
}
return nil
}

View file

@ -0,0 +1,32 @@
package standard
import (
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// RunTasks executes P2: Tasks phase
// Calls the Tasks Agent to break down goals into executable tasks
//
// Input:
// - Goals (from P1)
// - Available resources (Agents, MCP tools)
//
// Output:
// - List of Task objects with executor assignments
//
// TODO: Implement real Agent call
func (e *Executor) RunTasks(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
e.simulateStreamDelay()
exec.Tasks = []robottypes.Task{
{
ID: "task-1",
GoalRef: "Goal 1",
Source: robottypes.TaskSourceAuto,
ExecutorType: robottypes.ExecutorAssistant,
ExecutorID: "default-assistant",
Status: robottypes.TaskPending,
},
}
return nil
}

View file

@ -1,52 +0,0 @@
package executor
import (
"github.com/yaoapp/yao/agent/robot/types"
)
// RunTasks executes P2: Tasks phase
//
// Reads Goals markdown and breaks into executable tasks.
// Each task specifies executor type (assistant/mcp/process) and arguments.
//
// Implementation (TODO Phase 6):
// 1. Build prompt with Goals
// 2. Call Task Planning Agent via Assistant.Stream()
// 3. Parse response to []Task (structured)
func (e *Executor) RunTasks(_ *types.Context, exec *types.Execution, _ interface{}) error {
// TODO (Phase 6): Replace with real Agent call
// agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseTasks)
// messages := buildTasksMessages(exec.Goals, robot)
// response, err := callAgentStream(ctx, agentID, messages)
// if err != nil {
// return err
// }
// exec.Tasks = parseTasks(response)
// Simulate Agent Stream delay
e.simulateStreamDelay()
// Generate mock tasks
exec.Tasks = []types.Task{
{
ID: "task_1",
GoalRef: "Goal 1",
Source: types.TaskSourceAuto,
ExecutorType: types.ExecutorAssistant,
ExecutorID: "data-analyst",
Status: types.TaskPending,
Order: 0,
},
{
ID: "task_2",
GoalRef: "Goal 2",
Source: types.TaskSourceAuto,
ExecutorType: types.ExecutorAssistant,
ExecutorID: "report-writer",
Status: types.TaskPending,
Order: 1,
},
}
return nil
}

View file

@ -0,0 +1,33 @@
package types
import (
"time"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// BuildTriggerInput builds TriggerInput from trigger data
// Shared helper used by all executor implementations
func BuildTriggerInput(trigger robottypes.TriggerType, data interface{}) *robottypes.TriggerInput {
input := &robottypes.TriggerInput{}
switch trigger {
case robottypes.TriggerClock:
input.Clock = robottypes.NewClockContext(time.Now(), "")
case robottypes.TriggerHuman:
if req, ok := data.(*robottypes.InterveneRequest); ok {
input.Action = req.Action
input.Messages = req.Messages
}
case robottypes.TriggerEvent:
if req, ok := data.(*robottypes.EventRequest); ok {
input.Source = robottypes.EventSource(req.Source)
input.EventType = req.EventType
input.Data = req.Data
}
}
return input
}

View file

@ -0,0 +1,132 @@
package types
import (
"time"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// Executor defines the interface for robot phase execution
// Different implementations provide different execution strategies:
// - Standard: Real Agent calls with full phase execution
// - DryRun: Plan-only mode, simulates execution without Agent calls
// - Sandbox: Isolated execution with resource limits and safety controls
type Executor interface {
// Execute runs a robot through all applicable phases
// ctx: Execution context with auth and logging
// robot: Robot configuration and state
// trigger: What triggered this execution (clock, human, event)
// data: Trigger-specific data (human input, event payload, etc.)
// Returns: Execution record with all phase outputs
Execute(ctx *robottypes.Context, robot *robottypes.Robot, trigger robottypes.TriggerType, data interface{}) (*robottypes.Execution, error)
// Metrics and control
ExecCount() int // Total execution count
CurrentCount() int // Currently running execution count
Reset() // Reset counters (for testing)
}
// PhaseExecutor defines the interface for individual phase execution
// Used internally by Executor implementations
type PhaseExecutor interface {
// RunInspiration executes P0: Inspiration phase
RunInspiration(ctx *robottypes.Context, exec *robottypes.Execution, data interface{}) error
// RunGoals executes P1: Goals phase
RunGoals(ctx *robottypes.Context, exec *robottypes.Execution, data interface{}) error
// RunTasks executes P2: Tasks phase
RunTasks(ctx *robottypes.Context, exec *robottypes.Execution, data interface{}) error
// RunExecution executes P3: Run phase (task execution)
RunExecution(ctx *robottypes.Context, exec *robottypes.Execution, data interface{}) error
// RunDelivery executes P4: Delivery phase
RunDelivery(ctx *robottypes.Context, exec *robottypes.Execution, data interface{}) error
// RunLearning executes P5: Learning phase
RunLearning(ctx *robottypes.Context, exec *robottypes.Execution, data interface{}) error
}
// Config holds common executor configuration
type Config struct {
// SkipJobIntegration skips job system integration (for testing)
SkipJobIntegration bool
// OnPhaseStart callback when a phase starts
OnPhaseStart func(phase robottypes.Phase)
// OnPhaseEnd callback when a phase ends
OnPhaseEnd func(phase robottypes.Phase)
}
// DryRunConfig holds dry-run specific configuration
type DryRunConfig struct {
Config
// Delay simulates execution delay for each phase
Delay time.Duration
// OnStart callback on execution start
OnStart func()
// OnEnd callback on execution end
OnEnd func()
}
// SandboxConfig holds sandbox specific configuration
//
// ⚠️ NOT IMPLEMENTED: These settings are placeholders for future
// container-based isolation. True sandbox requires infrastructure support
// (Docker/gVisor/Firecracker). Current implementation behaves like DryRun.
type SandboxConfig struct {
Config
// MaxDuration limits total execution time
MaxDuration time.Duration
// MaxMemory limits memory usage (bytes) - requires container runtime
MaxMemory int64
// AllowedAgents restricts which agents can be called
AllowedAgents []string
// AllowedTools restricts which tools can be used
AllowedTools []string
// NetworkAccess controls network access - requires container networking
NetworkAccess bool
// FileAccess controls file system access - requires container filesystem
FileAccess bool
}
// Mode represents the executor mode
type Mode string
const (
ModeStandard Mode = "standard" // Real Agent execution (production)
ModeDryRun Mode = "dryrun" // Simulated execution (testing/demo)
ModeSandbox Mode = "sandbox" // Container-isolated execution (NOT IMPLEMENTED)
)
// Setting holds executor settings from configuration
type Setting struct {
Mode Mode `json:"mode,omitempty" yaml:"mode,omitempty"` // Executor mode
MaxDuration time.Duration `json:"max_duration,omitempty" yaml:"max_duration,omitempty"` // Max execution time
MaxMemory int64 `json:"max_memory,omitempty" yaml:"max_memory,omitempty"` // Max memory (bytes)
AllowedAgents []string `json:"allowed_agents,omitempty" yaml:"allowed_agents,omitempty"` // Allowed agent IDs
NetworkAccess bool `json:"network_access,omitempty" yaml:"network_access,omitempty"` // Allow network
FileAccess bool `json:"file_access,omitempty" yaml:"file_access,omitempty"` // Allow file system
}
// DefaultSetting returns default executor settings
func DefaultSetting() *Setting {
return &Setting{
Mode: ModeStandard,
MaxDuration: 30 * time.Minute,
MaxMemory: 512 * 1024 * 1024, // 512MB
NetworkAccess: true,
FileAccess: false,
}
}

View file

@ -14,12 +14,25 @@ import (
"github.com/stretchr/testify/require"
"github.com/yaoapp/gou/model"
"github.com/yaoapp/xun/capsule"
"github.com/yaoapp/yao/agent/robot/executor"
"github.com/yaoapp/yao/agent/robot/manager"
"github.com/yaoapp/yao/agent/robot/pool"
"github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/agent/testutils"
)
// createClockTestManager creates a manager with mock executor for clock tests
func createClockTestManager(t *testing.T, tickInterval time.Duration, workerSize, queueSize int) (*manager.Manager, *executor.DryRunExecutor) {
exec := executor.NewDryRunWithDelay(0)
config := &manager.Config{
TickInterval: tickInterval,
PoolConfig: &pool.Config{WorkerSize: workerSize, QueueSize: queueSize},
Executor: exec,
}
m := manager.NewWithConfig(config)
return m, exec
}
// ==================== Times Mode Tests ====================
// TestIntegrationClockTimesMode tests the times mode clock trigger
@ -35,6 +48,9 @@ func TestIntegrationClockTimesMode(t *testing.T) {
defer cleanupIntegrationRobots(t)
t.Run("triggers at configured time", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_times1", "team_integ_clock", map[string]interface{}{
"mode": "times",
"times": []string{"09:00", "14:00", "17:00"},
@ -42,11 +58,7 @@ func TestIntegrationClockTimesMode(t *testing.T) {
"tz": "Asia/Shanghai",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
@ -56,7 +68,7 @@ func TestIntegrationClockTimesMode(t *testing.T) {
robot := m.Cache().Get("robot_integ_clock_times1")
require.NotNil(t, robot, "Robot should be loaded into cache")
m.Executor().Reset()
exec.Reset()
// Trigger at 09:00 on Wednesday
loc, _ := time.LoadLocation("Asia/Shanghai")
@ -67,10 +79,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
assert.NoError(t, err)
time.Sleep(300 * time.Millisecond)
assert.GreaterOrEqual(t, m.Executor().ExecCount(), 1, "Should trigger at 09:00")
assert.GreaterOrEqual(t, exec.ExecCount(), 1, "Should trigger at 09:00")
})
t.Run("does not trigger at non-configured time", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_times2", "team_integ_clock", map[string]interface{}{
"mode": "times",
"times": []string{"09:00", "14:00"},
@ -78,17 +93,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
"tz": "Asia/Shanghai",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
// Trigger at 10:30 (not configured)
loc, _ := time.LoadLocation("Asia/Shanghai")
@ -99,10 +110,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
assert.Equal(t, 0, m.Executor().ExecCount(), "Should not trigger at non-configured time")
assert.Equal(t, 0, exec.ExecCount(), "Should not trigger at non-configured time")
})
t.Run("does not trigger on non-configured day", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_times3", "team_integ_clock", map[string]interface{}{
"mode": "times",
"times": []string{"09:00"},
@ -110,17 +124,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
"tz": "Asia/Shanghai",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
// Trigger at 09:00 on Saturday (not configured)
loc, _ := time.LoadLocation("Asia/Shanghai")
@ -131,10 +141,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
assert.Equal(t, 0, m.Executor().ExecCount(), "Should not trigger on Saturday")
assert.Equal(t, 0, exec.ExecCount(), "Should not trigger on Saturday")
})
t.Run("wildcard days matches all days", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_times4", "team_integ_clock", map[string]interface{}{
"mode": "times",
"times": []string{"09:00"},
@ -142,17 +155,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
"tz": "Asia/Shanghai",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
// Trigger at 09:00 on Saturday
loc, _ := time.LoadLocation("Asia/Shanghai")
@ -163,10 +172,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
assert.NoError(t, err)
time.Sleep(300 * time.Millisecond)
assert.GreaterOrEqual(t, m.Executor().ExecCount(), 1, "Should trigger on Saturday with wildcard days")
assert.GreaterOrEqual(t, exec.ExecCount(), 1, "Should trigger on Saturday with wildcard days")
})
t.Run("dedup prevents double trigger in same minute", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_times5", "team_integ_clock", map[string]interface{}{
"mode": "times",
"times": []string{"09:00"},
@ -174,17 +186,13 @@ func TestIntegrationClockTimesMode(t *testing.T) {
"tz": "Asia/Shanghai",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
loc, _ := time.LoadLocation("Asia/Shanghai")
ctx := types.NewContext(context.Background(), nil)
@ -194,7 +202,7 @@ func TestIntegrationClockTimesMode(t *testing.T) {
err = m.Tick(ctx, now1)
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
firstCount := m.Executor().ExecCount()
firstCount := exec.ExecCount()
assert.GreaterOrEqual(t, firstCount, 1, "First tick should trigger")
// Second tick at 09:00:30 (same minute)
@ -204,7 +212,7 @@ func TestIntegrationClockTimesMode(t *testing.T) {
time.Sleep(200 * time.Millisecond)
// Should not trigger again in same minute
assert.Equal(t, firstCount, m.Executor().ExecCount(), "Should not trigger twice in same minute")
assert.Equal(t, firstCount, exec.ExecCount(), "Should not trigger twice in same minute")
})
}
@ -223,22 +231,21 @@ func TestIntegrationClockIntervalMode(t *testing.T) {
defer cleanupIntegrationRobots(t)
t.Run("triggers on first run", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_interval1", "team_integ_clock", map[string]interface{}{
"mode": "interval",
"every": "30m",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
now := time.Now()
@ -246,26 +253,25 @@ func TestIntegrationClockIntervalMode(t *testing.T) {
assert.NoError(t, err)
time.Sleep(300 * time.Millisecond)
assert.GreaterOrEqual(t, m.Executor().ExecCount(), 1, "Should trigger on first run")
assert.GreaterOrEqual(t, exec.ExecCount(), 1, "Should trigger on first run")
})
t.Run("triggers after interval passed", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_interval2", "team_integ_clock", map[string]interface{}{
"mode": "interval",
"every": "100ms", // Short interval for testing
})
config := &manager.Config{
TickInterval: 50 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 50*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
@ -274,7 +280,7 @@ func TestIntegrationClockIntervalMode(t *testing.T) {
err = m.Tick(ctx, now1)
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
firstCount := m.Executor().ExecCount()
firstCount := exec.ExecCount()
assert.GreaterOrEqual(t, firstCount, 1, "First tick should trigger")
// Wait for interval to pass
@ -287,26 +293,25 @@ func TestIntegrationClockIntervalMode(t *testing.T) {
time.Sleep(200 * time.Millisecond)
// Should have triggered again
assert.Greater(t, m.Executor().ExecCount(), firstCount, "Should trigger again after interval")
assert.Greater(t, exec.ExecCount(), firstCount, "Should trigger again after interval")
})
t.Run("does not trigger before interval passed", func(t *testing.T) {
// Clean up before each subtest to ensure isolation
cleanupIntegrationRobots(t)
setupClockTestRobot(t, "robot_integ_clock_interval3", "team_integ_clock", map[string]interface{}{
"mode": "interval",
"every": "1h", // Long interval
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
@ -315,7 +320,7 @@ func TestIntegrationClockIntervalMode(t *testing.T) {
err = m.Tick(ctx, now1)
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
firstCount := m.Executor().ExecCount()
firstCount := exec.ExecCount()
assert.GreaterOrEqual(t, firstCount, 1, "First tick should trigger")
// Second tick immediately (interval not passed)
@ -325,7 +330,7 @@ func TestIntegrationClockIntervalMode(t *testing.T) {
time.Sleep(200 * time.Millisecond)
// Should not trigger again
assert.Equal(t, firstCount, m.Executor().ExecCount(), "Should not trigger before interval")
assert.Equal(t, firstCount, exec.ExecCount(), "Should not trigger before interval")
})
}
@ -349,24 +354,20 @@ func TestIntegrationClockDaemonMode(t *testing.T) {
"timeout": "5m",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
err = m.Tick(ctx, time.Now())
assert.NoError(t, err)
time.Sleep(300 * time.Millisecond)
assert.GreaterOrEqual(t, m.Executor().ExecCount(), 1, "Daemon should trigger when idle")
assert.GreaterOrEqual(t, exec.ExecCount(), 1, "Daemon should trigger when idle")
})
t.Run("respects quota limit", func(t *testing.T) {
@ -378,17 +379,13 @@ func TestIntegrationClockDaemonMode(t *testing.T) {
},
1, 5, 5) // Max=1, Queue=5
config := &manager.Config{
TickInterval: 50 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 5, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 50*time.Millisecond, 5, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
@ -430,17 +427,13 @@ func TestIntegrationClockTimezone(t *testing.T) {
"tz": "Asia/Shanghai",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
@ -452,7 +445,7 @@ func TestIntegrationClockTimezone(t *testing.T) {
assert.NoError(t, err)
time.Sleep(300 * time.Millisecond)
assert.GreaterOrEqual(t, m.Executor().ExecCount(), 1, "Should trigger at 09:00 Shanghai time")
assert.GreaterOrEqual(t, exec.ExecCount(), 1, "Should trigger at 09:00 Shanghai time")
})
t.Run("different timezone same UTC time", func(t *testing.T) {
@ -472,17 +465,13 @@ func TestIntegrationClockTimezone(t *testing.T) {
"tz": "America/New_York",
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
m := manager.NewWithConfig(config)
m, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
m.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
@ -494,7 +483,7 @@ func TestIntegrationClockTimezone(t *testing.T) {
time.Sleep(300 * time.Millisecond)
// Only Shanghai robot should trigger
execCount := m.Executor().ExecCount()
execCount := exec.ExecCount()
assert.GreaterOrEqual(t, execCount, 1, "Shanghai robot should trigger")
// New York robot should not trigger (it's 20:00 in NY)
})
@ -548,17 +537,13 @@ func TestIntegrationClockEdgeCases(t *testing.T) {
})
require.NoError(t, err)
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
mgr := manager.NewWithConfig(config)
mgr, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err = mgr.Start()
require.NoError(t, err)
defer mgr.Stop()
mgr.Executor().Reset()
exec.Reset()
// Trigger at matching time
loc, _ := time.LoadLocation("Asia/Shanghai")
@ -569,7 +554,7 @@ func TestIntegrationClockEdgeCases(t *testing.T) {
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
assert.Equal(t, 0, mgr.Executor().ExecCount(), "Clock disabled robot should not trigger")
assert.Equal(t, 0, exec.ExecCount(), "Clock disabled robot should not trigger")
})
t.Run("paused robot is skipped", func(t *testing.T) {
@ -606,17 +591,13 @@ func TestIntegrationClockEdgeCases(t *testing.T) {
})
require.NoError(t, err)
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
mgr := manager.NewWithConfig(config)
mgr, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err = mgr.Start()
require.NoError(t, err)
defer mgr.Stop()
mgr.Executor().Reset()
exec.Reset()
// Trigger at matching time
loc, _ := time.LoadLocation("Asia/Shanghai")
@ -627,7 +608,7 @@ func TestIntegrationClockEdgeCases(t *testing.T) {
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
assert.Equal(t, 0, mgr.Executor().ExecCount(), "Paused robot should not trigger")
assert.Equal(t, 0, exec.ExecCount(), "Paused robot should not trigger")
})
t.Run("robot without clock config is skipped", func(t *testing.T) {
@ -660,24 +641,20 @@ func TestIntegrationClockEdgeCases(t *testing.T) {
})
require.NoError(t, err)
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 3, QueueSize: 20},
}
mgr := manager.NewWithConfig(config)
mgr, exec := createClockTestManager(t, 100*time.Millisecond, 3, 20)
err = mgr.Start()
require.NoError(t, err)
defer mgr.Stop()
mgr.Executor().Reset()
exec.Reset()
ctx := types.NewContext(context.Background(), nil)
err = mgr.Tick(ctx, time.Now())
assert.NoError(t, err)
time.Sleep(200 * time.Millisecond)
assert.Equal(t, 0, mgr.Executor().ExecCount(), "Robot without clock config should not trigger")
assert.Equal(t, 0, exec.ExecCount(), "Robot without clock config should not trigger")
})
}

View file

@ -50,7 +50,7 @@ func TestIntegrationConcurrentExecution(t *testing.T) {
var maxConcurrent int32
var currentConcurrent int32
exec := executor.NewWithCallback(100*time.Millisecond,
exec := executor.NewDryRunWithCallbacks(100*time.Millisecond,
func() {
curr := atomic.AddInt32(&currentConcurrent, 1)
for {
@ -109,7 +109,7 @@ func TestIntegrationConcurrentExecution(t *testing.T) {
t.Run("same robot multiple triggers", func(t *testing.T) {
setupConcurrentTestRobot(t, "robot_integ_conc_same", "team_integ_conc", 3, 20)
exec := executor.NewWithDelay(50 * time.Millisecond)
exec := executor.NewDryRunWithDelay(50 * time.Millisecond)
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
@ -160,7 +160,7 @@ func TestIntegrationQuotaEnforcement(t *testing.T) {
var maxConcurrent int32
var currentConcurrent int32
exec := executor.NewWithCallback(200*time.Millisecond,
exec := executor.NewDryRunWithCallbacks(200*time.Millisecond,
func() {
curr := atomic.AddInt32(&currentConcurrent, 1)
for {
@ -210,7 +210,7 @@ func TestIntegrationQuotaEnforcement(t *testing.T) {
// Create robot with Max=1, Queue=3
setupConcurrentTestRobot(t, "robot_integ_quota_queue", "team_integ_quota", 1, 3)
exec := executor.NewWithDelay(300 * time.Millisecond) // Slow execution
exec := executor.NewDryRunWithDelay(300 * time.Millisecond) // Slow execution
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
@ -337,7 +337,7 @@ func TestIntegrationGlobalPoolLimit(t *testing.T) {
var maxConcurrent int32
var currentConcurrent int32
exec := executor.NewWithCallback(200*time.Millisecond,
exec := executor.NewDryRunWithCallbacks(200*time.Millisecond,
func() {
curr := atomic.AddInt32(&currentConcurrent, 1)
for {
@ -391,7 +391,7 @@ func TestIntegrationGlobalPoolLimit(t *testing.T) {
setupConcurrentTestRobot(t, memberID, "team_integ_pool", 5, 20)
}
exec := executor.NewWithDelay(500 * time.Millisecond) // Slow execution
exec := executor.NewDryRunWithDelay(500 * time.Millisecond) // Slow execution
config := &manager.Config{
TickInterval: 100 * time.Millisecond,

View file

@ -233,22 +233,21 @@ func TestIntegrationPhaseProgression(t *testing.T) {
// Track phases executed
phasesExecuted := make([]types.Phase, 0)
exec := executor.NewWithConfig(executor.Config{
SkipJobIntegration: true,
OnPhaseStart: func(phase types.Phase) {
phasesExecuted = append(phasesExecuted, phase)
exec := executor.NewDryRunWithConfig(executor.DryRunConfig{
Config: executor.Config{
OnPhaseStart: func(phase types.Phase) {
phasesExecuted = append(phasesExecuted, phase)
},
},
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 2, QueueSize: 20},
Executor: exec,
}
m := manager.NewWithConfig(config)
// Replace executor
m.Pool().SetExecutor(exec)
err := m.Start()
require.NoError(t, err)
defer m.Stop()
@ -272,22 +271,21 @@ func TestIntegrationPhaseProgression(t *testing.T) {
// Track phases executed
phasesExecuted := make([]types.Phase, 0)
exec := executor.NewWithConfig(executor.Config{
SkipJobIntegration: true,
OnPhaseStart: func(phase types.Phase) {
phasesExecuted = append(phasesExecuted, phase)
exec := executor.NewDryRunWithConfig(executor.DryRunConfig{
Config: executor.Config{
OnPhaseStart: func(phase types.Phase) {
phasesExecuted = append(phasesExecuted, phase)
},
},
})
config := &manager.Config{
TickInterval: 100 * time.Millisecond,
PoolConfig: &pool.Config{WorkerSize: 2, QueueSize: 20},
Executor: exec,
}
m := manager.NewWithConfig(config)
// Replace executor
m.Pool().SetExecutor(exec)
err := m.Start()
require.NoError(t, err)
defer m.Stop()

View file

@ -20,8 +20,9 @@ const (
// Config holds manager configuration
type Config struct {
TickInterval time.Duration // how often to check clock triggers (default: 1 minute)
PoolConfig *pool.Config // worker pool configuration
TickInterval time.Duration // how often to check clock triggers (default: 1 minute)
PoolConfig *pool.Config // worker pool configuration
Executor types.Executor // optional: custom executor (default: real executor)
}
// DefaultConfig returns default manager configuration
@ -38,7 +39,7 @@ type Manager struct {
config *Config
cache *cache.Cache
pool *pool.Pool
executor *executor.Executor
executor types.Executor
// Execution control for pause/resume/stop
execController *trigger.ExecutionController
@ -75,12 +76,37 @@ func NewWithConfig(config *Config) *Manager {
// Create components
c := cache.New()
p := pool.NewWithConfig(config.PoolConfig)
e := executor.New()
ec := trigger.NewExecutionController()
// Use custom executor if provided, otherwise create default
var e types.Executor
if config.Executor != nil {
e = config.Executor
} else {
e = executor.New()
}
// Wire up pool with executor
p.SetExecutor(e)
// Create shared executor instances for each mode
// These are reused across all executions to maintain accurate counters
dryRunExecutor := executor.NewDryRun()
// Set executor factory for mode-specific executors
p.SetExecutorFactory(func(mode types.ExecutorMode) types.Executor {
switch mode {
case types.ExecutorDryRun:
return dryRunExecutor
case types.ExecutorSandbox:
// Sandbox not implemented, fall back to DryRun
return dryRunExecutor
default:
// Standard mode or empty - use the configured executor
return e
}
})
return &Manager{
config: config,
cache: c,
@ -425,8 +451,11 @@ func (m *Manager) Intervene(ctx *types.Context, req *types.InterveneRequest) (*t
}, nil
}
// Submit to pool
execID, err := m.pool.Submit(ctx, robot, types.TriggerHuman, triggerInput)
// Determine executor mode: request > robot config > default
executorMode := m.resolveExecutorMode(req.ExecutorMode, robot)
// Submit to pool with executor mode
execID, err := m.pool.SubmitWithMode(ctx, robot, types.TriggerHuman, triggerInput, executorMode)
if err != nil {
return nil, err
}
@ -477,8 +506,11 @@ func (m *Manager) HandleEvent(ctx *types.Context, req *types.EventRequest) (*typ
// Build trigger input
triggerInput := trigger.BuildEventInput(req)
// Submit to pool
execID, err := m.pool.Submit(ctx, robot, types.TriggerEvent, triggerInput)
// Determine executor mode: request > robot config > default
executorMode := m.resolveExecutorMode(req.ExecutorMode, robot)
// Submit to pool with executor mode
execID, err := m.pool.SubmitWithMode(ctx, robot, types.TriggerEvent, triggerInput, executorMode)
if err != nil {
return nil, err
}
@ -529,6 +561,25 @@ func (m *Manager) ListExecutionsByMember(memberID string) []*trigger.ControlledE
return m.execController.ListByMember(memberID)
}
// ==================== Helper Methods ====================
// resolveExecutorMode determines the executor mode to use
// Priority: request > robot config > default (standard)
func (m *Manager) resolveExecutorMode(requestMode types.ExecutorMode, robot *types.Robot) types.ExecutorMode {
// Request mode takes precedence
if requestMode != "" && requestMode.IsValid() {
return requestMode
}
// Robot config mode
if robot != nil && robot.Config != nil && robot.Config.Executor != nil {
return robot.Config.Executor.GetMode()
}
// Default: standard
return types.ExecutorStandard
}
// ==================== Getters for internal components ====================
// These are exposed for testing and advanced use cases
@ -543,7 +594,7 @@ func (m *Manager) Pool() *pool.Pool {
}
// Executor returns the internal executor
func (m *Manager) Executor() *executor.Executor {
func (m *Manager) Executor() types.Executor {
return m.executor
}

View file

@ -42,7 +42,7 @@ func TestPoolNoGoroutineLeak(t *testing.T) {
baseline := getGoroutineCount()
// Create and start pool
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 5,
QueueSize: 100,
@ -81,7 +81,7 @@ func TestPoolMultipleStartStop(t *testing.T) {
time.Sleep(50 * time.Millisecond)
baseline := getGoroutineCount()
exec := executor.NewWithDelay(5 * time.Millisecond)
exec := executor.NewDryRunWithDelay(5 * time.Millisecond)
for i := 0; i < 5; i++ {
p := pool.NewWithConfig(&pool.Config{
@ -139,7 +139,7 @@ func TestPoolStopWithPendingJobs(t *testing.T) {
baseline := getGoroutineCount()
// Use slow executor so jobs stay in queue
exec := executor.NewWithDelay(500 * time.Millisecond)
exec := executor.NewDryRunWithDelay(500 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1, // only 1 worker
QueueSize: 100,
@ -170,7 +170,7 @@ func TestPoolConcurrentStartStop(t *testing.T) {
time.Sleep(50 * time.Millisecond)
baseline := getGoroutineCount()
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 5,
QueueSize: 100,
@ -223,7 +223,7 @@ func TestWorkerGoroutinesCleanup(t *testing.T) {
time.Sleep(50 * time.Millisecond)
baseline := getGoroutineCount()
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
// Create pool with many workers
p := pool.NewWithConfig(&pool.Config{
@ -253,7 +253,7 @@ func TestPoolLongRunningJobsNoLeak(t *testing.T) {
time.Sleep(50 * time.Millisecond)
baseline := getGoroutineCount()
exec := executor.NewWithDelay(200 * time.Millisecond)
exec := executor.NewDryRunWithDelay(200 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 3,
QueueSize: 100,

View file

@ -28,17 +28,21 @@ func DefaultConfig() *Config {
}
}
// ExecutorFactory creates an executor based on the mode
type ExecutorFactory func(mode types.ExecutorMode) types.Executor
// Pool implements types.Pool interface
// Manages a pool of workers that execute robot jobs from a priority queue
type Pool struct {
size int // number of workers
queue *PriorityQueue // priority queue for pending jobs
executor types.Executor // executor for running jobs
workers []*Worker // worker goroutines
running atomic.Int32 // number of currently running jobs
wg sync.WaitGroup // wait group for graceful shutdown
started bool // whether pool has been started
mu sync.RWMutex // protects started flag
size int // number of workers
queue *PriorityQueue // priority queue for pending jobs
executor types.Executor // default executor for running jobs
executorFactory ExecutorFactory // optional: factory for mode-specific executors
workers []*Worker // worker goroutines
running atomic.Int32 // number of currently running jobs
wg sync.WaitGroup // wait group for graceful shutdown
started bool // whether pool has been started
mu sync.RWMutex // protects started flag
}
// New creates a new pool instance with default configuration
@ -69,12 +73,29 @@ func NewWithConfig(config *Config) *Pool {
}
}
// SetExecutor sets the executor for the pool
// SetExecutor sets the default executor for the pool
// Must be called before Start()
func (p *Pool) SetExecutor(executor types.Executor) {
p.executor = executor
}
// SetExecutorFactory sets the executor factory for mode-specific executors
// If set, the factory is used to create executors based on ExecutorMode
func (p *Pool) SetExecutorFactory(factory ExecutorFactory) {
p.executorFactory = factory
}
// GetExecutor returns the appropriate executor for the given mode
// If factory is set and mode is specified, uses factory; otherwise uses default
func (p *Pool) GetExecutor(mode types.ExecutorMode) types.Executor {
// If factory is set and mode is specified, use factory
if p.executorFactory != nil && mode != "" {
return p.executorFactory(mode)
}
// Otherwise use default executor
return p.executor
}
// Start starts the worker pool
func (p *Pool) Start() error {
p.mu.Lock()
@ -91,7 +112,7 @@ func (p *Pool) Start() error {
// Create and start workers
p.workers = make([]*Worker, p.size)
for i := 0; i < p.size; i++ {
worker := newWorker(i+1, p, p.executor, &p.wg)
worker := newWorker(i+1, p, &p.wg)
p.workers[i] = worker
worker.start()
}
@ -125,6 +146,13 @@ func (p *Pool) Stop() error {
// Submit submits a robot execution to the pool
// Returns execution ID if successfully queued, error otherwise
func (p *Pool) Submit(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}) (string, error) {
return p.SubmitWithMode(ctx, robot, trigger, data, "")
}
// SubmitWithMode submits a robot execution with specified executor mode
// executorMode: optional, overrides robot's config if provided
// Returns execution ID if successfully queued, error otherwise
func (p *Pool) SubmitWithMode(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}, executorMode types.ExecutorMode) (string, error) {
p.mu.RLock()
if !p.started {
p.mu.RUnlock()
@ -138,10 +166,11 @@ func (p *Pool) Submit(ctx *types.Context, robot *types.Robot, trigger types.Trig
// Create queue item
item := &QueueItem{
Robot: robot,
Ctx: ctx,
Trigger: trigger,
Data: data,
Robot: robot,
Ctx: ctx,
Trigger: trigger,
Data: data,
ExecutorMode: executorMode,
}
// Try to add to queue

View file

@ -95,7 +95,7 @@ func TestPoolSubmitNilRobot(t *testing.T) {
// TestPoolBasicExecution tests basic job execution
func TestPoolBasicExecution(t *testing.T) {
exec := executor.NewWithDelay(50 * time.Millisecond)
exec := executor.NewDryRunWithDelay(50 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 5,
QueueSize: 100,
@ -125,7 +125,7 @@ func TestPoolBasicExecution(t *testing.T) {
// TestPoolConcurrencyLimit tests global worker limit
func TestPoolConcurrencyLimit(t *testing.T) {
exec := executor.NewWithDelay(200 * time.Millisecond) // longer delay
exec := executor.NewDryRunWithDelay(200 * time.Millisecond) // longer delay
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 3, // only 3 workers
QueueSize: 100,
@ -170,7 +170,7 @@ func TestPoolConcurrencyLimit(t *testing.T) {
// TestRobotConcurrencyLimit tests per-robot concurrent execution limit
func TestRobotConcurrencyLimit(t *testing.T) {
exec := executor.NewWithDelay(100 * time.Millisecond)
exec := executor.NewDryRunWithDelay(100 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 10, // plenty of workers
QueueSize: 100,
@ -207,7 +207,7 @@ func TestRobotConcurrencyLimit(t *testing.T) {
// TestRobotQueueLimit tests per-robot queue limit
func TestRobotQueueLimit(t *testing.T) {
exec := executor.NewWithDelay(200 * time.Millisecond)
exec := executor.NewDryRunWithDelay(200 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 2,
QueueSize: 100, // global queue is large
@ -238,7 +238,7 @@ func TestRobotQueueLimit(t *testing.T) {
// TestGlobalQueueLimit tests global queue limit
func TestGlobalQueueLimit(t *testing.T) {
exec := executor.NewWithDelay(500 * time.Millisecond) // slow execution
exec := executor.NewDryRunWithDelay(500 * time.Millisecond) // slow execution
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1, // only 1 worker
QueueSize: 5, // small global queue
@ -272,7 +272,7 @@ func TestGlobalQueueLimit(t *testing.T) {
// TestPriorityOrder tests that higher priority jobs execute first
func TestPriorityOrder(t *testing.T) {
exec := executor.NewWithDelay(50 * time.Millisecond)
exec := executor.NewDryRunWithDelay(50 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1, // single worker to ensure order
QueueSize: 100,
@ -302,7 +302,7 @@ func TestPriorityOrder(t *testing.T) {
// TestTriggerTypePriority tests that human triggers have higher priority than clock
func TestTriggerTypePriority(t *testing.T) {
exec := executor.NewWithDelay(50 * time.Millisecond)
exec := executor.NewDryRunWithDelay(50 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1, // single worker
QueueSize: 100,
@ -328,7 +328,7 @@ func TestTriggerTypePriority(t *testing.T) {
// TestMultipleRobotsFairness tests that multiple robots get fair access
func TestMultipleRobotsFairness(t *testing.T) {
exec := executor.NewWithDelay(30 * time.Millisecond)
exec := executor.NewDryRunWithDelay(30 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 5,
QueueSize: 100,
@ -361,7 +361,7 @@ func TestMultipleRobotsFairness(t *testing.T) {
// TestGracefulShutdown tests that pool waits for running jobs on shutdown
func TestGracefulShutdown(t *testing.T) {
exec := executor.NewWithDelay(200 * time.Millisecond)
exec := executor.NewDryRunWithDelay(200 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 2,
QueueSize: 10,

View file

@ -10,13 +10,14 @@ import (
// QueueItem represents a job waiting in the queue
type QueueItem struct {
Robot *types.Robot
Ctx *types.Context
Trigger types.TriggerType
Data interface{}
EnqueueTime time.Time
Priority int // calculated priority for sorting
Index int // index in heap (managed by container/heap)
Robot *types.Robot
Ctx *types.Context
Trigger types.TriggerType
Data interface{}
ExecutorMode types.ExecutorMode // optional: override robot's executor mode
EnqueueTime time.Time
Priority int // calculated priority for sorting
Index int // index in heap (managed by container/heap)
}
// PriorityQueue implements a priority queue for robot executions

View file

@ -12,17 +12,15 @@ import (
type Worker struct {
id int
pool *Pool
executor types.Executor
stopChan chan struct{}
wg *sync.WaitGroup
}
// newWorker creates a new worker
func newWorker(id int, pool *Pool, executor types.Executor, wg *sync.WaitGroup) *Worker {
func newWorker(id int, pool *Pool, wg *sync.WaitGroup) *Worker {
return &Worker{
id: id,
pool: pool,
executor: executor,
stopChan: make(chan struct{}),
wg: wg,
}
@ -78,9 +76,12 @@ func (w *Worker) execute(item *QueueItem) {
w.pool.incrementRunning()
defer w.pool.decrementRunning()
// Get executor based on mode (uses factory if available, otherwise default)
exec := w.pool.GetExecutor(item.ExecutorMode)
// Execute via Executor interface
// Note: Executor.Execute() does atomic quota check via TryAcquireSlot()
execution, err := w.executor.Execute(item.Ctx, item.Robot, item.Trigger, item.Data)
execution, err := exec.Execute(item.Ctx, item.Robot, item.Trigger, item.Data)
if err != nil {
// Check if it's a quota error (race condition - another worker got the slot)

View file

@ -16,7 +16,7 @@ import (
// TestWorkerExecutesJob tests that worker executes a job from queue
func TestWorkerExecutesJob(t *testing.T) {
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,
@ -39,7 +39,7 @@ func TestWorkerExecutesJob(t *testing.T) {
// TestWorkerMultipleJobs tests worker processes multiple jobs sequentially
func TestWorkerMultipleJobs(t *testing.T) {
exec := executor.NewWithDelay(20 * time.Millisecond)
exec := executor.NewDryRunWithDelay(20 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1, // single worker
QueueSize: 10,
@ -68,7 +68,7 @@ func TestWorkerMultipleJobs(t *testing.T) {
// TestWorkerRespectsRobotQuota tests worker re-enqueues when robot quota is full
func TestWorkerRespectsRobotQuota(t *testing.T) {
// This test verifies that all jobs eventually complete even when robot quota limits concurrency
exec := executor.NewWithDelay(100 * time.Millisecond)
exec := executor.NewDryRunWithDelay(100 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 5, // multiple workers
@ -98,7 +98,7 @@ func TestWorkerRespectsRobotQuota(t *testing.T) {
// TestWorkerReenqueueOnQuotaFull tests that jobs are re-enqueued when quota is full
func TestWorkerReenqueueOnQuotaFull(t *testing.T) {
exec := executor.NewWithDelay(100 * time.Millisecond)
exec := executor.NewDryRunWithDelay(100 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 3,
QueueSize: 100,
@ -132,7 +132,7 @@ func TestWorkersConcurrentExecution(t *testing.T) {
var maxConcurrent int32
var currentConcurrent int32
exec := executor.NewWithCallback(100*time.Millisecond, func() {
exec := executor.NewDryRunWithCallbacks(100*time.Millisecond, func() {
current := atomic.AddInt32(&currentConcurrent, 1)
// Update max if current is higher
for {
@ -174,7 +174,7 @@ func TestWorkersDoNotExceedPoolSize(t *testing.T) {
var currentConcurrent int32
var mu sync.Mutex
exec := executor.NewWithCallback(50*time.Millisecond, func() {
exec := executor.NewDryRunWithCallbacks(50*time.Millisecond, func() {
mu.Lock()
currentConcurrent++
if currentConcurrent > maxConcurrent {
@ -214,7 +214,7 @@ func TestWorkersDoNotExceedPoolSize(t *testing.T) {
// TestWorkerStopsGracefully tests worker stops when signaled
func TestWorkerStopsGracefully(t *testing.T) {
exec := executor.NewWithDelay(50 * time.Millisecond)
exec := executor.NewDryRunWithDelay(50 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 2,
QueueSize: 10,
@ -242,7 +242,7 @@ func TestWorkerStopsGracefully(t *testing.T) {
// TestWorkerCompletesCurrentJobOnStop tests worker completes current job before stopping
func TestWorkerCompletesCurrentJobOnStop(t *testing.T) {
exec := executor.NewWithDelay(100 * time.Millisecond)
exec := executor.NewDryRunWithDelay(100 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,
@ -270,7 +270,7 @@ func TestWorkerCompletesCurrentJobOnStop(t *testing.T) {
// TestWorkerHandlesExecutorError tests worker continues after executor error
func TestWorkerHandlesExecutorError(t *testing.T) {
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,
@ -299,7 +299,7 @@ func TestWorkerHandlesExecutorError(t *testing.T) {
// TestWorkerRunningCounterAccurate tests running counter is accurate
func TestWorkerRunningCounterAccurate(t *testing.T) {
exec := executor.NewWithDelay(100 * time.Millisecond)
exec := executor.NewDryRunWithDelay(100 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 3,
QueueSize: 10,
@ -330,7 +330,7 @@ func TestWorkerRunningCounterAccurate(t *testing.T) {
// TestWorkerRunningCounterDecrementsOnError tests running counter decrements on error
func TestWorkerRunningCounterDecrementsOnError(t *testing.T) {
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,
@ -356,7 +356,7 @@ func TestWorkerRunningCounterDecrementsOnError(t *testing.T) {
// TestWorkerProcessesDifferentTriggers tests worker handles all trigger types
func TestWorkerProcessesDifferentTriggers(t *testing.T) {
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,
@ -384,7 +384,7 @@ func TestWorkerProcessesDifferentTriggers(t *testing.T) {
// TestWorkerPollsQueuePeriodically tests worker polls queue at regular intervals
func TestWorkerPollsQueuePeriodically(t *testing.T) {
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,
@ -408,7 +408,7 @@ func TestWorkerPollsQueuePeriodically(t *testing.T) {
// TestWorkerContinuesAfterEmptyQueue tests worker continues polling after empty queue
func TestWorkerContinuesAfterEmptyQueue(t *testing.T) {
exec := executor.NewWithDelay(10 * time.Millisecond)
exec := executor.NewDryRunWithDelay(10 * time.Millisecond)
p := pool.NewWithConfig(&pool.Config{
WorkerSize: 1,
QueueSize: 10,

View file

@ -17,7 +17,7 @@ var (
globalPool *pool.Pool
globalDedup *dedup.Dedup
globalStore *store.Store
globalExecutor *executor.Executor
globalExecutor executor.Executor
globalPlan *plan.Plan
)

View file

@ -7,16 +7,43 @@ import (
// Config - robot_config in __yao.member
type Config struct {
Triggers *Triggers `json:"triggers,omitempty"`
Clock *Clock `json:"clock,omitempty"`
Identity *Identity `json:"identity"`
Quota *Quota `json:"quota,omitempty"`
KB *KB `json:"kb,omitempty"` // shared knowledge base (same as assistant)
DB *DB `json:"db,omitempty"` // shared database (same as assistant)
Learn *Learn `json:"learn,omitempty"` // learning config for private KB
Resources *Resources `json:"resources,omitempty"`
Delivery *Delivery `json:"delivery,omitempty"`
Events []Event `json:"events,omitempty"`
Triggers *Triggers `json:"triggers,omitempty"`
Clock *Clock `json:"clock,omitempty"`
Identity *Identity `json:"identity"`
Quota *Quota `json:"quota,omitempty"`
KB *KB `json:"kb,omitempty"` // shared knowledge base (same as assistant)
DB *DB `json:"db,omitempty"` // shared database (same as assistant)
Learn *Learn `json:"learn,omitempty"` // learning config for private KB
Resources *Resources `json:"resources,omitempty"`
Delivery *Delivery `json:"delivery,omitempty"`
Events []Event `json:"events,omitempty"`
Executor *ExecutorConfig `json:"executor,omitempty"` // executor mode settings
}
// ExecutorConfig - executor settings
type ExecutorConfig struct {
Mode ExecutorMode `json:"mode,omitempty"` // standard | dryrun | sandbox
MaxDuration string `json:"max_duration,omitempty"` // max execution time (e.g., "30m")
}
// GetMode returns the executor mode (default: standard)
func (e *ExecutorConfig) GetMode() ExecutorMode {
if e == nil || e.Mode == "" {
return ExecutorStandard
}
return e.Mode
}
// GetMaxDuration returns the max duration (default: 30m)
func (e *ExecutorConfig) GetMaxDuration() time.Duration {
if e == nil || e.MaxDuration == "" {
return 30 * time.Minute
}
d, err := time.ParseDuration(e.MaxDuration)
if err != nil {
return 30 * time.Minute
}
return d
}
// Validate validates the config

View file

@ -250,3 +250,67 @@ func TestResourcesGetPhaseAgent(t *testing.T) {
assert.Equal(t, "__yao.learning", resources.GetPhaseAgent(types.PhaseLearning))
})
}
func TestExecutorConfigGetMode(t *testing.T) {
t.Run("nil config - returns default", func(t *testing.T) {
var config *types.ExecutorConfig
assert.Equal(t, types.ExecutorStandard, config.GetMode())
})
t.Run("empty mode - returns default", func(t *testing.T) {
config := &types.ExecutorConfig{}
assert.Equal(t, types.ExecutorStandard, config.GetMode())
})
t.Run("standard mode", func(t *testing.T) {
config := &types.ExecutorConfig{Mode: types.ExecutorStandard}
assert.Equal(t, types.ExecutorStandard, config.GetMode())
})
t.Run("dryrun mode", func(t *testing.T) {
config := &types.ExecutorConfig{Mode: types.ExecutorDryRun}
assert.Equal(t, types.ExecutorDryRun, config.GetMode())
})
t.Run("sandbox mode", func(t *testing.T) {
config := &types.ExecutorConfig{Mode: types.ExecutorSandbox}
assert.Equal(t, types.ExecutorSandbox, config.GetMode())
})
}
func TestExecutorConfigGetMaxDuration(t *testing.T) {
t.Run("nil config - returns default 30m", func(t *testing.T) {
var config *types.ExecutorConfig
assert.Equal(t, 30*time.Minute, config.GetMaxDuration())
})
t.Run("empty duration - returns default 30m", func(t *testing.T) {
config := &types.ExecutorConfig{}
assert.Equal(t, 30*time.Minute, config.GetMaxDuration())
})
t.Run("custom duration", func(t *testing.T) {
config := &types.ExecutorConfig{MaxDuration: "10m"}
assert.Equal(t, 10*time.Minute, config.GetMaxDuration())
})
t.Run("invalid duration - returns default", func(t *testing.T) {
config := &types.ExecutorConfig{MaxDuration: "invalid"}
assert.Equal(t, 30*time.Minute, config.GetMaxDuration())
})
t.Run("various valid durations", func(t *testing.T) {
tests := []struct {
input string
expected time.Duration
}{
{"1h", time.Hour},
{"30s", 30 * time.Second},
{"2h30m", 2*time.Hour + 30*time.Minute},
}
for _, tt := range tests {
config := &types.ExecutorConfig{MaxDuration: tt.input}
assert.Equal(t, tt.expected, config.GetMaxDuration(), "for input %s", tt.input)
}
})
}

View file

@ -189,3 +189,34 @@ const (
InsertNext InsertPosition = "next" // insert after current task
InsertAt InsertPosition = "at" // insert at specific index (use AtIndex)
)
// ExecutorMode - executor mode for robot execution
type ExecutorMode string
// ExecutorMode constants define the executor modes
const (
// ExecutorStandard uses real Agent calls (production mode)
ExecutorStandard ExecutorMode = "standard"
// ExecutorDryRun simulates execution without LLM calls (testing/demo)
ExecutorDryRun ExecutorMode = "dryrun"
// ExecutorSandbox runs in container-isolated environment (NOT IMPLEMENTED)
// Requires Docker/gVisor/Firecracker infrastructure
ExecutorSandbox ExecutorMode = "sandbox"
)
// IsValid checks if the executor mode is valid
func (m ExecutorMode) IsValid() bool {
switch m {
case ExecutorStandard, ExecutorDryRun, ExecutorSandbox, "":
return true
}
return false
}
// GetDefault returns the default executor mode if empty
func (m ExecutorMode) GetDefault() ExecutorMode {
if m == "" {
return ExecutorStandard
}
return m
}

View file

@ -132,3 +132,47 @@ func TestInsertPositionEnum(t *testing.T) {
assert.Equal(t, types.InsertPosition("next"), types.InsertNext)
assert.Equal(t, types.InsertPosition("at"), types.InsertAt)
}
func TestExecutorModeEnum(t *testing.T) {
assert.Equal(t, types.ExecutorMode("standard"), types.ExecutorStandard)
assert.Equal(t, types.ExecutorMode("dryrun"), types.ExecutorDryRun)
assert.Equal(t, types.ExecutorMode("sandbox"), types.ExecutorSandbox)
}
func TestExecutorModeIsValid(t *testing.T) {
tests := []struct {
mode types.ExecutorMode
valid bool
}{
{types.ExecutorStandard, true},
{types.ExecutorDryRun, true},
{types.ExecutorSandbox, true},
{"", true}, // empty is valid (defaults to standard)
{types.ExecutorMode("invalid"), false},
{types.ExecutorMode("unknown"), false},
}
for _, tt := range tests {
t.Run(string(tt.mode), func(t *testing.T) {
assert.Equal(t, tt.valid, tt.mode.IsValid())
})
}
}
func TestExecutorModeGetDefault(t *testing.T) {
tests := []struct {
mode types.ExecutorMode
expected types.ExecutorMode
}{
{"", types.ExecutorStandard},
{types.ExecutorStandard, types.ExecutorStandard},
{types.ExecutorDryRun, types.ExecutorDryRun},
{types.ExecutorSandbox, types.ExecutorSandbox},
}
for _, tt := range tests {
t.Run(string(tt.mode), func(t *testing.T) {
assert.Equal(t, tt.expected, tt.mode.GetDefault())
})
}
}

View file

@ -17,6 +17,11 @@ type Manager interface {
// Executor - executes robot phases
type Executor interface {
Execute(ctx *Context, robot *Robot, trigger TriggerType, data interface{}) (*Execution, error)
// Metrics and control (for monitoring and testing)
ExecCount() int // total execution count
CurrentCount() int // currently running count
Reset() // reset counters
}
// Pool - worker pool for concurrent execution

View file

@ -8,19 +8,21 @@ import (
// InterveneRequest - human intervention request
type InterveneRequest struct {
TeamID string `json:"team_id"`
MemberID string `json:"member_id"`
Action InterventionAction `json:"action"`
Messages []agentcontext.Message `json:"messages"` // user input (text, images, files)
PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan
TeamID string `json:"team_id"`
MemberID string `json:"member_id"`
Action InterventionAction `json:"action"`
Messages []agentcontext.Message `json:"messages"` // user input (text, images, files)
PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan
ExecutorMode ExecutorMode `json:"executor_mode,omitempty"` // optional: override robot config
}
// EventRequest - event trigger request
type EventRequest struct {
MemberID string `json:"member_id"`
Source string `json:"source"` // webhook path or table name
EventType string `json:"event_type"` // lead.created, etc.
Data map[string]interface{} `json:"data"`
MemberID string `json:"member_id"`
Source string `json:"source"` // webhook path or table name
EventType string `json:"event_type"` // lead.created, etc.
Data map[string]interface{} `json:"data"`
ExecutorMode ExecutorMode `json:"executor_mode,omitempty"` // optional: override robot config
}
// ExecutionResult - trigger result

View file

@ -149,6 +149,16 @@ type Execution struct {
robot *Robot `json:"-"`
}
// GetRobot returns the robot associated with this execution
func (e *Execution) GetRobot() *Robot {
return e.robot
}
// SetRobot sets the robot associated with this execution
func (e *Execution) SetRobot(robot *Robot) {
e.robot = robot
}
// TriggerInput - stored trigger input for traceability
type TriggerInput struct {
// For human intervention