From bb2715c4d3101443e8a5a55312fcebe773ac8625 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Wed, 25 Feb 2026 02:48:10 +0900 Subject: [PATCH] feat: wire orchestration event broadcaster end-to-end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- pkg/miniapp/miniapp.go | 91 +++++++++++++++++++++++++++-- pkg/orch/broadcaster.go | 123 ++++++++++++++++++++++++++++++++++++++++ pkg/tools/subagent.go | 51 +++++++++++++++++ pkg/tools/toolloop.go | 15 ++++- 4 files changed, 272 insertions(+), 8 deletions(-) create mode 100644 pkg/orch/broadcaster.go diff --git a/pkg/miniapp/miniapp.go b/pkg/miniapp/miniapp.go index 6f1590d04..95e035bfd 100644 --- a/pkg/miniapp/miniapp.go +++ b/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 { diff --git a/pkg/orch/broadcaster.go b/pkg/orch/broadcaster.go new file mode 100644 index 000000000..fe89509bb --- /dev/null +++ b/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) + } + } +} diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go index 91ebff636..634d087ed 100644 --- a/pkg/tools/subagent.go +++ b/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", diff --git a/pkg/tools/toolloop.go b/pkg/tools/toolloop.go index cdfe0d6ce..6eb62ef23 100644 --- a/pkg/tools/toolloop.go +++ b/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