From 8ebfd853c6be5dae770e6576124bf23ce9d61471 Mon Sep 17 00:00:00 2001 From: dj-oyu <68707227+dj-oyu@users.noreply.github.com> Date: Sun, 22 Feb 2026 08:22:38 +0900 Subject: [PATCH] feat: add SessionTracker for plan coordination and session observability Independent SessionTracker tracks tool-call activity per session, enabling heartbeat to skip plan preamble when a chat session is actively working in the same directory. Adds /miniapp/api/sessions endpoint and Active Sessions card to the Mini App Session tab. Co-Authored-By: Claude Opus 4.6 --- README.md | 4 +- cmd/picoclaw/cmd_gateway.go | 16 ++++ pkg/agent/context.go | 5 ++ pkg/agent/loop.go | 33 +++++++- pkg/agent/memory.go | 16 +++- pkg/agent/memory_test.go | 2 +- pkg/agent/session_tracker.go | 98 ++++++++++++++++++++++++ pkg/agent/session_tracker_test.go | 123 ++++++++++++++++++++++++++++++ pkg/miniapp/miniapp.go | 20 +++++ pkg/miniapp/static/index.html | 52 ++++++++++++- 10 files changed, 359 insertions(+), 10 deletions(-) create mode 100644 pkg/agent/session_tracker.go create mode 100644 pkg/agent/session_tracker_test.go diff --git a/README.md b/README.md index aff790c58..c662015b5 100644 --- a/README.md +++ b/README.md @@ -308,7 +308,7 @@ picoclaw gateway #### Mini App (Dashboard) -PicoClaw includes a Telegram Mini App that provides a GUI dashboard directly inside Telegram. It shows plan progress, available skills, session stats, and lets you send commands without typing. +PicoClaw includes a Telegram Mini App that provides a GUI dashboard directly inside Telegram. It shows plan progress, available skills, active sessions, session stats, and lets you send commands without typing. **How it works:** @@ -318,7 +318,7 @@ When Telegram is enabled, PicoClaw automatically registers a "Dashboard" menu bu |-----|-------------| | **Plan** | View plan phases/steps as a checklist, tap to mark done, start new plans | | **Skills** | Browse and invoke skills with a message input | -| **Session** | View token usage stats (requires `--stats` flag) | +| **Session** | View active sessions and token usage stats (requires `--stats` flag for stats) | | **Config** | Quick command buttons and custom command input | **Setup — Tailscale (recommended for self-hosting):** diff --git a/cmd/picoclaw/cmd_gateway.go b/cmd/picoclaw/cmd_gateway.go index f943b35ef..0d8e8ce6f 100644 --- a/cmd/picoclaw/cmd_gateway.go +++ b/cmd/picoclaw/cmd_gateway.go @@ -333,6 +333,22 @@ func (p *agentLoopDataProvider) GetSessionStats() *stats.Stats { return p.loop.GetSessionStats() } +func (p *agentLoopDataProvider) GetActiveSessions() []miniapp.SessionInfo { + entries := p.loop.GetActiveSessions() + result := make([]miniapp.SessionInfo, len(entries)) + for i, e := range entries { + result[i] = miniapp.SessionInfo{ + SessionKey: e.SessionKey, + Channel: e.Channel, + ChatID: e.ChatID, + TouchDir: e.TouchDir, + LastSeenAt: e.LastSeenAt.Format(time.RFC3339), + AgeSec: int(time.Since(e.LastSeenAt).Seconds()), + } + } + return result +} + // telegramCommandSender injects Mini App commands into the message bus. type telegramCommandSender struct { bus *bus.MessageBus diff --git a/pkg/agent/context.go b/pkg/agent/context.go index 258134766..4e4835ada 100644 --- a/pkg/agent/context.go +++ b/pkg/agent/context.go @@ -424,6 +424,11 @@ func (cb *ContextBuilder) SetPlanStatus(status string) error { return cb.memory.SetStatus(status) } +// GetPlanWorkDir returns the WorkDir from the plan metadata, or "". +func (cb *ContextBuilder) GetPlanWorkDir() string { + return cb.memory.GetPlanWorkDir() +} + // GetSkillsInfo returns information about loaded skills. func (cb *ContextBuilder) GetSkillsInfo() map[string]any { allSkills := cb.skillsLoader.ListSkills() diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 11c0ba506..7d82f12d9 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -86,6 +86,7 @@ type AgentLoop struct { planStartPending bool // set by /plan start to trigger LLM execution sessionLocks sync.Map // sessionKey → *sessionSemaphore activeTasks sync.Map // sessionKey → *activeTask + sessions *SessionTracker } // processOptions configures how a message is processed @@ -137,6 +138,7 @@ func NewAgentLoop(cfg *config.Config, msgBus *bus.MessageBus, provider providers summarizing: sync.Map{}, fallback: fallbackChain, providerCache: providerCache, + sessions: NewSessionTracker(), } } @@ -753,8 +755,11 @@ func (al *AgentLoop) runAgentLoop(ctx context.Context, agent *AgentInstance, opt // 2c. Background plan preamble: append to system prompt (high attention) // so the LLM knows from the start that it must mark steps [x]. + // Skip if a chat session is actively working on the plan directory. if opts.Background && agent.ContextBuilder.HasActivePlan() && agent.ContextBuilder.GetPlanStatus() == "executing" { - if len(messages) > 0 && messages[0].Role == "system" { + planDir := agent.ContextBuilder.GetPlanWorkDir() + skipPreamble := planDir != "" && al.sessions.IsActiveInDir(planDir, "heartbeat") + if !skipPreamble && len(messages) > 0 && messages[0].Role == "system" { var sb strings.Builder sb.WriteString(messages[0].Content) sb.WriteString("\n\n## Background Execution\n") @@ -1616,6 +1621,25 @@ func (al *AgentLoop) runLLMIteration( } } + // Record session activity for heartbeat/plan coordination + for _, tc := range normalizedToolCalls { + var detectedDir string + if tc.Name == "exec" { + detectedDir = extractExecProjectDir(tc.Arguments) + } + if detectedDir == "" { + switch tc.Name { + case "read_file", "write_file", "edit_file", "append_file", "list_dir": + if p, _ := tc.Arguments["path"].(string); p != "" { + detectedDir = fileParentRelDir(p, agent.Workspace) + } + } + } + if detectedDir != "" { + al.sessions.Touch(opts.SessionKey, opts.Channel, opts.ChatID, detectedDir) + } + } + // Build assistant message with tool calls assistantMsg := providers.Message{ Role: "assistant", @@ -1988,6 +2012,11 @@ func (al *AgentLoop) GetPlanPhases() []PlanPhase { return mem.GetPlanPhases() } +// GetActiveSessions returns currently active sessions for the mini app API. +func (al *AgentLoop) GetActiveSessions() []SessionEntry { + return al.sessions.ListActive() +} + // GetSessionStats returns the current session statistics snapshot, or nil if stats tracking is disabled. func (al *AgentLoop) GetSessionStats() *stats.Stats { if al.stats == nil { @@ -2552,7 +2581,7 @@ func (al *AgentLoop) expandPlanCommand(msg bus.InboundMessage) (expanded string, } // Write the interview seed - seed := BuildInterviewSeed(task) + seed := BuildInterviewSeed(task, agent.Workspace) if err := agent.ContextBuilder.WriteMemory(seed); err != nil { return "", "", false } diff --git a/pkg/agent/memory.go b/pkg/agent/memory.go index 92c097354..2aa51075a 100644 --- a/pkg/agent/memory.go +++ b/pkg/agent/memory.go @@ -141,6 +141,7 @@ var ( rePhaseHeader = regexp.MustCompile(`(?m)^## Phase (\d+):\s*(.*)`) reStepDone = regexp.MustCompile(`(?m)^- \[x\] `) reStepTodo = regexp.MustCompile(`(?m)^- \[ \] `) + reWorkDir = regexp.MustCompile(`(?m)^> WorkDir:\s*(.+)`) ) // HasActivePlan returns true if MEMORY.md contains an active plan. @@ -408,17 +409,28 @@ func (ms *MemoryStore) AddStep(phase int, desc string) error { // ---------- Selective injection methods ---------- +// GetPlanWorkDir returns the WorkDir from the plan metadata, or "". +func (ms *MemoryStore) GetPlanWorkDir() string { + content := ms.ReadLongTerm() + m := reWorkDir.FindStringSubmatch(content) + if len(m) < 2 { + return "" + } + return strings.TrimSpace(m[1]) +} + // interviewSeed is the initial content written to MEMORY.md when /plan starts. const interviewSeedTemplate = `# Active Plan > Task: %s +> WorkDir: %s > Status: interviewing > Phase: 1 ` // BuildInterviewSeed creates the initial plan seed for a given task description. -func BuildInterviewSeed(task string) string { - return fmt.Sprintf(interviewSeedTemplate, task) +func BuildInterviewSeed(task, workDir string) string { + return fmt.Sprintf(interviewSeedTemplate, task, workDir) } // GetInterviewContext returns context for injection during the interviewing phase. diff --git a/pkg/agent/memory_test.go b/pkg/agent/memory_test.go index 5d21994cb..12b77c418 100644 --- a/pkg/agent/memory_test.go +++ b/pkg/agent/memory_test.go @@ -458,7 +458,7 @@ func TestGetMemoryContext_RegularMemory(t *testing.T) { } func TestBuildInterviewSeed(t *testing.T) { - seed := BuildInterviewSeed("Deploy monitoring stack") + seed := BuildInterviewSeed("Deploy monitoring stack", "/home/user/project") if !strings.Contains(seed, "# Active Plan") { t.Error("expected '# Active Plan' header") diff --git a/pkg/agent/session_tracker.go b/pkg/agent/session_tracker.go new file mode 100644 index 000000000..44adf9771 --- /dev/null +++ b/pkg/agent/session_tracker.go @@ -0,0 +1,98 @@ +package agent + +import ( + "sort" + "strings" + "sync" + "time" +) + +// SessionEntry represents an active or recently-active session. +type SessionEntry struct { + SessionKey string `json:"session_key"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + TouchDir string `json:"touch_dir"` + LastSeenAt time.Time `json:"last_seen_at"` +} + +// SessionTracker tracks per-session tool-call activity. +// Thread-safe; used by AgentLoop for plan coordination and by the mini app API for observability. +type SessionTracker struct { + entries sync.Map // sessionKey → *SessionEntry +} + +// NewSessionTracker creates a new tracker. +func NewSessionTracker() *SessionTracker { + return &SessionTracker{} +} + +const sessionActivityTimeout = 15 * time.Minute + +// Touch records a tool-call activity for a session. +// dir is the workspace-relative directory the tool call targeted. +// If dir is empty, only LastSeenAt is updated. +func (st *SessionTracker) Touch(sessionKey, channel, chatID, dir string) { + now := time.Now() + val, loaded := st.entries.Load(sessionKey) + if loaded { + entry := val.(*SessionEntry) + entry.LastSeenAt = now + if dir != "" { + entry.TouchDir = dir + } + if channel != "" { + entry.Channel = channel + } + if chatID != "" { + entry.ChatID = chatID + } + return + } + st.entries.Store(sessionKey, &SessionEntry{ + SessionKey: sessionKey, + Channel: channel, + ChatID: chatID, + TouchDir: dir, + LastSeenAt: now, + }) +} + +// IsActiveInDir returns true if any session (excluding those matching excludeKey) +// has touched a directory overlapping with dir within sessionActivityTimeout. +// Overlap = either is a prefix of the other (parent/child relationship). +func (st *SessionTracker) IsActiveInDir(dir, excludeKey string) bool { + cutoff := time.Now().Add(-sessionActivityTimeout) + active := false + st.entries.Range(func(key, val any) bool { + if key.(string) == excludeKey { + return true + } + entry := val.(*SessionEntry) + if entry.LastSeenAt.After(cutoff) && entry.TouchDir != "" && + (strings.HasPrefix(entry.TouchDir, dir) || strings.HasPrefix(dir, entry.TouchDir)) { + active = true + return false + } + return true + }) + return active +} + +// ListActive returns all sessions seen within sessionActivityTimeout, +// sorted by LastSeenAt descending (most recent first). +func (st *SessionTracker) ListActive() []SessionEntry { + cutoff := time.Now().Add(-sessionActivityTimeout) + var result []SessionEntry + st.entries.Range(func(key, val any) bool { + entry := val.(*SessionEntry) + if entry.LastSeenAt.After(cutoff) { + result = append(result, *entry) // copy + } + return true + }) + sort.Slice(result, func(i, j int) bool { + return result[i].LastSeenAt.After(result[j].LastSeenAt) + }) + return result +} diff --git a/pkg/agent/session_tracker_test.go b/pkg/agent/session_tracker_test.go new file mode 100644 index 000000000..0882d0513 --- /dev/null +++ b/pkg/agent/session_tracker_test.go @@ -0,0 +1,123 @@ +package agent + +import ( + "testing" + "time" +) + +func TestTouch(t *testing.T) { + st := NewSessionTracker() + + // Basic touch creates entry + st.Touch("sess1", "telegram", "123", "projects/myapp") + entries := st.ListActive() + if len(entries) != 1 { + t.Fatalf("expected 1 entry, got %d", len(entries)) + } + if entries[0].SessionKey != "sess1" { + t.Errorf("expected session_key=sess1, got %s", entries[0].SessionKey) + } + if entries[0].Channel != "telegram" { + t.Errorf("expected channel=telegram, got %s", entries[0].Channel) + } + if entries[0].TouchDir != "projects/myapp" { + t.Errorf("expected touch_dir=projects/myapp, got %s", entries[0].TouchDir) + } + + // Touch again with new dir overwrites TouchDir + st.Touch("sess1", "", "", "projects/other") + entries = st.ListActive() + if len(entries) != 1 { + t.Fatalf("expected 1 entry, got %d", len(entries)) + } + if entries[0].TouchDir != "projects/other" { + t.Errorf("expected touch_dir=projects/other, got %s", entries[0].TouchDir) + } + // Channel should remain from first touch + if entries[0].Channel != "telegram" { + t.Errorf("expected channel=telegram (unchanged), got %s", entries[0].Channel) + } + + // Touch with empty dir does not overwrite TouchDir + st.Touch("sess1", "", "", "") + entries = st.ListActive() + if entries[0].TouchDir != "projects/other" { + t.Errorf("expected touch_dir unchanged, got %s", entries[0].TouchDir) + } +} + +func TestIsActiveInDir(t *testing.T) { + st := NewSessionTracker() + + // Setup: sess1 touches "projects/myapp" + st.Touch("sess1", "telegram", "123", "projects/myapp") + + // Same dir, excluding sess1 → false + if st.IsActiveInDir("projects/myapp", "sess1") { + t.Error("expected false when excluding the only active session") + } + + // Same dir, excluding different key → true + if !st.IsActiveInDir("projects/myapp", "heartbeat") { + t.Error("expected true for exact dir match") + } + + // Parent dir match: "projects" is prefix of "projects/myapp" + if !st.IsActiveInDir("projects", "heartbeat") { + t.Error("expected true for parent dir match") + } + + // Child dir match: "projects/myapp/src" has prefix "projects/myapp" + if !st.IsActiveInDir("projects/myapp/src", "heartbeat") { + t.Error("expected true for child dir match") + } + + // Unrelated dir → false + if st.IsActiveInDir("other/stuff", "heartbeat") { + t.Error("expected false for unrelated dir") + } + + // Stale entry (manually set LastSeenAt to past) + val, _ := st.entries.Load("sess1") + entry := val.(*SessionEntry) + entry.LastSeenAt = time.Now().Add(-sessionActivityTimeout - time.Minute) + + if st.IsActiveInDir("projects/myapp", "heartbeat") { + t.Error("expected false for stale session") + } +} + +func TestListActive(t *testing.T) { + st := NewSessionTracker() + + // Add two sessions + st.Touch("sess1", "telegram", "123", "projects/a") + time.Sleep(5 * time.Millisecond) // ensure different timestamps + st.Touch("sess2", "discord", "456", "projects/b") + + entries := st.ListActive() + if len(entries) != 2 { + t.Fatalf("expected 2 entries, got %d", len(entries)) + } + + // Most recent first + if entries[0].SessionKey != "sess2" { + t.Errorf("expected sess2 first (most recent), got %s", entries[0].SessionKey) + } + if entries[1].SessionKey != "sess1" { + t.Errorf("expected sess1 second, got %s", entries[1].SessionKey) + } + + // Make sess1 stale + val, _ := st.entries.Load("sess1") + entry := val.(*SessionEntry) + entry.LastSeenAt = time.Now().Add(-sessionActivityTimeout - time.Minute) + + entries = st.ListActive() + if len(entries) != 1 { + t.Fatalf("expected 1 active entry after stale, got %d", len(entries)) + } + if entries[0].SessionKey != "sess2" { + t.Errorf("expected only sess2, got %s", entries[0].SessionKey) + } +} diff --git a/pkg/miniapp/miniapp.go b/pkg/miniapp/miniapp.go index 924a5750d..14d804a18 100644 --- a/pkg/miniapp/miniapp.go +++ b/pkg/miniapp/miniapp.go @@ -44,11 +44,22 @@ type PlanInfo struct { Phases []PlanPhase `json:"phases"` } +// 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"` + LastSeenAt string `json:"last_seen_at"` + AgeSec int `json:"age_sec"` +} + // 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 } // CommandSender injects a command into the message bus on behalf of a user. @@ -78,6 +89,7 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("/miniapp/api/skills", h.requireAuth(h.apiSkills)) mux.HandleFunc("/miniapp/api/plan", h.requireAuth(h.apiPlan)) 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)) } @@ -116,6 +128,14 @@ func (h *Handler) apiPlan(w http.ResponseWriter, r *http.Request) { 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 { diff --git a/pkg/miniapp/static/index.html b/pkg/miniapp/static/index.html index e8e287534..f6cfdfe1b 100644 --- a/pkg/miniapp/static/index.html +++ b/pkg/miniapp/static/index.html @@ -657,6 +657,44 @@ async function loadSkills() { } // ── Session tab ── +function formatAge(sec) { + if (sec < 60) return sec + 's ago'; + if (sec < 3600) return Math.floor(sec / 60) + 'm ago'; + return Math.floor(sec / 3600) + 'h ago'; +} + +function shortSessionKey(key) { + // "agent:default:telegram:123" → "telegram:123" + var parts = key.split(':'); + if (parts.length > 2) return parts.slice(2).join(':'); + return key; +} + +function renderActiveSessions(sessions) { + if (!sessions || sessions.length === 0) { + return '
' + + '
Active Sessions
' + + '
No active sessions
' + + '
'; + } + var html = '
Active Sessions
'; + for (var i = 0; i < sessions.length; i++) { + var s = sessions[i]; + var touchDir = s.touch_dir || '\u2014'; + html += '
' + + '
' + + '\u25CF' + + '' + escapeHtml(shortSessionKey(s.session_key)) + '' + + '' + formatAge(s.age_sec) + '' + + '
' + + '
touch: ' + escapeHtml(touchDir) + '
' + + '
'; + } + html += ''; + html += '
'; + return html; +} + async function loadSession() { const loading = document.getElementById('session-loading'); const el = document.getElementById('session-content'); @@ -665,17 +703,23 @@ async function loadSession() { el.style.display = 'none'; try { - const data = await apiFetch('/miniapp/api/session'); + const [data, sessions] = 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') { - el.innerHTML = '
Stats tracking not enabled.
Start gateway with --stats flag.
'; + 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'; - el.innerHTML = + html += '
' + '
Today
' + '
Prompts' + (data.today?.prompts || 0) + '
' + @@ -690,6 +734,8 @@ async function loadSession() { '
Prompt Tokens' + formatTokens(data.total_prompt_tokens || 0) + '
' + '
Completion Tokens' + formatTokens(data.total_completion_tokens || 0) + '
' + '
'; + + el.innerHTML = html; } catch (e) { loading.textContent = 'Failed to load session.'; loading.style.display = 'block';