Task Orchestrator v2.0 - Unified Task Orchestration Hub

Core Mission: Enable agents to learn fundamental task creation and execution capabilities, including task identification, standardized setup, and execution tracking

v2.0 New Features: Dynamic configuration loading, deadlock prevention mechanism, intelligent aggregation rules, 4-level risk confirmation boundaries

🎯 Core Capabilities

1. Task Type Identification (Enhanced)

Automatically identify user request task types and route to correct execution mode:

2. Task Identification Flowchart (Enhanced)

User Request
    ↓
[Task Analysis]
    ├── Has exact time expression? → Cron Task
    ├── Has "heartbeat/regular"? → Heartbeat Task
    ├── Has "multi-step/complex"? → Subagent Task
    ├── Has "continuous/track"? → Task Record
    └── No special markers → Immediate Task
    ↓
[6-Point Checklist]
    ├── Task objective clear?
    ├── Constraints defined?
    ├── Complexity assessment
    ├── Question points identified
    ├── Resource requirements confirmed
    └── Risk points assessed
    ↓
[⚡ Clarification Confirmation] → Has questions/ambiguity/conflict → Generate clarification questions → User confirms
    ↓
[Show Task Plan] → User confirms
    ↓
[Execute] → Create/Dispatch/Record
    ↓
[Track] → Update status → Report results

📋 Task Identification Rules (Enhanced)

Rule 1: Cron Task Identification

Trigger Keywords:

• Time expressions: "daily", "weekly", "monthly", "at X o'clock", "in X minutes"
• Scheduling: "schedule", "remind", "alarm", "cron"
• Cron expression format: "0 9 * * *"

Judgment Logic:

def is_cron_task(user_input):
    # Check time expressions
    time_patterns = [
        r'daily\s*at\s*\d+',           # daily at 9
        r'every\s*(mon|tue|wed|thu|fri|sat|sun)',  # every Monday
        r'\d+\s*(minute|hour|day|week)s?\s*(later|from\s*now)',  # 10 minutes later
        r'in\s+\d+\s*(minute|hour|day|week)s?',  # in 10 minutes
        r'\d+\s+[*\d\s]+\d+',    # Cron expression
    ]
    
    # Check keywords
    cron_keywords = ['schedule', 'remind', 'alarm', 'cron', 'timer']
    
    has_time_expr = any(re.search(p, user_input, re.IGNORECASE) for p in time_patterns)
    has_keyword = any(k in user_input.lower() for k in cron_keywords)
    
    return has_time_expr or has_keyword

Examples:

• ✅ "Push news every day at 9am" → Cron
• ✅ "Remind me to meet in 10 minutes" → Cron (one-time)
• ✅ "Weekly meeting every Monday at 10am" → Cron (periodic)

Rule 2: Heartbeat Task Identification

Trigger Keywords:

• "heartbeat", "regular", "every check", "periodic"
• "monitor", "polling", "cycle"
• "keep online", "auto check"

Judgment Logic:

def is_heartbeat_task(user_input):
    # First exclude Cron tasks (have exact time)
    if is_cron_task(user_input):
        return False
    
    heartbeat_keywords = [
        'heartbeat', 'regular', 'every', 'periodic',
        'monitor', 'polling', 'cycle',
        'keep online', 'auto check', 'background'
    ]
    
    return any(k in user_input.lower() for k in heartbeat_keywords)

Examples:

• ✅ "Check unread email every heartbeat" → Heartbeat
• ✅ "Regularly check API status" → Heartbeat
• ❌ "Check at 9am daily" → Cron (has exact time)

Rule 3: Subagent Task Identification (Enhanced)

Trigger Keywords:

• "multi-step", "break down", "step by step"
• "complex", "research", "analyze"
• "parallel", "multiple agents", "collaborate"

Complexity Scoring (Enhanced):

def calculate_complexity(user_input):
    score = 0
    
    # Action count (0-4 points)
    actions = count_actions(user_input)
    score += min(actions, 4)
    
    # Dependency relations (0-3 points)
    dependency_keywords = ['based on', 'using', 'then', 'after', 'step']
    if any(k in user_input.lower() for k in dependency_keywords):
        score += 2
        if 'step' in user_input.lower() or 'step-by-step' in user_input:
            score += 1
    
    # Scope size (0-3 points)
    scope_keywords = ['entire', 'complete', 'all', 'full', 'whole']
    if any(k in user_input.lower() for k in scope_keywords):
        score += 2
    
    return min(score, 10)

def is_subagent_task(user_input):
    # Action count >= 3 or complexity >= 5
    return count_actions(user_input) >= 3 or calculate_complexity(user_input) >= 5

Complexity Levels:

Examples:

• ✅ "Research competitors and write analysis report" → Subagent (complexity 6)
• ✅ "Refactor this project, including tests and docs" → Subagent (complexity 8)
• ❌ "Check weather" → Immediate (complexity 1)

Rule 4: Task Record Identification

Trigger Keywords:

• "continuous", "track", "tracking"
• "long-term", "ongoing", "project"
• "record to task system"

Examples:

• ✅ "Continuously monitor this stock trend" → Task Record
• ✅ "Track this project progress" → Task Record

🏗️ Standardized Setup Process (Enhanced)

Process 1: Cron Task Setup (Enhanced)

Step 1: Requirements Analysis (6-Point Checklist)

Checklist:

1. Task Objective - What needs to be achieved?
2. Constraints - Time, scope, quality requirements?
3. Complexity - Simple/Normal/Complex?
4. Question Points - Information gaps need confirmation?
5. Resource Requirements - Which agents/tools needed?
6. Risk Points - Potential issues?

Step 2: Risk Classification (4 Levels)

Step 3: Generate Task Plan

Output Template:

## ⏰ Cron Task Plan

### Task Information
- **Task Type**: {One-time | Periodic}
- **Task Name**: {Short name}
- **Execution Time**: {ISO time or Cron expression}
- **Timezone**: Asia/Shanghai
- **Risk Level**: {🟢/🟡/🔴/⚫}

### Execution Content
{Specific task description to execute}

### Push Configuration
- **Channel**: Feishu
- **Recipient**: Boss (ou_xxx)

### Confirmation
Please confirm and I'll create this scheduled task ✓

Step 4: Create Task

Standard Command Format:

# One-time task
openclaw cron add \
  --name "{Task Name}" \
  --at "{ISO time or relative time}" \
  --session isolated \
  --channel feishu \
  --to "user:ou_642f1cb74d63462d1037375caac5ea2d" \
  --message "{Execution content}" \
  --tz "Asia/Shanghai" \
  --description "{Task description}"

# Periodic task
openclaw cron add \
  --name "{Task Name}" \
  --cron "{Cron expression}" \
  --tz "Asia/Shanghai" \
  --session isolated \
  --channel feishu \
  --to "user:ou_642f1cb74d63462d1037375caac5ea2d" \
  --message "{Execution content}" \
  --description "{Task description}"

Step 5: Verify and Track

# Verify task created
openclaw cron list

# Test task (optional)
openclaw cron run {task_id}

# Record to task system
Update TASKS.md, mark task type=Cron

Process 2: Heartbeat Task Setup (Enhanced)

Step 1: Requirements Analysis

Checklist:

• Monitoring objective clear
• Check frequency reasonable (recommend 30-60 minutes)
• Output format defined (HEARTBEAT_OK or specific content)
• Cost assessment (involves API calls?)
• Quiet hours defined (default 22:00-08:00)

Step 2: Select Heartbeat Mode

3 Modes:

Step 3: Generate HEARTBEAT.md Template

Standard Template (Enhanced):

# HEARTBEAT.md

## Core Task: {Task Name}

## Heartbeat Mode
- **Mode**: {Conservative | Balanced | Aggressive}
- **Default Interval**: {N} minutes
- **Quiet Hours**: 22:00-08:00

## Scheduled Task: {Task Description}

Execute following checks every heartbeat:

### Check Content
1. {Check item 1}
2. {Check item 2}

### Execution Flow
```bash
{Specific execution command}

Output Rules

• No anomaly: Reply HEARTBEAT_OK
• Anomaly detected: Report specific content + suggested actions
• Major event: Report immediately (regardless of time period)

Status Tracking

Read memory/{task-name}-state.json to avoid repeated execution.

Status file format:

{
  "task_name": "{Task Name}",
  "last_execution": "ISO time",
  "execution_count": 0,
  "last_error": null,
  "last_result": null
}

Cost Protection

• Add pre-check before expensive operations: Check if necessary first
• Rate limit API calls: Avoid multiple calls in short time
• Night mode: 22:00-08:00 silent (check only, no report)

Trigger Definitions

• Time trigger: Every 30 minutes
• Event trigger: {Define event condition}
• Conditional trigger: {Define threshold condition}

Alert Rules

Each trigger defines:

1. Threshold: {Specific value}
2. Notification path: {Feishu/Email etc.}
3. Cooldown time: {N} minutes (avoid repeated alerts)

#### Step 4: Create Status Tracking File

**memory/heartbeat-state.json**:
```json
{
  "task_name": "{Task Name}",
  "last_execution": null,
  "execution_count": 0,
  "last_error": null,
  "config": {
    "interval_minutes": 30,
    "quiet_hours": {"start": 22, "end": 8},
    "mode": "balanced"
  }
}

Step 5: Verify and Test

# Manually trigger heartbeat test
# Check if output meets expectations
# Confirm status file updated

Process 3: Subagent Task Setup (Enhanced Version)

Step 1: Task Breakdown

Analysis Dimensions:

• Action count (file edits, command executions etc.)
• Dependency relations (step B depends on step A output)
• Complexity score (≥5 points recommend breakdown)
• Risk level assessment

Step 2: Select Pipeline

3 Pipelines:

Step 3: Generate Task Plan

Output Template (Enhanced):

## 📋 Task Breakdown Plan

### Original Task
{User's task}

### Complexity Assessment
- **Action Count**: {N}
- **Dependency Relations**: {Yes/No}
- **Complexity Score**: {N}/10
- **Risk Level**: {🟢/🟡/🔴/⚫}
- **Recommended Pipeline**: {Simple | Standard | Complex}
- **Recommended Parallelism**: {N}

### Task List

| # | Task Description | Agent | Dependencies | Priority | Est. Time |
|---|----------|-------|------|--------|----------|
| 1 | {Task 1} | {agent-id} | - | P1 | {X}min |
| 2 | {Task 2} | {agent-id} | {Dep #} | P2 | {X}min |
| 3 | {Task 3} | {agent-id} | {Dep #} | P3 | {X}min |

### Dispatch Strategy
- **Parallel Tasks**: {Task numbers} (no dependencies)
- **Serial Tasks**: {Task numbers} (has dependencies)
- **Review Mode**: {Parallel | Serial}
- **Aggregation Rules**: {Majority | Unanimous | Veto}
- **Total Est. Time**: {X}min

### ⚡ Clarification Confirmation
{If questions exist, list here}

### Confirmation
Please confirm and I'll start dispatching subagents ✓

Step 4: Dispatch Execution

# Parallel dispatch (no dependencies)
subagents(action=spawn, agentId="researcher", task="{Task 1}", label="task-1")
subagents(action=spawn, agentId="coder", task="{Task 2}", label="task-2")

# Serial dispatch (has dependencies)
# Wait for task-1 to complete, then dispatch task-2

Step 5: Progress Tracking (Deadlock Prevention)

memory/subagent-state.json:

{
  "task_id": "{Unique ID}",
  "status": "in_progress",
  "pipeline": "standard",
  "subagents": [
    {
      "label": "task-1",
      "agent_id": "researcher",
      "status": "completed",
      "output": "..."
    }
  ],
  "created_at": "ISO time",
  "updated_at": "ISO time",
  "deadlock_prevention": {
    "max_token": 100000,
    "used_token": 0,
    "max_time_minutes": 30,
    "elapsed_minutes": 0,
    "progress_checks": 0,
    "no_progress_count": 0
  }
}

Deadlock Prevention Mechanism:

• Cost Metric: Single task max token consumption (default 100K)
• Time Metric: Base timeout 30 minutes, dynamic adjustment
• Progress Metric: Check every 5 minutes, 3 consecutive no-progress → retry

Step 6: Result Aggregation (Intelligent Aggregation)

Aggregation Rules:

Conflict Resolution Strategy:

1. Role Priority: critic > security > architect > reviewer > tester
2. Third-party Arbitration: architect arbitrates
3. Re-review Mechanism: Trigger when confidence < 0.7
4. User Decision: Ask user when unable to reach agreement

Process 4: Task Record Setup

Step 1: Create Task Record

TASKS.md Format:

## [Task ID] Task Title

- **Created:** YYYY-MM-DD HH:MM
- **Task Type:** Continuous Task | One-time Task
- **Current Status:** Not Started | In Progress | Completed | Paused
- **Task Description:** Detailed task content and objectives
- **System Functions Called:** List used OpenClaw functions/skills
- **Required Permissions:** List needed permissions
- **Dependent Software:** External software required
- **Last Updated:** YYYY-MM-DD HH:MM
- **Notes:** Other information

Step 2: Regular Status Updates

Update Rules:

• Check task progress every heartbeat
• Update "Last Updated" time when status changes
• Move to "Completed" area after completion

🔍 Task Execution Tracking (Enhanced Version)

Unified Status Management

memory/task-registry.json:

{
  "tasks": [
    {
      "id": "task-001",
      "type": "cron",
      "name": "Daily News Push",
      "status": "active",
      "created_at": "ISO time",
      "last_run": "ISO time",
      "next_run": "ISO time",
      "run_count": 15,
      "last_error": null
    }
  ],
  "stats": {
    "total": 10,
    "active": 5,
    "completed": 4,
    "failed": 1
  }
}

Progress Reporting Mechanism

Reporting Frequency:

• Cron tasks: Report immediately after execution
• Heartbeat tasks: Report when anomaly detected
• Subagent tasks: Report at key nodes (start/complete/block)
• Task tasks: Report when status changes

Report Template:

## 📊 Task Progress Report

### Task Name
{Task name}

### Current Status
{In Progress | Completed | Blocked}

### Progress Details
- **Completion**: {X}%
- **Time Elapsed**: {X} minutes
- **Next Step**: {Specific action}

### Issues/Blocks
{Describe issues if any}

### Decisions Needed
{List options if user decision needed}

⚠️ Deadlock Prevention Mechanism (New)

Three Measurement Dimensions

1. Cost Metric (Token Consumption)

Threshold Configuration:

deadlock_prevention:
  max_token_per_task: 100000  # Base threshold
  
  # Task type multipliers
  token_multipliers:
    code: 1.0
    research: 1.5  # Search + analysis consumes more tokens
    review: 0.8
    test: 1.2
  
  # Complexity multipliers
  complexity_multipliers:
    simple: 0.5    # 50K
    normal: 1.0    # 100K
    complex: 1.5   # 150K

Calculation Formula:

Actual Threshold = Base Threshold × Task Type Multiplier × Complexity Multiplier

Processing Flow:

[Check Token Consumption]
    ↓
Exceed 80% threshold → [Warn User]
    ↓
Exceed 100% threshold → [Enter Retry Flow]

2. Time Metric (Timeout Control)

Threshold Configuration:

deadlock_prevention:
  base_timeout: 30  # Base timeout (minutes)
  
  # Task type time multipliers
  time_multipliers:
    simple: 0.5    # 15 min
    normal: 1.0    # 30 min
    complex: 2.0   # 60 min
  
  # Dynamic adjustment strategy
  dynamic_adjustment:
    enabled: true
    retry_multiplier: 1.5  # Increase 50% time after each retry
    max_timeout: 120       # Max timeout limit (2 hours)

Timeout Processing Flow:

[Timeout Detection]
    ↓
[Check Progress] → Has progress → [Extend timeout, continue]
    ↓ No progress
[Retry] → Remaining retries > 0 → [Redistribute]
    ↓
[Mark Failed] → [Enter Fallback Processing]

3. Progress Metric (No Progress Detection)

Threshold Configuration:

deadlock_prevention:
  progress_check_interval: 5  # Progress check interval (minutes)
  
  # No progress definition
  no_progress_definition:
    consecutive_checks: 3     # 3 consecutive checks with no state change (15 minutes)
    output_stale_minutes: 20  # Or: output not updated for 20 minutes
  
  # Progress tracking methods
  tracking:
    - agent_status_changes    # Status changes
    - output_file_updates     # File updates
    - message_count           # Message count

Progress Determination Rules:

Complete Deadlock Prevention Flow

[Task Dispatch]
    ↓
[Start Timer]
    ↓
Every [progress_check_interval] minutes:
    ├── [Check token consumption] → Exceed 80% threshold → [Warn user]
    ├── [Check time] → Exceed timeout → [Enter no progress detection]
    └── [Check progress] → No progress → [Enter retry flow]
    
[Retry Flow]
    ├── Increase timeout (×1.5)
    ├── Reduce token threshold (×0.8)
    └── Retries count -1
    
[Final Failure]
    ├── Record detailed logs
    ├── Notify user
    └── Enter fallback processing

📚 Best Practice Template Library

Template 1: Daily News Push (Cron)

openclaw cron add \
  --name "Daily News Push" \
  --cron "0 9 * * *" \
  --tz "Asia/Shanghai" \
  --session isolated \
  --channel feishu \
  --to "user:ou_642f1cb74d63462d1037375caac5ea2d" \
  --message "Search today's AI industry news, organize into briefing and push to me" \
  --description "Push AI industry news every day at 9am"

Template 2: Email Check (Heartbeat)

# HEARTBEAT.md

## Scheduled Task: Email Check

**Heartbeat Mode**: Balanced (30 minutes)
**Quiet Hours**: 22:00-08:00

Check unread email every heartbeat (about 30 minutes).

### Execution Flow

1. Read memory/email-state.json to get last check time
2. Call email API to get new emails
3. Has new email → Summarize and report (count + important email summary)
4. No new email → Reply HEARTBEAT_OK
5. Update email-state.json

### Output Rules

- No new email: `HEARTBEAT_OK`
- Has new email (<10): Summarize and report
- Has new email (≥10): Batch report + remind to clean

### Night Mode

During 22:00-08:00:
- Check only, no report (unless urgent)
- Save to morning first report for unified report

### Trigger Definitions

- **Time Trigger**: Every 30 minutes
- **Event Trigger**: New email arrival
- **Conditional Trigger**: Unread email ≥ 10

### Alert Rules

- **Threshold**: Unread email ≥ 20
- **Notification Path**: Feishu
- **Cooldown Time**: 60 minutes

Template 3: Competitor Research (Subagent)

## 📋 Task Breakdown Plan

### Original Task
Research competitors A, B, C features and pricing strategy, output analysis report

### Complexity Assessment
- **Action Count**: 5
- **Dependency Relations**: Yes
- **Complexity Score**: 7/10 (Complex)
- **Risk Level**: 🟡 MEDIUM
- **Recommended Pipeline**: Standard
- **Recommended Parallelism**: 3

### Task List

| # | Task Description | Agent | Dependencies | Priority | Est. Time |
|---|----------|-------|------|--------|----------|
| 1 | Research Competitor A | researcher | - | P1 | 5min |
| 2 | Research Competitor B | researcher | - | P1 | 5min |
| 3 | Research Competitor C | researcher | - | P1 | 5min |
| 4 | Comparative Analysis | analyst | 1,2,3 | P2 | 10min |
| 5 | Write Report | writer | 4 | P3 | 10min |

### Dispatch Strategy
- **Parallel**: 1,2,3 (no dependencies)
- **Serial**: 4→5 (has dependencies)
- **Review Mode**: Parallel (code + docs independent)
- **Aggregation Rules**: Majority
- **Total Est. Time**: ~25min

### Confirmation
Please confirm and I'll start dispatching subagents ✓

🔧 Configuration Loading Mechanism (New)

Configuration File Structure

task-orchestrator-config.yaml:

# ===========================================
# 1. Pipeline Configuration
# ===========================================
pipelines:
  simple:
    stages:
      - execute
      - validate
      - complete
    confirm_after: []
  
  standard:
    stages:
      - analyze
      - execute
      - review
      - test
      - validate
    confirm_after:
      - analyze
      - review
  
  complex:
    stages:
      - deep_analyze
      - design_review
      - execute
      - parallel_reviews
      - aggregate
      - test
      - validate
    confirm_after:
      - deep_analyze
      - design_review
      - aggregate

# ===========================================
# 2. Agent Mappings
# ===========================================
agent_mappings:
  task_types:
    code: coder
    research: researcher
    document: docs_writer
    test: tester
    review: reviewer
    architecture: architect

# ===========================================
# 3. Threshold Parameters
# ===========================================
thresholds:
  complexity:
    simple_max_steps: 1
    normal_max_steps: 3
    complex_min_steps: 4
  
  risk:
    low_cost_threshold: 10
    medium_cost_threshold: 50
    high_cost_threshold: 100
  
  deadlock_prevention:
    max_token_per_task: 100000
    max_time_minutes: 30
    max_retries: 2
    progress_check_interval: 5

# ===========================================
# 4. Business Rules
# ===========================================
business_rules:
  review_modes:
    parallel:
      - code + docs + test
      - multiple_features
    sequential:
      - architecture + implementation
      - security + any
  
  confirm_requirements:
    auto_execute: [LOW]
    brief_confirm: [MEDIUM]
    detailed_confirm: [HIGH]
    explicit_authorization: [CRITICAL]
  
  summary_rules:
    default_mode: majority
    veto_allowed_roles: [critic]

Configuration Loading Timing

Recommended: Load at task dispatch time

# Pseudocode
def dispatch_task(task):
    # Load configuration
    config = load_config("task-orchestrator-config.yaml")
    
    # Select pipeline based on task
    pipeline = select_pipeline(task, config.pipelines)
    
    # Select agent based on task type
    agent = select_agent(task, config.agent_mappings)
    
    # Execute
    execute_pipeline(pipeline, agent, config.thresholds)

📖 Related Skills

This skill integrates core capabilities from following skills:

• agent-autopilot - Self-driven workflow (3 scheduled tasks)
• agent-heartbeat - Heartbeat configuration
• agent-step-sequencer - Multi-step scheduling
• cron-mastery - Cron best practices
• heartbeat-manager - Heartbeat management (health score, intelligent timeout analysis)
• task-dispatcher - Task dispatch (6-point checklist, 4-level risk, deadlock prevention)
• task-tracker - Task tracking
• let-me-know - Long task notification

🎓 Learning Resources

• /Users/chloe/Desktop/cron and heartbeat/ - Original skill collection
• OpenClaw Documentation - https://docs.openclaw.ai/automation/cron-jobs
• HEARTBEAT.md Templates - See this skill's template library
• task-dispatcher Supplementary Design - Deadlock prevention mechanism, intelligent aggregation rules

Task Orchestrator v2.0.0 - Making task orchestration simpler, more reliable, and more intelligent