From 2eaf6984a6489951b2b07ffa58d7d009c46fbbfe Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Sun, 22 Feb 2026 10:09:45 +0900 Subject: [PATCH] feat: add SSE real-time updates for MiniApp dashboard Push plan, session, and skills state changes to the MiniApp via Server-Sent Events instead of requiring manual refresh. A lightweight StateNotifier fans out notifications from AgentLoop's three mutation points (post-LLM state machine, /plan commands, plan creation) to all connected SSE clients, with JSON diff dedup to avoid redundant writes. Co-Authored-By: Claude Opus 4.6 --- cmd/picoclaw/cmd_gateway.go | 4 +- pkg/agent/loop.go | 17 ++- pkg/agent/loop_test.go | 3 + pkg/miniapp/miniapp.go | 96 ++++++++++++- pkg/miniapp/miniapp_test.go | 167 ++++++++++++++++++++++ pkg/miniapp/static/index.html | 257 ++++++++++++++++++++-------------- 6 files changed, 431 insertions(+), 113 deletions(-) diff --git a/cmd/picoclaw/cmd_gateway.go b/cmd/picoclaw/cmd_gateway.go index 0d8e8ce6f..47c889718 100644 --- a/cmd/picoclaw/cmd_gateway.go +++ b/cmd/picoclaw/cmd_gateway.go @@ -222,7 +222,9 @@ func gatewayCmd() { if webAppURL != "" { provider := &agentLoopDataProvider{loop: agentLoop} sender := &telegramCommandSender{bus: msgBus} - handler := miniapp.NewHandler(provider, sender, cfg.Channels.Telegram.Token) + notifier := miniapp.NewStateNotifier() + handler := miniapp.NewHandler(provider, sender, cfg.Channels.Telegram.Token, notifier) + agentLoop.OnStateChange = notifier.Notify handler.RegisterRoutes(healthServer.Mux()) fmt.Printf("✓ Mini App registered at %s\n", webAppURL) } diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 380e5e4ca..8989ee020 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -87,6 +87,7 @@ type AgentLoop struct { sessionLocks sync.Map // sessionKey → *sessionSemaphore activeTasks sync.Map // sessionKey → *activeTask sessions *SessionTracker + OnStateChange func() // called on plan/session/skills mutations } // processOptions configures how a message is processed @@ -142,6 +143,12 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers } } +func (al *AgentLoop) notifyStateChange() { + if al.OnStateChange != nil { + al.OnStateChange() + } +} + // registerSharedTools registers tools that are shared across all agents (web, message, spawn). func registerSharedTools( cfg *config.Config, @@ -866,6 +873,8 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt } } + al.notifyStateChange() + // 5b. Interview staleness detection: compare MEMORY.md size after iteration. if agent.ContextBuilder.GetPlanStatus() == "interviewing" { postMemoryLen := len(agent.ContextBuilder.ReadMemory()) @@ -2327,6 +2336,9 @@ func (al *AgentLoop) handleCommand(ctx context.Context, msg bus.InboundMessage) case "/plan": resp, handled := al.handlePlanCommand(args) + if handled { + al.notifyStateChange() + } return resp, handled } @@ -2551,9 +2563,9 @@ func isPlanPreExecution(status string) bool { func isToolAllowedDuringInterview(toolName string, args map[string]interface{}) bool { norm := tools.NormalizeToolName(toolName) - // Read-type tools: always allowed + // Read-type tools and communication: always allowed switch norm { - case "readfile", "listdir", "websearch", "webfetch": + case "readfile", "listdir", "websearch", "webfetch", "message": return true } @@ -2609,6 +2621,7 @@ func (al *AgentLoop) expandPlanCommand(msg bus.InboundMessage) (expanded string, if err := agent.ContextBuilder.WriteMemory(seed); err != nil { return "", "", false } + al.notifyStateChange() // Expanded: the task description goes to LLM. // The system prompt already contains the interview guide. diff --git a/pkg/agent/loop_test.go b/pkg/agent/loop_test.go index 72e7d7ec4..df588df72 100644 --- a/pkg/agent/loop_test.go +++ b/pkg/agent/loop_test.go @@ -1391,6 +1391,9 @@ func TestIsToolAllowedDuringInterview_FuzzyNames(t *testing.T) { {"listdir", nil, true}, {"websearch", nil, true}, {"webfetch", nil, true}, + // Message tool — allowed (needed for interview questions) + {"message", nil, true}, + {"Message", nil, true}, // Write to MEMORY.md — allowed {"edit_file", map[string]interface{}{"path": "/ws/memory/MEMORY.md"}, true}, {"editfile", map[string]interface{}{"path": "/ws/memory/MEMORY.md"}, true}, diff --git a/pkg/miniapp/miniapp.go b/pkg/miniapp/miniapp.go index 14d804a18..dda3c2eb2 100644 --- a/pkg/miniapp/miniapp.go +++ b/pkg/miniapp/miniapp.go @@ -1,6 +1,7 @@ package miniapp import ( + "bytes" "crypto/hmac" "crypto/sha256" "embed" @@ -12,6 +13,8 @@ import ( "net/url" "sort" "strings" + "sync" + "time" "github.com/sipeed/picoclaw/pkg/skills" "github.com/sipeed/picoclaw/pkg/stats" @@ -67,19 +70,60 @@ type CommandSender interface { SendCommand(senderID, chatID, command string) } +// StateNotifier broadcasts state-change signals to SSE subscribers. +type StateNotifier struct { + mu sync.Mutex + subs map[chan struct{}]struct{} +} + +// NewStateNotifier creates a new StateNotifier. +func NewStateNotifier() *StateNotifier { + return &StateNotifier{subs: make(map[chan struct{}]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() +} + +// 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: + } + } +} + // Handler serves the Mini App HTML and API endpoints. type Handler struct { provider DataProvider sender CommandSender botToken string + notifier *StateNotifier } // NewHandler creates a new Mini App handler. -func NewHandler(provider DataProvider, sender CommandSender, botToken string) *Handler { +func NewHandler(provider DataProvider, sender CommandSender, botToken string, notifier *StateNotifier) *Handler { return &Handler{ provider: provider, sender: sender, botToken: botToken, + notifier: notifier, } } @@ -91,6 +135,7 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("/miniapp/api/session", h.requireAuth(h.apiSession)) mux.HandleFunc("/miniapp/api/sessions", h.requireAuth(h.apiSessions)) mux.HandleFunc("/miniapp/api/command", h.requireAuth(h.apiCommand)) + mux.HandleFunc("/miniapp/api/events", h.requireAuth(h.apiEvents)) } func (h *Handler) serveIndex(w http.ResponseWriter, r *http.Request) { @@ -204,6 +249,55 @@ func extractUserFromInitData(initData string) (userID, chatID string) { 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 []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) + + for { + select { + case <-r.Context().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) + } + } +} + +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) diff --git a/pkg/miniapp/miniapp_test.go b/pkg/miniapp/miniapp_test.go index 46f5904b3..95f5f88bd 100644 --- a/pkg/miniapp/miniapp_test.go +++ b/pkg/miniapp/miniapp_test.go @@ -1,14 +1,21 @@ package miniapp import ( + "bufio" "crypto/hmac" "crypto/sha256" "encoding/hex" "fmt" + "net/http" + "net/http/httptest" "net/url" "sort" "strings" "testing" + "time" + + "github.com/sipeed/picoclaw/pkg/skills" + "github.com/sipeed/picoclaw/pkg/stats" ) // buildInitData constructs a valid initData string from params and a bot token. @@ -97,3 +104,163 @@ func TestValidateInitData(t *testing.T) { } }) } + +// ── StateNotifier tests ── + +func TestStateNotifier_FanOut(t *testing.T) { + n := NewStateNotifier() + ch1 := n.Subscribe() + ch2 := n.Subscribe() + defer n.Unsubscribe(ch1) + defer n.Unsubscribe(ch2) + + n.Notify() + + select { + case <-ch1: + case <-time.After(100 * time.Millisecond): + t.Error("ch1 did not receive notification") + } + select { + case <-ch2: + case <-time.After(100 * time.Millisecond): + t.Error("ch2 did not receive notification") + } +} + +func TestStateNotifier_Coalesce(t *testing.T) { + n := NewStateNotifier() + ch := n.Subscribe() + defer n.Unsubscribe(ch) + + // Multiple rapid notifications should coalesce into one + n.Notify() + n.Notify() + n.Notify() + + select { + case <-ch: + case <-time.After(100 * time.Millisecond): + t.Error("ch did not receive notification") + } + + // Channel should be empty now (coalesced) + select { + case <-ch: + t.Error("expected no second notification (should coalesce)") + case <-time.After(50 * time.Millisecond): + } +} + +// ── SSE endpoint tests ── + +type mockDataProvider struct{} + +func (m *mockDataProvider) ListSkills() []skills.SkillInfo { + return []skills.SkillInfo{{Name: "test-skill", Description: "A test", Source: "local"}} +} +func (m *mockDataProvider) GetPlanInfo() PlanInfo { + return PlanInfo{HasPlan: false, Status: "none"} +} +func (m *mockDataProvider) GetSessionStats() *stats.Stats { + return nil +} +func (m *mockDataProvider) GetActiveSessions() []SessionInfo { + return []SessionInfo{} +} + +type mockSender struct{} + +func (m *mockSender) SendCommand(senderID, chatID, command string) {} + +const testBotToken = "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11" + +func testInitData() string { + return buildInitData(map[string]string{ + "user": `{"id":279058397,"first_name":"Test"}`, + "auth_date": "1234567890", + }, testBotToken) +} + +func TestSSE_AuthRequired(t *testing.T) { + notifier := NewStateNotifier() + h := NewHandler(&mockDataProvider{}, &mockSender{}, testBotToken, notifier) + mux := http.NewServeMux() + h.RegisterRoutes(mux) + + req := httptest.NewRequest("GET", "/miniapp/api/events", nil) + w := httptest.NewRecorder() + mux.ServeHTTP(w, req) + + if w.Code != http.StatusUnauthorized { + t.Errorf("expected 401, got %d", w.Code) + } +} + +func TestSSE_Headers(t *testing.T) { + notifier := NewStateNotifier() + h := NewHandler(&mockDataProvider{}, &mockSender{}, testBotToken, notifier) + mux := http.NewServeMux() + h.RegisterRoutes(mux) + + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/miniapp/api/events?initData=" + url.QueryEscape(testInitData())) + if err != nil { + t.Fatalf("GET failed: %v", err) + } + defer resp.Body.Close() + + if ct := resp.Header.Get("Content-Type"); ct != "text/event-stream" { + t.Errorf("expected Content-Type text/event-stream, got %q", ct) + } + if cc := resp.Header.Get("Cache-Control"); cc != "no-cache" { + t.Errorf("expected Cache-Control no-cache, got %q", cc) + } +} + +func TestSSE_InitialEvents(t *testing.T) { + notifier := NewStateNotifier() + h := NewHandler(&mockDataProvider{}, &mockSender{}, testBotToken, notifier) + mux := http.NewServeMux() + h.RegisterRoutes(mux) + + ts := httptest.NewServer(mux) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/miniapp/api/events?initData=" + url.QueryEscape(testInitData())) + if err != nil { + t.Fatalf("GET failed: %v", err) + } + defer resp.Body.Close() + + scanner := bufio.NewScanner(resp.Body) + events := make(map[string]bool) + deadline := time.After(2 * time.Second) + + for len(events) < 3 { + done := make(chan bool, 1) + go func() { + done <- scanner.Scan() + }() + select { + case ok := <-done: + if !ok { + t.Fatalf("scanner ended early: %v", scanner.Err()) + } + case <-deadline: + t.Fatalf("timed out waiting for events, got: %v", events) + } + line := scanner.Text() + if strings.HasPrefix(line, "event: ") { + events[strings.TrimPrefix(line, "event: ")] = true + } + } + + for _, name := range []string{"plan", "session", "skills"} { + if !events[name] { + t.Errorf("missing initial event %q", name) + } + } +} diff --git a/pkg/miniapp/static/index.html b/pkg/miniapp/static/index.html index 8a276429d..9321383b7 100644 --- a/pkg/miniapp/static/index.html +++ b/pkg/miniapp/static/index.html @@ -581,49 +581,50 @@ async function apiFetch(path) { } // ── Plan tab ── +function renderPlanFromData(data) { + var loading = document.getElementById('plan-loading'); + var el = document.getElementById('plan-content'); + loading.style.display = 'none'; + el.style.display = 'block'; + + if (!data.has_plan) { + el.innerHTML = + '
No active plan.
' + + '
' + + '
Start a Plan
' + + '
' + + '' + + '' + + '
' + + '
'; + return; + } + + var html = ''; + html += '
' + + '
Status
' + + '
' + escapeHtml(data.status) + '
' + + '
Phase ' + data.current_phase + ' / ' + data.total_phases + '
' + + '
'; + + if (data.phases && data.phases.length > 0) { + html += renderPhases(data.phases, data.current_phase); + } + + html += ''; + el.innerHTML = html; +} + async function loadPlan() { - const loading = document.getElementById('plan-loading'); - const el = document.getElementById('plan-content'); + var loading = document.getElementById('plan-loading'); + var el = document.getElementById('plan-content'); loading.style.display = 'block'; loading.textContent = 'Loading plan...'; el.style.display = 'none'; try { - const data = await apiFetch('/miniapp/api/plan'); - loading.style.display = 'none'; - el.style.display = 'block'; - - if (!data.has_plan) { - el.innerHTML = - '
No active plan.
' + - '
' + - '
Start a Plan
' + - '
' + - '' + - '' + - '
' + - '
'; - return; - } - - let html = ''; - - // Status header card - html += '
' + - '
Status
' + - '
' + escapeHtml(data.status) + '
' + - '
Phase ' + data.current_phase + ' / ' + data.total_phases + '
' + - '
'; - - // Phase/step list - if (data.phases && data.phases.length > 0) { - html += renderPhases(data.phases, data.current_phase); - } - - // Refresh button (in-app, not sendData) - html += ''; - - el.innerHTML = html; + var data = await apiFetch('/miniapp/api/plan'); + renderPlanFromData(data); } catch (e) { loading.textContent = 'Failed to load plan.'; loading.style.display = 'block'; @@ -687,51 +688,55 @@ document.getElementById('plan-content').addEventListener('click', function(e) { }); // ── Skills tab ── +function renderSkillsFromData(data) { + var loading = document.getElementById('skills-loading'); + var el = document.getElementById('skills-list'); + loading.style.display = 'none'; + el.style.display = 'block'; + + if (!data || data.length === 0) { + el.innerHTML = '
No skills installed.
'; + return; + } + + el.innerHTML = data.map(function(s) { + return '
' + + '
' + + '
' + escapeHtml(s.name) + '
' + + '
' + escapeHtml(s.description || 'No description') + '
' + + '' + escapeHtml(s.source) + '' + + '
' + + '\u203A' + + '
'; + }).join(''); + + if (selectedSkill) { + var prev = el.querySelector('[data-skill="' + CSS.escape(selectedSkill) + '"]'); + if (prev) prev.classList.add('selected'); + } + + el.querySelectorAll('.skill-item').forEach(function(item) { + item.addEventListener('click', function() { + el.querySelectorAll('.skill-item').forEach(function(i) { i.classList.remove('selected'); }); + item.classList.add('selected'); + selectedSkill = item.dataset.skill; + document.getElementById('send-bar').style.display = 'flex'; + document.getElementById('skill-msg').placeholder = 'Message for /' + selectedSkill + '...'; + document.getElementById('skill-msg').focus(); + }); + }); +} + async function loadSkills() { - const loading = document.getElementById('skills-loading'); - const el = document.getElementById('skills-list'); + var loading = document.getElementById('skills-loading'); + var el = document.getElementById('skills-list'); loading.style.display = 'block'; loading.textContent = 'Loading skills...'; el.style.display = 'none'; try { - const data = await apiFetch('/miniapp/api/skills'); - loading.style.display = 'none'; - el.style.display = 'block'; - - if (!data || data.length === 0) { - el.innerHTML = '
No skills installed.
'; - return; - } - - el.innerHTML = data.map(s => - '
' + - '
' + - '
' + escapeHtml(s.name) + '
' + - '
' + escapeHtml(s.description || 'No description') + '
' + - '' + escapeHtml(s.source) + '' + - '
' + - '\u203A' + - '
' - ).join(''); - - // Restore selection if still valid - if (selectedSkill) { - const prev = el.querySelector('[data-skill="' + CSS.escape(selectedSkill) + '"]'); - if (prev) prev.classList.add('selected'); - } - - el.querySelectorAll('.skill-item').forEach(item => { - item.addEventListener('click', () => { - el.querySelectorAll('.skill-item').forEach(i => i.classList.remove('selected')); - item.classList.add('selected'); - selectedSkill = item.dataset.skill; - document.getElementById('send-bar').style.display = 'flex'; - document.getElementById('skill-msg').placeholder = - 'Message for /' + selectedSkill + '...'; - document.getElementById('skill-msg').focus(); - }); - }); + var data = await apiFetch('/miniapp/api/skills'); + renderSkillsFromData(data); } catch (e) { loading.textContent = 'Failed to load skills.'; loading.style.display = 'block'; @@ -778,47 +783,54 @@ function renderActiveSessions(sessions) { return html; } +function renderSessionFromData(sessions, stats) { + var loading = document.getElementById('session-loading'); + var el = document.getElementById('session-content'); + loading.style.display = 'none'; + el.style.display = 'block'; + + var html = renderActiveSessions(sessions); + + if (!stats || stats.status === 'stats not enabled') { + html += '
Stats tracking not enabled.
Start gateway with --stats flag.
'; + el.innerHTML = html; + return; + } + + var since = stats.since ? new Date(stats.since).toLocaleDateString() : 'N/A'; + var today = stats.today || {}; + html += + '
' + + '
Today
' + + '
Prompts' + (today.prompts || 0) + '
' + + '
Requests' + (today.requests || 0) + '
' + + '
Tokens' + formatTokens(today.total_tokens || 0) + '
' + + '
' + + '
' + + '
All Time (since ' + escapeHtml(since) + ')
' + + '
Prompts' + (stats.total_prompts || 0) + '
' + + '
Requests' + (stats.total_requests || 0) + '
' + + '
Total Tokens' + formatTokens(stats.total_tokens || 0) + '
' + + '
Prompt Tokens' + formatTokens(stats.total_prompt_tokens || 0) + '
' + + '
Completion Tokens' + formatTokens(stats.total_completion_tokens || 0) + '
' + + '
'; + + el.innerHTML = html; +} + async function loadSession() { - const loading = document.getElementById('session-loading'); - const el = document.getElementById('session-content'); + var loading = document.getElementById('session-loading'); + var el = document.getElementById('session-content'); loading.style.display = 'block'; loading.textContent = 'Loading session...'; el.style.display = 'none'; try { - const [data, sessions] = await Promise.all([ + var results = await Promise.all([ apiFetch('/miniapp/api/session'), apiFetch('/miniapp/api/sessions').catch(function() { return []; }), ]); - loading.style.display = 'none'; - el.style.display = 'block'; - - var html = renderActiveSessions(sessions); - - if (data.status === 'stats not enabled') { - html += '
Stats tracking not enabled.
Start gateway with --stats flag.
'; - el.innerHTML = html; - return; - } - - const since = data.since ? new Date(data.since).toLocaleDateString() : 'N/A'; - html += - '
' + - '
Today
' + - '
Prompts' + (data.today?.prompts || 0) + '
' + - '
Requests' + (data.today?.requests || 0) + '
' + - '
Tokens' + formatTokens(data.today?.total_tokens || 0) + '
' + - '
' + - '
' + - '
All Time (since ' + escapeHtml(since) + ')
' + - '
Prompts' + (data.total_prompts || 0) + '
' + - '
Requests' + (data.total_requests || 0) + '
' + - '
Total Tokens' + formatTokens(data.total_tokens || 0) + '
' + - '
Prompt Tokens' + formatTokens(data.total_prompt_tokens || 0) + '
' + - '
Completion Tokens' + formatTokens(data.total_completion_tokens || 0) + '
' + - '
'; - - el.innerHTML = html; + renderSessionFromData(results[1], results[0]); } catch (e) { loading.textContent = 'Failed to load session.'; loading.style.display = 'block'; @@ -842,7 +854,34 @@ function escapeAttr(s) { return s.replace(/&/g, '&').replace(/"/g, '"').replace(/'/g, '''); } -// Initial load +// ── SSE real-time updates ── +var eventSource = null; + +function connectSSE() { + if (eventSource) eventSource.close(); + eventSource = new EventSource( + API_BASE + '/miniapp/api/events?initData=' + encodeURIComponent(initData) + ); + eventSource.addEventListener('plan', function(e) { + try { renderPlanFromData(JSON.parse(e.data)); } catch(err) {} + }); + eventSource.addEventListener('session', function(e) { + try { + var d = JSON.parse(e.data); + renderSessionFromData(d.sessions, d.stats); + } catch(err) {} + }); + eventSource.addEventListener('skills', function(e) { + try { renderSkillsFromData(JSON.parse(e.data)); } catch(err) {} + }); + eventSource.onerror = function() { + // Browser will auto-reconnect EventSource + }; +} + +connectSSE(); + +// Initial load (fallback for tabs not covered by initial SSE burst) loadPlan();