JarvisHub开源画布原生Agent框架深度解析：长程多模态创作的状态管理与Agent协作新范式

// canvas/state.go — Canvas State 核心实现

package canvas

import (
	"errors"
	"fmt"
	"sync"
	"time"
)

// NodeKind 定义画布节点类型
type NodeKind string

const (
	NodeKindText        NodeKind = "text"
	NodeKindImage       NodeKind = "image"
	NodeKindVideo       NodeKind = "video"
	NodeKindAudio       NodeKind = "audio"
	NodeKindStoryboard  NodeKind = "storyboard"
	NodeKindWebPage     NodeKind = "webpage"
	NodeKindSlide       NodeKind = "slide"
	NodeKindCharacter   NodeKind = "character"
	NodeKindScene       NodeKind = "scene"
	NodeKindReference   NodeKind = "reference"
	NodeKindCandidate   NodeKind = "candidate"
	NodeKindFeedback    NodeKind = "feedback"
)

// RuntimeStatus 节点运行时状态
type RuntimeStatus string

const (
	StatusPlanned   RuntimeStatus = "planned"
	StatusRunning   RuntimeStatus = "running"
	StatusCompleted RuntimeStatus = "completed"
	StatusFailed    RuntimeStatus = "failed"
	StatusAccepted  RuntimeStatus = "accepted"
	StatusRejected  RuntimeStatus = "rejected"
	StatusPendingReview RuntimeStatus = "pending_review"
)

// EdgeKind 边类型
type EdgeKind string

const (
	EdgeReference    EdgeKind = "reference"
	EdgeVersion      EdgeKind = "version"
	EdgeDependency   EdgeKind = "dependency"
	EdgeGroup        EdgeKind = "group"
	EdgeContinuation EdgeKind = "continuation"
)

// Node 画布节点
type Node struct {
	ID          string                 `json:"id"`
	Kind        NodeKind               `json:"kind"`
	Position    Position               `json:"position"`
	Input       map[string]interface{} `json:"input,omitempty"`
	Output      map[string]interface{} `json:"output,omitempty"`
	Metadata    Metadata               `json:"metadata"`
	Status      RuntimeStatus          `json:"status"`
	CreatedAt   time.Time              `json:"created_at"`
	UpdatedAt   time.Time              `json:"updated_at"`
}

// Position 节点位置
type Position struct {
	X       float64 `json:"x"`
	Y       float64 `json:"y"`
	Width   float64 `json:"width,omitempty"`
	Height  float64 `json:"height,omitempty"`
	GroupID string  `json:"group_id,omitempty"`
}

// Metadata 溯源与执行元数据
type Metadata struct {
	Provenance   string            `json:"provenance,omitempty"`
	ModelName    string            `json:"model_name,omitempty"`
	ModelParams  map[string]string `json:"model_params,omitempty"`
	ToolCallID   string            `json:"tool_call_id,omitempty"`
	ErrorMessage string            `json:"error_message,omitempty"`
	Duration     time.Duration     `json:"duration,omitempty"`
	Version      int               `json:"version"`
}

// Edge 有向边
type Edge struct {
	SourceID string   `json:"source_id"`
	TargetID string   `json:"target_id"`
	Kind     EdgeKind `json:"kind"`
	Label    string   `json:"label,omitempty"`
}

// CanvasState 画布状态
type CanvasState struct {
	mu       sync.RWMutex
	Nodes    map[string]*Node   `json:"nodes"`
	Edges    []Edge             `json:"edges"`
	UserData map[string]interface{} `json:"user_data,omitempty"`
	Version  int                `json:"version"`
}

// NewCanvasState 创建新的画布状态
func NewCanvasState() *CanvasState {
	return &CanvasState{
		Nodes:    make(map[string]*Node),
		Edges:    make([]Edge, 0),
		UserData: make(map[string]interface{}),
		Version:  0,
	}
}

// AddNode 添加节点
func (cs *CanvasState) AddNode(n *Node) error {
	cs.mu.Lock()
	defer cs.mu.Unlock()

	if _, exists := cs.Nodes[n.ID]; exists {
		return fmt.Errorf("node %s already exists", n.ID)
	}
	n.CreatedAt = time.Now()
	n.UpdatedAt = time.Now()
	cs.Nodes[n.ID] = n
	cs.Version++
	return nil
}

// GetNode 获取节点
func (cs *CanvasState) GetNode(id string) (*Node, error) {
	cs.mu.RLock()
	defer cs.mu.RUnlock()

	n, exists := cs.Nodes[id]
	if !exists {
		return nil, fmt.Errorf("node %s not found", id)
	}
	return n, nil
}

// UpdateNode 更新节点内容
func (cs *CanvasState) UpdateNode(id string, updateFn func(*Node)) error {
	cs.mu.Lock()
	defer cs.mu.Unlock()

	n, exists := cs.Nodes[id]
	if !exists {
		return fmt.Errorf("node %s not found", id)
	}
	updateFn(n)
	n.UpdatedAt = time.Now()
	cs.Version++
	return nil
}

// AddEdge 添加边
func (cs *CanvasState) AddEdge(sourceID, targetID string, kind EdgeKind, label string) error {
	cs.mu.Lock()
	defer cs.mu.Unlock()

	if _, ok := cs.Nodes[sourceID]; !ok {
		return fmt.Errorf("source node %s not found", sourceID)
	}
	if _, ok := cs.Nodes[targetID]; !ok {
		return fmt.Errorf("target node %s not found", targetID)
	}

	cs.Edges = append(cs.Edges, Edge{
		SourceID: sourceID,
		TargetID: targetID,
		Kind:     kind,
		Label:    label,
	})
	cs.Version++
	return nil
}

// GetDependents 获取依赖于指定节点的所有下游节点
func (cs *CanvasState) GetDependents(nodeID string) []*Node {
	cs.mu.RLock()
	defer cs.mu.RUnlock()

	var dependents []*Node
	for _, e := range cs.Edges {
		if e.SourceID == nodeID && e.Kind == EdgeDependency {
			if n, ok := cs.Nodes[e.TargetID]; ok {
				dependents = append(dependents, n)
			}
		}
	}
	return dependents
}

// GetUpstreamNodes 获取指定节点的所有上游依赖
func (cs *CanvasState) GetUpstreamNodes(nodeID string) []*Node {
	cs.mu.RLock()
	defer cs.mu.RUnlock()

	var upstream []*Node
	for _, e := range cs.Edges {
		if e.TargetID == nodeID && e.Kind == EdgeDependency {
			if n, ok := cs.Nodes[e.SourceID]; ok {
				upstream = append(upstream, n)
			}
		}
	}
	return upstream
}

// Snapshot 创建画布快照（用于检查点/恢复）
func (cs *CanvasState) Snapshot() *CanvasState {
	cs.mu.RLock()
	defer cs.mu.RUnlock()

	snap := &CanvasState{
		Nodes:    make(map[string]*Node),
		Edges:    make([]Edge, len(cs.Edges)),
		UserData: make(map[string]interface{}),
		Version:  cs.Version,
	}
	for k, v := range cs.Nodes {
		cp := *v
		snap.Nodes[k] = &cp
	}
	copy(snap.Edges, cs.Edges)
	for k, v := range cs.UserData {
		snap.UserData[k] = v
	}
	return snap
}

# canvas_state.py — Python 画布状态实现

from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
from datetime import datetime
import copy
import threading


class NodeKind(str, Enum):
    TEXT = "text"
    IMAGE = "image"
    VIDEO = "video"
    AUDIO = "audio"
    STORYBOARD = "storyboard"
    WEBPAGE = "webpage"
    SLIDE = "slide"
    CHARACTER = "character"
    SCENE = "scene"
    REFERENCE = "reference"
    CANDIDATE = "candidate"
    FEEDBACK = "feedback"


class RuntimeStatus(str, Enum):
    PLANNED = "planned"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    PENDING_REVIEW = "pending_review"


class EdgeKind(str, Enum):
    REFERENCE = "reference"
    VERSION = "version"
    DEPENDENCY = "dependency"
    GROUP = "group"
    CONTINUATION = "continuation"


@dataclass
class Position:
    x: float
    y: float
    width: Optional[float] = None
    height: Optional[float] = None
    group_id: Optional[str] = None


@dataclass
class Metadata:
    provenance: Optional[str] = None
    model_name: Optional[str] = None
    model_params: Dict[str, str] = field(default_factory=dict)
    tool_call_id: Optional[str] = None
    error_message: Optional[str] = None
    duration_ms: Optional[float] = None
    version: int = 1


@dataclass
class Node:
    id: str
    kind: NodeKind
    position: Position
    input: Dict[str, Any] = field(default_factory=dict)
    output: Dict[str, Any] = field(default_factory=dict)
    metadata: Metadata = field(default_factory=Metadata)
    status: RuntimeStatus = RuntimeStatus.PLANNED
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None

    def __post_init__(self):
        now = datetime.now()
        self.created_at = self.created_at or now
        self.updated_at = self.updated_at or now


@dataclass
class Edge:
    source_id: str
    target_id: str
    kind: EdgeKind
    label: Optional[str] = None


class CanvasState:
    """画布状态 —— 线程安全的有类型资产图"""

    def __init__(self):
        self._lock = threading.RLock()
        self.nodes: Dict[str, Node] = {}
        self.edges: List[Edge] = []
        self.user_data: Dict[str, Any] = {}
        self.version: int = 0

    def add_node(self, node: Node) -> None:
        with self._lock:
            if node.id in self.nodes:
                raise ValueError(f"Node {node.id} already exists")
            node.created_at = datetime.now()
            node.updated_at = datetime.now()
            self.nodes[node.id] = node
            self.version += 1

    def get_node(self, node_id: str) -> Optional[Node]:
        with self._lock:
            return self.nodes.get(node_id)

    def update_node(self, node_id: str, **updates) -> None:
        with self._lock:
            node = self.nodes.get(node_id)
            if node is None:
                raise KeyError(f"Node {node_id} not found")
            for key, value in updates.items():
                if hasattr(node, key):
                    setattr(node, key, value)
            node.updated_at = datetime.now()
            self.version += 1

    def add_edge(self, source_id: str, target_id: str,
                 kind: EdgeKind, label: Optional[str] = None) -> None:
        with self._lock:
            if source_id not in self.nodes:
                raise KeyError(f"Source node {source_id} not found")
            if target_id not in self.nodes:
                raise KeyError(f"Target node {target_id} not found")
            self.edges.append(Edge(source_id, target_id, kind, label))
            self.version += 1

    def get_dependents(self, node_id: str) -> List[Node]:
        """获取依赖指定节点的所有下游节点"""
        with self._lock:
            return [
                self.nodes[e.target_id]
                for e in self.edges
                if e.source_id == node_id and e.kind == EdgeKind.DEPENDENCY
                and e.target_id in self.nodes
            ]

    def get_upstream(self, node_id: str) -> List[Node]:
        """获取指定节点的所有上游依赖"""
        with self._lock:
            return [
                self.nodes[e.source_id]
                for e in self.edges
                if e.target_id == node_id and e.kind == EdgeKind.DEPENDENCY
                and e.source_id in self.nodes
            ]

    def get_nodes_by_status(self, status: RuntimeStatus) -> List[Node]:
        with self._lock:
            return [n for n in self.nodes.values() if n.status == status]

    def get_nodes_by_kind(self, kind: NodeKind) -> List[Node]:
        with self._lock:
            return [n for n in self.nodes.values() if n.kind == kind]

    def snapshot(self) -> 'CanvasState':
        """创建不可变快照用于检查点恢复"""
        with self._lock:
            snap = CanvasState()
            snap.nodes = copy.deepcopy(self.nodes)
            snap.edges = copy.deepcopy(self.edges)
            snap.user_data = copy.deepcopy(self.user_data)
            snap.version = self.version
            return snap

// bridge/protocol.go — Protocol Bridge 核心实现

package bridge

import (
	"errors"
	"fmt"
	"time"

	"jarvishub/canvas"
)

// OperationType 画布操作类型
type OperationType string

const (
	OpCreateNode  OperationType = "create_node"
	OpUpdateNode  OperationType = "update_node"
	OpDeleteNode  OperationType = "delete_node"
	OpConnectEdge OperationType = "connect_edge"
	OpGroupNodes  OperationType = "group_nodes"
	OpSelectNode  OperationType = "select_node"
	OpBranchNode  OperationType = "branch_node"
)

// ToolFamily 工具族
type ToolFamily string

const (
	ToolCanvas     ToolFamily = "canvas"
	ToolGeneration ToolFamily = "generation"
	ToolNative     ToolFamily = "native"
	ToolRecovery   ToolFamily = "recovery"
	ToolMCP        ToolFamily = "mcp"
)

// CapabilityManifest 能力清单
type CapabilityManifest struct {
	AllowedNodeKinds []canvas.NodeKind   `json:"allowed_node_kinds"`
	AllowedOps       []OperationType     `json:"allowed_ops"`
	AvailableTools   []ToolFamily        `json:"available_tools"`
	ArtifactHandles  []string            `json:"artifact_handles"`
	Constraints      map[string]string   `json:"constraints,omitempty"`
}

// ExecutionGrant 执行授权
type ExecutionGrant struct {
	MaxNodes       int              `json:"max_nodes,omitempty"`
	AllowedActions []OperationType  `json:"allowed_actions"`
	ToolGrants     map[ToolFamily]bool `json:"tool_grants"`
	MaxTokens      int              `json:"max_tokens,omitempty"`
	TimeoutMs      int64            `json:"timeout_ms,omitempty"`
}

// Action 代理动作
type Action struct {
	Type      OperationType          `json:"type"`
	Payload   map[string]interface{} `json:"payload"`
	ToolCalls []ToolCall             `json:"tool_calls,omitempty"`
	Timestamp time.Time              `json:"timestamp"`
}

// ToolCall 工具调用
type ToolCall struct {
	Family    ToolFamily             `json:"family"`
	Name      string                 `json:"name"`
	Params    map[string]interface{} `json:"params"`
	RequestID string                 `json:"request_id"`
}

// ToolObservation 工具观察结果
type ToolObservation struct {
	ToolCallID string                 `json:"tool_call_id"`
	Success    bool                   `json:"success"`
	Output     map[string]interface{} `json:"output,omitempty"`
	Artifacts  []string               `json:"artifacts,omitempty"`
	Error      string                 `json:"error,omitempty"`
	DurationMs int64                  `json:"duration_ms"`
}

// FeedbackSignal 反馈信号
type FeedbackSignal struct {
	Source    string                 `json:"source"` // "human", "model_critic", "subagent", "evaluator"
	NodeID    string                 `json:"node_id,omitempty"`
	Verdict   string                 `json:"verdict"` // "accept", "reject", "partial", "clarify"
	Comments  string                 `json:"comments,omitempty"`
	Timestamp time.Time              `json:"timestamp"`
	Edits     map[string]interface{} `json:"edits,omitempty"`
}

// RepairDecision 修复决策
type RepairDecision struct {
	Action      string                 `json:"action"` // "continue", "repair", "clarify", "stop"
	TargetNodes []string               `json:"target_nodes,omitempty"`
	Strategy    string                 `json:"strategy,omitempty"`
	Params      map[string]interface{} `json:"params,omitempty"`
}

// StateTransition 状态转换记录
type StateTransition struct {
	Turn        int                `json:"turn"`
	BeforeState *canvas.CanvasState `json:"before_state"`
	Action      *Action            `json:"action"`
	Observation *ToolObservation   `json:"observation,omitempty"`
	Feedback    *FeedbackSignal    `json:"feedback,omitempty"`
	Repair      *RepairDecision    `json:"repair,omitempty"`
	AfterState  *canvas.CanvasState `json:"after_state"`
	Timestamp   time.Time          `json:"timestamp"`
}

// ProtocolBridge 协议桥
type ProtocolBridge struct {
	state       *canvas.CanvasState
	manifest    *CapabilityManifest
	grant       *ExecutionGrant
	transitions []StateTransition
	turnCount   int
}

// NewProtocolBridge 创建协议桥
func NewProtocolBridge(state *canvas.CanvasState) *ProtocolBridge {
	return &ProtocolBridge{
		state:       state,
		transitions: make([]StateTransition, 0),
	}
}

// BuildManifest 构建当前轮次的能力清单
func (pb *ProtocolBridge) BuildManifest(projectType string) *CapabilityManifest {
	// 根据项目类型动态构建清单
	manifest := &CapabilityManifest{
		AllowedOps: []OperationType{
			OpCreateNode, OpUpdateNode, OpConnectEdge,
		},
		AvailableTools: []ToolFamily{
			ToolCanvas, ToolGeneration, ToolNative, ToolRecovery,
		},
		ArtifactHandles: make([]string, 0),
	}

	// 从画布状态中收集已有资产
	for id := range pb.state.Nodes {
		manifest.ArtifactHandles = append(manifest.ArtifactHandles, id)
	}

	// 项目类型特定配置
	switch projectType {
	case "narrative_media":
		manifest.AllowedNodeKinds = []canvas.NodeKind{
			canvas.NodeKindText, canvas.NodeKindImage,
			canvas.NodeKindVideo, canvas.NodeKindAudio,
			canvas.NodeKindStoryboard, canvas.NodeKindCharacter,
			canvas.NodeKindScene, canvas.NodeKindReference,
			canvas.NodeKindCandidate,
		}
	case "web_development":
		manifest.AllowedNodeKinds = []canvas.NodeKind{
			canvas.NodeKindText, canvas.NodeKindImage,
			canvas.NodeKindWebPage, canvas.NodeKindReference,
			canvas.NodeKindCandidate,
		}
		manifest.AvailableTools = append(manifest.AvailableTools, ToolMCP)
	case "presentation":
		manifest.AllowedNodeKinds = []canvas.NodeKind{
			canvas.NodeKindText, canvas.NodeKindImage,
			canvas.NodeKindSlide, canvas.NodeKindReference,
			canvas.NodeKindCandidate,
		}
	}

	pb.manifest = manifest
	return manifest
}

// DeriveGrant 推导执行授权
func (pb *ProtocolBridge) DeriveGrant(userQuery string) *ExecutionGrant {
	// 根据用户查询和当前画布状态推导授权
	grant := &ExecutionGrant{
		MaxNodes: 100,
		AllowedActions: []OperationType{
			OpCreateNode, OpUpdateNode, OpConnectEdge,
		},
		ToolGrants: map[ToolFamily]bool{
			ToolCanvas:     true,
			ToolGeneration: true,
			ToolNative:     true,
			ToolRecovery:   true,
		},
		TimeoutMs: 300000, // 5分钟
	}
	pb.grant = grant
	return grant
}

// ValidateAction 校验动作是否合法
func (pb *ProtocolBridge) ValidateAction(action *Action) error {
	// 检查操作类型是否被授权
	allowed := false
	for _, op := range pb.grant.AllowedActions {
		if op == action.Type {
			allowed = true
			break
		}
	}
	if !allowed {
		return fmt.Errorf("operation %s is not in current execution grant", action.Type)
	}

	// 检查工具调用是否被授权
	for _, tc := range action.ToolCalls {
		if granted, ok := pb.grant.ToolGrants[tc.Family]; !ok || !granted {
			return fmt.Errorf("tool family %s is not granted", tc.Family)
		}
	}

	return nil
}

// CommitTransition 提交状态转换并记录轨迹
func (pb *ProtocolBridge) CommitTransition(
	before *canvas.CanvasState,
	action *Action,
	observation *ToolObservation,
	feedback *FeedbackSignal,
	repair *RepairDecision,
	after *canvas.CanvasState,
) StateTransition {
	pb.turnCount++
	transition := StateTransition{
		Turn:        pb.turnCount,
		BeforeState: before,
		Action:      action,
		Observation: observation,
		Feedback:    feedback,
		Repair:      repair,
		AfterState:  after,
		Timestamp:   time.Now(),
	}
	pb.transitions = append(pb.transitions, transition)
	return transition
}

// GetTrajectory 获取完整执行轨迹
func (pb *ProtocolBridge) GetTrajectory() []StateTransition {
	return pb.transitions
}

# runtime/agent_runtime.py — Agent Runtime 核心实现

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Any
from enum import Enum
import time
import logging

from canvas_state import CanvasState, Node, NodeKind, RuntimeStatus, EdgeKind
from protocol_bridge import (ProtocolBridge, CapabilityManifest, ExecutionGrant,
                             Action, OperationType, ToolCall, ToolObservation,
                             FeedbackSignal, RepairDecision, ToolFamily)


logger = logging.getLogger(__name__)


@dataclass
class RuntimeContext:
    """运行时上下文"""
    user_query: str
    canvas_state: CanvasState
    manifest: CapabilityManifest
    grant: ExecutionGrant
    skills: Dict[str, 'Skill']
    memory: Dict[str, Any]
    subagents: List['SubAgent']


class AgentRuntime:
    """Agent执行运行时"""

    def __init__(self, bridge: ProtocolBridge, state: CanvasState):
        self.bridge = bridge
        self.state = state
        self.skills: Dict[str, Skill] = {}
        self.memory: Dict[str, Any] = {}
        self.subagents: List[SubAgent] = []
        self.tool_handlers: Dict[ToolFamily, ToolHandler] = {}
        self.turn_count = 0

    def register_tool_handler(self, family: ToolFamily, handler: ToolHandler):
        self.tool_handlers[family] = handler

    def register_skill(self, name: str, skill: Skill):
        self.skills[name] = skill

    def execute_turn(self, user_query: str) -> Dict[str, Any]:
        """执行一轮运行时循环"""
        self.turn_count += 1
        logger.info(f"=== Turn {self.turn_count} ===")

        # 步骤1: 构建能力清单
        manifest = self._build_manifest()

        # 步骤2: 推导执行授权
        grant = self._derive_grant(user_query)

        # 步骤3: 创建运行时上下文
        ctx = RuntimeContext(
            user_query=user_query,
            canvas_state=self.state.snapshot(),
            manifest=manifest,
            grant=grant,
            skills=self.skills,
            memory=self.memory,
            subagents=self.subagents,
        )

        # 步骤4: 选择动作（由LLM驱动，这里模拟）
        action = self._select_action(ctx)

        # 步骤5: 校验动作
        validation_error = self._validate_action(action, grant)
        if validation_error:
            return {"error": validation_error, "turn": self.turn_count}

        # 步骤6: 执行动作
        before_state = self.state.snapshot()
        observation = self._execute_action(action)

        # 步骤7: 接收反馈
        feedback = self._collect_feedback(observation, ctx)

        # 步骤8: 确定修复决策
        repair = self._decide_repair(feedback, before_state)

        # 步骤9: 执行修复
        if repair and repair.action == "repair":
            self._apply_repair(repair, before_state)
        elif repair and repair.action == "stop":
            logger.warning("Agent stopped due to insufficient evidence")
            return {"status": "stopped", "reason": repair.strategy}

        # 步骤10: 提交状态转换
        after_state = self.state.snapshot()
        transition = self.bridge.commit_transition(
            before=before_state,
            action=action,
            observation=observation,
            feedback=feedback,
            repair=repair,
            after=after_state,
        )

        return {
            "turn": self.turn_count,
            "action": action,
            "observation": observation,
            "feedback": feedback,
            "repair": repair,
            "transition_recorded": True,
        }

    def _build_manifest(self) -> CapabilityManifest:
        project_type = self.memory.get("project_type", "general")
        return self.bridge.build_manifest(project_type)

    def _derive_grant(self, query: str) -> ExecutionGrant:
        return self.bridge.derive_grant(query)

    def _select_action(self, ctx: RuntimeContext) -> Action:
        """
        选择动作。在实际系统中由LLM Agent驱动，
        这里展示核心逻辑结构。
        """
        # 1. 检查是否有匹配的Skill
        for skill_name, skill in self.skills.items():
            if skill.should_apply(ctx):
                logger.info(f"Applying skill: {skill_name}")
                return skill.generate_action(ctx)

        # 2. 默认：创建新节点并调用生成工具
        return Action(
            type=OperationType.CREATE_NODE,
            payload={
                "kind": "image",
                "prompt": ctx.user_query,
            },
            tool_calls=[
                ToolCall(
                    family=ToolFamily.GENERATION,
                    name="generate_image",
                    params={"prompt": ctx.user_query},
                    request_id=f"req_{self.turn_count}",
                )
            ],
        )

    def _validate_action(self, action: Action, grant: ExecutionGrant) -> Optional[str]:
        try:
            return self.bridge.validate_action(action)
        except Exception as e:
            return str(e)

    def _execute_action(self, action: Action) -> ToolObservation:
        """执行动作并返回观察结果"""
        start = time.time()

        all_results = []
        for tc in action.tool_calls:
            handler = self.tool_handlers.get(tc.family)
            if handler is None:
                return ToolObservation(
                    tool_call_id=tc.request_id,
                    success=False,
                    error=f"No handler for tool family: {tc.family}",
                    duration_ms=int((time.time() - start) * 1000),
                )
            try:
                result = handler.execute(tc)
                all_results.append(result)
            except Exception as e:
                all_results.append(ToolObservation(
                    tool_call_id=tc.request_id,
                    success=False,
                    error=str(e),
                    duration_ms=int((time.time() - start) * 1000),
                ))

        # 如果是创建节点的操作，将结果写入画布
        if action.type == OperationType.CREATE_NODE and all_results:
            result = all_results[0]
            if result.success:
                node = Node(
                    id=f"node_{self.turn_count}_{int(time.time())}",
                    kind=NodeKind(action.payload.get("kind", "text")),
                    position={"x": 100, "y": 100 * self.turn_count},
                    input=action.payload,
                    output=result.output or {},
                    status=RuntimeStatus.COMPLETED,
                )
                self.state.add_node(node)

        # 合并观察结果
        merged = ToolObservation(
            tool_call_id=action.tool_calls[0].request_id if action.tool_calls else "",
            success=all(r.success for r in all_results),
            output={},
            artifacts=[],
            duration_ms=int((time.time() - start) * 1000),
        )
        for r in all_results:
            if r.output:
                merged.output.update(r.output)
            if r.artifacts:
                merged.artifacts.extend(r.artifacts)

        return merged

    def _collect_feedback(self, obs: ToolObservation, ctx: RuntimeContext) -> Optional[FeedbackSignal]:
        """收集反馈信号（来自用户、模型评估器或子Agent）"""
        if not obs.success:
            return FeedbackSignal(
                source="evaluator",
                verdict="reject",
                comments=f"Tool execution failed: {obs.error}",
            )
        return None

    def _decide_repair(self, feedback: Optional[FeedbackSignal],
                       state: CanvasState) -> Optional[RepairDecision]:
        """根据反馈决定修复策略"""
        if feedback is None:
            return RepairDecision(action="continue")

        if feedback.verdict == "reject":
            # 查找失败的节点
            failed_nodes = state.get_nodes_by_status(RuntimeStatus.FAILED)
            if failed_nodes:
                return RepairDecision(
                    action="repair",
                    target_nodes=[n.id for n in failed_nodes],
                    strategy="regenerate_with_feedback",
                    params={"feedback": feedback.comments},
                )
            return RepairDecision(action="clarify", strategy="ask_user_for_guidance")

        return RepairDecision(action="continue")

    def _apply_repair(self, repair: RepairDecision, state: CanvasState):
        """应用修复决策"""
        for node_id in repair.target_nodes:
            node = state.get_node(node_id)
            if node:
                # 标记为待修复，重新规划
                state.update_node(node_id, status=RuntimeStatus.PLANNED)
                logger.info(f"Repair scheduled for node {node_id}: {repair.strategy}")

// runtime/runtime.go — Agent Runtime Go实现

package runtime

import (
	"context"
	"fmt"
	"log"
	"time"

	"jarvishub/bridge"
	"jarvishub/canvas"
)

// ToolHandler 工具处理函数
type ToolHandler func(ctx context.Context, call bridge.ToolCall) (*bridge.ToolObservation, error)

// RuntimeConfig 运行时配置
type RuntimeConfig struct {
	MaxTurns       int
	DefaultTimeout time.Duration
	ModelName      string
}

// AgentRuntime 代理运行时
type AgentRuntime struct {
	config       RuntimeConfig
	bridge       *bridge.ProtocolBridge
	state        *canvas.CanvasState
	skills       map[string]Skill
	memory       map[string]interface{}
	toolHandlers map[bridge.ToolFamily]ToolHandler
	turnCount    int
}

// Skill 技能接口
type Skill interface {
	Name() string
	ShouldApply(ctx context.Context, query string, state *canvas.CanvasState) bool
	Execute(ctx context.Context, query string, state *canvas.CanvasState) (*bridge.Action, error)
}

// NewAgentRuntime 创建Agent运行时
func NewAgentRuntime(config RuntimeConfig, bridge *bridge.ProtocolBridge, state *canvas.CanvasState) *AgentRuntime {
	return &AgentRuntime{
		config:       config,
		bridge:       bridge,
		state:        state,
		skills:       make(map[string]Skill),
		memory:       make(map[string]interface{}),
		toolHandlers: make(map[bridge.ToolFamily]ToolHandler),
	}
}

// RegisterToolHandler 注册工具处理函数
func (rt *AgentRuntime) RegisterToolHandler(family bridge.ToolFamily, handler ToolHandler) {
	rt.toolHandlers[family] = handler
}

// RegisterSkill 注册技能
func (rt *AgentRuntime) RegisterSkill(skill Skill) {
	rt.skills[skill.Name()] = skill
}

// ExecuteTurn 执行一轮运行时循环
func (rt *AgentRuntime) ExecuteTurn(ctx context.Context, userQuery string) (map[string]interface{}, error) {
	rt.turnCount++
	log.Printf("=== Turn %d ===\n", rt.turnCount)

	// 1. 构建能力清单
	projectType := "general"
	if pt, ok := rt.memory["project_type"].(string); ok {
		projectType = pt
	}
	manifest := rt.bridge.BuildManifest(projectType)

	// 2. 推导执行授权
	grant := rt.bridge.DeriveGrant(userQuery)

	// 3. 选择动作
	action, err := rt.selectAction(ctx, userQuery, manifest, grant)
	if err != nil {
		return nil, fmt.Errorf("action selection failed: %w", err)
	}

	// 4. 校验动作
	if err := rt.bridge.ValidateAction(action); err != nil {
		return nil, fmt.Errorf("action validation failed: %w", err)
	}

	// 5. 执行动作
	beforeState := rt.state.Snapshot()
	observation, err := rt.executeAction(ctx, action)
	if err != nil {
		observation = &bridge.ToolObservation{
			ToolCallID: action.ToolCalls[0].RequestID,
			Success:    false,
			Error:      err.Error(),
		}
	}

	// 6. 收集反馈
	feedback := rt.collectFeedback(observation)

	// 7. 决定修复策略
	repair := rt.decideRepair(feedback, beforeState)

	// 8. 应用修复
	if repair != nil && repair.Action == "repair" {
		rt.applyRepair(ctx, repair, beforeState)
	}

	// 9. 提交状态转换
	afterState := rt.state.Snapshot()
	rt.bridge.CommitTransition(beforeState, action, observation, feedback, repair, afterState)

	return map[string]interface{}{
		"turn":       rt.turnCount,
		"action":     action,
		"observation": observation,
	}, nil
}

func (rt *AgentRuntime) selectAction(
	ctx context.Context,
	query string,
	manifest *bridge.CapabilityManifest,
	grant *bridge.ExecutionGrant,
) (*bridge.Action, error) {
	// 检查是否有匹配的Skill
	for name, skill := range rt.skills {
		if skill.ShouldApply(ctx, query, rt.state) {
			log.Printf("Applying skill: %s\n", name)
			return skill.Execute(ctx, query, rt.state)
		}
	}

	// 默认动作：创建节点
	return &bridge.Action{
		Type: bridge.OpCreateNode,
		Payload: map[string]interface{}{
			"kind":   "image",
			"prompt": query,
		},
		ToolCalls: []bridge.ToolCall{
			{
				Family:    bridge.ToolGeneration,
				Name:      "generate_image",
				Params:    map[string]interface{}{"prompt": query},
				RequestID: fmt.Sprintf("req_%d", rt.turnCount),
			},
		},
	}, nil
}

func (rt *AgentRuntime) executeAction(ctx context.Context, action *bridge.Action) (*bridge.ToolObservation, error) {
	start := time.Now()

	for _, tc := range action.ToolCalls {
		handler, ok := rt.toolHandlers[tc.Family]
		if !ok {
			return nil, fmt.Errorf("no handler for tool family: %s", tc.Family)
		}
		obs, err := handler(ctx, tc)
		if err != nil {
			return nil, err
		}
		_ = obs
	}

	return &bridge.ToolObservation{
		ToolCallID: action.ToolCalls[0].RequestID,
		Success:    true,
		DurationMs: time.Since(start).Milliseconds(),
	}, nil
}

func (rt *AgentRuntime) collectFeedback(obs *bridge.ToolObservation) *bridge.FeedbackSignal {
	if !obs.Success {
		return &bridge.FeedbackSignal{
			Source:   "evaluator",
			Verdict:  "reject",
			Comments: fmt.Sprintf("Tool execution failed: %s", obs.Error),
		}
	}
	return nil
}

func (rt *AgentRuntime) decideRepair(feedback *bridge.FeedbackSignal, state *canvas.CanvasState) *bridge.RepairDecision {
	if feedback == nil {
		return &bridge.RepairDecision{Action: "continue"}
	}
	if feedback.Verdict == "reject" {
		return &bridge.RepairDecision{
			Action:   "repair",
			Strategy: "regenerate_with_feedback",
			Params:   map[string]interface{}{"feedback": feedback.Comments},
		}
	}
	return &bridge.RepairDecision{Action: "continue"}
}

func (rt *AgentRuntime) applyRepair(ctx context.Context, repair *bridge.RepairDecision, state *canvas.CanvasState) {
	for _, nodeID := range repair.TargetNodes {
		node, err := state.GetNode(nodeID)
		if err == nil && node != nil {
			state.UpdateNode(nodeID, func(n *canvas.Node) {
				n.Status = canvas.StatusPlanned
			})
			log.Printf("Repair scheduled for node %s: %s\n", nodeID, repair.Strategy)
		}
	}
}

# repair/local_repair.py — 局部修复算法实现

from typing import List, Optional, Set
from dataclasses import dataclass

from canvas_state import (
    CanvasState, Node, NodeKind, RuntimeStatus, EdgeKind
)


@dataclass
class RepairPlan:
    """修复计划"""
    failed_node_id: str
    affected_downstream: List[str]  # 需要级联修复的节点
    preserved_upstream: List[str]   # 保留的上游依赖
    repair_strategy: str
    feedback_context: str


class LocalRepairEngine:
    """局部修复引擎"""

    def __init__(self, state: CanvasState):
        self.state = state

    def analyze_failure(self, feedback: str, failed_node_id: str) -> RepairPlan:
        """
        分析失败并制定修复计划

        核心算法:
        1. 定位失败节点及其下游依赖
        2. 保留上游依赖（不重新生成）
        3. 只标记受影响的下游节点需要修复
        """
        node = self.state.get_node(failed_node_id)
        if not node:
            raise ValueError(f"Node {failed_node_id} not found")

        # 找到所有需要级联修复的下游节点
        affected = self._find_affected_downstream(failed_node_id)

        # 找到可以保留的上游依赖
        preserved = self._find_preserved_upstream(failed_node_id)

        # 确定修复策略
        strategy = self._determine_strategy(node, feedback)

        return RepairPlan(
            failed_node_id=failed_node_id,
            affected_downstream=affected,
            preserved_upstream=preserved,
            repair_strategy=strategy,
            feedback_context=feedback,
        )

    def _find_affected_downstream(self, node_id: str) -> List[str]:
        """
        BFS遍历下游依赖图，找到所有受影响的节点
        """
        affected: Set[str] = set()
        queue = [node_id]

        while queue:
            current = queue.pop(0)
            dependents = self.state.get_dependents(current)
            for dep in dependents:
                if dep.id not in affected:
                    affected.add(dep.id)
                    queue.append(dep.id)

        return list(affected)

    def _find_preserved_upstream(self, node_id: str) -> List[str]:
        """
        找到可以保留的上游依赖（这些节点不受影响）
        """
        preserved: Set[str] = set()
        queue = [node_id]

        while queue:
            current = queue.pop(0)
            upstream = self.state.get_upstream(current)
            for up in upstream:
                if up.id not in preserved and up.status == RuntimeStatus.COMPLETED:
                    preserved.add(up.id)
                    queue.append(up.id)

        return list(preserved)

    def _determine_strategy(self, node: Node, feedback: str) -> str:
        """根据节点类型和反馈确定修复策略"""
        strategy_map = {
            NodeKind.IMAGE: "regenerate_with_feedback",
            NodeKind.VIDEO: "regenerate_segment",
            NodeKind.STORYBOARD: "revise_layout",
            NodeKind.TEXT: "revise_with_feedback",
            NodeKind.WEBPAGE: "patch_element",
            NodeKind.SLIDE: "revise_slide",
        }
        return strategy_map.get(node.kind, "regenerate_with_feedback")

    def apply_repair(self, plan: RepairPlan) -> int:
        """
        应用修复计划

        返回: 修复的节点数
        """
        repair_count = 0

        # 标记失败节点为待修复
        self.state.update_node(
            plan.failed_node_id,
            status=RuntimeStatus.PLANNED,
            metadata={
                "repair_strategy": plan.repair_strategy,
                "feedback_context": plan.feedback_context,
                "repair_attempts": 1,
            }
        )
        repair_count += 1

        # 标记受影响的下游节点为待修复
        for node_id in plan.affected_downstream:
            node = self.state.get_node(node_id)
            if node and node.status in (RuntimeStatus.COMPLETED, RuntimeStatus.ACCEPTED):
                # 标记为待修复，但保留输出历史
                self.state.update_node(
                    node_id,
                    status=RuntimeStatus.PLANNED,
                )
                repair_count += 1

        # 确认上游依赖保持不变
        logger.info(
            f"Repair plan: {repair_count} nodes to repair, "
            f"{len(plan.preserved_upstream)} upstream nodes preserved"
        )

        return repair_count