= 0 {
- // Find the closing '>' of the tag
- closeIdx := bytes.IndexByte(lower[idx:], '>')
- if closeIdx >= 0 {
- insertAt := idx + closeIdx + 1
- out := make([]byte, 0, len(html)+len(script))
- out = append(out, html[:insertAt]...)
- out = append(out, script...)
- out = append(out, html[insertAt:]...)
- return out
- }
- }
-
- // Priority 3: prepend
- out := make([]byte, 0, len(html)+len(script))
- out = append(out, script...)
- out = append(out, html...)
- return out
-}
-
-// escapeHTMLString escapes HTML special characters in a string.
-func escapeHTMLString(s string) string {
- s = strings.ReplaceAll(s, "&", "&")
- s = strings.ReplaceAll(s, "<", "<")
- s = strings.ReplaceAll(s, ">", ">")
- s = strings.ReplaceAll(s, "\"", """)
- 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.
@@ -541,6 +73,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)
}
@@ -554,647 +87,3 @@ func (h *Handler) serveIndex(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Write(data)
}
-
-func (h *Handler) requireAuth(next http.HandlerFunc) http.HandlerFunc {
- return func(w http.ResponseWriter, r *http.Request) {
- initData := r.URL.Query().Get("initData")
- if initData == "" {
- http.Error(w, `{"error":"missing initData"}`, http.StatusUnauthorized)
- return
- }
- if !ValidateInitData(initData, h.botToken) {
- http.Error(w, `{"error":"invalid initData"}`, http.StatusUnauthorized)
- return
- }
- if len(h.allowList) > 0 {
- userID, _ := extractUserFromInitData(initData)
- if userID == "" || !isAllowed(userID, h.allowList) {
- http.Error(w, `{"error":"forbidden"}`, http.StatusForbidden)
- return
- }
- }
- next(w, r)
- }
-}
-
-// isAllowed checks whether userID matches any entry in the allow list.
-// Logic mirrors BaseChannel.IsAllowed without importing channels package.
-func isAllowed(userID string, allowList []string) bool {
- if len(allowList) == 0 {
- return true
- }
- for _, allowed := range allowList {
- trimmed := strings.TrimPrefix(allowed, "@")
- allowedID := trimmed
- if idx := strings.Index(trimmed, "|"); idx > 0 {
- allowedID = trimmed[:idx]
- }
- if userID == allowed || userID == trimmed || userID == allowedID {
- return true
- }
- }
- return false
-}
-
-func (h *Handler) apiSkills(w http.ResponseWriter, r *http.Request) {
- skillsList := h.provider.ListSkills()
- writeJSON(w, skillsList)
-}
-
-func (h *Handler) apiPlan(w http.ResponseWriter, r *http.Request) {
- info := h.provider.GetPlanInfo()
- writeJSON(w, info)
-}
-
-func (h *Handler) apiSessions(w http.ResponseWriter, r *http.Request) {
- sessions := h.provider.GetActiveSessions()
- if sessions == nil {
- sessions = []SessionInfo{}
- }
- writeJSON(w, sessions)
-}
-
-func (h *Handler) apiSession(w http.ResponseWriter, r *http.Request) {
- s := h.provider.GetSessionStats()
- if s == nil {
- writeJSON(w, map[string]string{"status": "stats not enabled"})
- return
- }
- writeJSON(w, s)
-}
-
-func (h *Handler) apiContext(w http.ResponseWriter, r *http.Request) {
- writeJSON(w, h.provider.GetContextInfo())
-}
-
-func (h *Handler) apiPrompt(w http.ResponseWriter, r *http.Request) {
- writeJSON(w, map[string]string{"prompt": h.provider.GetSystemPrompt()})
-}
-
-func (h *Handler) apiGit(w http.ResponseWriter, r *http.Request) {
- repo := r.URL.Query().Get("repo")
- if repo == "" {
- writeJSON(w, h.provider.GetGitRepos())
- } else {
- writeJSON(w, h.provider.GetGitRepoDetail(repo))
- }
-}
-
-func (h *Handler) apiCommand(w http.ResponseWriter, r *http.Request) {
- if r.Method != http.MethodPost {
- http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
- return
- }
-
- body, err := io.ReadAll(io.LimitReader(r.Body, 4096))
- if err != nil {
- http.Error(w, `{"error":"bad request"}`, http.StatusBadRequest)
- return
- }
-
- var req struct {
- Command string `json:"command"`
- }
- if err := json.Unmarshal(body, &req); err != nil || req.Command == "" {
- http.Error(w, `{"error":"missing command"}`, http.StatusBadRequest)
- return
- }
-
- if !strings.HasPrefix(req.Command, "/") {
- http.Error(w, `{"error":"command must start with /"}`, http.StatusBadRequest)
- return
- }
-
- // Extract user ID from initData to identify the sender
- initData := r.URL.Query().Get("initData")
- userID, chatID := extractUserFromInitData(initData)
- if userID == "" {
- http.Error(w, `{"error":"cannot identify user"}`, http.StatusBadRequest)
- return
- }
-
- h.sender.SendCommand(userID, chatID, req.Command)
- writeJSON(w, map[string]string{"status": "ok"})
-}
-
-func (h *Handler) apiDev(w http.ResponseWriter, r *http.Request) {
- switch r.Method {
- case http.MethodGet:
- writeJSON(w, h.devStatus())
- case http.MethodPost:
- body, err := io.ReadAll(io.LimitReader(r.Body, 4096))
- if err != nil {
- http.Error(w, `{"error":"bad request"}`, http.StatusBadRequest)
- return
- }
- var req struct {
- Action string `json:"action"`
- ID string `json:"id"`
- }
- if err := json.Unmarshal(body, &req); err != nil {
- http.Error(w, `{"error":"invalid JSON"}`, http.StatusBadRequest)
- return
- }
- switch req.Action {
- case "activate":
- if req.ID == "" {
- writeJSON(w, map[string]any{"error": "id is required"})
- return
- }
- if err := h.ActivateDevTarget(req.ID); err != nil {
- writeJSON(w, map[string]any{"error": err.Error()})
- return
- }
- case "deactivate":
- if err := h.DeactivateDevTarget(); err != nil {
- writeJSON(w, map[string]any{"error": err.Error()})
- return
- }
- case "unregister":
- if req.ID == "" {
- writeJSON(w, map[string]any{"error": "id is required"})
- return
- }
- if err := h.UnregisterDevTarget(req.ID); err != nil {
- writeJSON(w, map[string]any{"error": err.Error()})
- return
- }
- default:
- writeJSON(w, map[string]any{"error": "unknown action"})
- return
- }
- writeJSON(w, h.devStatus())
- default:
- http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
- }
-}
-
-func (h *Handler) serveDevProxy(w http.ResponseWriter, r *http.Request) {
- h.devMu.RLock()
- proxy := h.devProxy
- h.devMu.RUnlock()
-
- if proxy == nil {
- http.Error(w, "dev proxy not configured", http.StatusServiceUnavailable)
- return
- }
-
- // Strip /miniapp/dev prefix so /miniapp/dev/foo â /foo
- r.URL.Path = strings.TrimPrefix(r.URL.Path, "/miniapp/dev")
- if r.URL.Path == "" {
- r.URL.Path = "/"
- }
- proxy.ServeHTTP(w, r)
-}
-
-// extractUserFromInitData parses user.id from the initData query string.
-// initData contains a "user" param with JSON like {"id":123456,...}.
-func extractUserFromInitData(initData string) (userID, chatID string) {
- values, err := url.ParseQuery(initData)
- if err != nil {
- return "", ""
- }
- userJSON := values.Get("user")
- if userJSON == "" {
- return "", ""
- }
- var user struct {
- ID int64 `json:"id"`
- }
- if err := json.Unmarshal([]byte(userJSON), &user); err != nil || user.ID == 0 {
- return "", ""
- }
- id := fmt.Sprintf("%d", user.ID)
- // For Mini App commands, chatID = userID (private chat)
- return id, id
-}
-
-func (h *Handler) apiEvents(w http.ResponseWriter, r *http.Request) {
- flusher, ok := w.(http.Flusher)
- if !ok {
- http.Error(w, `{"error":"streaming not supported"}`, http.StatusInternalServerError)
- return
- }
- rc := http.NewResponseController(w)
- _ = rc.SetWriteDeadline(time.Time{})
-
- w.Header().Set("Content-Type", "text/event-stream")
- w.Header().Set("Cache-Control", "no-cache")
- w.Header().Set("Connection", "keep-alive")
- w.Header().Set("X-Accel-Buffering", "no")
-
- ch := h.notifier.Subscribe()
- defer h.notifier.Unsubscribe(ch)
-
- var lastPlan, lastSession, lastSkills, lastDev, lastContext, lastPrompt []byte
-
- // Send initial state immediately
- sendSSEIfChanged(w, flusher, "plan", h.provider.GetPlanInfo(), &lastPlan)
- sendSSEIfChanged(w, flusher, "session",
- map[string]any{"stats": h.provider.GetSessionStats(), "sessions": h.provider.GetActiveSessions()},
- &lastSession)
- sendSSEIfChanged(w, flusher, "skills", h.provider.ListSkills(), &lastSkills)
- sendSSEIfChanged(w, flusher, "dev", h.devStatus(), &lastDev)
- sendSSEIfChanged(w, flusher, "context", h.provider.GetContextInfo(), &lastContext)
- sendSSEIfChanged(w, flusher, "prompt", map[string]string{"prompt": h.provider.GetSystemPrompt()}, &lastPrompt)
-
- for {
- select {
- case <-r.Context().Done():
- return
- case <-h.notifier.Done():
- return
- case <-ch:
- sendSSEIfChanged(w, flusher, "plan", h.provider.GetPlanInfo(), &lastPlan)
- sendSSEIfChanged(w, flusher, "session",
- map[string]any{"stats": h.provider.GetSessionStats(), "sessions": h.provider.GetActiveSessions()},
- &lastSession)
- sendSSEIfChanged(w, flusher, "skills", h.provider.ListSkills(), &lastSkills)
- sendSSEIfChanged(w, flusher, "dev", h.devStatus(), &lastDev)
- sendSSEIfChanged(w, flusher, "context", h.provider.GetContextInfo(), &lastContext)
- sendSSEIfChanged(w, flusher, "prompt", map[string]string{"prompt": h.provider.GetSystemPrompt()}, &lastPrompt)
- }
- }
-}
-
-func (h *Handler) devStatus() map[string]any {
- h.devMu.RLock()
- defer h.devMu.RUnlock()
-
- active := h.devTarget != nil
- target := ""
- if h.devTarget != nil {
- target = h.devTargets[h.devActiveID].Target // original URL before IPv6 rewrite
- }
-
- targets := make([]DevTarget, 0, len(h.devTargets))
- for _, dt := range h.devTargets {
- targets = append(targets, *dt)
- }
- sort.Slice(targets, func(i, j int) bool { return targets[i].ID < targets[j].ID })
-
- return map[string]any{
- "active": active,
- "active_id": h.devActiveID,
- "target": target,
- "targets": targets,
- }
-}
-
-func sendSSEIfChanged(w http.ResponseWriter, f http.Flusher, event string, v any, last *[]byte) {
- data, _ := json.Marshal(v)
- if !bytes.Equal(data, *last) {
- fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, data)
- f.Flush()
- *last = data
- }
-}
-
-func writeJSON(w http.ResponseWriter, v any) {
- w.Header().Set("Content-Type", "application/json")
- json.NewEncoder(w).Encode(v)
-}
-
-// apiDevConsole receives console output from dev preview iframes.
-func (h *Handler) apiDevConsole(w http.ResponseWriter, r *http.Request) {
- if r.Method != http.MethodPost {
- http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
- return
- }
-
- // Only accept console posts when dev proxy is active
- if h.GetDevTarget() == "" {
- http.Error(w, `{"error":"not available"}`, http.StatusNotFound)
- return
- }
-
- // Simple rate limit: max 10 requests per second
- now := time.Now().Unix()
- h.consoleMu.Lock()
- if h.consoleReqSec != now {
- h.consoleReqSec = now
- h.consoleReqCount = 0
- }
- h.consoleReqCount++
- over := h.consoleReqCount > 10
- h.consoleMu.Unlock()
- if over {
- http.Error(w, `{"error":"rate limit"}`, http.StatusTooManyRequests)
- return
- }
-
- body, err := io.ReadAll(io.LimitReader(r.Body, 32*1024))
- if err != nil {
- http.Error(w, `{"error":"bad request"}`, http.StatusBadRequest)
- return
- }
-
- var entries []struct {
- Level string `json:"level"`
- Message string `json:"message"`
- }
- if err := json.Unmarshal(body, &entries); err != nil {
- http.Error(w, `{"error":"invalid JSON"}`, http.StatusBadRequest)
- return
- }
-
- // Cap at 20 entries per batch
- if len(entries) > 20 {
- entries = entries[:20]
- }
-
- for _, e := range entries {
- msg := e.Message
- if len(msg) > 1024 {
- msg = msg[:1024]
- }
- switch e.Level {
- case "warn":
- logger.WarnC("dev-console", msg)
- case "error":
- logger.ErrorC("dev-console", msg)
- default:
- logger.InfoC("dev-console", msg)
- }
- }
-
- w.WriteHeader(http.StatusNoContent)
-}
-
-// wsLogs serves a WebSocket endpoint that streams log entries in real time.
-func (h *Handler) wsLogs(w http.ResponseWriter, r *http.Request) {
- // Parse filter params
- component := r.URL.Query().Get("component")
- levelStr := r.URL.Query().Get("level")
- minLevel := logger.INFO
- if levelStr != "" {
- minLevel = logger.ParseLevel(levelStr)
- }
-
- // Clear HTTP server deadlines before WebSocket hijack
- rc := http.NewResponseController(w)
- _ = rc.SetWriteDeadline(time.Time{})
- _ = rc.SetReadDeadline(time.Time{})
-
- conn, err := wsUpgrader.Upgrade(w, r, nil)
- if err != nil {
- return
- }
-
- client := &wsClient{conn: conn}
-
- // Enforce max WS clients: evict oldest if full
- h.wsClientsMu.Lock()
- if len(h.wsClients) >= maxWSClients {
- oldest := h.wsClients[0]
- h.wsClients = h.wsClients[1:]
- oldest.conn.Close()
- }
- h.wsClients = append(h.wsClients, client)
- h.wsClientsMu.Unlock()
-
- defer func() {
- h.wsClientsMu.Lock()
- for i, c := range h.wsClients {
- if c == client {
- h.wsClients = append(h.wsClients[:i], h.wsClients[i+1:]...)
- break
- }
- }
- h.wsClientsMu.Unlock()
- conn.Close()
- }()
-
- // Build filter function
- filter := func(e logger.LogEntry) bool {
- if lvl := logger.ParseLevel(e.Level); lvl < minLevel {
- return false
- }
- if component != "" && e.Component != component {
- return false
- }
- return true
- }
-
- sub := logger.Subscribe(filter)
- defer logger.Unsubscribe(sub)
-
- // Configure ping/pong to detect dead connections
- conn.SetReadDeadline(time.Now().Add(wsPongWait))
- conn.SetPongHandler(func(string) error {
- conn.SetReadDeadline(time.Now().Add(wsPongWait))
- return nil
- })
-
- // Send initial data
- initial := logger.RecentLogs(minLevel, component, 50)
- if err := conn.WriteJSON(map[string]any{"type": "init", "entries": initial}); err != nil {
- return
- }
-
- // Close detection goroutine
- done := make(chan struct{})
- go func() {
- defer close(done)
- for {
- if _, _, err := conn.ReadMessage(); err != nil {
- return
- }
- }
- }()
-
- // Stream loop with periodic pings
- ticker := time.NewTicker(wsPingPeriod)
- defer ticker.Stop()
-
- for {
- select {
- case entry, ok := <-sub.Ch:
- if !ok {
- return
- }
- entry.Caller = "" // strip for security
- entry.Fields = logger.SanitizeFields(entry.Fields) // mask sensitive values
- if err := conn.WriteJSON(map[string]any{"type": "entry", "entry": entry}); 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 {
- http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
- return
- }
-
- entries := logger.RecentLogs(logger.DEBUG, "", 300)
-
- snapshotDir := filepath.Join(h.workspace, "logs", "snapshots")
- if err := os.MkdirAll(snapshotDir, 0o755); err != nil {
- http.Error(w, `{"error":"cannot create snapshot dir"}`, http.StatusInternalServerError)
- return
- }
-
- id := time.Now().UTC().Format("20060102-150405")
- filename := fmt.Sprintf("picoclaw-logs-%s.tar.gz", id)
- snapshotPath := filepath.Join(snapshotDir, filename)
-
- // Create tar.gz
- f, err := os.Create(snapshotPath)
- if err != nil {
- http.Error(w, `{"error":"cannot create snapshot file"}`, http.StatusInternalServerError)
- return
- }
-
- gw := gzip.NewWriter(f)
- tw := tar.NewWriter(gw)
-
- prefix := fmt.Sprintf("picoclaw-logs-%s/", id)
-
- // logs.json
- logsJSON, _ := json.MarshalIndent(entries, "", " ")
- _ = tw.WriteHeader(&tar.Header{
- Name: prefix + "logs.json",
- Size: int64(len(logsJSON)),
- Mode: 0o644,
- ModTime: time.Now(),
- })
- _, _ = tw.Write(logsJSON)
-
- // metadata.json
- hostname, _ := os.Hostname()
- meta := map[string]any{
- "version": "1",
- "hostname": hostname,
- "timestamp": time.Now().UTC().Format(time.RFC3339),
- "entry_count": len(entries),
- }
- metaJSON, _ := json.MarshalIndent(meta, "", " ")
- _ = tw.WriteHeader(&tar.Header{
- Name: prefix + "metadata.json",
- Size: int64(len(metaJSON)),
- Mode: 0o644,
- ModTime: time.Now(),
- })
- _, _ = tw.Write(metaJSON)
-
- tw.Close()
- gw.Close()
- f.Close()
-
- // Cleanup old snapshots (>14 days)
- go cleanOldSnapshots(snapshotDir, 14*24*time.Hour)
-
- downloadURL := fmt.Sprintf("/miniapp/api/logs/snapshot/%s", id)
- writeJSON(w, map[string]string{"id": id, "download_url": downloadURL})
-}
-
-// apiLogsSnapshotDownload serves a snapshot tar.gz file.
-func (h *Handler) apiLogsSnapshotDownload(w http.ResponseWriter, r *http.Request) {
- if r.Method != http.MethodGet {
- http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed)
- return
- }
-
- id := strings.TrimPrefix(r.URL.Path, "/miniapp/api/logs/snapshot/")
- id = filepath.Base(id) // path traversal prevention
-
- if id == "" || id == "." || id == ".." {
- http.Error(w, `{"error":"invalid id"}`, http.StatusBadRequest)
- return
- }
-
- filename := fmt.Sprintf("picoclaw-logs-%s.tar.gz", id)
- snapshotPath := filepath.Join(h.workspace, "logs", "snapshots", filename)
-
- if _, err := os.Stat(snapshotPath); os.IsNotExist(err) {
- http.Error(w, `{"error":"not found"}`, http.StatusNotFound)
- return
- }
-
- w.Header().Set("Content-Type", "application/gzip")
- w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename))
- http.ServeFile(w, r, snapshotPath)
-}
-
-// cleanOldSnapshots removes snapshot files older than maxAge.
-func cleanOldSnapshots(dir string, maxAge time.Duration) {
- entries, err := os.ReadDir(dir)
- if err != nil {
- return
- }
- cutoff := time.Now().Add(-maxAge)
- for _, e := range entries {
- if e.IsDir() {
- continue
- }
- info, err := e.Info()
- if err != nil {
- continue
- }
- if info.ModTime().Before(cutoff) {
- os.Remove(filepath.Join(dir, e.Name()))
- }
- }
-}
-
-// initDataMaxAge is the maximum age of initData before it is considered expired.
-const initDataMaxAge = 24 * time.Hour
-
-// ValidateInitData verifies the Telegram WebApp initData HMAC-SHA256 signature
-// and checks that auth_date is not older than initDataMaxAge.
-// See https://core.telegram.org/bots/webapps#validating-data-received-via-the-mini-app
-func ValidateInitData(initData, botToken string) bool {
- values, err := url.ParseQuery(initData)
- if err != nil {
- return false
- }
-
- receivedHash := values.Get("hash")
- if receivedHash == "" {
- return false
- }
-
- // Check auth_date freshness
- if authDateStr := values.Get("auth_date"); authDateStr != "" {
- authDate, err := strconv.ParseInt(authDateStr, 10, 64)
- if err != nil {
- return false
- }
- if time.Since(time.Unix(authDate, 0)) > initDataMaxAge {
- return false
- }
- }
-
- // Build the data-check-string: sort all key=value pairs except "hash",
- // join with newlines.
- var pairs []string
- for key := range values {
- if key == "hash" {
- continue
- }
- pairs = append(pairs, fmt.Sprintf("%s=%s", key, values.Get(key)))
- }
- sort.Strings(pairs)
- dataCheckString := strings.Join(pairs, "\n")
-
- // secret_key = HMAC-SHA256("WebAppData", bot_token)
- secretKeyMac := hmac.New(sha256.New, []byte("WebAppData"))
- secretKeyMac.Write([]byte(botToken))
- secretKey := secretKeyMac.Sum(nil)
-
- // hash = HMAC-SHA256(secret_key, data_check_string)
- hashMac := hmac.New(sha256.New, secretKey)
- hashMac.Write([]byte(dataCheckString))
- computedHash := hex.EncodeToString(hashMac.Sum(nil))
-
- return hmac.Equal([]byte(computedHash), []byte(receivedHash))
-}
diff --git a/pkg/miniapp/static/map-preview.html b/pkg/miniapp/static/map-preview.html
new file mode 100644
index 000000000..5cadcae5d
--- /dev/null
+++ b/pkg/miniapp/static/map-preview.html
@@ -0,0 +1,458 @@
+
+
+
+
+
+ Orchestration Room
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ demo: idle
+ ⬡ fast = toolcall
+ ⬡ slow = llm wait
+
+
+
+
+
diff --git a/pkg/miniapp/static/map.js b/pkg/miniapp/static/map.js
new file mode 100644
index 000000000..74b56e433
--- /dev/null
+++ b/pkg/miniapp/static/map.js
@@ -0,0 +1,204 @@
+// map.js â Orchestration Room
+//
+// External asset: drop map.png (320Ã320px) next to index.html to replace
+// the procedural fallback. Character positions (MAP_POSITIONS) are defined
+// in canvas-pixel coordinates and remain valid regardless of which rendering
+// path is used â just make sure your map.png matches them.
+//
+// Usage:
+// loadMapAsset(function() { drawMap(ctx); }); // call once on init
+// drawMap(ctx); // call each frame
+
+// âââ Character home positions (px, canvas 320Ã320) âââââââââââââââââââââââââ
+//
+// ââââââââââââââââââââââââââââââââ
+// â [conductor desk] â y â 20â50
+// â ð(160,58) ð©âðŒ(108,58) â
+// â [carpet] â
+// â [WS1] [WS2] [WS3] â y â 80â100
+// â ð40 ð»144 ð248 â y = 106
+// â [meeting area] â y â 130â192
+// â [WS4] [WS5] â y â 200â220
+// â ð§40 ð¯144 â y = 222
+// â ðª(160,308) â door
+// ââââââââââââââââââââââââââââââââ
+
+var MAP_POSITIONS = {
+ door: { x: 160, y: 314 }, // entry / exit point
+ conductor: { x: 160, y: 58 },
+ secretary: { x: 108, y: 58 },
+ meeting: { x: 160, y: 161 }, // neutral zone for conversations
+ stations: [
+ { x: 40, y: 106 }, // S0 scout
+ { x: 144, y: 106 }, // S1 analyst
+ { x: 248, y: 106 }, // S2 coder
+ { x: 40, y: 222 }, // S3 worker
+ { x: 144, y: 222 }, // S4 coordinator
+ ],
+};
+
+// âââ Asset loading ââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ
+
+var _mapImage = null;
+
+// Call once before first draw. cb() is invoked when ready (image or fallback).
+function loadMapAsset(cb) {
+ var img = new Image();
+ img.onload = function() { _mapImage = img; cb(); };
+ img.onerror = function() { cb(); }; // no map.png â use fallback
+ img.src = './map.png';
+}
+
+// âââ Public draw entry point ââââââââââââââââââââââââââââââââââââââââââââââââ
+
+function drawMap(ctx) {
+ ctx.imageSmoothingEnabled = false;
+ if (_mapImage) {
+ ctx.drawImage(_mapImage, 0, 0, 320, 320);
+ } else {
+ _drawMapFallback(ctx);
+ }
+}
+
+// âââ Procedural fallback ââââââââââââââââââââââââââââââââââââââââââââââââââââ
+
+var _C = {
+ wallDark: '#0c1018',
+ wallHighlight: '#252d3f',
+ floorA: '#171b2c',
+ floorB: '#1b2033',
+ carpetBase: '#1a2050',
+ carpetBorder: '#2a3480',
+ deskBack: '#2c3e6b',
+ deskTop: '#3a50a0',
+ deskEdge: '#4a6ac0',
+ deskShadow: '#1a2448',
+ monitorFrame: '#070b14',
+ monitorBlue: '#1040a0',
+ monitorGlow: '#4488ff',
+ wsBase: '#162818',
+ wsTop: '#1e3822',
+ wsEdge: '#2a5030',
+ termGlow: '#00dd55',
+ rugFill: '#1c2248',
+ rugBorder: '#283070',
+ doorMid: '#8a5818',
+ doorLight: '#a06820',
+ doorGold: '#c8940a',
+};
+
+function _r(ctx, color, x, y, w, h, alpha) {
+ ctx.globalAlpha = alpha === undefined ? 1 : alpha;
+ ctx.fillStyle = color;
+ ctx.fillRect(x, y, w, h);
+ ctx.globalAlpha = 1;
+}
+
+function _b(ctx, color, x, y, w, h) {
+ ctx.strokeStyle = color;
+ ctx.lineWidth = 1;
+ ctx.strokeRect(x + 0.5, y + 0.5, w - 1, h - 1);
+}
+
+function _dot(ctx, color, x, y) {
+ ctx.fillStyle = color;
+ ctx.fillRect(x, y, 2, 2);
+}
+
+function _workstation(ctx, x, y) {
+ _r(ctx, _C.wsBase, x, y, 48, 20);
+ _r(ctx, _C.wsTop, x, y, 48, 8);
+ _r(ctx, _C.wsEdge, x, y, 2, 20);
+ _r(ctx, _C.wsEdge, x+46, y, 2, 20);
+ _r(ctx, _C.wsEdge, x, y, 48, 2);
+ // terminal screen
+ _r(ctx, _C.monitorFrame, x+16, y+2, 16, 12);
+ _r(ctx, '#041008', x+17, y+3, 14, 10);
+ _r(ctx, '#003315', x+18, y+4, 12, 8);
+ _r(ctx, _C.termGlow, x+20, y+6, 8, 3);
+ _dot(ctx, '#00ff88', x+22, y+6);
+}
+
+function _drawMapFallback(ctx) {
+ var T = 16;
+
+ // floor tiles
+ for (var ty = 0; ty < 20; ty++) {
+ for (var tx = 0; tx < 20; tx++) {
+ ctx.fillStyle = (tx + ty) % 2 === 0 ? _C.floorA : _C.floorB;
+ ctx.fillRect(tx * T, ty * T, T, T);
+ }
+ }
+
+ // conductor carpet
+ _r(ctx, _C.carpetBase, 16, 16, 288, 50);
+ _b(ctx, _C.carpetBorder, 18, 18, 284, 46);
+
+ // conductor desk
+ _r(ctx, _C.deskBack, 96, 20, 128, 30);
+ _r(ctx, _C.deskTop, 96, 20, 128, 12);
+ _r(ctx, _C.deskEdge, 96, 20, 128, 2);
+ _r(ctx, _C.deskEdge, 96, 20, 2, 30);
+ _r(ctx, _C.deskEdge, 222, 20, 2, 30);
+ _r(ctx, _C.deskShadow,96,48, 128, 4);
+ // monitor
+ _r(ctx, _C.monitorFrame, 138, 22, 44, 14);
+ _r(ctx, _C.monitorBlue, 140, 23, 40, 12);
+ _r(ctx, _C.monitorGlow, 156, 26, 8, 6);
+ _r(ctx, '#6699ff', 158, 27, 4, 3);
+
+ // workstations
+ _workstation(ctx, 16, 80); // S0
+ _workstation(ctx, 128, 80); // S1 (x+24 = 152 â 144 center)
+ _workstation(ctx, 224, 80); // S2
+ _workstation(ctx, 16, 200); // S3
+ _workstation(ctx, 128, 200); // S4
+
+ // meeting rug
+ _r(ctx, _C.rugFill, 64, 130, 192, 62, 0.55);
+ _b(ctx, _C.rugBorder, 66, 132, 188, 58);
+ _b(ctx, '#202860', 70, 136, 180, 50);
+
+ // bulletin board (left wall)
+ _r(ctx, '#2c1a06', 18, 148, 36, 44);
+ _r(ctx, '#3a2508', 20, 150, 32, 40);
+ _r(ctx, '#cc9900', 22, 153, 12, 8);
+ _r(ctx, '#dd8800', 22, 164, 10, 6);
+ _r(ctx, '#bb7700', 34, 155, 13, 8);
+ _r(ctx, '#ccaa00', 33, 165, 11, 6);
+ _dot(ctx, '#ff4444', 28, 153);
+ _dot(ctx, '#44aaff', 41, 158);
+ _dot(ctx, '#44ff88', 27, 165);
+
+ // server rack (right wall)
+ _r(ctx, '#111122', 285, 80, 18, 112);
+ _r(ctx, '#181830', 287, 82, 14, 108);
+ for (var i = 0; i < 10; i++) {
+ var ry = 85 + i * 10;
+ _r(ctx, '#0a0a12', 288, ry, 12, 8);
+ var lc = ['#00ff44','#0044ff','#ff3300','#111111'][i % 4];
+ _r(ctx, lc, 296, ry + 2, 3, 4);
+ }
+
+ // walls (drawn last to cover any overruns)
+ _r(ctx, _C.wallDark, 0, 0, 320, 16);
+ _r(ctx, _C.wallHighlight, 0, 14, 320, 2);
+ _r(ctx, _C.wallDark, 0, 0, 16, 320);
+ _r(ctx, _C.wallHighlight,14, 0, 2, 320);
+ _r(ctx, _C.wallDark, 304, 0, 16, 320);
+ _r(ctx, _C.wallHighlight,304, 0, 2, 320);
+ _r(ctx, _C.wallDark, 0, 304, 144, 16);
+ _r(ctx, _C.wallDark, 176, 304, 144, 16);
+ _r(ctx, _C.wallHighlight, 0, 304, 144, 2);
+ _r(ctx, _C.wallHighlight,176, 304, 144, 2);
+
+ // door
+ _r(ctx, '#0a0808', 144, 292, 32, 12); // outside (dark)
+ _r(ctx, _C.doorMid, 144, 280, 32, 24);
+ _r(ctx, _C.doorLight, 144, 280, 32, 3);
+ _r(ctx, _C.doorLight, 144, 280, 3, 24);
+ _r(ctx, _C.doorLight, 173, 280, 3, 24);
+ _r(ctx, '#4a2408', 146, 284, 12, 16); // door panels
+ _r(ctx, '#4a2408', 162, 284, 12, 16);
+ _r(ctx, _C.doorGold, 170, 291, 5, 5); // handle
+}
diff --git a/pkg/miniapp/types.go b/pkg/miniapp/types.go
new file mode 100644
index 000000000..2e2fd1016
--- /dev/null
+++ b/pkg/miniapp/types.go
@@ -0,0 +1,180 @@
+package miniapp
+
+import (
+ "sync"
+
+ "github.com/sipeed/picoclaw/pkg/skills"
+ "github.com/sipeed/picoclaw/pkg/stats"
+)
+
+// PlanPhase mirrors agent.PlanPhase for JSON serialization.
+type PlanPhase struct {
+ Number int `json:"number"`
+ Title string `json:"title"`
+ Steps []PlanStep `json:"steps"`
+}
+
+// PlanStep mirrors agent.PlanStep for JSON serialization.
+type PlanStep struct {
+ Index int `json:"index"`
+ Description string `json:"description"`
+ Done bool `json:"done"`
+}
+
+// PlanInfo represents the plan state exposed via the API.
+type PlanInfo struct {
+ HasPlan bool `json:"has_plan"`
+ Status string `json:"status"`
+ CurrentPhase int `json:"current_phase"`
+ TotalPhases int `json:"total_phases"`
+ Display string `json:"display"`
+ Phases []PlanPhase `json:"phases"`
+ Memory string `json:"memory"`
+}
+
+// SessionInfo represents an active session entry for the API response.
+type SessionInfo struct {
+ SessionKey string `json:"session_key"`
+ Channel string `json:"channel"`
+ ChatID string `json:"chat_id"`
+ TouchDir string `json:"touch_dir"`
+ ProjectPath string `json:"project_path,omitempty"`
+ Purpose string `json:"purpose,omitempty"`
+ Branch string `json:"branch,omitempty"`
+ LastSeenAt string `json:"last_seen_at"`
+ AgeSec int `json:"age_sec"`
+}
+
+// GitRepoSummary represents a lightweight repo entry for the list view.
+type GitRepoSummary struct {
+ Name string `json:"name"`
+ Branch string `json:"branch"`
+}
+
+// GitInfo represents the git repository state exposed via the API.
+type GitInfo struct {
+ Name string `json:"name"`
+ Branch string `json:"branch"`
+ Commits []GitCommit `json:"commits"`
+ Modified []GitChange `json:"modified"`
+}
+
+// GitCommit represents a single commit entry.
+type GitCommit struct {
+ Hash string `json:"hash"`
+ Subject string `json:"subject"`
+ Author string `json:"author"`
+ Date string `json:"date"`
+}
+
+// GitChange represents a modified/untracked file entry.
+type GitChange struct {
+ Status string `json:"status"`
+ Path string `json:"path"`
+}
+
+// BootstrapFileInfo describes a resolved bootstrap file for the context API.
+type BootstrapFileInfo struct {
+ Name string `json:"name"`
+ Path string `json:"path"`
+ Scope string `json:"scope"`
+}
+
+// ContextInfo describes the agent's directory context and bootstrap file resolution.
+type ContextInfo struct {
+ WorkDir string `json:"work_dir"`
+ PlanWorkDir string `json:"plan_work_dir"`
+ Workspace string `json:"workspace"`
+ Bootstrap []BootstrapFileInfo `json:"bootstrap"`
+}
+
+// DataProvider is the read-only interface to agent state for the Mini App API.
+type DataProvider interface {
+ ListSkills() []skills.SkillInfo
+ GetPlanInfo() PlanInfo
+ GetSessionStats() *stats.Stats
+ GetActiveSessions() []SessionInfo
+ GetGitRepos() []GitRepoSummary
+ GetGitRepoDetail(name string) GitInfo
+ GetContextInfo() ContextInfo
+ GetSystemPrompt() string
+}
+
+// CommandSender injects a command into the message bus on behalf of a user.
+type CommandSender interface {
+ SendCommand(senderID, chatID, command string)
+}
+
+// DevTarget represents a registered dev server target.
+type DevTarget struct {
+ ID string `json:"id"`
+ Name string `json:"name"` // display name (e.g. "frontend")
+ Target string `json:"target"` // URL (e.g. "http://localhost:3000")
+}
+
+// DevTargetManager allows tools to register, activate, and deactivate dev proxy targets.
+type DevTargetManager interface {
+ RegisterDevTarget(name, target string) (id string, err error)
+ UnregisterDevTarget(id string) error
+ ActivateDevTarget(id string) error
+ DeactivateDevTarget() error
+ GetDevTarget() string
+ ListDevTargets() []DevTarget
+}
+
+// StateNotifier broadcasts state-change signals to SSE subscribers.
+type StateNotifier struct {
+ mu sync.Mutex
+ subs map[chan struct{}]struct{}
+ done chan struct{}
+}
+
+// NewStateNotifier creates a new StateNotifier.
+func NewStateNotifier() *StateNotifier {
+ return &StateNotifier{
+ subs: make(map[chan struct{}]struct{}),
+ done: make(chan struct{}),
+ }
+}
+
+// Subscribe returns a channel that receives a signal on each state change.
+func (n *StateNotifier) Subscribe() chan struct{} {
+ ch := make(chan struct{}, 1)
+ n.mu.Lock()
+ n.subs[ch] = struct{}{}
+ n.mu.Unlock()
+ return ch
+}
+
+// Unsubscribe removes a subscriber channel.
+func (n *StateNotifier) Unsubscribe(ch chan struct{}) {
+ n.mu.Lock()
+ delete(n.subs, ch)
+ n.mu.Unlock()
+}
+
+// Close signals all SSE handlers to exit.
+func (n *StateNotifier) Close() {
+ select {
+ case <-n.done:
+ default:
+ close(n.done)
+ }
+}
+
+// Done returns a channel that is closed when the notifier is shut down.
+func (n *StateNotifier) Done() <-chan struct{} {
+ return n.done
+}
+
+// Notify sends a signal to all subscribers, coalescing rapid notifications.
+func (n *StateNotifier) Notify() {
+ n.mu.Lock()
+ defer n.mu.Unlock()
+ for ch := range n.subs {
+ select {
+ case ch <- struct{}{}:
+ default:
+ }
+ }
+}
diff --git a/pkg/miniapp/ws.go b/pkg/miniapp/ws.go
new file mode 100644
index 000000000..d754c6b7c
--- /dev/null
+++ b/pkg/miniapp/ws.go
@@ -0,0 +1,224 @@
+package miniapp
+
+import (
+ "net/http"
+ "net/url"
+ "strings"
+ "time"
+
+ "github.com/gorilla/websocket"
+ "github.com/sipeed/picoclaw/pkg/logger"
+)
+
+
+const maxWSClients = 4
+
+const (
+ wsPongWait = 60 * time.Second
+ wsPingPeriod = 54 * time.Second // must be less than wsPongWait
+)
+
+type wsClient struct {
+ conn *websocket.Conn
+}
+
+
+var wsUpgrader = websocket.Upgrader{
+ CheckOrigin: func(r *http.Request) bool {
+ origin := r.Header.Get("Origin")
+ if origin == "" {
+ return true // non-browser clients (e.g. curl)
+ }
+ // Allow same-origin requests (e.g. Tailscale direct access)
+ if u, err := url.Parse(origin); err == nil && u.Host == r.Host {
+ return true
+ }
+ // Allow Telegram WebApp origins and localhost for dev
+ return strings.HasSuffix(origin, ".telegram.org") ||
+ strings.HasSuffix(origin, ".t.me") ||
+ strings.HasPrefix(origin, "http://localhost") ||
+ strings.HasPrefix(origin, "http://127.0.0.1")
+ },
+}
+
+// NewHandler creates a new Mini App handler.
+
+
+// wsLogs serves a WebSocket endpoint that streams log entries in real time.
+func (h *Handler) wsLogs(w http.ResponseWriter, r *http.Request) {
+ // Parse filter params
+ component := r.URL.Query().Get("component")
+ levelStr := r.URL.Query().Get("level")
+ minLevel := logger.INFO
+ if levelStr != "" {
+ minLevel = logger.ParseLevel(levelStr)
+ }
+
+ // Clear HTTP server deadlines before WebSocket hijack
+ rc := http.NewResponseController(w)
+ _ = rc.SetWriteDeadline(time.Time{})
+ _ = rc.SetReadDeadline(time.Time{})
+
+ conn, err := wsUpgrader.Upgrade(w, r, nil)
+ if err != nil {
+ return
+ }
+
+ client := &wsClient{conn: conn}
+
+ // Enforce max WS clients: evict oldest if full
+ h.wsClientsMu.Lock()
+ if len(h.wsClients) >= maxWSClients {
+ oldest := h.wsClients[0]
+ h.wsClients = h.wsClients[1:]
+ oldest.conn.Close()
+ }
+ h.wsClients = append(h.wsClients, client)
+ h.wsClientsMu.Unlock()
+
+ defer func() {
+ h.wsClientsMu.Lock()
+ for i, c := range h.wsClients {
+ if c == client {
+ h.wsClients = append(h.wsClients[:i], h.wsClients[i+1:]...)
+ break
+ }
+ }
+ h.wsClientsMu.Unlock()
+ conn.Close()
+ }()
+
+ // Build filter function
+ filter := func(e logger.LogEntry) bool {
+ if lvl := logger.ParseLevel(e.Level); lvl < minLevel {
+ return false
+ }
+ if component != "" && e.Component != component {
+ return false
+ }
+ return true
+ }
+
+ sub := logger.Subscribe(filter)
+ defer logger.Unsubscribe(sub)
+
+ // Configure ping/pong to detect dead connections
+ conn.SetReadDeadline(time.Now().Add(wsPongWait))
+ conn.SetPongHandler(func(string) error {
+ conn.SetReadDeadline(time.Now().Add(wsPongWait))
+ return nil
+ })
+
+ // Send initial data
+ initial := logger.RecentLogs(minLevel, component, 50)
+ if err := conn.WriteJSON(map[string]any{"type": "init", "entries": initial}); err != nil {
+ return
+ }
+
+ // Close detection goroutine
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ for {
+ if _, _, err := conn.ReadMessage(); err != nil {
+ return
+ }
+ }
+ }()
+
+ // Stream loop with periodic pings
+ ticker := time.NewTicker(wsPingPeriod)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case entry, ok := <-sub.Ch:
+ if !ok {
+ return
+ }
+ entry.Caller = "" // strip for security
+ entry.Fields = logger.SanitizeFields(entry.Fields) // mask sensitive values
+ if err := conn.WriteJSON(map[string]any{"type": "entry", "entry": entry}); 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.
+
+// wsOrchestration streams live orchestration events (agent spawn/state/gc and
+// conductor<->agent conversations) to the canvas UI.
+//
+// Protocol:
+//
+// {"type":"init","agents":[...orch.AgentInfo]} -- sent once on connect
+// {"type":"event","event":{...orch.Event}} -- 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
+ }
+ }
+}
diff --git a/pkg/orch/broadcaster.go b/pkg/orch/broadcaster.go
new file mode 100644
index 000000000..70fff1881
--- /dev/null
+++ b/pkg/orch/broadcaster.go
@@ -0,0 +1,143 @@
+// 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
+}
+
+// ReportSpawn implements AgentReporter.
+func (b *Broadcaster) ReportSpawn(id, label, task string) {
+ b.Publish(Event{Type: "agent_spawn", ID: id, Label: label, Task: task})
+}
+
+// ReportStateChange implements AgentReporter.
+func (b *Broadcaster) ReportStateChange(id, state, tool string) {
+ b.Publish(Event{Type: "agent_state", ID: id, State: state, Tool: tool})
+}
+
+// ReportConversation implements AgentReporter.
+func (b *Broadcaster) ReportConversation(from, to, text string) {
+ b.Publish(Event{Type: "conversation", From: from, To: to, Text: text})
+}
+
+// ReportGC implements AgentReporter.
+func (b *Broadcaster) ReportGC(id, reason string) {
+ b.Publish(Event{Type: "agent_gc", ID: id, Reason: reason})
+}
+
+// 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/orch/broadcaster_test.go b/pkg/orch/broadcaster_test.go
new file mode 100644
index 000000000..54ae8af02
--- /dev/null
+++ b/pkg/orch/broadcaster_test.go
@@ -0,0 +1,140 @@
+package orch
+
+import (
+ "testing"
+ "time"
+)
+
+func TestBroadcasterSpawnAndGC(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ b.Publish(Event{Type: "agent_spawn", ID: "t1", Label: "scout", Task: "do something"})
+
+ select {
+ case ev := <-sub.Ch:
+ if ev.Type != "agent_spawn" || ev.ID != "t1" {
+ t.Fatalf("expected agent_spawn for t1, got %+v", ev)
+ }
+ case <-time.After(100 * time.Millisecond):
+ t.Fatal("timeout waiting for agent_spawn event")
+ }
+
+ snap := b.Snapshot()
+ if len(snap) != 1 || snap[0].ID != "t1" {
+ t.Fatalf("expected 1 agent in snapshot, got %v", snap)
+ }
+
+ b.Publish(Event{Type: "agent_gc", ID: "t1", Reason: "completed"})
+
+ select {
+ case ev := <-sub.Ch:
+ if ev.Type != "agent_gc" || ev.Reason != "completed" {
+ t.Fatalf("expected agent_gc/completed, got %+v", ev)
+ }
+ case <-time.After(100 * time.Millisecond):
+ t.Fatal("timeout waiting for agent_gc event")
+ }
+
+ if len(b.Snapshot()) != 0 {
+ t.Fatal("snapshot should be empty after agent_gc")
+ }
+}
+
+func TestBroadcasterAgentState(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ b.Publish(Event{Type: "agent_spawn", ID: "t1"})
+ <-sub.Ch // consume spawn
+
+ b.Publish(Event{Type: "agent_state", ID: "t1", State: "toolcall", Tool: "bash"})
+
+ select {
+ case ev := <-sub.Ch:
+ if ev.State != "toolcall" || ev.Tool != "bash" {
+ t.Fatalf("unexpected state event: %+v", ev)
+ }
+ case <-time.After(100 * time.Millisecond):
+ t.Fatal("timeout waiting for agent_state event")
+ }
+
+ snap := b.Snapshot()
+ if len(snap) == 0 || snap[0].State != "toolcall" || snap[0].Tool != "bash" {
+ t.Fatalf("snapshot state not updated: %v", snap)
+ }
+}
+
+func TestBroadcasterNonBlocking(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe() // do NOT read from sub.Ch
+ defer b.Unsubscribe(sub)
+
+ // Fill buffer beyond capacity (cap=32) â must not block or deadlock
+ done := make(chan struct{})
+ go func() {
+ for i := 0; i < 50; i++ {
+ b.Publish(Event{Type: "agent_state", ID: "t1", State: "waiting"})
+ }
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(500 * time.Millisecond):
+ t.Fatal("Publish blocked on slow subscriber")
+ }
+}
+
+func TestBroadcasterMultipleSubscribers(t *testing.T) {
+ b := NewBroadcaster()
+ s1 := b.Subscribe()
+ s2 := b.Subscribe()
+ defer b.Unsubscribe(s1)
+ defer b.Unsubscribe(s2)
+
+ b.Publish(Event{Type: "agent_spawn", ID: "t1", Label: "worker"})
+
+ for _, sub := range []*Subscriber{s1, s2} {
+ select {
+ case ev := <-sub.Ch:
+ if ev.Type != "agent_spawn" {
+ t.Fatalf("expected agent_spawn, got %s", ev.Type)
+ }
+ case <-time.After(100 * time.Millisecond):
+ t.Fatal("timeout: not all subscribers received event")
+ }
+ }
+}
+
+func TestBroadcasterUnsubscribe(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ b.Unsubscribe(sub)
+
+ b.Publish(Event{Type: "agent_spawn", ID: "t1"})
+
+ select {
+ case ev := <-sub.Ch:
+ t.Fatalf("received event after unsubscribe: %+v", ev)
+ case <-time.After(50 * time.Millisecond):
+ // correct: nothing delivered after unsubscribe
+ }
+}
+
+func TestBroadcasterTimestampAutoSet(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ before := time.Now().UnixMilli()
+ b.Publish(Event{Type: "agent_spawn", ID: "t1"}) // Created == 0
+ after := time.Now().UnixMilli()
+
+ ev := <-sub.Ch
+ if ev.Created < before || ev.Created > after {
+ t.Fatalf("Created timestamp %d not in [%d, %d]", ev.Created, before, after)
+ }
+}
diff --git a/pkg/orch/reporter.go b/pkg/orch/reporter.go
new file mode 100644
index 000000000..b8229314d
--- /dev/null
+++ b/pkg/orch/reporter.go
@@ -0,0 +1,21 @@
+package orch
+
+// AgentReporter is the interface for reporting agent lifecycle events.
+// Both Broadcaster (real events) and noopReporter (disabled) implement this.
+type AgentReporter interface {
+ ReportSpawn(id, label, task string)
+ ReportStateChange(id, state, tool string)
+ ReportConversation(from, to, text string)
+ ReportGC(id, reason string)
+}
+
+type noopReporter struct{}
+
+func (n *noopReporter) ReportSpawn(id, label, task string) {}
+func (n *noopReporter) ReportStateChange(id, state, tool string) {}
+func (n *noopReporter) ReportConversation(from, to, text string) {}
+func (n *noopReporter) ReportGC(id, reason string) {}
+
+// Noop is the AgentReporter to use when orchestration is disabled.
+// Allows nil-free code in callers.
+var Noop AgentReporter = &noopReporter{}
diff --git a/pkg/orch/reporter_test.go b/pkg/orch/reporter_test.go
new file mode 100644
index 000000000..120f03c26
--- /dev/null
+++ b/pkg/orch/reporter_test.go
@@ -0,0 +1,102 @@
+package orch
+
+import "testing"
+
+// Compile-time: Broadcaster must satisfy AgentReporter.
+var _ AgentReporter = (*Broadcaster)(nil)
+
+// TestNoop_AllMethods_NoPanic verifies that orch.Noop can be called for all
+// four methods without panic. This is the nil-safe baseline for disabled
+// orchestration mode.
+func TestNoop_AllMethods_NoPanic(t *testing.T) {
+ Noop.ReportSpawn("id", "label", "task")
+ Noop.ReportStateChange("id", "waiting", "")
+ Noop.ReportStateChange("id", "toolcall", "bash")
+ Noop.ReportConversation("conductor", "sub-1", "do something")
+ Noop.ReportGC("id", "completed")
+}
+
+// TestBroadcaster_ReportSpawn_MapsToAgentSpawnEvent verifies that ReportSpawn
+// publishes an Event with Type="agent_spawn" and the correct ID/Label/Task
+// fields, and that the agent appears in the Snapshot immediately.
+func TestBroadcaster_ReportSpawn_MapsToAgentSpawnEvent(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ b.ReportSpawn("agent-1", "scout", "find all TODOs")
+
+ ev := <-sub.Ch
+ if ev.Type != "agent_spawn" {
+ t.Fatalf("want agent_spawn, got %q", ev.Type)
+ }
+ if ev.ID != "agent-1" || ev.Label != "scout" || ev.Task != "find all TODOs" {
+ t.Fatalf("field mismatch: %+v", ev)
+ }
+ snap := b.Snapshot()
+ if len(snap) != 1 || snap[0].ID != "agent-1" || snap[0].Label != "scout" {
+ t.Fatalf("snapshot not updated correctly: %v", snap)
+ }
+}
+
+// TestBroadcaster_ReportStateChange_MapsToAgentStateEvent verifies that
+// ReportStateChange publishes agent_state and updates the live snapshot.
+func TestBroadcaster_ReportStateChange_MapsToAgentStateEvent(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ b.ReportSpawn("agent-1", "coder", "implement it")
+ <-sub.Ch // consume spawn
+
+ b.ReportStateChange("agent-1", "toolcall", "bash")
+ ev := <-sub.Ch
+ if ev.Type != "agent_state" || ev.State != "toolcall" || ev.Tool != "bash" {
+ t.Fatalf("unexpected event: %+v", ev)
+ }
+ snap := b.Snapshot()
+ if snap[0].State != "toolcall" || snap[0].Tool != "bash" {
+ t.Fatalf("snapshot state not updated: %v", snap)
+ }
+}
+
+// TestBroadcaster_ReportConversation_MapsToConversationEvent verifies that
+// ReportConversation publishes a conversation event with correct From/To/Text
+// fields and does NOT modify the agent snapshot (conversation is not a state
+// change of any agent).
+func TestBroadcaster_ReportConversation_MapsToConversationEvent(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ b.ReportConversation("conductor", "sub-1", "please do the task")
+
+ ev := <-sub.Ch
+ if ev.Type != "conversation" || ev.From != "conductor" || ev.To != "sub-1" || ev.Text != "please do the task" {
+ t.Fatalf("unexpected event: %+v", ev)
+ }
+ if len(b.Snapshot()) != 0 {
+ t.Fatal("conversation event must not modify agent snapshot")
+ }
+}
+
+// TestBroadcaster_ReportGC_RemovesAgentFromSnapshot verifies that ReportGC
+// publishes agent_gc with the correct Reason and removes the agent from the
+// live snapshot so new WS connections no longer see it.
+func TestBroadcaster_ReportGC_RemovesAgentFromSnapshot(t *testing.T) {
+ b := NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ b.ReportSpawn("agent-1", "scout", "task")
+ <-sub.Ch // consume spawn
+
+ b.ReportGC("agent-1", "completed")
+ ev := <-sub.Ch
+ if ev.Type != "agent_gc" || ev.ID != "agent-1" || ev.Reason != "completed" {
+ t.Fatalf("unexpected event: %+v", ev)
+ }
+ if len(b.Snapshot()) != 0 {
+ t.Fatal("agent must be removed from snapshot after ReportGC")
+ }
+}
diff --git a/pkg/tools/subagent.go b/pkg/tools/subagent.go
index 91ebff636..3956cf0fd 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,13 +37,18 @@ type SubagentManager struct {
hasMaxTokens bool
hasTemperature bool
nextID int
+ reporter orch.AgentReporter
}
func NewSubagentManager(
provider providers.LLMProvider,
defaultModel, workspace string,
bus *bus.MessageBus,
+ reporter orch.AgentReporter,
) *SubagentManager {
+ if reporter == nil {
+ reporter = orch.Noop
+ }
return &SubagentManager{
tasks: make(map[string]*SubagentTask),
provider: provider,
@@ -52,6 +58,7 @@ func NewSubagentManager(
tools: NewToolRegistry(),
maxIterations: 10,
nextID: 1,
+ reporter: reporter,
}
}
@@ -103,6 +110,8 @@ func (sm *SubagentManager) Spawn(
}
sm.tasks[taskID] = subagentTask
+ sm.reporter.ReportSpawn(taskID, label, task)
+
// Start task in background with context cancellation support
go sm.runTask(ctx, subagentTask, callback)
@@ -114,7 +123,6 @@ func (sm *SubagentManager) Spawn(
func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, callback AsyncCallback) {
task.Status = "running"
- task.Created = time.Now().UnixMilli()
// Build system prompt for subagent
systemPrompt := `You are a subagent. Complete the given task independently and report the result.
@@ -164,12 +172,17 @@ After completing the task, provide a clear summary of what was done.`
}
}
+ // Notify conductor that the subagent is starting
+ sm.reporter.ReportConversation("conductor", task.ID, task.Task)
+
loopResult, err := RunToolLoop(ctx, ToolLoopConfig{
Provider: sm.provider,
Model: sm.defaultModel,
Tools: tools,
MaxIterations: maxIter,
LLMOptions: llmOptions,
+ Reporter: sm.reporter,
+ AgentID: task.ID,
}, messages, task.OriginChannel, task.OriginChatID)
sm.mu.Lock()
@@ -186,10 +199,13 @@ 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.reporter.ReportGC(task.ID, gcReason)
result = &ToolResult{
ForLLM: task.Result,
ForUser: "",
@@ -201,6 +217,9 @@ 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.reporter.ReportConversation(task.ID, "conductor", loopResult.Content)
+ sm.reporter.ReportGC(task.ID, "completed")
result = &ToolResult{
ForLLM: fmt.Sprintf(
"Subagent '%s' completed (iterations: %d): %s",
diff --git a/pkg/tools/subagent_reporter_test.go b/pkg/tools/subagent_reporter_test.go
new file mode 100644
index 000000000..aaf3bd7a4
--- /dev/null
+++ b/pkg/tools/subagent_reporter_test.go
@@ -0,0 +1,243 @@
+package tools
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/sipeed/picoclaw/pkg/orch"
+ "github.com/sipeed/picoclaw/pkg/providers"
+)
+
+// blockingProvider blocks inside Chat until the context is cancelled.
+// The ready channel is closed the moment Chat is entered, so callers can
+// synchronise before cancelling the context.
+type blockingProvider struct {
+ ready chan struct{}
+}
+
+func newBlockingProvider() *blockingProvider {
+ return &blockingProvider{ready: make(chan struct{})}
+}
+
+func (p *blockingProvider) Chat(ctx context.Context, _ []providers.Message, _ []providers.ToolDefinition, _ string, _ map[string]any) (*providers.LLMResponse, error) {
+ close(p.ready) // signal: we are now blocking
+ <-ctx.Done()
+ return nil, ctx.Err()
+}
+
+func (p *blockingProvider) GetDefaultModel() string { return "test" }
+
+// TestSubagentManager_Spawn_EmitsLifecycleEvents verifies that Spawn() fires
+// the correct sequence of orchestration events through a real Broadcaster:
+//
+// agent_spawn â conversation(conductorâsub) â agent_state(waiting) â
+// conversation(subâconductor) â agent_gc(completed)
+//
+// It also verifies that the snapshot is empty after ReportGC and that the
+// completion callback is invoked.
+func TestSubagentManager_Spawn_EmitsLifecycleEvents(t *testing.T) {
+ b := orch.NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ provider := &MockLLMProvider{}
+ mgr := NewSubagentManager(provider, "test-model", "/tmp/test", nil, b)
+
+ var callbackCalled int32
+ cb := AsyncCallback(func(_ context.Context, _ *ToolResult) {
+ atomic.StoreInt32(&callbackCalled, 1)
+ })
+
+ _, err := mgr.Spawn(
+ context.Background(),
+ "say hello", "hello-task", "", "cli", "direct",
+ cb,
+ )
+ if err != nil {
+ t.Fatalf("Spawn() error: %v", err)
+ }
+
+ // Collect events until agent_gc or timeout.
+ var events []orch.Event
+ deadline := time.After(3 * time.Second)
+loop:
+ for {
+ select {
+ case ev := <-sub.Ch:
+ events = append(events, ev)
+ if ev.Type == "agent_gc" {
+ break loop
+ }
+ case <-deadline:
+ t.Fatalf("timed out waiting for agent_gc; events so far: %+v", events)
+ }
+ }
+
+ // 1. First event must be agent_spawn with the correct label.
+ if len(events) == 0 || events[0].Type != "agent_spawn" {
+ t.Fatalf("first event must be agent_spawn, got: %+v", events)
+ }
+ if events[0].Label != "hello-task" {
+ t.Errorf("agent_spawn label = %q, want %q", events[0].Label, "hello-task")
+ }
+ spawnedID := events[0].ID
+
+ // 2. There must be a conversation from conductor â subagent.
+ var hasConvToSub bool
+ for _, ev := range events {
+ if ev.Type == "conversation" && ev.From == "conductor" && ev.To == spawnedID {
+ hasConvToSub = true
+ break
+ }
+ }
+ if !hasConvToSub {
+ t.Errorf("missing conversation(conductor â %s); events: %+v", spawnedID, events)
+ }
+
+ // 3. There must be at least one agent_state(waiting) for the subagent.
+ var hasWaiting bool
+ for _, ev := range events {
+ if ev.Type == "agent_state" && ev.ID == spawnedID && ev.State == "waiting" {
+ hasWaiting = true
+ break
+ }
+ }
+ if !hasWaiting {
+ t.Errorf("missing agent_state(waiting) for %s; events: %+v", spawnedID, events)
+ }
+
+ // 4. Last event must be agent_gc with reason "completed".
+ last := events[len(events)-1]
+ if last.Type != "agent_gc" || last.ID != spawnedID || last.Reason != "completed" {
+ t.Errorf("last event must be agent_gc(completed), got: %+v", last)
+ }
+
+ // 5. Snapshot must be empty after GC (agent removed from live map).
+ if snap := b.Snapshot(); len(snap) != 0 {
+ t.Errorf("snapshot must be empty after agent_gc, got: %v", snap)
+ }
+
+ // 6. Callback must be called. The callback fires in the same goroutine
+ // as ReportGC (after the deferred unlock), so we poll briefly.
+ for i := 0; i < 100; i++ {
+ if atomic.LoadInt32(&callbackCalled) == 1 {
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ if atomic.LoadInt32(&callbackCalled) != 1 {
+ t.Error("completion callback was not called after agent_gc")
+ }
+}
+
+// TestSubagentManager_Spawn_SnapshotLiveDuringExecution verifies that the
+// Broadcaster snapshot contains the agent between agent_spawn and agent_gc.
+// Because Publish() updates the agent map before dispatching to subscribers,
+// the snapshot is guaranteed to be non-empty as soon as agent_spawn is
+// received on the channel.
+func TestSubagentManager_Spawn_SnapshotLiveDuringExecution(t *testing.T) {
+ b := orch.NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ provider := &MockLLMProvider{}
+ mgr := NewSubagentManager(provider, "test-model", "/tmp/test", nil, b)
+
+ _, err := mgr.Spawn(
+ context.Background(),
+ "any task", "live-test", "", "cli", "direct",
+ nil,
+ )
+ if err != nil {
+ t.Fatalf("Spawn() error: %v", err)
+ }
+
+ // Wait for agent_spawn, then immediately check snapshot.
+ deadline := time.After(2 * time.Second)
+ for {
+ select {
+ case ev := <-sub.Ch:
+ if ev.Type == "agent_spawn" {
+ snap := b.Snapshot()
+ if len(snap) == 0 {
+ t.Error("snapshot must contain the spawned agent after agent_spawn event")
+ }
+ return // test complete; background goroutine drains safely
+ }
+ case <-deadline:
+ t.Fatal("timed out waiting for agent_spawn event")
+ }
+ }
+}
+
+// TestSubagentManager_Spawn_CancelledDuringExecution verifies that when the
+// context is cancelled while a subagent's LLM call is in progress, the
+// Broadcaster receives agent_gc with reason="cancelled" and the agent is
+// removed from the snapshot.
+//
+// Synchronisation:
+// 1. blockingProvider.ready is closed when Chat() is entered (goroutine is
+// now blocked inside the LLM call).
+// 2. Only then is the context cancelled, so there is no race between spawn
+// and cancellation.
+func TestSubagentManager_Spawn_CancelledDuringExecution(t *testing.T) {
+ b := orch.NewBroadcaster()
+ sub := b.Subscribe()
+ defer b.Unsubscribe(sub)
+
+ bp := newBlockingProvider()
+ mgr := NewSubagentManager(bp, "test-model", "/tmp/test", nil, b)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ _, err := mgr.Spawn(ctx, "long task", "cancel-me", "", "cli", "direct", nil)
+ if err != nil {
+ t.Fatalf("Spawn() error: %v", err)
+ }
+
+ // Wait until the subagent goroutine is inside Chat (blocking on ctx).
+ select {
+ case <-bp.ready:
+ case <-time.After(3 * time.Second):
+ t.Fatal("timed out waiting for blockingProvider to enter Chat")
+ }
+
+ // Now cancel â the LLM call unblocks with ctx.Err().
+ cancel()
+
+ // Collect events until agent_gc.
+ var events []orch.Event
+ deadline := time.After(3 * time.Second)
+loop:
+ for {
+ select {
+ case ev := <-sub.Ch:
+ events = append(events, ev)
+ if ev.Type == "agent_gc" {
+ break loop
+ }
+ case <-deadline:
+ t.Fatalf("timed out waiting for agent_gc; events so far: %+v", events)
+ }
+ }
+
+ // Locate agent_gc and verify reason = "cancelled".
+ var gcEv orch.Event
+ for _, ev := range events {
+ if ev.Type == "agent_gc" {
+ gcEv = ev
+ break
+ }
+ }
+ if gcEv.Reason != "cancelled" {
+ t.Errorf("agent_gc reason = %q, want %q; events: %+v", gcEv.Reason, "cancelled", events)
+ }
+
+ // Snapshot must be empty after the GC event.
+ if snap := b.Snapshot(); len(snap) != 0 {
+ t.Errorf("snapshot must be empty after agent_gc(cancelled), got: %v", snap)
+ }
+}
diff --git a/pkg/tools/subagent_tool_test.go b/pkg/tools/subagent_tool_test.go
index 59bfdffae..30838d843 100644
--- a/pkg/tools/subagent_tool_test.go
+++ b/pkg/tools/subagent_tool_test.go
@@ -6,6 +6,7 @@ import (
"testing"
"github.com/sipeed/picoclaw/pkg/bus"
+ "github.com/sipeed/picoclaw/pkg/orch"
"github.com/sipeed/picoclaw/pkg/providers"
)
@@ -47,7 +48,7 @@ func (m *MockLLMProvider) GetContextWindow() int {
func TestSubagentManager_SetLLMOptions_AppliesToRunToolLoop(t *testing.T) {
provider := &MockLLMProvider{}
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, orch.Noop)
manager.SetLLMOptions(2048, 0.6)
tool := NewSubagentTool(manager)
tool.SetContext("cli", "direct")
@@ -74,7 +75,7 @@ func TestSubagentManager_SetLLMOptions_AppliesToRunToolLoop(t *testing.T) {
// TestSubagentTool_Name verifies tool name
func TestSubagentTool_Name(t *testing.T) {
provider := &MockLLMProvider{}
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, orch.Noop)
tool := NewSubagentTool(manager)
if tool.Name() != "subagent" {
@@ -85,7 +86,7 @@ func TestSubagentTool_Name(t *testing.T) {
// TestSubagentTool_Description verifies tool description
func TestSubagentTool_Description(t *testing.T) {
provider := &MockLLMProvider{}
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, orch.Noop)
tool := NewSubagentTool(manager)
desc := tool.Description()
@@ -100,7 +101,7 @@ func TestSubagentTool_Description(t *testing.T) {
// TestSubagentTool_Parameters verifies tool parameters schema
func TestSubagentTool_Parameters(t *testing.T) {
provider := &MockLLMProvider{}
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, orch.Noop)
tool := NewSubagentTool(manager)
params := tool.Parameters()
@@ -150,7 +151,7 @@ func TestSubagentTool_Parameters(t *testing.T) {
// TestSubagentTool_SetContext verifies context setting
func TestSubagentTool_SetContext(t *testing.T) {
provider := &MockLLMProvider{}
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, orch.Noop)
tool := NewSubagentTool(manager)
tool.SetContext("test-channel", "test-chat")
@@ -164,7 +165,7 @@ func TestSubagentTool_SetContext(t *testing.T) {
func TestSubagentTool_Execute_Success(t *testing.T) {
provider := &MockLLMProvider{}
msgBus := bus.NewMessageBus()
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus, orch.Noop)
tool := NewSubagentTool(manager)
tool.SetContext("telegram", "chat-123")
@@ -220,7 +221,7 @@ func TestSubagentTool_Execute_Success(t *testing.T) {
func TestSubagentTool_Execute_NoLabel(t *testing.T) {
provider := &MockLLMProvider{}
msgBus := bus.NewMessageBus()
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus, orch.Noop)
tool := NewSubagentTool(manager)
ctx := context.Background()
@@ -243,7 +244,7 @@ func TestSubagentTool_Execute_NoLabel(t *testing.T) {
// TestSubagentTool_Execute_MissingTask tests error handling for missing task
func TestSubagentTool_Execute_MissingTask(t *testing.T) {
provider := &MockLLMProvider{}
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", nil, orch.Noop)
tool := NewSubagentTool(manager)
ctx := context.Background()
@@ -294,7 +295,7 @@ func TestSubagentTool_Execute_NilManager(t *testing.T) {
func TestSubagentTool_Execute_ContextPassing(t *testing.T) {
provider := &MockLLMProvider{}
msgBus := bus.NewMessageBus()
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus, orch.Noop)
tool := NewSubagentTool(manager)
// Set context
@@ -323,7 +324,7 @@ func TestSubagentTool_ForUserTruncation(t *testing.T) {
// Create a mock provider that returns very long content
provider := &MockLLMProvider{}
msgBus := bus.NewMessageBus()
- manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus)
+ manager := NewSubagentManager(provider, "test-model", "/tmp/test", msgBus, orch.Noop)
tool := NewSubagentTool(manager)
ctx := context.Background()
diff --git a/pkg/tools/toolloop.go b/pkg/tools/toolloop.go
index cdfe0d6ce..d8793caef 100644
--- a/pkg/tools/toolloop.go
+++ b/pkg/tools/toolloop.go
@@ -12,6 +12,7 @@ import (
"fmt"
"github.com/sipeed/picoclaw/pkg/logger"
+ "github.com/sipeed/picoclaw/pkg/orch"
"github.com/sipeed/picoclaw/pkg/providers"
"github.com/sipeed/picoclaw/pkg/utils"
)
@@ -23,6 +24,12 @@ type ToolLoopConfig struct {
Tools *ToolRegistry
MaxIterations int
LLMOptions map[string]any
+ // Reporter and AgentID replace the old OnStateChange func.
+ // Reporter is called with ReportStateChange("waiting","") before each LLM
+ // call and ReportStateChange("toolcall", toolName) when each tool starts.
+ // Pass nil or orch.Noop to disable. nil is treated as orch.Noop internally.
+ Reporter orch.AgentReporter
+ AgentID string
}
// ToolLoopResult contains the result of running the tool loop.
@@ -39,6 +46,11 @@ func RunToolLoop(
messages []providers.Message,
channel, chatID string,
) (*ToolLoopResult, error) {
+ reporter := config.Reporter
+ if reporter == nil {
+ reporter = orch.Noop
+ }
+
iteration := 0
var finalContent string
@@ -62,7 +74,8 @@ func RunToolLoop(
if llmOpts == nil {
llmOpts = map[string]any{}
}
- // 3. Call LLM
+ // 3. Call LLM (hook: waiting for response)
+ reporter.ReportStateChange(config.AgentID, "waiting", "")
response, err := config.Provider.Chat(ctx, messages, providerToolDefs, config.Model, llmOpts)
if err != nil {
logger.ErrorCF("toolloop", "LLM call failed",
@@ -121,7 +134,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 +143,7 @@ func RunToolLoop(
"tool": tc.Name,
"iteration": iteration,
})
+ reporter.ReportStateChange(config.AgentID, "toolcall", tc.Name)
// Execute tool (no async callback for subagents - they run independently)
var toolResult *ToolResult
diff --git a/pkg/tools/toolloop_reporter_test.go b/pkg/tools/toolloop_reporter_test.go
new file mode 100644
index 000000000..38284b722
--- /dev/null
+++ b/pkg/tools/toolloop_reporter_test.go
@@ -0,0 +1,170 @@
+package tools
+
+import (
+ "context"
+ "sync"
+ "testing"
+
+ "github.com/sipeed/picoclaw/pkg/orch"
+ "github.com/sipeed/picoclaw/pkg/providers"
+)
+
+// reporterSpy records every ReportStateChange call in order.
+// Spawn/Conversation/GC are not needed for toolloop tests.
+type reporterSpy struct {
+ mu sync.Mutex
+ calls []spyCall
+}
+
+type spyCall struct {
+ state string
+ tool string
+}
+
+func (r *reporterSpy) ReportSpawn(id, label, task string) {}
+func (r *reporterSpy) ReportConversation(from, to, text string) {}
+func (r *reporterSpy) ReportGC(id, reason string) {}
+func (r *reporterSpy) ReportStateChange(id, state, tool string) {
+ r.mu.Lock()
+ r.calls = append(r.calls, spyCall{state, tool})
+ r.mu.Unlock()
+}
+func (r *reporterSpy) snapshot() []spyCall {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ out := make([]spyCall, len(r.calls))
+ copy(out, r.calls)
+ return out
+}
+
+// sequenceMockProvider returns a tool call on the first Chat() call and a
+// plain text response on all subsequent calls. Used to exercise the
+// waiting â toolcall â waiting event sequence in RunToolLoop.
+type sequenceMockProvider struct {
+ mu sync.Mutex
+ callCount int
+}
+
+func (m *sequenceMockProvider) Chat(
+ _ context.Context,
+ _ []providers.Message,
+ _ []providers.ToolDefinition,
+ _ string,
+ _ map[string]any,
+) (*providers.LLMResponse, error) {
+ m.mu.Lock()
+ m.callCount++
+ n := m.callCount
+ m.mu.Unlock()
+ if n == 1 {
+ return &providers.LLMResponse{
+ ToolCalls: []providers.ToolCall{
+ {ID: "tc-1", Name: "echo_tool", Arguments: map[string]any{"msg": "hi"}},
+ },
+ }, nil
+ }
+ return &providers.LLMResponse{Content: "done"}, nil
+}
+func (m *sequenceMockProvider) GetDefaultModel() string { return "test" }
+func (m *sequenceMockProvider) SupportsTools() bool { return true }
+func (m *sequenceMockProvider) GetContextWindow() int { return 4096 }
+
+// echoTool is a minimal Tool stub registered as "echo_tool".
+type echoTool struct{}
+
+func (t *echoTool) Name() string { return "echo_tool" }
+func (t *echoTool) Description() string { return "echo" }
+func (t *echoTool) Parameters() map[string]any {
+ return map[string]any{"type": "object", "properties": map[string]any{}}
+}
+func (t *echoTool) Execute(_ context.Context, _ map[string]any) *ToolResult {
+ return &ToolResult{ForLLM: "echoed"}
+}
+
+// TestToolLoop_NilReporter_FallsBackToNoop ensures that passing nil as
+// Reporter does not panic â the loop must substitute orch.Noop internally.
+func TestToolLoop_NilReporter_FallsBackToNoop(t *testing.T) {
+ _, err := RunToolLoop(context.Background(), ToolLoopConfig{
+ Provider: &MockLLMProvider{},
+ Model: "test",
+ MaxIterations: 1,
+ Reporter: nil, // must not panic
+ }, []providers.Message{{Role: "user", Content: "hi"}}, "cli", "direct")
+ if err != nil {
+ t.Fatalf("unexpected error with nil reporter: %v", err)
+ }
+}
+
+// TestToolLoop_Reporter_WaitingBeforeLLM verifies that ReportStateChange is
+// called with state="waiting" before the first LLM call. The mock provider
+// returns a direct text answer (no tool calls), so exactly one waiting event
+// is expected.
+func TestToolLoop_Reporter_WaitingBeforeLLM(t *testing.T) {
+ rep := &reporterSpy{}
+ _, err := RunToolLoop(context.Background(), ToolLoopConfig{
+ Provider: &MockLLMProvider{},
+ Model: "test",
+ MaxIterations: 1,
+ Reporter: rep,
+ AgentID: "sess-1",
+ }, []providers.Message{{Role: "user", Content: "hi"}}, "cli", "direct")
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ calls := rep.snapshot()
+ if len(calls) == 0 {
+ t.Fatal("expected at least one ReportStateChange call")
+ }
+ if calls[0].state != "waiting" {
+ t.Fatalf("first call must be state=waiting, got %+v", calls[0])
+ }
+}
+
+// TestToolLoop_Reporter_ToolcallOrderedAfterWaiting verifies the canonical
+// two-iteration sequence:
+//
+// waiting (before 1st LLM call)
+// toolcall(echo_tool) (before tool execution)
+// waiting (before 2nd LLM call)
+//
+// The sequenceMockProvider returns a tool call on iteration 1 and a text
+// response on iteration 2, driving exactly this path.
+func TestToolLoop_Reporter_ToolcallOrderedAfterWaiting(t *testing.T) {
+ rep := &reporterSpy{}
+ reg := NewToolRegistry()
+ reg.Register(&echoTool{})
+
+ _, err := RunToolLoop(context.Background(), ToolLoopConfig{
+ Provider: &sequenceMockProvider{},
+ Model: "test",
+ Tools: reg,
+ MaxIterations: 5,
+ Reporter: rep,
+ AgentID: "sess-1",
+ }, []providers.Message{{Role: "user", Content: "do it"}}, "cli", "direct")
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+
+ calls := rep.snapshot()
+ if len(calls) < 3 {
+ t.Fatalf("expected at least 3 calls, got %d: %+v", len(calls), calls)
+ }
+ if calls[0].state != "waiting" {
+ t.Fatalf("calls[0] must be waiting, got %+v", calls[0])
+ }
+ if calls[1].state != "toolcall" || calls[1].tool != "echo_tool" {
+ t.Fatalf("calls[1] must be toolcall(echo_tool), got %+v", calls[1])
+ }
+ if calls[2].state != "waiting" {
+ t.Fatalf("calls[2] must be waiting (2nd LLM iteration), got %+v", calls[2])
+ }
+}
+
+// TestToolLoop_Reporter_NoopImplementsInterface is a compile-time check that
+// orch.Noop satisfies the orch.AgentReporter interface accepted by
+// ToolLoopConfig.Reporter. If Noop ever stops implementing the interface the
+// build will fail here before any test runs.
+func TestToolLoop_Reporter_NoopImplementsInterface(t *testing.T) {
+ var _ orch.AgentReporter = orch.Noop
+}