荣耀MagicOS 11系统级Agent Harness架构深度解析——业界首个移动OS层面Agent框架，YOYO超百步长程任务与端侧AI架构重构

"""
YOYO Harness 感知引擎取件码场景模拟
模拟：用户到达快递站附近 → 感知引擎识别位置上下文 → 提取快递短信中的取件码 → 主动推送卡片
"""

import re
import json
from dataclasses import dataclass, field
from typing import Optional, List
from enum import Enum
import time


class LocationType(Enum):
    """位置类型枚举"""
    EXPRESS_STATION = "express_station"
    RESTAURANT = "restaurant"
    AIRPORT = "airport"
    GROCERY = "grocery"
    UNKNOWN = "unknown"


class ContentCategory(Enum):
    """内容分类枚举"""
    EXPRESS_PICKUP = "express_pickup"  # 快递取件码
    FOOD_ORDER = "food_order"          # 餐饮取餐码
    TICKET = "ticket"                  # 票务信息
    APPOINTMENT = "appointment"        # 日程/预约
    PAYMENT_CODE = "payment_code"      # 付款码
    UNKNOWN = "unknown"


@dataclass
class PerceptionContext:
    """感知上下文，融合多源感知数据"""
    location: str = ""
    location_type: LocationType = LocationType.UNKNOWN
    screen_content: str = ""
    recent_notifications: List[str] = field(default_factory=list)
    wifi_ssid: str = ""
    bluetooth_devices: List[str] = field(default_factory=list)
    time_of_day: str = ""
    battery_level: int = 100
    activity_type: str = ""  # walking, driving, stationary, etc.
    user_last_interaction: str = ""


@dataclass
class PerceptionEvent:
    """统一感知事件输出"""
    event_id: str = ""
    category: ContentCategory = ContentCategory.UNKNOWN
    confidence: float = 0.0
    extracted_data: dict = field(default_factory=dict)
    source: str = ""
    timestamp: int = 0


class SensoryHub:
    """
    感知中枢：融合多源感知数据，输出结构化感知事件
    YOYO Harness 感知引擎的核心入口
    """

    def __init__(self):
        # 已注册的地点围栏（坐标简化为地名字符串）
        self.location_fences = {
            "蜂鸟快递柜-望京SOHO": {"type": LocationType.EXPRESS_STATION, "radius_m": 50},
            "菜鸟驿站-中关村": {"type": LocationType.EXPRESS_STATION, "radius_m": 80},
            "顺丰速运-国贸": {"type": LocationType.EXPRESS_STATION, "radius_m": 60},
        }
        # 用户短期记忆缓存
        self.short_term_memory: List[PerceptionEvent] = []
        self.max_memory_size = 100

    def perceive(self, ctx: PerceptionContext) -> List[PerceptionEvent]:
        """
        主感知入口：并行启动多路感知通道
        返回当前上下文下产生的一组感知事件
        """
        events: List[PerceptionEvent] = []

        # 1. 位置感知通道
        location_events = self._location_perception(ctx)
        events.extend(location_events)

        # 2. 内容感知通道（屏幕/通知）
        content_events = self._content_perception(ctx)
        events.extend(content_events)

        # 3. 情境感知通道（时间/活动/设备状态）
        context_events = self._context_perception(ctx)
        events.extend(context_events)

        # 融合去重 & 置信度排序
        events = self._fusion_and_rank(events)

        # 更新短期记忆
        for evt in events:
            self.short_term_memory.append(evt)
            if len(self.short_term_memory) > self.max_memory_size:
                self.short_term_memory.pop(0)

        return events

    def _location_perception(self, ctx: PerceptionContext) -> List[PerceptionEvent]:
        """位置感知：判断是否进入已知地点围栏"""
        events = []
        for place_name, fence in self.location_fences.items():
            # 简化位置匹配（实际使用GPS坐标距离计算）
            if ctx.location == place_name or ctx.location in place_name:
                events.append(PerceptionEvent(
                    event_id=f"loc_{int(time.time())}",
                    category=ContentCategory.EXPRESS_PICKUP,
                    confidence=0.75,
                    extracted_data={
                        "place": place_name,
                        "type": fence["type"].value,
                        "status": "entered_fence"
                    },
                    source="location_perception",
                    timestamp=int(time.time())
                ))
        return events

    def _content_perception(self, ctx: PerceptionContext) -> List[PerceptionEvent]:
        """内容感知：从通知和屏幕内容中提取结构化信息"""
        events = []

        # 快递取件码提取模式
        pickup_patterns = [
            (r'(?:取件码|取货码|提件码)[：:\s]*(\d{4,8})', ContentCategory.EXPRESS_PICKUP),
            (r'(\d{4,8})\s*(?:为|是)\s*(?:您的\s*)?(?:取件|取货)码', ContentCategory.EXPRESS_PICKUP),
            (r'包裹\s*(?:已到|已达|到达)[^。！\n]*?(\d{4,8})', ContentCategory.EXPRESS_PICKUP),
        ]

        for notification in ctx.recent_notifications:
            for pattern, category in pickup_patterns:
                match = re.search(pattern, notification)
                if match:
                    code = match.group(1)
                    # 区分快递类型
                    item_type = self._classify_express_item(notification)
                    events.append(PerceptionEvent(
                        event_id=f"content_{int(time.time())}_{len(events)}",
                        category=ContentCategory.EXPRESS_PICKUP,
                        confidence=0.92 if len(code) >= 6 else 0.80,
                        extracted_data={
                            "pickup_code": code,
                            "item_type": item_type,
                            "source_text": notification[:80],
                            "is_fresh": self._is_recent_notification(notification)
                        },
                        source="content_perception",
                        timestamp=int(time.time())
                    ))
                    break

        return events

    def _classify_express_item(self, text: str) -> str:
        """识别快递物品类型：生鲜/贵重/大件/普通"""
        if any(kw in text for kw in ["生鲜", "冷链", "冷藏", "冷冻", "水果", "食品"]):
            return "fresh_food"  # 生鲜 - 需要及时取件提醒
        if any(kw in text for kw in ["贵重", "保价", "保险"]):
            return "valuable"    # 贵重 - 需要特别标记
        if any(kw in text for kw in ["大件", "家具", "家电"]):
            return "bulky"      # 大件 - 需要准备搬运
        return "normal"

    def _is_recent_notification(self, text: str) -> bool:
        """判断通知是否为近期（简化实现）"""
        # 实际会读取通知的时间戳
        return True

    def _context_perception(self, ctx: PerceptionContext) -> List[PerceptionEvent]:
        """情境感知：时间、活动类型、设备状态等"""
        events = []
        # 例如：检测到用户在快递站附近行走，提升取件码推送优先级
        if ctx.location_type == LocationType.EXPRESS_STATION and \
           ctx.activity_type == "walking":
            events.append(PerceptionEvent(
                event_id=f"context_{int(time.time())}",
                category=ContentCategory.EXPRESS_PICKUP,
                confidence=0.85,
                extracted_data={"context_hint": "user_arriving_express_station"},
                source="context_perception",
                timestamp=int(time.time())
            ))
        return events

    def _fusion_and_rank(self, events: List[PerceptionEvent]) -> List[PerceptionEvent]:
        """融合排序：去重、合并关联事件、按置信度排序"""
        # 按event_id去重
        seen = set()
        deduped = []
        for evt in events:
            key = (evt.category.value, json.dumps(evt.extracted_data, sort_keys=True))
            if key not in seen:
                seen.add(key)
                deduped.append(evt)

        # 合并同类别事件（取最高置信度）
        merged = {}
        for evt in deduped:
            cat = evt.category.value
            if cat not in merged or evt.confidence > merged[cat].confidence:
                merged[cat] = evt

        # 按置信度降序排列
        result = sorted(merged.values(), key=lambda e: e.confidence, reverse=True)
        return result


# ============ 模拟运行 ============
if __name__ == "__main__":
    hub = SensoryHub()

    # 模拟场景：用户走到蜂鸟快递柜附近，收到快递通知
    test_context = PerceptionContext(
        location="蜂鸟快递柜-望京SOHO",
        location_type=LocationType.EXPRESS_STATION,
        recent_notifications=[
            "【蜂鸟快递】您的包裹已到蜂鸟快递柜，取件码：668899，请及时取件。（生鲜食品，建议2小时内取出）",
            "【天气预报】明天多云转晴，气温18-25℃",
        ],
        wifi_ssid="HONOR-WiFi",
        activity_type="walking",
        time_of_day="18:30",
        battery_level=85,
        user_last_interaction="查看短信"
    )

    perceived_events = hub.perceive(test_context)

    print("=== 感知引擎输出事件 ===")
    for evt in perceived_events:
        print(f"[{evt.category.value}] 置信度: {evt.confidence:.0%}")
        print(f"  提取数据: {json.dumps(evt.extracted_data, ensure_ascii=False, indent=2)}")
        print(f"  来源: {evt.source}")
        print()

package main

import (
	"container/list"
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// TaskNode 表示任务图中的一个节点
type TaskNode struct {
	ID         string            `json:"id"`
	Desc       string            `json:"description"`
	Tool       string            `json:"tool"`
	Status     string            `json:"status"` // pending, running, success, failed, skipped
	Deps       []string          `json:"dependencies"`
	Params     map[string]string `json:"params"`
	Retries    int               `json:"retries"`
	MaxRetries int               `json:"max_retries"`
	Result     string            `json:"result,omitempty"`
	ErrorCount int               `json:"error_count"`
}

// TaskGraph 表示由规划引擎生成的DAG任务图
type TaskGraph struct {
	Nodes map[string]*TaskNode `json:"nodes"`
	mu    sync.RWMutex
}

// NewTaskGraph 创建一个新的任务图
func NewTaskGraph() *TaskGraph {
	return &TaskGraph{
		Nodes: make(map[string]*TaskNode),
	}
}

// AddNode 添加任务节点
func (g *TaskGraph) AddNode(node *TaskNode) {
	g.mu.Lock()
	defer g.mu.Unlock()
	g.Nodes[node.ID] = node
}

// TopoSort 拓扑排序：找到可并行执行的任务层级
func (g *TaskGraph) TopoSort() [][]string {
	g.mu.RLock()
	defer g.mu.RUnlock()

	// 计算每个节点的入度
	inDegree := make(map[string]int)
	for id, node := range g.Nodes {
		if _, ok := inDegree[id]; !ok {
			inDegree[id] = 0
		}
		for _, dep := range node.Deps {
			inDegree[id]++
		}
	}

	// Kahn 拓扑排序：分层次输出可并行任务
	queue := list.New()
	for id, degree := range inDegree {
		if degree == 0 {
			queue.PushBack(id)
		}
	}

	var levels [][]string
	for queue.Len() > 0 {
		levelSize := queue.Len()
		level := make([]string, 0, levelSize)

		for i := 0; i < levelSize; i++ {
			elem := queue.Front()
			queue.Remove(elem)
			nodeID := elem.Value.(string)
			level = append(level, nodeID)

			// 减少下游节点的入度
			for _, candidateID := range g.getDependents(nodeID) {
				inDegree[candidateID]--
				if inDegree[candidateID] == 0 {
					queue.PushBack(candidateID)
				}
			}
		}
		levels = append(levels, level)
	}

	return levels
}

// getDependents 获取依赖当前节点的所有下游节点
func (g *TaskGraph) getDependents(nodeID string) []string {
	var deps []string
	for id, node := range g.Nodes {
		for _, dep := range node.Deps {
			if dep == nodeID {
				deps = append(deps, id)
			}
		}
	}
	return deps
}

// TaskExecutor 任务执行器：按拓扑顺序调度执行
type TaskExecutor struct {
	Graph *TaskGraph
}

// ExecuteResult 记录单个节点的执行结果
type ExecuteResult struct {
	NodeID string `json:"node_id"`
	Status string `json:"status"`
	Output string `json:"output"`
	CostMs int64  `json:"cost_ms"`
}

// Execute 按DAG拓扑顺序执行任务，支持同层并行
func (e *TaskExecutor) Execute() []ExecuteResult {
	levels := e.Graph.TopoSort()
	var results []ExecuteResult

	fmt.Printf("=== YOYO Planner: 任务分解为 %d 个层级 ===\n", len(levels))
	for i, level := range levels {
		fmt.Printf("  第%d层 [并行执行]: %v\n", i+1, level)

		var wg sync.WaitGroup
		resultCh := make(chan ExecuteResult, len(level))

		for _, nodeID := range level {
			wg.Add(1)
			go func(nid string) {
				defer wg.Done()
				node, ok := e.Graph.Nodes[nid]
				if !ok {
					return
				}
				result := e.executeNode(node)
				resultCh <- result
			}(nodeID)
		}

		wg.Wait()
		close(resultCh)

		for r := range resultCh {
			results = append(results, r)
		}
	}

	return results
}

// executeNode 执行单个任务节点（模拟跨应用调用）
func (e *TaskExecutor) executeNode(node *TaskNode) ExecuteResult {
	start := time.Now()
	node.Status = "running"

	fmt.Printf("    ▶ 执行: [%s] %s (工具: %s)\n", node.ID, node.Desc, node.Tool)

	// 模拟执行耗时
	time.Sleep(time.Duration(200+len(node.Deps)*50) * time.Millisecond)

	// 模拟执行结果：87%成功率
	success := true
	if len(node.Params) > 0 {
		// 复杂的参数场景有13%的概率失败
		if node.ErrorCount > 2 {
			success = false
		}
	}

	cost := time.Since(start).Milliseconds()

	if success {
		node.Status = "success"
		node.Result = fmt.Sprintf("✅ %s 执行完成，输出: %s_%s_result",
			node.Tool, node.Tool, node.ID)
		fmt.Printf("    ✅ [%s] 完成 (%dms)\n", node.ID, cost)
	} else {
		node.Status = "failed"
		node.ErrorCount++
		if node.ErrorCount <= node.MaxRetries {
			// 自动重试
			fmt.Printf("    🔄 [%s] 失败，第%d次重试\n", node.ID, node.ErrorCount)
			return e.executeNode(node)
		}
		fmt.Printf("    ❌ [%s] 执行失败\n", node.ID)
	}

	return ExecuteResult{
		NodeID: node.ID,
		Status: node.Status,
		Output: node.Result,
		CostMs: cost,
	}
}

// YOYOPlanner YOYO规划引擎主类
type YOYOPlanner struct {
	ModelEndpoint string
	MaxTaskSteps  int
}

// NewYOYOPlanner 创建YOYO规划引擎实例
func NewYOYOPlanner(endpoint string, maxSteps int) *YOYOPlanner {
	return &YOYOPlanner{
		ModelEndpoint: endpoint,
		MaxTaskSteps:  maxSteps,
	}
}

// Plan 从自然语言生成任务图
// 实际场景中：调用端侧/云侧LLM进行意图理解和任务拆解
func (p *YOYOPlanner) Plan(userIntent string) *TaskGraph {
	fmt.Println("\n=== YOYO Planner: 开始规划 ===")
	fmt.Printf("用户意图: %s\n", userIntent)

	graph := NewTaskGraph()

	// 模拟LLM任务拆解结果
	// 以"演唱会+抢票+练歌+日程"复合任务为例
	graph.AddNode(&TaskNode{
		ID: "task_001", Desc: "打开购票App搜索演唱会信息",
		Tool: "购票App", Status: "pending", Deps: []string{},
		Params: map[string]string{"action": "search", "keywords": "演唱会"},
		MaxRetries: 2,
	})
	graph.AddNode(&TaskNode{
		ID: "task_002", Desc: "获取演唱会时间地点详情",
		Tool: "购票App", Status: "pending", Deps: []string{"task_001"},
		Params:     map[string]string{"action": "get_detail"},
		MaxRetries: 2,
	})
	graph.AddNode(&TaskNode{
		ID: "task_003", Desc: "设置抢票闹钟（提前5分钟）",
		Tool: "系统闹钟", Status: "pending", Deps: []string{"task_002"},
		Params:     map[string]string{"offset": "-5min"},
		MaxRetries: 1,
	})
	graph.AddNode(&TaskNode{
		ID: "task_004", Desc: "搜索10首热门相关歌曲",
		Tool: "音乐App", Status: "pending", Deps: []string{},
		Params:     map[string]string{"genre": "热门演唱会歌曲", "count": "10"},
		MaxRetries: 2,
	})
	graph.AddNode(&TaskNode{
		ID: "task_005", Desc: "制定练歌计划",
		Tool: "备忘录", Status: "pending", Deps: []string{"task_004"},
		Params:     map[string]string{"template": "练习计划"},
		MaxRetries: 2,
	})
	graph.AddNode(&TaskNode{
		ID: "task_006", Desc: "将练歌计划写入日程",
		Tool: "系统日历", Status: "pending", Deps: []string{"task_002", "task_005"},
		Params:     map[string]string{"sync": "true"},
		MaxRetries: 1,
	})

	fmt.Printf("规划完成: %d个任务节点\n", len(graph.Nodes))
	return graph
}

// PipelineStat 记录整个pipeline的执行统计
type PipelineStat struct {
	TotalSteps    int     `json:"total_steps"`
	SuccessSteps  int     `json:"success_steps"`
	FailedSteps   int     `json:"failed_steps"`
	SuccessRate   float64 `json:"success_rate"`
	TotalCostMs   int64   `json:"total_cost_ms"`
}

func main() {
	fmt.Println("=== YOYO Harness 规划引擎模拟 ===")
	fmt.Println("版本: MagicOS 11 | Agent Harness v2.0")
	fmt.Println()

	// 初始化规划引擎
	planner := NewYOYOPlanner("endpoint://yoyo-harness/planner", 100)

	// 用户复合指令
	userIntent := "打开App查演唱会时间、提前5分钟定抢票闹钟、找10首相关热门歌曲指定练歌计划、把计划写到日程里"

	// 阶段1: 规划 -> 生成DAG
	taskGraph := planner.Plan(userIntent)

	// 显示拓扑结构
	levels := taskGraph.TopoSort()
	fmt.Printf("\n拓扑层级 (共%d层):\n", len(levels))
	for i, l := range levels {
		fmt.Printf("  层%d: %v\n", i+1, l)
	}

	// 阶段2: 执行DAG
	fmt.Println("\n=== 开始任务执行 ===")
	executor := &TaskExecutor{Graph: taskGraph}
	results := executor.Execute()

	// 统计
	var stats PipelineStat
	for _, r := range results {
		stats.TotalSteps++
		stats.TotalCostMs += r.CostMs
		if r.Status == "success" {
			stats.SuccessSteps++
		} else {
			stats.FailedSteps++
		}
	}
	stats.SuccessRate = float64(stats.SuccessSteps) / float64(stats.TotalSteps) * 100

	fmt.Printf("\n=== 执行统计 ===\n")
	fmt.Printf("总步骤: %d | 成功: %d | 失败: %d\n", stats.TotalSteps, stats.SuccessSteps, stats.FailedSteps)
	fmt.Printf("成功率: %.1f%%\n", stats.SuccessRate)
	fmt.Printf("总执行耗时: %dms\n", stats.TotalCostMs)

	// JSON输出
	statsJSON, _ := json.MarshalIndent(stats, "", "  ")
	fmt.Printf("\n统计JSON:\n%s\n", string(statsJSON))
}

"""
YOYO Harness 跨应用执行适配器
支持 MCP / A2A / GUI 三种执行路径的自动选择与降级
"""

import abc
import enum
import json
import random
import time
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field


class ExecPath(enum.Enum):
    """执行路径类型"""
    MCP = "mcp"       # Model Context Protocol - 模型原生工具调用
    A2A = "a2a"       # Agent-to-Agent - Agent间协作协议
    GUI = "gui"       # GUI Simulation - 视觉坐标模拟


class ExecPriority(enum.IntEnum):
    """执行优先级：数字越大优先级越高"""
    MCP_HIGH = 30
    A2A_MEDIUM = 20
    GUI_LOW = 10
    NONE = 0


@dataclass
class ExecCapability:
    """目标App的执行能力描述"""
    app_name: str
    supports_mcp: bool = False
    supports_a2a: bool = False
    supports_gui_sim: bool = True  # 所有App原则上都支持GUI模拟
    mcp_tools: List[str] = field(default_factory=list)
    a2a_endpoint: Optional[str] = None
    gui_complexity: int = 5  # 1(简单) ~ 10(极高) GUI交互复杂度


@dataclass
class ExecRequest:
    """执行请求"""
    intent: str
    target_app: str
    action: str
    params: Dict[str, Any] = field(default_factory=dict)
    timeout_ms: int = 30000


@dataclass
class ExecResult:
    """执行结果"""
    success: bool
    path: ExecPath
    output: str = ""
    error: str = ""
    cost_ms: int = 0
    fallback_from: List[ExecPath] = field(default_factory=list)


class PathAdapter(abc.ABC):
    """执行路径适配器基类"""

    @abc.abstractmethod
    def can_handle(self, capability: ExecCapability) -> ExecPriority:
        """判断是否能处理该目标App，返回优先级"""
        pass

    @abc.abstractmethod
    def execute(self, request: ExecRequest) -> ExecResult:
        """执行请求"""
        pass


class MCPAdapter(PathAdapter):
    """MCP协议适配器 - 最优先路径"""

    def can_handle(self, capability: ExecCapability) -> ExecPriority:
        if capability.supports_mcp and len(capability.mcp_tools) > 0:
            return ExecPriority.MCP_HIGH
        return ExecPriority.NONE

    def execute(self, request: ExecRequest) -> ExecResult:
        start = time.time()
        print(f"  [MCP] 调用 {request.target_app}.{request.action}()")
        print(f"        params: {json.dumps(request.params, ensure_ascii=False)}")

        # 模拟MCP调用：通过工具定义直接调用
        time.sleep(random.uniform(0.3, 0.8))

        # 95% MCP调用成功率
        success = random.random() < 0.95
        cost = int((time.time() - start) * 1000)

        return ExecResult(
            success=success,
            path=ExecPath.MCP,
            output=f"MCP调用{request.target_app}.{request.action}完成" if success else "",
            error="" if success else "MCP调用超时或返回异常",
            cost_ms=cost
        )


class A2AAdapter(PathAdapter):
    """A2A代理间通信适配器"""

    def can_handle(self, capability: ExecCapability) -> ExecPriority:
        if capability.supports_a2a and capability.a2a_endpoint:
            return ExecPriority.A2A_MEDIUM
        return ExecPriority.NONE

    def execute(self, request: ExecRequest) -> ExecResult:
        start = time.time()
        print(f"  [A2A] 发送 Agent 间请求到 {request.target_app}")
        print(f"        endpoint: {json.dumps({'intent': request.intent})}")

        # 模拟A2A通信
        time.sleep(random.uniform(0.5, 1.2))

        # 90% A2A成功率
        success = random.random() < 0.90
        cost = int((time.time() - start) * 1000)

        return ExecResult(
            success=success,
            path=ExecPath.A2A,
            output=f"A2A Agent协作完成: {request.intent}" if success else "",
            error="" if success else "A2A Agent无响应",
            cost_ms=cost
        )


class GUIAdapter(PathAdapter):
    """GUI模拟适配器 - 兜底路径（无需App授权）"""

    def can_handle(self, capability: ExecCapability) -> ExecPriority:
        # 所有App都支持GUI模拟
        priority = max(ExecPriority.GUI_LOW.value - capability.gui_complexity, 1)
        return ExecPriority(priority)

    def execute(self, request: ExecRequest) -> ExecResult:
        start = time.time()
        print(f"  [GUI] 视觉坐标模拟: {request.target_app}")

        # 解析UI布局，找到目标控件
        gui_steps = [
            "  1. 截取当前屏幕",
            "  2. CNN目标检测 → 定位目标控件坐标",
            "  3. 模拟点击 (x, y)",
            f"  4. 输入参数: {json.dumps(request.params, ensure_ascii=False)[:50]}...",
            "  5. 确认操作完成",
        ]
        for step in gui_steps:
            print(f"     {step}")
            time.sleep(0.1)

        # GUI模拟受App更新影响，成功率相对较低
        time.sleep(random.uniform(0.8, 1.5))
        success = random.random() < 0.85
        cost = int((time.time() - start) * 1000)

        return ExecResult(
            success=success,
            path=ExecPath.GUI,
            output=f"GUI模拟完成: {request.action}" if success else "",
            error="" if success else "GUI控件未找到或交互超时",
            cost_ms=cost
        )


class ExecutorScheduler:
    """
    执行调度器：根据App能力自动选择最优执行路径
    并支持路径失败时的自动降级 (fallback)
    """

    def __init__(self):
        self.adapters: List[PathAdapter] = [
            MCPAdapter(),
            A2AAdapter(),
            GUIAdapter(),
        ]
        # App能力注册表
        self.capability_registry: Dict[str, ExecCapability] = {}

    def register_app(self, capability: ExecCapability):
        """注册App的执行能力"""
        self.capability_registry[capability.app_name] = capability

    def execute(self, request: ExecRequest) -> ExecResult:
        """智能路由执行：按优先级尝试，失败自动降级"""
        capability = self.capability_registry.get(request.target_app)
        if not capability:
            # 未知App，使用默认GUI兜底
            capability = ExecCapability(
                app_name=request.target_app,
                gui_complexity=7
            )

        # 按优先级排序适配器
        ranked = []
        for adapter in self.adapters:
            priority = adapter.can_handle(capability)
            if priority > ExecPriority.NONE:
                ranked.append((priority, adapter))

        ranked.sort(key=lambda x: x[0].value, reverse=True)

        if not ranked:
            return ExecResult(
                success=False,
                path=ExecPath.GUI,
                error="无可用执行路径"
            )

        # 按优先级依次尝试，失败后fallback
        fallback_chain = []
        for priority, adapter in ranked:
            print(f"\n  尝试路径: {type(adapter).__name__} (优先级={priority.value})")
            result = adapter.execute(request)
            if result.success:
                result.fallback_from = fallback_chain
                return result
            fallback_chain.append(type(adapter).__name__)
            print(f"  ⚠ {type(adapter).__name__} 失败，降级到下一路径")

        # 所有路径都失败
        return ExecResult(
            success=False,
            path=ExecPath.GUI,
            error=f"所有执行路径均失败: {', '.join(fallback_chain)}",
            fallback_from=fallback_chain
        )


# ============ 模拟运行 ============
if __name__ == "__main__":
    scheduler = ExecutorScheduler()

    # 注册App能力
    scheduler.register_app(ExecCapability(
        app_name="购票App",
        supports_mcp=True,
        supports_a2a=True,
        mcp_tools=["search_events", "book_ticket", "get_detail"],
        a2a_endpoint="agent://ticket-agent/handle",
        gui_complexity=6
    ))
    scheduler.register_app(ExecCapability(
        app_name="音乐App",
        supports_mcp=False,  # 不支持MCP
        supports_a2a=True,
        mcp_tools=[],
        a2a_endpoint="agent://music-agent/handle",
        gui_complexity=4
    ))
    scheduler.register_app(ExecCapability(
        app_name="某小众记账App",
        supports_mcp=False,
        supports_a2a=False,
        gui_complexity=8  # UI复杂，但不支持任何协议
    ))

    # 测试1: 购票App - 应走MCP路径
    print("=" * 60)
    print("场景1: 购票App (支持MCP+A2A)")
    result1 = scheduler.execute(ExecRequest(
        intent="搜索最近演唱会",
        target_app="购票App",
        action="search_events",
        params={"keywords": "演唱会", "city": "北京"}
    ))
    print(f"结果: {'✅' if result1.success else '❌'} 路径={result1.path.value} 耗时={result1.cost_ms}ms")

    # 测试2: 音乐App - MCP不可用，走A2A
    print("\n" + "=" * 60)
    print("场景2: 音乐App (不支持MCP，支持A2A)")
    result2 = scheduler.execute(ExecRequest(
        intent="搜索10首热门演唱会歌曲",
        target_app="音乐App",
        action="search_songs",
        params={"genre": "演唱会热门", "count": 10}
    ))
    print(f"结果: {'✅' if result2.success else '❌'} 路径={result2.path.value} 耗时={result2.cost_ms}ms")

    # 测试3: 小众App - 仅支持GUI模拟
    print("\n" + "=" * 60)
    print("场景3: 小众记账App (仅支持GUI)")
    result3 = scheduler.execute(ExecRequest(
        intent="记录一笔今日支出",
        target_app="某小众记账App",
        action="add_expense",
        params={"amount": 39.9, "category": "餐饮"}
    ))
    print(f"结果: {'✅' if result3.success else '❌'} 路径={result3.path.value} fallback={result3.fallback_from} 耗时={result3.cost_ms}ms")

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// TaskComplexity 任务复杂度等级
type TaskComplexity int

const (
	Simple TaskComplexity = iota
	Medium
	Complex
	VeryComplex
)

// ModelProfile 模型性能画像
type ModelProfile struct {
	Name          string
	Location      string // "ondevice" or "cloud"
	LatencyMs     int    // 平均推理延迟
	MaxTokens     int    // 最大输出长度
	SupportsVLM   bool   // 是否支持视觉理解
	SupportsPlan  bool   // 是否支持任务规划
}

// InferenceRequest 推理请求
type InferenceRequest struct {
	Input      string
	Complexity TaskComplexity
	DeadlineMs int // 用户可接受的最大等待时间
	NeedVLM    bool
}

// InferenceResult 推理结果
type InferenceResult struct {
	ModelUsed  string
	Output     string
	LatencyMs  int
	FromCloud  bool
}

// HybridInferenceScheduler 混合推理调度器
type HybridInferenceScheduler struct {
	onDevice ModelProfile
	cloud    ModelProfile
	mu       sync.RWMutex

	// 统计
	onDeviceCalls int
	cloudCalls    int
	totalLatency  int
}

func NewHybridInferenceScheduler() *HybridInferenceScheduler {
	return &HybridInferenceScheduler{
		onDevice: ModelProfile{
			Name:        "Honor Magic LLM (端侧)",
			Location:    "ondevice",
			LatencyMs:   250,
			MaxTokens:   2048,
			SupportsVLM: true,
			SupportsPlan: false,
		},
		cloud: ModelProfile{
			Name:        "Honor-Alibaba Joint Model (云侧)",
			Location:    "cloud",
			LatencyMs:   1200,
			MaxTokens:   8192,
			SupportsVLM: true,
			SupportsPlan: true,
		},
	}
}

// Route 根据任务特征路由到最优模型
func (s *HybridInferenceScheduler) Route(req InferenceRequest) ModelProfile {
	// 规则1: VLM任务优先端侧（减少网络传输图片）
	if req.NeedVLM && req.Complexity <= Medium {
		return s.onDevice
	}

	// 规则2: 简单任务且延迟敏感 → 端侧
	if req.Complexity == Simple && req.DeadlineMs < 500 {
		return s.onDevice
	}

	// 规则3: 需要复杂规划 → 云侧
	if req.Complexity >= VeryComplex || req.Complexity == Complex {
		return s.cloud
	}

	// 规则4: 长输出 → 云侧
	if len(req.Input) > 1000 {
		return s.cloud
	}

	// 默认: 端侧
	return s.onDevice
}

// Infer 执行推理（模拟）
func (s *HybridInferenceScheduler) Infer(req InferenceRequest) InferenceResult {
	model := s.Route(req)

	// 模拟推理耗时
	latency := model.LatencyMs + rand.Intn(model.LatencyMs/2)

	// 如果是云端，加上网络传输时间
	if model.Location == "cloud" {
		latency += 150 + rand.Intn(100)
	}

	if model.Location == "ondevice" {
		s.mu.Lock()
		s.onDeviceCalls++
		s.totalLatency += latency
		s.mu.Unlock()
	} else {
		s.mu.Lock()
		s.cloudCalls++
		s.totalLatency += latency
		s.mu.Unlock()
	}

	return InferenceResult{
		ModelUsed: model.Name,
		Output:    fmt.Sprintf("[%s推理完成] 输入长度=%d字, 复杂度=%d",
			model.Location, len(req.Input), req.Complexity),
		LatencyMs: latency,
		FromCloud: model.Location == "cloud",
	}
}

// GetStats 获取调度统计
func (s *HybridInferenceScheduler) GetStats() map[string]interface{} {
	s.mu.RLock()
	defer s.mu.RUnlock()
	total := s.onDeviceCalls + s.cloudCalls
	avgLatency := 0
	if total > 0 {
		avgLatency = s.totalLatency / total
	}
	return map[string]interface{}{
		"total_calls":        total,
		"ondevice_calls":     s.onDeviceCalls,
		"cloud_calls":        s.cloudCalls,
		"ondevice_ratio":     float64(s.onDeviceCalls) / float64(total) * 100,
		"avg_latency_ms":     avgLatency,
	}
}

func main() {
	rand.Seed(time.Now().UnixNano())
	scheduler := NewHybridInferenceScheduler()

	// 模拟多种场景的推理请求
	scenarios := []InferenceRequest{
		{"打开手电筒", Simple, 200, false},
		{"明天天气如何", Simple, 300, false},
		{"帮我设置明天早上8点的闹钟", Simple, 300, false},
		{"这张图片里是什么", Medium, 1000, true},
		{"帮我搜索最近的演唱会信息", Medium, 2000, false},
		{"帮我预定9月29日晚北京直飞大理的机票，规划5天大理旅游攻略", VeryComplex, 5000, false},
		{"分析我的相册，找出所有包含猫的照片并按时间排序", Complex, 3000, true},
		{"播放周杰伦的音乐", Simple, 500, false},
		{"打开Chrome浏览器", Simple, 200, false},
		{"帮我写一篇5000字的市场分析报告", VeryComplex, 10000, false},
	}

	fmt.Println("=== YOYO Harness 端云协同推理调度模拟 ===")
	fmt.Printf("%-60s %-12s %-15s\n", "场景", "复杂度", "路由目标")
	fmt.Println("──────────────────────────────────────────────────────────────────────────────")

	for i, req := range scenarios {
		result := scheduler.Infer(req)
		fmt.Printf("[%2d] %-55s %-12d %s (%dms)\n",
			i+1, req.Input[:min(len(req.Input), 50)], req.Complexity,
			map[bool]string{true: "☁️ 云侧", false: "📱 端侧"}[result.FromCloud],
			result.LatencyMs)
	}

	fmt.Println("\n=== 调度统计 ===")
	stats := scheduler.GetStats()
	fmt.Printf("总请求: %d\n", stats["total_calls"])
	fmt.Printf("端侧推理: %d (%.1f%%)\n", stats["ondevice_calls"], stats["ondevice_ratio"])
	fmt.Printf("云侧推理: %d (%.1f%%)\n", stats["cloud_calls"], 100-stats["ondevice_ratio"].(float64))
	fmt.Printf("平均延迟: %dms\n", stats["avg_latency_ms"])
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

"""
YOYO 任务引擎 - 条件触发型自动化执行
支持40+触发条件和130+执行动作的灵活组合
"""

import abc
import enum
import time
import threading
import json
from typing import Optional, Callable, List, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta


class TaskStatus(enum.Enum):
    CREATED = "created"
    ACTIVE = "active"
    PAUSED = "paused"
    IDLE = "idle"
    EXECUTING = "executing"
    COMPLETED = "completed"
    RETRYING = "retrying"
    DELETED = "deleted"


class TriggerType(enum.Enum):
    TIME = "time"               # 定时触发
    LOCATION = "location"       # 地理围栏触发
    NETWORK = "network"         # 网络状态变更 (WiFi/蜂窝)
    BLUETOOTH = "bluetooth"     # 蓝牙设备连接/断开
    APP = "app"                 # 应用状态 (打开/关闭/时长)
    BATTERY = "battery"         # 电量阈值
    DEVICE = "device"           # 设备状态 (充电/横竖屏/运动)
    NOTIFICATION = "notification"  # 收到特定通知
    SYSTEM = "system"           # 系统事件 (开机/锁屏/解锁)
    GEOFENCE = "geofence"       # 地理围栏 (进入/离开)


@dataclass
class TriggerCondition:
    """触发条件"""
    trigger_type: TriggerType
    params: Dict[str, Any] = field(default_factory=dict)
    description: str = ""

    def matches(self, context: Dict[str, Any]) -> bool:
        """判断当前上下文是否满足触发条件"""
        if self.trigger_type == TriggerType.TIME:
            return self._check_time(context)
        elif self.trigger_type == TriggerType.LOCATION:
            return self._check_location(context)
        elif self.trigger_type == TriggerType.NETWORK:
            return self._check_network(context)
        elif self.trigger_type == TriggerType.BATTERY:
            return self._check_battery(context)
        elif self.trigger_type == TriggerType.APP:
            return self._check_app(context)
        elif self.trigger_type == TriggerType.BLUETOOTH:
            return self._check_bluetooth(context)
        return False

    def _check_time(self, ctx: Dict) -> bool:
        """时间条件匹配: 支持cron-like表达式"""
        current = datetime.now()
        if "days_of_week" in self.params:
            # 如: [1,3,5] 表示周一三五
            if current.weekday() not in self.params["days_of_week"]:
                return False
        if "hour" in self.params and "minute" in self.params:
            target = current.replace(
                hour=self.params["hour"],
                minute=self.params["minute"],
                second=0, microsecond=0
            )
            # 允许1分钟误差窗口
            return abs((current - target).total_seconds()) < 60
        return True

    def _check_location(self, ctx: Dict) -> bool:
        """地理围栏匹配"""
        if "fence_name" not in self.params:
            return False
        current_location = ctx.get("location", "")
        fence = self.params["fence_name"]
        return fence in current_location or current_location in fence

    def _check_network(self, ctx: Dict) -> bool:
        """网络状态匹配: connected_wifi / disconnected / specific_ssid"""
        expected = self.params.get("state", "")
        actual = ctx.get("wifi_state", "")
        if expected == "connected_wifi":
            return actual == "connected"
        elif expected == "specific_ssid":
            return ctx.get("wifi_ssid") == self.params.get("ssid")
        return False

    def _check_battery(self, ctx: Dict) -> bool:
        """电量条件匹配"""
        threshold = self.params.get("threshold", 20)
        direction = self.params.get("direction", "below")  # below/above
        current = ctx.get("battery_level", 100)
        if direction == "below":
            return current <= threshold
        return current >= threshold

    def _check_app(self, ctx: Dict) -> bool:
        """应用状态条件匹配"""
        app_name = self.params.get("app_name", "")
        app_state = self.params.get("app_state", "opened")  # opened/closed/duration
        current_app = ctx.get("foreground_app", "")
        if app_state == "opened":
            return current_app == app_name
        return False

    def _check_bluetooth(self, ctx: Dict) -> bool:
        """蓝牙设备条件匹配"""
        device = self.params.get("device_name", "")
        action = self.params.get("action", "connected")  # connected/disconnected
        devices = ctx.get("bluetooth_devices", [])
        if action == "connected":
            return device in devices
        return device not in devices


@dataclass
class ExecutionAction:
    """执行动作"""
    action_type: str  # e.g., "play_music", "close_app", "send_note", "price_protection"
    params: Dict[str, Any] = field(default_factory=dict)
    description: str = ""

    def execute(self) -> bool:
        """执行该动作（模拟）"""
        print(f"      执行动作: [{self.action_type}] {self.description}")
        print(f"        参数: {json.dumps(self.params, ensure_ascii=False)}")
        # 模拟执行耗时
        time.sleep(0.2)
        return True


@dataclass
class YOYOTask:
    """YOYO 任务定义"""
    id: str
    name: str
    description: str
    triggers: List[TriggerCondition] = field(default_factory=list)
    actions: List[ExecutionAction] = field(default_factory=list)
    status: TaskStatus = TaskStatus.CREATED
    max_retries: int = 3
    retry_count: int = 0
    created_at: str = ""
    last_executed_at: Optional[str] = None
    execute_count: int = 0
    is_recurring: bool = True  # 是否重复执行

    def should_trigger(self, context: Dict[str, Any]) -> bool:
        """判断是否应该触发（任意条件满足即可）"""
        for trigger in self.triggers:
            if trigger.matches(context):
                print(f"  ✅ 触发条件匹配: {trigger.description}")
                return True
        return False


class YOYOTaskEngine:
    """
    YOYO 任务引擎
    管理所有任务的生命周期，持续监控系统上下文，自动触发执行
    """

    def __init__(self):
        self.tasks: Dict[str, YOYOTask] = {}
        self._running = False
        self._monitor_thread: Optional[threading.Thread] = None
        self._current_context: Dict[str, Any] = {}

    def create_task(self, task: YOYOTask):
        """创建新任务"""
        task.status = TaskStatus.CREATED
        self.tasks[task.id] = task
        print(f"  📋 任务创建: [{task.id}] {task.name}")

    def activate_task(self, task_id: str):
        """激活任务，开始监控触发条件"""
        task = self.tasks.get(task_id)
        if task and task.status in (TaskStatus.CREATED, TaskStatus.PAUSED):
            task.status = TaskStatus.ACTIVE
            print(f"  ▶️ 任务激活: [{task_id}]")

    def pause_task(self, task_id: str):
        """暂停任务"""
        task = self.tasks.get(task_id)
        if task:
            task.status = TaskStatus.PAUSED

    def update_context(self, context: Dict[str, Any]):
        """更新系统上下文（由感知引擎推送）"""
        self._current_context = context

    def start_monitoring(self):
        """启动监控循环（模拟）"""
        self._running = True
        self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self._monitor_thread.start()

    def _monitor_loop(self):
        """监控循环：每秒检查一次所有激活任务的触发条件"""
        while self._running:
            for task in list(self.tasks.values()):
                if task.status != TaskStatus.ACTIVE:
                    continue

                if task.should_trigger(self._current_context):
                    self._execute_task(task)

            time.sleep(1)

    def _execute_task(self, task: YOYOTask):
        """执行任务（串行执行所有动作）"""
        task.status = TaskStatus.EXECUTING
        task.execute_count += 1
        task.last_executed_at = datetime.now().isoformat()

        print(f"\n  ⚡ 任务执行: [{task.id}] {task.name}")
        print(f"     步骤数: {len(task.actions)}")

        success = True
        for i, action in enumerate(task.actions):
            print(f"    步骤 {i+1}/{len(task.actions)}: ", end="")
            action_success = action.execute()
            if not action_success:
                print(f"    ❌ 步骤 {i+1} 失败")
                success = False
                break

        if success:
            task.status = TaskStatus.COMPLETED if not task.is_recurring else TaskStatus.ACTIVE
            print(f"  ✅ 任务[{task.id}] 执行完成 (共执行{task.execute_count}次)")
        else:
            task.retry_count += 1
            if task.retry_count <= task.max_retries:
                task.status = TaskStatus.RETRYING
                print(f"  🔄 任务[{task.id}] 失败，第{task.retry_count}次重试")
            else:
                task.status = TaskStatus.PAUSED
                print(f"  ❌ 任务[{task.id}] 超过最大重试次数，已暂停")


# ============ 模拟场景 ============
if __name__ == "__main__":
    engine = YOYOTaskEngine()

    # 场景A: 每晚8点京东一键价保 + 领加补券
    engine.create_task(YOYOTask(
        id="task_a001",
        name="京东每晚价保",
        description="每天晚上8点去京东一键价保，并领取限时加补券",
        triggers=[
            TriggerCondition(
                trigger_type=TriggerType.TIME,
                params={"hour": 20, "minute": 0},
                description="每晚20:00触发"
            )
        ],
        actions=[
            ExecutionAction(
                action_type="open_app",
                params={"app_name": "京东"},
                description="打开京东App"
            ),
            ExecutionAction(
                action_type="navigate",
                params={"page": "price_protection"},
                description="进入价保中心"
            ),
            ExecutionAction(
                action_type="click",
                params={"target": "one_click_protection"},
                description="一键价保"
            ),
            ExecutionAction(
                action_type="navigate",
                params={"page": "coupon_center"},
                description="领取加补券"
            ),
        ],
        is_recurring=True
    ))

    # 场景B: 连接蓝牙耳机时自动播放歌单
    engine.create_task(YOYOTask(
        id="task_a002",
        name="耳机连接自动播放",
        description="当蓝牙耳机连接时自动播放收藏歌单",
        triggers=[
            TriggerCondition(
                trigger_type=TriggerType.BLUETOOTH,
                params={"device_name": "Honor Earbuds 3 Pro", "action": "connected"},
                description="蓝牙耳机连接"
            )
        ],
        actions=[
            ExecutionAction(
                action_type="open_app",
                params={"app_name": "音乐App"},
                description="打开音乐App"
            ),
            ExecutionAction(
                action_type="play_playlist",
                params={"playlist_name": "我喜欢的音乐", "shuffle": True},
                description="播放收藏歌单（随机顺序）"
            ),
        ],
        is_recurring=True
    ))

    # 场景C: 刷抖音超30分钟时提醒并关闭
    engine.create_task(YOYOTask(
        id="task_a003",
        name="抖音使用时长控制",
        description="刷抖音超过30分钟时提醒并关闭",
        triggers=[
            TriggerCondition(
                trigger_type=TriggerType.APP,
                params={"app_name": "抖音", "app_state": "duration", "threshold_min": 30},
                description="抖音使用超过30分钟"
            )
        ],
        actions=[
            ExecutionAction(
                action_type="show_alert",
                params={"title": "休息提醒", "message": "您已刷抖音30分钟，建议休息一下"},
                description="弹出休息提醒"
            ),
            ExecutionAction(
                action_type="close_app",
                params={"app_name": "抖音"},
                description="关闭抖音"
            ),
        ],
        is_recurring=True
    ))

    # 激活所有任务
    print("\n=== 激活YOYO任务 ===")
    for task_id in engine.tasks:
        engine.activate_task(task_id)

    # 模拟系统上下文变化
    print("\n=== 模拟触发场景: 蓝牙耳机连接 ===\n")
    engine.update_context({
        "bluetooth_devices": ["Honor Earbuds 3 Pro"],
        "wifi_state": "connected",
        "battery_level": 85,
        "location": "家",
        "foreground_app": "桌面",
    })
    engine._execute_task(engine.tasks["task_a002"])

    print("\n=== 任务状态 ===")
    for task_id, task in engine.tasks.items():
        print(f"  [{task.status.value}] {task.name} (执行{task.execute_count}次)")