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
' +
+ '
' +
+ '' +
+ '' +
+ '
' +
+ '
' +
+ '
Status
' +
+ '
' + escapeHtml(data.status) + '
' +
+ '
Phase ' + data.current_phase + ' / ' + data.total_phases + '
' +
+ '
No active plan.
' +
- '' +
- '
Start a Plan
' +
- '
' +
- '' +
- '' +
- '
' +
- '
' +
- '
Status
' +
- '
' + escapeHtml(data.status) + '
' +
- '
Phase ' + data.current_phase + ' / ' + data.total_phases + '
' +
- '
No skills installed.
';
+ return;
+ }
+
+ el.innerHTML = data.map(function(s) {
+ return '' +
+ '
' +
+ '
' + escapeHtml(s.name) + '
' +
+ '
' + escapeHtml(s.description || 'No description') + '
' +
+ '
' + escapeHtml(s.source) + '' +
+ '
' +
+ '
\u203A' +
+ '
No skills installed.
';
- return;
- }
-
- el.innerHTML = data.map(s =>
- '' +
- '
' +
- '
' + escapeHtml(s.name) + '
' +
- '
' + escapeHtml(s.description || 'No description') + '
' +
- '
' + escapeHtml(s.source) + '' +
- '
' +
- '
\u203A' +
- '
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) + '
' +
+ '
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) + '
' +
- '