JarvisHub Deep Dive: Canvas-Native State Management and Agent Orchestration for Long-Horizon Multimodal Creation

// canvas/state.go — Canvas State Core Implementation

package canvas

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

// NodeKind defines canvas node types
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 node runtime status
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 edge type
type EdgeKind string

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

// Node canvas 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 node 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 provenance and execution 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 directed edge
type Edge struct {
	SourceID string   `json:"source_id"`
	TargetID string   `json:"target_id"`
	Kind     EdgeKind `json:"kind"`
	Label    string   `json:"label,omitempty"`
}

// CanvasState canvas state
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 creates a new canvas state
func NewCanvasState() *CanvasState {
	return &CanvasState{
		Nodes:    make(map[string]*Node),
		Edges:    make([]Edge, 0),
		UserData: make(map[string]interface{}),
		Version:  0,
	}
}

// AddNode adds a node to the canvas
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 retrieves a node by ID
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 applies an update function to a node
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 adds a directed edge between two nodes
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 returns all downstream nodes depending on nodeID
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 returns all upstream dependencies of nodeID
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 creates an immutable checkpoint for recovery
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 Canvas State Implementation

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:
    """Thread-safe typed artifact graph for canvas state management"""

    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':
        """Create immutable checkpoint for recovery"""
        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 Implementation

package bridge

import (
	"fmt"
	"time"

	"jarvishub/canvas"
)

// OperationType canvas operation type
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 tool family
type ToolFamily string

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

// CapabilityManifest capability manifest
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 execution grant
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 agent 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 tool invocation
type ToolCall struct {
	Family    ToolFamily             `json:"family"`
	Name      string                 `json:"name"`
	Params    map[string]interface{} `json:"params"`
	RequestID string                 `json:"request_id"`
}

// ToolObservation tool observation result
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 feedback signal
type FeedbackSignal struct {
	Source    string                 `json:"source"`
	NodeID    string                 `json:"node_id,omitempty"`
	Verdict   string                 `json:"verdict"`
	Comments  string                 `json:"comments,omitempty"`
	Timestamp time.Time              `json:"timestamp"`
	Edits     map[string]interface{} `json:"edits,omitempty"`
}

// RepairDecision repair decision
type RepairDecision struct {
	Action      string                 `json:"action"`
	TargetNodes []string               `json:"target_nodes,omitempty"`
	Strategy    string                 `json:"strategy,omitempty"`
	Params      map[string]interface{} `json:"params,omitempty"`
}

// StateTransition state transition record
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 protocol bridge
type ProtocolBridge struct {
	state       *canvas.CanvasState
	manifest    *CapabilityManifest
	grant       *ExecutionGrant
	transitions []StateTransition
	turnCount   int
}

// NewProtocolBridge creates a new protocol bridge
func NewProtocolBridge(state *canvas.CanvasState) *ProtocolBridge {
	return &ProtocolBridge{
		state:       state,
		transitions: make([]StateTransition, 0),
	}
}

// BuildManifest builds the capability manifest for the current turn
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 derives the execution grant for the current turn
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,
	}
	pb.grant = grant
	return grant
}

// ValidateAction validates an action against the current grant
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 commits a state transition to the trajectory log
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 returns the full execution trajectory
func (pb *ProtocolBridge) GetTrajectory() []StateTransition {
	return pb.transitions
}

# runtime/agent_runtime.py — Agent Runtime Core Implementation

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:
    """Runtime execution context"""
    user_query: str
    canvas_state: CanvasState
    manifest: CapabilityManifest
    grant: ExecutionGrant
    skills: Dict[str, 'Skill']
    memory: Dict[str, Any]
    subagents: List['SubAgent']


class AgentRuntime:
    """Agent execution runtime"""

    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]:
        """Execute one turn of the runtime loop"""
        self.turn_count += 1
        logger.info(f"=== Turn {self.turn_count} ===")

        # Step 1: Build capability manifest
        manifest = self._build_manifest()

        # Step 2: Derive execution grant
        grant = self._derive_grant(user_query)

        # Step 3: Create runtime context
        ctx = RuntimeContext(
            user_query=user_query,
            canvas_state=self.state.snapshot(),
            manifest=manifest,
            grant=grant,
            skills=self.skills,
            memory=self.memory,
            subagents=self.subagents,
        )

        # Step 4: Select action (LLM-driven in production, simulated here)
        action = self._select_action(ctx)

        # Step 5: Validate action
        validation_error = self._validate_action(action, grant)
        if validation_error:
            return {"error": validation_error, "turn": self.turn_count}

        # Step 6: Execute action
        before_state = self.state.snapshot()
        observation = self._execute_action(action)

        # Step 7: Collect feedback
        feedback = self._collect_feedback(observation, ctx)

        # Step 8: Determine repair decision
        repair = self._decide_repair(feedback, before_state)

        # Step 9: Apply repair
        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}

        # Step 10: Commit state transition
        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:
        """Select action. In production, LLM-driven."""
        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)
        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]:
        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 Implementation

package runtime

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

	"jarvishub/bridge"
	"jarvishub/canvas"
)

// ToolHandler tool handler function type
type ToolHandler func(ctx context.Context, call bridge.ToolCall) (*bridge.ToolObservation, error)

// RuntimeConfig runtime configuration
type RuntimeConfig struct {
	MaxTurns       int
	DefaultTimeout time.Duration
	ModelName      string
}

// AgentRuntime agent execution runtime
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 interface for reusable creative procedures
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 creates a new agent runtime
func NewAgentRuntime(config RuntimeConfig, b *bridge.ProtocolBridge, s *canvas.CanvasState) *AgentRuntime {
	return &AgentRuntime{
		config:       config,
		bridge:       b,
		state:        s,
		skills:       make(map[string]Skill),
		memory:       make(map[string]interface{}),
		toolHandlers: make(map[bridge.ToolFamily]ToolHandler),
	}
}

// RegisterToolHandler registers a tool handler
func (rt *AgentRuntime) RegisterToolHandler(family bridge.ToolFamily, handler ToolHandler) {
	rt.toolHandlers[family] = handler
}

// RegisterSkill registers a skill
func (rt *AgentRuntime) RegisterSkill(skill Skill) {
	rt.skills[skill.Name()] = skill
}

// ExecuteTurn executes one turn of the runtime loop
func (rt *AgentRuntime) ExecuteTurn(ctx context.Context, userQuery string) (map[string]interface{}, error) {
	rt.turnCount++
	log.Printf("=== Turn %d ===\n", rt.turnCount)

	projectType := "general"
	if pt, ok := rt.memory["project_type"].(string); ok {
		projectType = pt
	}
	manifest := rt.bridge.BuildManifest(projectType)
	grant := rt.bridge.DeriveGrant(userQuery)

	action, err := rt.selectAction(ctx, userQuery, manifest, grant)
	if err != nil {
		return nil, fmt.Errorf("action selection failed: %w", err)
	}
	if err := rt.bridge.ValidateAction(action); err != nil {
		return nil, fmt.Errorf("action validation failed: %w", err)
	}

	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(),
		}
	}

	feedback := rt.collectFeedback(observation)
	repair := rt.decideRepair(feedback, beforeState)
	if repair != nil && repair.Action == "repair" {
		rt.applyRepair(ctx, repair, beforeState)
	}

	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) {
	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)
		}
		if _, err := handler(ctx, tc); err != nil {
			return nil, err
		}
	}
	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 — Local Repair Algorithm

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

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


@dataclass
class RepairPlan:
    """Repair plan"""
    failed_node_id: str
    affected_downstream: List[str]
    preserved_upstream: List[str]
    repair_strategy: str
    feedback_context: str


class LocalRepairEngine:
    """Local repair engine"""

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

    def analyze_failure(self, feedback: str, failed_node_id: str) -> RepairPlan:
        """
        Analyze failure and formulate repair plan.

        Core algorithm:
        1. Locate the failed node and its downstream dependencies
        2. Preserve upstream dependencies (no regeneration needed)
        3. Only mark affected downstream nodes for repair
        """
        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 traversal of downstream dependency graph"""
        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]:
        """Find upstream dependencies that can be preserved"""
        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:
        """Apply the repair plan. Returns count of repaired nodes."""
        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