feat: wire orchestration event broadcaster end-to-end

- Add pkg/orch.Broadcaster (extracted to break tools↔miniapp import cycle):
  agent_spawn, agent_state, conversation, agent_gc events; live agent
  snapshot for WS initial-state delivery; non-blocking fan-out (drop on
  slow subscriber)

- pkg/tools/toolloop.go: OnStateChange hook on ToolLoopConfig — emits
  ("waiting","") before LLM call and ("toolcall", name) per tool execution

- pkg/tools/subagent.go: Spawn() emits agent_spawn; runTask() emits
  conversation (conductor→agent), wires OnStateChange into ToolLoopConfig,
  emits conversation (agent→conductor) + agent_gc on completion/failure/cancel

- pkg/miniapp/miniapp.go: SetOrchBroadcaster(), wsOrchestration() handler
  at /miniapp/api/orchestration/ws — sends snapshot on connect, streams
  events with ping/pong keepalive; no polling

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

pkg/miniapp/miniapp.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/gorilla/websocket"
 	"github.com/sipeed/picoclaw/pkg/logger"
+	"github.com/sipeed/picoclaw/pkg/orch"
 	"github.com/sipeed/picoclaw/pkg/skills"
 	"github.com/sipeed/picoclaw/pkg/stats"
 )
@@ -206,12 +207,13 @@ type DevTargetManager interface {
 
 // Handler serves the Mini App HTML and API endpoints.
 type Handler struct {
-	provider  DataProvider
-	sender    CommandSender
-	botToken  string
-	notifier  *StateNotifier
-	allowList []string
-	workspace string
+	provider        DataProvider
+	sender          CommandSender
+	botToken        string
+	notifier        *StateNotifier
+	allowList       []string
+	workspace       string
+	orchBroadcaster *orch.Broadcaster
 
 	devMu       sync.RWMutex
 	devTarget   *url.URL
@@ -525,6 +527,12 @@ func escapeHTMLString(s string) string {
 	return s
 }
 
+// SetOrchBroadcaster wires the orchestration broadcaster so the Mini App can
+// push live agent state to the canvas UI via WebSocket.
+func (h *Handler) SetOrchBroadcaster(b *orch.Broadcaster) {
+	h.orchBroadcaster = b
+}
+
 // RegisterRoutes registers Mini App routes on the given mux.
 func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
 	mux.HandleFunc("/miniapp", h.serveIndex)
@@ -541,6 +549,7 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) {
 	mux.HandleFunc("/miniapp/api/logs/ws", h.requireAuth(h.wsLogs))
 	mux.HandleFunc("/miniapp/api/logs/snapshot", h.requireAuth(h.apiLogsSnapshot))
 	mux.HandleFunc("/miniapp/api/logs/snapshot/", h.requireAuth(h.apiLogsSnapshotDownload))
+	mux.HandleFunc("/miniapp/api/orchestration/ws", h.requireAuth(h.wsOrchestration))
 	mux.HandleFunc("/miniapp/dev/console", h.apiDevConsole)
 	mux.HandleFunc("/miniapp/dev/", h.serveDevProxy)
 }
@@ -1028,6 +1037,76 @@ func (h *Handler) wsLogs(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// wsOrchestration streams live orchestration events (agent spawn/state/gc and
+// conductor↔agent conversations) to the canvas UI.
+//
+// Protocol:
+//
+//	{"type":"init","agents":[...OrchAgentInfo]}   — sent once on connect
+//	{"type":"event","event":{...OrchEvent}}        — pushed on each state change
+func (h *Handler) wsOrchestration(w http.ResponseWriter, r *http.Request) {
+	if h.orchBroadcaster == nil {
+		http.Error(w, `{"error":"orchestration not enabled"}`, http.StatusServiceUnavailable)
+		return
+	}
+
+	rc := http.NewResponseController(w)
+	_ = rc.SetWriteDeadline(time.Time{})
+	_ = rc.SetReadDeadline(time.Time{})
+
+	conn, err := wsUpgrader.Upgrade(w, r, nil)
+	if err != nil {
+		return
+	}
+	defer conn.Close()
+
+	sub := h.orchBroadcaster.Subscribe()
+	defer h.orchBroadcaster.Unsubscribe(sub)
+
+	// Send current agent snapshot so the canvas can populate immediately
+	snapshot := h.orchBroadcaster.Snapshot()
+	if err := conn.WriteJSON(map[string]any{"type": "init", "agents": snapshot}); err != nil {
+		return
+	}
+
+	conn.SetReadDeadline(time.Now().Add(wsPongWait))
+	conn.SetPongHandler(func(string) error {
+		conn.SetReadDeadline(time.Now().Add(wsPongWait))
+		return nil
+	})
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for {
+			if _, _, err := conn.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}()
+
+	ticker := time.NewTicker(wsPingPeriod)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case ev, ok := <-sub.Ch:
+			if !ok {
+				return
+			}
+			if err := conn.WriteJSON(map[string]any{"type": "event", "event": ev}); err != nil {
+				return
+			}
+		case <-ticker.C:
+			if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				return
+			}
+		case <-done:
+			return
+		}
+	}
+}
+
 // apiLogsSnapshot creates a tar.gz snapshot of the current log buffer.
 func (h *Handler) apiLogsSnapshot(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {

pkg/orch/broadcaster.go

@@ -0,0 +1,123 @@
+// Package orch provides the orchestration event broadcaster used by the
+// subagent system and the Mini App WebSocket UI.
+package orch
+
+import (
+	"sync"
+	"time"
+)
+
+// Event is a single orchestration event pushed over WebSocket to the UI.
+// type values: "agent_spawn" | "agent_state" | "conversation" | "agent_gc"
+type Event struct {
+	Type    string `json:"type"`
+	ID      string `json:"id,omitempty"`
+	Label   string `json:"label,omitempty"`
+	Task    string `json:"task,omitempty"`
+	State   string `json:"state,omitempty"`  // waiting | toolcall | idle
+	Tool    string `json:"tool,omitempty"`   // tool name during toolcall
+	From    string `json:"from,omitempty"`
+	To      string `json:"to,omitempty"`
+	Text    string `json:"text,omitempty"`
+	Reason  string `json:"reason,omitempty"` // agent_gc: completed | failed | cancelled
+	Created int64  `json:"created,omitempty"`
+}
+
+// AgentInfo is the live snapshot of one active agent.
+// Kept inside Broadcaster so new WS connections can get current state.
+type AgentInfo struct {
+	ID      string `json:"id"`
+	Label   string `json:"label"`
+	Task    string `json:"task"`
+	State   string `json:"state"`
+	Tool    string `json:"tool,omitempty"`
+	Created int64  `json:"created"`
+}
+
+// Subscriber is a single WebSocket client subscription.
+type Subscriber struct {
+	Ch chan Event
+}
+
+// Broadcaster distributes orchestration events to all connected WS clients.
+// It also maintains a live agent snapshot for initial-state delivery on connect.
+//
+// Publish is non-blocking: events are dropped if a subscriber's buffer is full
+// (same pattern as pkg/logger).
+type Broadcaster struct {
+	mu     sync.Mutex
+	subs   map[*Subscriber]struct{}
+	agents map[string]*AgentInfo // live agents, keyed by task ID
+}
+
+func NewBroadcaster() *Broadcaster {
+	return &Broadcaster{
+		subs:   make(map[*Subscriber]struct{}),
+		agents: make(map[string]*AgentInfo),
+	}
+}
+
+func (b *Broadcaster) Subscribe() *Subscriber {
+	sub := &Subscriber{Ch: make(chan Event, 32)}
+	b.mu.Lock()
+	b.subs[sub] = struct{}{}
+	b.mu.Unlock()
+	return sub
+}
+
+func (b *Broadcaster) Unsubscribe(sub *Subscriber) {
+	b.mu.Lock()
+	delete(b.subs, sub)
+	b.mu.Unlock()
+}
+
+// Snapshot returns the current set of active agents.
+// Called once on new WS connection to send initial state.
+func (b *Broadcaster) Snapshot() []AgentInfo {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	out := make([]AgentInfo, 0, len(b.agents))
+	for _, a := range b.agents {
+		out = append(out, *a)
+	}
+	return out
+}
+
+// Publish updates internal agent state and fans out to all subscribers.
+func (b *Broadcaster) Publish(ev Event) {
+	if ev.Created == 0 {
+		ev.Created = time.Now().UnixMilli()
+	}
+
+	b.mu.Lock()
+	switch ev.Type {
+	case "agent_spawn":
+		b.agents[ev.ID] = &AgentInfo{
+			ID:      ev.ID,
+			Label:   ev.Label,
+			Task:    ev.Task,
+			State:   "idle",
+			Created: ev.Created,
+		}
+	case "agent_state":
+		if a, ok := b.agents[ev.ID]; ok {
+			a.State = ev.State
+			a.Tool = ev.Tool
+		}
+	case "agent_gc":
+		delete(b.agents, ev.ID)
+	}
+	// snapshot subs while holding lock, then release before sending
+	subs := make([]*Subscriber, 0, len(b.subs))
+	for sub := range b.subs {
+		subs = append(subs, sub)
+	}
+	b.mu.Unlock()
+
+	for _, sub := range subs {
+		select {
+		case sub.Ch <- ev:
+		default: // subscriber slow — drop (non-blocking)
+		}
+	}
+}

pkg/tools/subagent.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/sipeed/picoclaw/pkg/bus"
+	"github.com/sipeed/picoclaw/pkg/orch"
 	"github.com/sipeed/picoclaw/pkg/providers"
 )
 
@@ -36,6 +37,7 @@ type SubagentManager struct {
 	hasMaxTokens   bool
 	hasTemperature bool
 	nextID         int
+	broadcaster    *orch.Broadcaster
 }
 
 func NewSubagentManager(
@@ -52,9 +54,16 @@ func NewSubagentManager(
 		tools:         NewToolRegistry(),
 		maxIterations: 10,
 		nextID:        1,
+		broadcaster:   orch.NewBroadcaster(),
 	}
 }
 
+// GetBroadcaster returns the Broadcaster so the miniapp handler can
+// subscribe to real-time orchestration events.
+func (sm *SubagentManager) GetBroadcaster() *orch.Broadcaster {
+	return sm.broadcaster
+}
+
 // SetLLMOptions sets max tokens and temperature for subagent LLM calls.
 func (sm *SubagentManager) SetLLMOptions(maxTokens int, temperature float64) {
 	sm.mu.Lock()
@@ -103,6 +112,13 @@ func (sm *SubagentManager) Spawn(
 	}
 	sm.tasks[taskID] = subagentTask
 
+	sm.broadcaster.Publish(orch.Event{
+		Type:  "agent_spawn",
+		ID:    taskID,
+		Label: label,
+		Task:  task,
+	})
+
 	// Start task in background with context cancellation support
 	go sm.runTask(ctx, subagentTask, callback)
 
@@ -164,12 +180,28 @@ After completing the task, provide a clear summary of what was done.`
 		}
 	}
 
+	// Notify conductor that the subagent is starting
+	sm.broadcaster.Publish(orch.Event{
+		Type: "conversation",
+		From: "conductor",
+		To:   task.ID,
+		Text: task.Task,
+	})
+
 	loopResult, err := RunToolLoop(ctx, ToolLoopConfig{
 		Provider:      sm.provider,
 		Model:         sm.defaultModel,
 		Tools:         tools,
 		MaxIterations: maxIter,
 		LLMOptions:    llmOptions,
+		OnStateChange: func(state, tool string) {
+			sm.broadcaster.Publish(orch.Event{
+				Type:  "agent_state",
+				ID:    task.ID,
+				State: state,
+				Tool:  tool,
+			})
+		},
 	}, messages, task.OriginChannel, task.OriginChatID)
 
 	sm.mu.Lock()
@@ -186,10 +218,17 @@ After completing the task, provide a clear summary of what was done.`
 		task.Status = "failed"
 		task.Result = fmt.Sprintf("Error: %v", err)
 		// Check if it was cancelled
+		gcReason := "failed"
 		if ctx.Err() != nil {
 			task.Status = "cancelled"
 			task.Result = "Task cancelled during execution"
+			gcReason = "cancelled"
 		}
+		sm.broadcaster.Publish(orch.Event{
+			Type:   "agent_gc",
+			ID:     task.ID,
+			Reason: gcReason,
+		})
 		result = &ToolResult{
 			ForLLM:  task.Result,
 			ForUser: "",
@@ -201,6 +240,18 @@ After completing the task, provide a clear summary of what was done.`
 	} else {
 		task.Status = "completed"
 		task.Result = loopResult.Content
+		// Notify conductor of the result
+		sm.broadcaster.Publish(orch.Event{
+			Type: "conversation",
+			From: task.ID,
+			To:   "conductor",
+			Text: loopResult.Content,
+		})
+		sm.broadcaster.Publish(orch.Event{
+			Type:   "agent_gc",
+			ID:     task.ID,
+			Reason: "completed",
+		})
 		result = &ToolResult{
 			ForLLM: fmt.Sprintf(
 				"Subagent '%s' completed (iterations: %d): %s",

pkg/tools/toolloop.go

@@ -23,6 +23,11 @@ type ToolLoopConfig struct {
 	Tools         *ToolRegistry
 	MaxIterations int
 	LLMOptions    map[string]any
+	// OnStateChange is an optional hook for UI feedback.
+	// Called with ("waiting","") before each LLM call and
+	// ("toolcall", toolName) when each tool starts executing.
+	// nil is safe to pass.
+	OnStateChange func(state, tool string)
 }
 
 // ToolLoopResult contains the result of running the tool loop.
@@ -62,7 +67,10 @@ func RunToolLoop(
 		if llmOpts == nil {
 			llmOpts = map[string]any{}
 		}
-		// 3. Call LLM
+		// 3. Call LLM  (hook: waiting for response)
+		if config.OnStateChange != nil {
+			config.OnStateChange("waiting", "")
+		}
 		response, err := config.Provider.Chat(ctx, messages, providerToolDefs, config.Model, llmOpts)
 		if err != nil {
 			logger.ErrorCF("toolloop", "LLM call failed",
@@ -121,7 +129,7 @@ func RunToolLoop(
 		}
 		messages = append(messages, assistantMsg)
 
-		// 7. Execute tool calls
+		// 7. Execute tool calls  (hook: toolcall per tool)
 		for _, tc := range normalizedToolCalls {
 			argsJSON, _ := json.Marshal(tc.Arguments)
 			argsPreview := utils.Truncate(string(argsJSON), 200)
@@ -130,6 +138,9 @@ func RunToolLoop(
 					"tool":      tc.Name,
 					"iteration": iteration,
 				})
+			if config.OnStateChange != nil {
+				config.OnStateChange("toolcall", tc.Name)
+			}
 
 			// Execute tool (no async callback for subagents - they run independently)
 			var toolResult *ToolResult