diff --git a/cmd/picoclaw/internal/gateway/helpers.go b/cmd/picoclaw/internal/gateway/helpers.go index 952b1c794..850b6b47f 100644 --- a/cmd/picoclaw/internal/gateway/helpers.go +++ b/cmd/picoclaw/internal/gateway/helpers.go @@ -63,6 +63,8 @@ const ( type gatewayServices struct { CronService *cron.CronService HeartbeatService *heartbeat.HeartbeatService + ResearchStore *research.ResearchStore + ResearchFocus *research.FocusTracker MediaStore media.MediaStore ChannelManager *channels.Manager DeviceService *devices.Service @@ -340,8 +342,13 @@ func setupAndStartServices( if rsErr != nil { logger.ErrorCF("research", "Failed to open research store", map[string]any{"error": rsErr.Error()}) } else { - agentLoop.RegisterTool(tools.NewResearchTool(researchStore, cfg.WorkspacePath())) + focusTracker := research.NewFocusTracker() + agentLoop.RegisterTool(tools.NewResearchTool(researchStore, cfg.WorkspacePath(), focusTracker)) handler.SetResearchStore(researchStore) + handler.SetResearchFocus(focusTracker) + services.ResearchStore = researchStore + services.ResearchFocus = focusTracker + services.HeartbeatService.SetResearchStore(researchStore) fmt.Println("✓ Research store initialized") } @@ -547,6 +554,9 @@ func restartServices( cfg.Heartbeat.Enabled, ) services.HeartbeatService.SetBus(msgBus) + if services.ResearchStore != nil { + services.HeartbeatService.SetResearchStore(services.ResearchStore) + } services.HeartbeatService.SetHandler(func(prompt, channel, chatID string) *tools.ToolResult { if channel == "" || chatID == "" { channel, chatID = "cli", "direct" diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index a16fc5cec..512c93f5c 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -25,6 +25,7 @@ import ( "github.com/sipeed/picoclaw/pkg/logger" "github.com/sipeed/picoclaw/pkg/media" "github.com/sipeed/picoclaw/pkg/providers" + "github.com/sipeed/picoclaw/pkg/research" "github.com/sipeed/picoclaw/pkg/routing" "github.com/sipeed/picoclaw/pkg/skills" "github.com/sipeed/picoclaw/pkg/state" @@ -911,6 +912,9 @@ func (al *AgentLoop) ProcessDirectWithChannel( // ProcessHeartbeat processes a heartbeat request without session history. // Each heartbeat is independent and doesn't accumulate context. func (al *AgentLoop) ProcessHeartbeat(ctx context.Context, content, channel, chatID string) (string, error) { + ctx = tools.WithHeartbeatContext(ctx) + ctx = tools.WithWebSearchQuota(ctx, research.DefaultHeartbeatSearchQuota) + agent := al.GetRegistry().GetDefaultAgent() if agent == nil { return "", fmt.Errorf("no default agent for heartbeat") diff --git a/pkg/heartbeat/service.go b/pkg/heartbeat/service.go index ef836d31e..2b4b34385 100644 --- a/pkg/heartbeat/service.go +++ b/pkg/heartbeat/service.go @@ -18,6 +18,7 @@ import ( "github.com/sipeed/picoclaw/pkg/constants" "github.com/sipeed/picoclaw/pkg/fileutil" "github.com/sipeed/picoclaw/pkg/logger" + "github.com/sipeed/picoclaw/pkg/research" "github.com/sipeed/picoclaw/pkg/state" "github.com/sipeed/picoclaw/pkg/tools" ) @@ -39,6 +40,7 @@ type HeartbeatService struct { bus *bus.MessageBus state *state.Manager handler HeartbeatHandler + researchStore *research.ResearchStore interval time.Duration enabled bool mu sync.RWMutex @@ -87,6 +89,13 @@ func (hs *HeartbeatService) SetHeartbeatThreadID(threadID int) { hs.heartbeatThreadID = threadID } +// SetResearchStore injects the research store for heartbeat-driven research. +func (hs *HeartbeatService) SetResearchStore(store *research.ResearchStore) { + hs.mu.Lock() + defer hs.mu.Unlock() + hs.researchStore = store +} + // ResetSuppression clears the notification suppression so the next // non-silent heartbeat result will be delivered to the user again. // Typically called when a user message arrives. @@ -321,6 +330,7 @@ func (hs *HeartbeatService) buildPrompt() string { } now := time.Now().Format("2006-01-02 15:04:05") + researchCtx := hs.buildResearchContext() return fmt.Sprintf(`# Heartbeat Check Current time: %s @@ -330,7 +340,95 @@ Review the following tasks and execute any necessary actions using available ski If there is nothing that requires attention, respond ONLY with: HEARTBEAT_OK %s -`, now, content) +%s`, now, content, researchCtx) +} + +const ( + maxResearchTasks = 3 + maxFindingsPerTask = 10 + maxSummaryLen = 200 +) + +// buildResearchContext generates a prompt section for incremental research progress. +// Only includes tasks that are "due" based on their interval setting. +func (hs *HeartbeatService) buildResearchContext() string { + hs.mu.RLock() + store := hs.researchStore + hs.mu.RUnlock() + + if store == nil { + return "" + } + + tasks, err := store.ListDueTasks(maxResearchTasks) + if err != nil { + hs.logErrorf("Failed to list due research tasks: %v", err) + return "" + } + + if len(tasks) == 0 { + return "" + } + + var b strings.Builder + b.WriteString("\n## Research Tasks (Incremental Progress)\n\n") + b.WriteString("You have research tasks due for progress. For each task below:\n") + b.WriteString("1. If pending, set status to 'active' first\n") + b.WriteString("2. Use web_search to find new information, then add_finding to record it\n") + b.WriteString( + fmt.Sprintf( + "3. **Web search budget: %d calls total this heartbeat** — use them wisely\n", + research.DefaultHeartbeatSearchQuota, + ), + ) + b.WriteString("4. Do NOT set status to 'completed' — research progresses incrementally across heartbeats\n\n") + + for _, task := range tasks { + b.WriteString(fmt.Sprintf("### %s [%s] (id: %s)\n", task.Title, task.Status, task.ID)) + b.WriteString(fmt.Sprintf("Interval: %s", task.Interval)) + if !task.LastResearchedAt.IsZero() { + b.WriteString(fmt.Sprintf(" | Last researched: %s", task.LastResearchedAt.Format("2006-01-02 15:04"))) + } + b.WriteString("\n") + if task.Description != "" { + b.WriteString(fmt.Sprintf("Description: %s\n", task.Description)) + } + + docs, err := store.ListDocuments(task.ID) + if err != nil { + b.WriteString("(error loading findings)\n\n") + continue + } + + if len(docs) == 0 { + b.WriteString("No findings yet — start researching this topic.\n\n") + continue + } + + shown := docs + if len(shown) > maxFindingsPerTask { + shown = shown[:maxFindingsPerTask] + } + + b.WriteString(fmt.Sprintf("Existing findings (%d total):\n", len(docs))) + for _, d := range shown { + summary := d.Summary + if len(summary) > maxSummaryLen { + summary = summary[:maxSummaryLen] + "..." + } + b.WriteString(fmt.Sprintf("- [%d] %s", d.Seq, d.Title)) + if summary != "" { + b.WriteString(fmt.Sprintf(" — %s", summary)) + } + b.WriteString("\n") + } + if len(docs) > maxFindingsPerTask { + b.WriteString(fmt.Sprintf(" ... and %d more findings\n", len(docs)-maxFindingsPerTask)) + } + b.WriteString("\nAdd NEW findings that build on and extend the above. Do not repeat existing findings.\n\n") + } + + return b.String() } // createDefaultHeartbeatTemplate creates the default HEARTBEAT.md file diff --git a/pkg/miniapp/api_research.go b/pkg/miniapp/api_research.go index 9560a521e..cebfbdaab 100644 --- a/pkg/miniapp/api_research.go +++ b/pkg/miniapp/api_research.go @@ -16,6 +16,11 @@ func (h *Handler) SetResearchStore(rs *research.ResearchStore) { h.researchStore = rs } +// SetResearchFocus injects the shared focus tracker for recall/forget control. +func (h *Handler) SetResearchFocus(ft *research.FocusTracker) { + h.researchFocus = ft +} + // apiResearch handles GET /miniapp/api/research (list) and POST (create). func (h *Handler) apiResearch(w http.ResponseWriter, r *http.Request) { if h.researchStore == nil { @@ -34,16 +39,19 @@ func (h *Handler) apiResearch(w http.ResponseWriter, r *http.Request) { } type researchTaskResponse struct { - ID string `json:"id"` - Title string `json:"title"` - Slug string `json:"slug"` - Description string `json:"description"` - Status string `json:"status"` - OutputDir string `json:"output_dir"` - CreatedAt string `json:"created_at"` - UpdatedAt string `json:"updated_at"` - CompletedAt string `json:"completed_at,omitempty"` - DocumentCount int `json:"document_count"` + ID string `json:"id"` + Title string `json:"title"` + Slug string `json:"slug"` + Description string `json:"description"` + Status string `json:"status"` + OutputDir string `json:"output_dir"` + Interval string `json:"interval"` + LastResearchedAt string `json:"last_researched_at,omitempty"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` + CompletedAt string `json:"completed_at,omitempty"` + DocumentCount int `json:"document_count"` + Focused bool `json:"focused"` } func taskToResponse(t *research.Task, docCount int) researchTaskResponse { @@ -54,10 +62,14 @@ func taskToResponse(t *research.Task, docCount int) researchTaskResponse { Description: t.Description, Status: string(t.Status), OutputDir: t.OutputDir, + Interval: t.Interval, CreatedAt: t.CreatedAt.Format("2006-01-02T15:04:05Z"), UpdatedAt: t.UpdatedAt.Format("2006-01-02T15:04:05Z"), DocumentCount: docCount, } + if !t.LastResearchedAt.IsZero() { + resp.LastResearchedAt = t.LastResearchedAt.Format("2006-01-02T15:04:05Z") + } if !t.CompletedAt.IsZero() { resp.CompletedAt = t.CompletedAt.Format("2006-01-02T15:04:05Z") } @@ -75,7 +87,11 @@ func (h *Handler) apiResearchList(w http.ResponseWriter, r *http.Request) { result := make([]researchTaskResponse, 0, len(tasks)) for _, t := range tasks { docCount, _ := h.researchStore.DocumentCount(t.ID) - result = append(result, taskToResponse(t, docCount)) + resp := taskToResponse(t, docCount) + if h.researchFocus != nil { + resp.Focused = h.researchFocus.IsFocused(t.ID) + } + result = append(result, resp) } writeJSON(w, result) } @@ -84,6 +100,7 @@ func (h *Handler) apiResearchCreate(w http.ResponseWriter, r *http.Request) { var req struct { Title string `json:"title"` Description string `json:"description"` + Interval string `json:"interval"` } if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { http.Error(w, `{"error":"invalid request body"}`, http.StatusBadRequest) @@ -94,7 +111,11 @@ func (h *Handler) apiResearchCreate(w http.ResponseWriter, r *http.Request) { return } - task, err := h.researchStore.CreateTask(strings.TrimSpace(req.Title), strings.TrimSpace(req.Description)) + task, err := h.researchStore.CreateTask( + strings.TrimSpace(req.Title), + strings.TrimSpace(req.Description), + strings.TrimSpace(req.Interval), + ) if err != nil { http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusInternalServerError) return @@ -180,8 +201,13 @@ func (h *Handler) apiResearchGetTask(w http.ResponseWriter, taskID string) { }) } + resp := taskToResponse(task, len(docs)) + if h.researchFocus != nil { + resp.Focused = h.researchFocus.IsFocused(taskID) + } + writeJSON(w, researchTaskDetailResponse{ - researchTaskResponse: taskToResponse(task, len(docs)), + researchTaskResponse: resp, Documents: docResponses, }) } @@ -191,6 +217,7 @@ func (h *Handler) apiResearchTaskAction(w http.ResponseWriter, r *http.Request, Action string `json:"action"` Title string `json:"title"` Description string `json:"description"` + Interval string `json:"interval"` } if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { http.Error(w, `{"error":"invalid request body"}`, http.StatusBadRequest) @@ -208,6 +235,16 @@ func (h *Handler) apiResearchTaskAction(w http.ResponseWriter, r *http.Request, http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusBadRequest) return } + case "activate": + if err := h.researchStore.SetTaskStatus(taskID, research.StatusActive); err != nil { + http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusBadRequest) + return + } + case "complete": + if err := h.researchStore.SetTaskStatus(taskID, research.StatusCompleted); err != nil { + http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusBadRequest) + return + } case "update": task, err := h.researchStore.GetTask(taskID) if err != nil { @@ -226,6 +263,15 @@ func (h *Handler) apiResearchTaskAction(w http.ResponseWriter, r *http.Request, http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusInternalServerError) return } + case "set_interval": + if req.Interval == "" { + http.Error(w, `{"error":"interval is required"}`, http.StatusBadRequest) + return + } + if err := h.researchStore.SetInterval(taskID, req.Interval); err != nil { + http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusBadRequest) + return + } default: http.Error(w, `{"error":"unknown action"}`, http.StatusBadRequest) return @@ -273,3 +319,65 @@ func (h *Handler) apiResearchGetDoc(w http.ResponseWriter, taskID, docID string) "content": string(content), }) } + +// apiResearchFocus handles GET/POST /miniapp/api/research/focus. +// GET returns the current focus state; POST sets focus/unfocus for a task. +func (h *Handler) apiResearchFocus(w http.ResponseWriter, r *http.Request) { + if h.researchFocus == nil { + http.Error(w, `{"error":"research focus not available"}`, http.StatusServiceUnavailable) + return + } + + switch r.Method { + case http.MethodGet: + h.apiResearchFocusGet(w) + case http.MethodPost: + h.apiResearchFocusSet(w, r) + default: + http.Error(w, `{"error":"method not allowed"}`, http.StatusMethodNotAllowed) + } +} + +func (h *Handler) apiResearchFocusGet(w http.ResponseWriter) { + taskID, title := h.researchFocus.Current() + writeJSON(w, map[string]any{ + "focused_id": taskID, + "focused_title": title, + }) +} + +func (h *Handler) apiResearchFocusSet(w http.ResponseWriter, r *http.Request) { + var req struct { + Action string `json:"action"` // "recall" or "forget" + TaskID string `json:"task_id"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<14)).Decode(&req); err != nil { + http.Error(w, `{"error":"invalid request body"}`, http.StatusBadRequest) + return + } + + switch req.Action { + case "recall": + if req.TaskID == "" { + http.Error(w, `{"error":"task_id is required"}`, http.StatusBadRequest) + return + } + task, err := h.researchStore.GetTask(req.TaskID) + if err != nil { + http.Error(w, `{"error":"task not found"}`, http.StatusNotFound) + return + } + h.researchFocus.Focus(task.ID, task.Title) + case "forget": + if req.TaskID == "" { + h.researchFocus.UnfocusAll() + } else { + h.researchFocus.Unfocus(req.TaskID) + } + default: + http.Error(w, `{"error":"action must be recall or forget"}`, http.StatusBadRequest) + return + } + + h.apiResearchFocusGet(w) +} diff --git a/pkg/miniapp/frontend/src/app.js b/pkg/miniapp/frontend/src/app.js index 03fe1066a..4a4c77a74 100644 --- a/pkg/miniapp/frontend/src/app.js +++ b/pkg/miniapp/frontend/src/app.js @@ -1381,15 +1381,22 @@ async function loadResearch() { } el.innerHTML = tasks.map(function(t) { var sc = STATUS_COLORS[t.status] || STATUS_COLORS.pending; - return '
' + + var focusBadge = t.focused ? 'focused' : ''; + return '
' + '
' + '' + esc(t.title) + '' + - '' + t.status + '' + + '
' + + focusBadge + + '' + t.status + '' + + '
' + '
' + (t.description ? '
' + esc(t.description).substring(0, 120) + '
' : '') + '
' + - t.document_count + ' docs · ' + new Date(t.created_at).toLocaleDateString() + + t.document_count + ' docs · ⏱ ' + ((t.interval === '24h' ? '1d' : t.interval) || '1d') + + (t.last_researched_at ? ' · last: ' + new Date(t.last_researched_at).toLocaleDateString() : '') + '
' + '
'; }).join(''); @@ -1429,16 +1436,36 @@ async function openResearchTask(id) { if (task.description) { html += '
' + esc(task.description) + '
'; } - html += '
' + + var curInterval = task.interval || '24h'; + if (curInterval === '24h') curInterval = '1d'; + html += '
' + + 'Interval:' + + '' + + (task.last_researched_at ? ' · Last: ' + new Date(task.last_researched_at).toLocaleString() + '' : ' · Not yet researched') + + '
'; + html += '
' + 'Created: ' + new Date(task.created_at).toLocaleString() + (task.completed_at ? ' · Completed: ' + new Date(task.completed_at).toLocaleString() : '') + '
'; - if (canCancel || canReopen) { - html += '
'; - if (canCancel) html += ''; - if (canReopen) html += ''; - html += '
'; + html += '
'; + if (task.focused) { + html += ''; + } else { + html += ''; } + if (task.status === 'pending') html += ''; + if (task.status === 'active') html += ''; + if (canCancel) html += ''; + if (canReopen) html += ''; + html += '
'; html += '
'; // Documents @@ -1461,6 +1488,13 @@ async function openResearchTask(id) { }).join(''); } el.innerHTML = html; + // Bind interval select via addEventListener (inline onchange unreliable in Telegram WebView) + var sel = document.getElementById('research-interval-select'); + if (sel) { + sel.addEventListener('change', function() { + researchSetInterval(sel.dataset.taskId, sel.value); + }); + } } catch (e) { el.innerHTML = '
Failed to load task.
'; } @@ -1489,6 +1523,18 @@ async function toggleResearchDoc(card, taskId, docId) { } } +async function researchSetInterval(taskId, interval) { + try { + await fetch( + API_BASE + '/miniapp/api/research/' + taskId + '?initData=' + encodeURIComponent(initData), + { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ action: 'set_interval', interval: interval }) } + ); + openResearchTask(taskId); + } catch (e) { + // ignore + } +} + async function researchAction(taskId, action) { try { await fetch( @@ -1535,12 +1581,27 @@ async function createResearchTask() { } } +async function researchSetFocus(taskId, recall) { + try { + await fetch( + API_BASE + '/miniapp/api/research/focus?initData=' + encodeURIComponent(initData), + { method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ action: recall ? 'recall' : 'forget', task_id: taskId }) } + ); + // Refresh the detail view to show updated focus state + openResearchTask(taskId); + } catch (e) { + // ignore + } +} + window.showNewTaskForm = showNewTaskForm; window.hideNewTaskForm = hideNewTaskForm; window.createResearchTask = createResearchTask; window.openResearchTask = openResearchTask; window.toggleResearchDoc = toggleResearchDoc; window.researchAction = researchAction; +window.researchSetFocus = researchSetFocus; window.showResearchList = showResearchList; diff --git a/pkg/miniapp/miniapp.go b/pkg/miniapp/miniapp.go index c9c6d0f8f..6b6d95077 100644 --- a/pkg/miniapp/miniapp.go +++ b/pkg/miniapp/miniapp.go @@ -41,6 +41,7 @@ type Handler struct { workspace string orchBroadcaster *orch.Broadcaster researchStore *research.ResearchStore + researchFocus *research.FocusTracker devMu sync.RWMutex devTarget *url.URL @@ -107,6 +108,7 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("/miniapp/dev/console", h.apiDevConsole) mux.HandleFunc("/miniapp/dev/", h.serveDevProxy) mux.HandleFunc("/miniapp/api/research", h.requireAuth(h.apiResearch)) + mux.HandleFunc("/miniapp/api/research/focus", h.requireAuth(h.apiResearchFocus)) mux.HandleFunc("/miniapp/api/research/", h.requireAuth(h.apiResearchDetail)) } diff --git a/pkg/miniapp/static/dist/app.js b/pkg/miniapp/static/dist/app.js index 9d7855aec..67d8bcb02 100644 --- a/pkg/miniapp/static/dist/app.js +++ b/pkg/miniapp/static/dist/app.js @@ -8638,7 +8638,8 @@ Please report this to https://github.com/markedjs/marked.`, e) { } el.innerHTML = tasks.map(function(t) { var sc = STATUS_COLORS[t.status] || STATUS_COLORS.pending; - return `
` + '
' + '' + esc(t.title) + "" + '' + t.status + "" + "
" + (t.description ? '
' + esc(t.description).substring(0, 120) + "
" : "") + '
' + t.document_count + " docs · " + new Date(t.created_at).toLocaleDateString() + "
" + "
"; + var focusBadge = t.focused ? 'focused' : ""; + return '
` + '
' + '' + esc(t.title) + "" + '
' + focusBadge + '' + t.status + "" + "
" + "
" + (t.description ? '
' + esc(t.description).substring(0, 120) + "
" : "") + '
' + t.document_count + " docs · " + new Date(t.created_at).toLocaleDateString() + "
" + "
"; }).join(""); } catch (e) { loading.classList.add("hidden"); @@ -8667,14 +8668,17 @@ Please report this to https://github.com/markedjs/marked.`, e) { html += '
' + esc(task.description) + "
"; } html += '
' + "Created: " + new Date(task.created_at).toLocaleString() + (task.completed_at ? " · Completed: " + new Date(task.completed_at).toLocaleString() : "") + "
"; - if (canCancel || canReopen) { - html += '
'; - if (canCancel) - html += ``; - if (canReopen) - html += ``; - html += "
"; + html += '
'; + if (task.focused) { + html += ``; + } else { + html += ``; } + if (canCancel) + html += ``; + if (canReopen) + html += ``; + html += "
"; html += "
"; html += '
Documents (' + task.documents.length + ")
"; if (task.documents.length === 0) { @@ -8741,11 +8745,22 @@ Please report this to https://github.com/markedjs/marked.`, e) { loadResearch(); } catch (e) {} } + async function researchSetFocus(taskId, recall) { + try { + await fetch(API_BASE + "/miniapp/api/research/focus?initData=" + encodeURIComponent(initData), { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ action: recall ? "recall" : "forget", task_id: taskId }) + }); + openResearchTask(taskId); + } catch (e) {} + } window.showNewTaskForm = showNewTaskForm; window.hideNewTaskForm = hideNewTaskForm; window.createResearchTask = createResearchTask; window.openResearchTask = openResearchTask; window.toggleResearchDoc = toggleResearchDoc; window.researchAction = researchAction; + window.researchSetFocus = researchSetFocus; window.showResearchList = showResearchList; })(); diff --git a/pkg/research/focus.go b/pkg/research/focus.go new file mode 100644 index 000000000..0fc008529 --- /dev/null +++ b/pkg/research/focus.go @@ -0,0 +1,65 @@ +package research + +import "sync" + +// FocusTracker tracks which single research task is currently "recalled" +// into the agent's active context. At most one task can be focused at a time +// to avoid context pollution. Thread-safe. +type FocusTracker struct { + mu sync.RWMutex + taskID string // currently focused task ID (empty = none) + title string // currently focused task title +} + +// NewFocusTracker creates a new FocusTracker. +func NewFocusTracker() *FocusTracker { + return &FocusTracker{} +} + +// Focus sets the single focused task, replacing any previous focus. +func (ft *FocusTracker) Focus(taskID, title string) { + ft.mu.Lock() + ft.taskID = taskID + ft.title = title + ft.mu.Unlock() +} + +// Unfocus removes a specific task from focus. +// Returns true if the task was focused. +func (ft *FocusTracker) Unfocus(taskID string) bool { + ft.mu.Lock() + defer ft.mu.Unlock() + if ft.taskID != taskID { + return false + } + ft.taskID = "" + ft.title = "" + return true +} + +// UnfocusAll clears focus. Returns 1 if something was focused, 0 otherwise. +func (ft *FocusTracker) UnfocusAll() int { + ft.mu.Lock() + defer ft.mu.Unlock() + if ft.taskID == "" { + return 0 + } + ft.taskID = "" + ft.title = "" + return 1 +} + +// Current returns the currently focused task ID and title. +// Returns empty strings if nothing is focused. +func (ft *FocusTracker) Current() (taskID, title string) { + ft.mu.RLock() + defer ft.mu.RUnlock() + return ft.taskID, ft.title +} + +// IsFocused returns whether a specific task is currently focused. +func (ft *FocusTracker) IsFocused(taskID string) bool { + ft.mu.RLock() + defer ft.mu.RUnlock() + return ft.taskID == taskID +} diff --git a/pkg/research/store.go b/pkg/research/store.go index e0e1281c4..073051eba 100644 --- a/pkg/research/store.go +++ b/pkg/research/store.go @@ -67,7 +67,69 @@ func OpenResearchStore(dbPath, workspace string) (*ResearchStore, error) { _ = db.Close() return nil, fmt.Errorf("create schema: %w", err) } - return &ResearchStore{db: db, workspace: workspace}, nil + s := &ResearchStore{db: db, workspace: workspace} + if err := s.migrateSchema(); err != nil { + _ = db.Close() + return nil, fmt.Errorf("migrate schema: %w", err) + } + return s, nil +} + +// migrateSchema adds columns introduced after the initial schema. +func (s *ResearchStore) migrateSchema() error { + migrations := []struct { + table, column, ddl string + }{ + { + "research_tasks", "interval", + `ALTER TABLE research_tasks ADD COLUMN interval TEXT NOT NULL DEFAULT '1d'`, + }, + { + "research_tasks", "last_researched_at", + `ALTER TABLE research_tasks ADD COLUMN last_researched_at TEXT NOT NULL DEFAULT ''`, + }, + } + for _, m := range migrations { + exists, err := s.columnExists(m.table, m.column) + if err != nil { + return err + } + if !exists { + if _, err := s.db.Exec(m.ddl); err != nil { + return fmt.Errorf("add column %s.%s: %w", m.table, m.column, err) + } + } + } + + // Normalize legacy '24h' default to '1d' + if _, err := s.db.Exec( + `UPDATE research_tasks SET interval = '1d' WHERE interval = '24h'`, + ); err != nil { + return fmt.Errorf("normalize interval 24h→1d: %w", err) + } + return nil +} + +func (s *ResearchStore) columnExists(table, column string) (bool, error) { + rows, err := s.db.Query(fmt.Sprintf("PRAGMA table_info(%s)", table)) + if err != nil { + return false, err + } + defer rows.Close() + for rows.Next() { + var cid int + var name, typ string + var notNull int + var dfltValue sql.NullString + var pk int + if err := rows.Scan(&cid, &name, &typ, ¬Null, &dfltValue, &pk); err != nil { + return false, err + } + if name == column { + return true, nil + } + } + return false, rows.Err() } // Close closes the database connection. @@ -76,7 +138,15 @@ func (s *ResearchStore) Close() error { } // CreateTask creates a new research task with auto-generated slug and output directory. -func (s *ResearchStore) CreateTask(title, description string) (*Task, error) { +// interval is optional; pass "" to use DefaultResearchInterval. +func (s *ResearchStore) CreateTask(title, description, interval string) (*Task, error) { + if interval == "" { + interval = DefaultResearchInterval + } + if _, err := ParseInterval(interval); err != nil { + return nil, fmt.Errorf("invalid interval %q: %w", interval, err) + } + id := uuid.New().String() slug := slugify(title) now := time.Now().UTC() @@ -90,9 +160,9 @@ func (s *ResearchStore) CreateTask(title, description string) (*Task, error) { nowStr := now.Format(time.RFC3339) _, err := s.db.Exec( - `INSERT INTO research_tasks (id, title, slug, description, status, output_dir, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, - id, title, slug, description, string(StatusPending), outputDir, nowStr, nowStr, + `INSERT INTO research_tasks (id, title, slug, description, status, output_dir, interval, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + id, title, slug, description, string(StatusPending), outputDir, interval, nowStr, nowStr, ) if err != nil { return nil, fmt.Errorf("insert task: %w", err) @@ -105,16 +175,18 @@ func (s *ResearchStore) CreateTask(title, description string) (*Task, error) { Description: description, Status: StatusPending, OutputDir: outputDir, + Interval: interval, CreatedAt: now, UpdatedAt: now, }, nil } +const taskColumns = `id, title, slug, description, status, output_dir, interval, last_researched_at, created_at, updated_at, completed_at` + // GetTask retrieves a single task by ID. func (s *ResearchStore) GetTask(id string) (*Task, error) { row := s.db.QueryRow( - `SELECT id, title, slug, description, status, output_dir, created_at, updated_at, completed_at - FROM research_tasks WHERE id = ?`, id) + `SELECT `+taskColumns+` FROM research_tasks WHERE id = ?`, id) return scanTask(row) } @@ -124,12 +196,10 @@ func (s *ResearchStore) ListTasks(status TaskStatus) ([]*Task, error) { var err error if status == "" { rows, err = s.db.Query( - `SELECT id, title, slug, description, status, output_dir, created_at, updated_at, completed_at - FROM research_tasks ORDER BY created_at DESC`) + `SELECT ` + taskColumns + ` FROM research_tasks ORDER BY created_at DESC`) } else { rows, err = s.db.Query( - `SELECT id, title, slug, description, status, output_dir, created_at, updated_at, completed_at - FROM research_tasks WHERE status = ? ORDER BY created_at DESC`, string(status)) + `SELECT `+taskColumns+` FROM research_tasks WHERE status = ? ORDER BY created_at DESC`, string(status)) } if err != nil { return nil, fmt.Errorf("list tasks: %w", err) @@ -147,6 +217,33 @@ func (s *ResearchStore) ListTasks(status TaskStatus) ([]*Task, error) { return tasks, rows.Err() } +// ListDueTasks returns active/pending tasks that are due for research (interval elapsed). +func (s *ResearchStore) ListDueTasks(maxTasks int) ([]*Task, error) { + rows, err := s.db.Query( + `SELECT ` + taskColumns + ` FROM research_tasks + WHERE status IN ('active', 'pending') + ORDER BY last_researched_at ASC, created_at ASC`) + if err != nil { + return nil, fmt.Errorf("list due tasks: %w", err) + } + defer rows.Close() + + var due []*Task + for rows.Next() { + t, err := scanTaskRow(rows) + if err != nil { + return nil, err + } + if t.IsDue() { + due = append(due, t) + if len(due) >= maxTasks { + break + } + } + } + return due, rows.Err() +} + // SetTaskStatus updates task status with transition validation. func (s *ResearchStore) SetTaskStatus(id string, status TaskStatus) error { task, err := s.GetTask(id) @@ -251,17 +348,65 @@ func (s *ResearchStore) DocumentCount(taskID string) (int, error) { return count, err } +// SearchTasks searches tasks by title or slug partial match (case-insensitive). +func (s *ResearchStore) SearchTasks(query string) ([]*Task, error) { + like := "%" + query + "%" + rows, err := s.db.Query( + `SELECT `+taskColumns+` FROM research_tasks + WHERE title LIKE ? COLLATE NOCASE OR slug LIKE ? COLLATE NOCASE + ORDER BY updated_at DESC`, like, like) + if err != nil { + return nil, fmt.Errorf("search tasks: %w", err) + } + defer rows.Close() + + var tasks []*Task + for rows.Next() { + t, err := scanTaskRow(rows) + if err != nil { + return nil, err + } + tasks = append(tasks, t) + } + return tasks, rows.Err() +} + +// TouchLastResearched updates the last_researched_at timestamp for a task. +func (s *ResearchStore) TouchLastResearched(taskID string) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.Exec( + `UPDATE research_tasks SET last_researched_at = ?, updated_at = ? WHERE id = ?`, + now, now, taskID) + return err +} + +// SetInterval updates the research interval for a task. +func (s *ResearchStore) SetInterval(taskID, interval string) error { + if _, err := ParseInterval(interval); err != nil { + return fmt.Errorf("invalid interval %q: %w", interval, err) + } + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.Exec( + `UPDATE research_tasks SET interval = ?, updated_at = ? WHERE id = ?`, + interval, now, taskID) + return err +} + // --- helpers --- func scanTask(row *sql.Row) (*Task, error) { var t Task - var statusStr, createdStr, updatedStr, completedStr string + var statusStr, lastResearchedStr, createdStr, updatedStr, completedStr string err := row.Scan(&t.ID, &t.Title, &t.Slug, &t.Description, &statusStr, - &t.OutputDir, &createdStr, &updatedStr, &completedStr) + &t.OutputDir, &t.Interval, &lastResearchedStr, &createdStr, &updatedStr, &completedStr) if err != nil { return nil, err } t.Status = TaskStatus(statusStr) + if t.Interval == "" { + t.Interval = DefaultResearchInterval + } + t.LastResearchedAt, _ = time.Parse(time.RFC3339, lastResearchedStr) t.CreatedAt, _ = time.Parse(time.RFC3339, createdStr) t.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr) if completedStr != "" { @@ -276,13 +421,17 @@ type rowScanner interface { func scanTaskRow(row rowScanner) (*Task, error) { var t Task - var statusStr, createdStr, updatedStr, completedStr string + var statusStr, lastResearchedStr, createdStr, updatedStr, completedStr string err := row.Scan(&t.ID, &t.Title, &t.Slug, &t.Description, &statusStr, - &t.OutputDir, &createdStr, &updatedStr, &completedStr) + &t.OutputDir, &t.Interval, &lastResearchedStr, &createdStr, &updatedStr, &completedStr) if err != nil { return nil, err } t.Status = TaskStatus(statusStr) + if t.Interval == "" { + t.Interval = DefaultResearchInterval + } + t.LastResearchedAt, _ = time.Parse(time.RFC3339, lastResearchedStr) t.CreatedAt, _ = time.Parse(time.RFC3339, createdStr) t.UpdatedAt, _ = time.Parse(time.RFC3339, updatedStr) if completedStr != "" { diff --git a/pkg/research/store_test.go b/pkg/research/store_test.go index 2f91e36d3..a90dd1e6e 100644 --- a/pkg/research/store_test.go +++ b/pkg/research/store_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "testing" + "time" ) func setupTestStore(t *testing.T) (*ResearchStore, string) { @@ -21,7 +22,7 @@ func setupTestStore(t *testing.T) (*ResearchStore, string) { func TestCreateAndGetTask(t *testing.T) { store, dir := setupTestStore(t) - task, err := store.CreateTask("Test Research", "A test description") + task, err := store.CreateTask("Test Research", "A test description", "") if err != nil { t.Fatalf("create task: %v", err) } @@ -50,8 +51,8 @@ func TestCreateAndGetTask(t *testing.T) { func TestListTasks(t *testing.T) { store, _ := setupTestStore(t) - store.CreateTask("Task A", "") - store.CreateTask("Task B", "") + store.CreateTask("Task A", "", "") + store.CreateTask("Task B", "", "") all, err := store.ListTasks("") if err != nil { @@ -81,7 +82,7 @@ func TestListTasks(t *testing.T) { func TestSetTaskStatus(t *testing.T) { store, _ := setupTestStore(t) - task, _ := store.CreateTask("Status Test", "") + task, _ := store.CreateTask("Status Test", "", "") // Valid: pending → active if err := store.SetTaskStatus(task.ID, StatusActive); err != nil { @@ -113,7 +114,7 @@ func TestSetTaskStatus(t *testing.T) { func TestAddAndListDocuments(t *testing.T) { store, dir := setupTestStore(t) - task, _ := store.CreateTask("Doc Test", "") + task, _ := store.CreateTask("Doc Test", "", "") // Create a test file filePath := filepath.Join(dir, task.OutputDir, "001-finding.md") @@ -144,7 +145,7 @@ func TestAddAndListDocuments(t *testing.T) { func TestDocumentCount(t *testing.T) { store, _ := setupTestStore(t) - task, _ := store.CreateTask("Count Test", "") + task, _ := store.CreateTask("Count Test", "", "") count, _ := store.DocumentCount(task.ID) if count != 0 { @@ -163,7 +164,7 @@ func TestDocumentCount(t *testing.T) { func TestDeleteTask(t *testing.T) { store, _ := setupTestStore(t) - task, _ := store.CreateTask("Delete Test", "") + task, _ := store.CreateTask("Delete Test", "", "") store.AddDocument(task.ID, "D1", "p1", "finding", "") if err := store.DeleteTask(task.ID); err != nil { @@ -180,7 +181,7 @@ func TestDeleteTask(t *testing.T) { func TestUpdateTask(t *testing.T) { store, _ := setupTestStore(t) - task, _ := store.CreateTask("Original", "desc") + task, _ := store.CreateTask("Original", "desc", "") if err := store.UpdateTask(task.ID, "Updated", "new desc"); err != nil { t.Fatalf("update: %v", err) } @@ -191,6 +192,106 @@ func TestUpdateTask(t *testing.T) { } } +func TestParseInterval(t *testing.T) { + cases := []struct { + input string + want string // duration string + err bool + }{ + {"1d", "24h0m0s", false}, + {"7d", "168h0m0s", false}, + {"6h", "6h0m0s", false}, + {"30m", "30m0s", false}, + {"", "", true}, + {"abc", "", true}, + } + for _, tc := range cases { + got, err := ParseInterval(tc.input) + if tc.err { + if err == nil { + t.Errorf("ParseInterval(%q) expected error", tc.input) + } + continue + } + if err != nil { + t.Errorf("ParseInterval(%q) error: %v", tc.input, err) + continue + } + if got.String() != tc.want { + t.Errorf("ParseInterval(%q) = %v, want %v", tc.input, got, tc.want) + } + } +} + +func TestTaskIsDue(t *testing.T) { + // Zero LastResearchedAt → always due + task := &Task{Interval: "1d"} + if !task.IsDue() { + t.Error("zero LastResearchedAt should be due") + } + + // Recently researched → not due + task.LastResearchedAt = time.Now().Add(-1 * time.Hour) + task.Interval = "1d" + if task.IsDue() { + t.Error("researched 1h ago with 1d interval should not be due") + } + + // Long ago → due + task.LastResearchedAt = time.Now().Add(-25 * time.Hour) + if !task.IsDue() { + t.Error("researched 25h ago with 1d interval should be due") + } +} + +func TestListDueTasks(t *testing.T) { + store, _ := setupTestStore(t) + + // Create tasks with different intervals + t1, _ := store.CreateTask("Fast", "", "1h") + t2, _ := store.CreateTask("Slow", "", "7d") + + // Both pending with no LastResearchedAt → both due + due, err := store.ListDueTasks(10) + if err != nil { + t.Fatalf("list due: %v", err) + } + if len(due) != 2 { + t.Errorf("expected 2 due tasks, got %d", len(due)) + } + + // Touch t1 → it should no longer be due (1h not elapsed) + store.TouchLastResearched(t1.ID) + due, _ = store.ListDueTasks(10) + if len(due) != 1 { + t.Errorf("expected 1 due task after touch, got %d", len(due)) + } + if len(due) > 0 && due[0].ID != t2.ID { + t.Errorf("expected Slow task to be due, got %s", due[0].Title) + } +} + +func TestSetInterval(t *testing.T) { + store, _ := setupTestStore(t) + task, _ := store.CreateTask("Interval Test", "", "") + if task.Interval != DefaultResearchInterval { + t.Errorf("default interval = %q, want %q", task.Interval, DefaultResearchInterval) + } + + if err := store.SetInterval(task.ID, "6h"); err != nil { + t.Fatalf("set interval: %v", err) + } + got, _ := store.GetTask(task.ID) + if got.Interval != "6h" { + t.Errorf("interval = %q, want 6h", got.Interval) + } + + // Invalid interval + if err := store.SetInterval(task.ID, "xyz"); err == nil { + t.Error("expected error for invalid interval") + } +} + func TestCanTransition(t *testing.T) { cases := []struct { from, to TaskStatus diff --git a/pkg/research/types.go b/pkg/research/types.go index ecbc5b4fd..dce557ba6 100644 --- a/pkg/research/types.go +++ b/pkg/research/types.go @@ -1,6 +1,11 @@ package research -import "time" +import ( + "fmt" + "strconv" + "strings" + "time" +) // TaskStatus represents the lifecycle state of a research task. type TaskStatus string @@ -13,17 +18,60 @@ const ( StatusCanceled TaskStatus = "canceled" ) +// MinFindingsForCompletion is the minimum number of findings required +// before a heartbeat execution can mark a research task as completed. +const MinFindingsForCompletion = 5 + +// DefaultResearchInterval is the default research interval for new tasks. +const DefaultResearchInterval = "1d" + +// DefaultHeartbeatSearchQuota is the max web searches per heartbeat execution. +const DefaultHeartbeatSearchQuota = 3 + // Task represents a research task tracked in the database. type Task struct { - ID string `json:"id"` - Title string `json:"title"` - Slug string `json:"slug"` - Description string `json:"description"` - Status TaskStatus `json:"status"` - OutputDir string `json:"output_dir"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` - CompletedAt time.Time `json:"completed_at,omitempty"` + ID string `json:"id"` + Title string `json:"title"` + Slug string `json:"slug"` + Description string `json:"description"` + Status TaskStatus `json:"status"` + OutputDir string `json:"output_dir"` + Interval string `json:"interval"` + LastResearchedAt time.Time `json:"last_researched_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + CompletedAt time.Time `json:"completed_at,omitempty"` +} + +// IsDue returns true if enough time has elapsed since the last research. +func (t *Task) IsDue() bool { + if t.LastResearchedAt.IsZero() { + return true + } + interval, err := ParseInterval(t.Interval) + if err != nil || interval <= 0 { + interval = 24 * time.Hour + } + return time.Since(t.LastResearchedAt) >= interval +} + +// ParseInterval parses a duration string with support for "d" (days) suffix. +// Examples: "30m", "6h", "1d", "7d", "24h". +func ParseInterval(s string) (time.Duration, error) { + s = strings.TrimSpace(s) + if s == "" { + return 0, fmt.Errorf("empty interval") + } + // Handle "d" suffix: e.g. "1d" -> "24h", "7d" -> "168h" + if strings.HasSuffix(s, "d") { + numStr := strings.TrimSuffix(s, "d") + days, err := strconv.Atoi(numStr) + if err != nil { + return 0, fmt.Errorf("invalid day interval %q: %w", s, err) + } + return time.Duration(days) * 24 * time.Hour, nil + } + return time.ParseDuration(s) } // Document represents a research output document linked to a task. diff --git a/pkg/tools/context_heartbeat.go b/pkg/tools/context_heartbeat.go new file mode 100644 index 000000000..25c975207 --- /dev/null +++ b/pkg/tools/context_heartbeat.go @@ -0,0 +1,59 @@ +package tools + +import ( + "context" + "sync/atomic" +) + +type heartbeatKey struct{} + +// WithHeartbeatContext marks the context as a heartbeat execution. +func WithHeartbeatContext(ctx context.Context) context.Context { + return context.WithValue(ctx, heartbeatKey{}, true) +} + +// IsHeartbeatContext returns true if the context was created by a heartbeat execution. +func IsHeartbeatContext(ctx context.Context) bool { + v, _ := ctx.Value(heartbeatKey{}).(bool) + return v +} + +// WebSearchQuota tracks remaining web searches for a heartbeat execution. +type WebSearchQuota struct { + max int32 + remaining atomic.Int32 +} + +// TryConsume atomically decrements the quota. Returns false if exhausted. +func (q *WebSearchQuota) TryConsume() bool { + for { + cur := q.remaining.Load() + if cur <= 0 { + return false + } + if q.remaining.CompareAndSwap(cur, cur-1) { + return true + } + } +} + +// Max returns the initial quota limit. +func (q *WebSearchQuota) Max() int32 { return q.max } + +// Remaining returns the current remaining quota. +func (q *WebSearchQuota) Remaining() int32 { return q.remaining.Load() } + +type searchQuotaKey struct{} + +// WithWebSearchQuota attaches a web search quota to the context. +func WithWebSearchQuota(ctx context.Context, quota int) context.Context { + q := &WebSearchQuota{max: int32(quota)} + q.remaining.Store(int32(quota)) + return context.WithValue(ctx, searchQuotaKey{}, q) +} + +// GetWebSearchQuota returns the search quota from the context, or nil if not set. +func GetWebSearchQuota(ctx context.Context) *WebSearchQuota { + q, _ := ctx.Value(searchQuotaKey{}).(*WebSearchQuota) + return q +} diff --git a/pkg/tools/context_heartbeat_test.go b/pkg/tools/context_heartbeat_test.go new file mode 100644 index 000000000..38d347c2d --- /dev/null +++ b/pkg/tools/context_heartbeat_test.go @@ -0,0 +1,82 @@ +package tools + +import ( + "context" + "sync" + "testing" +) + +func TestWebSearchQuotaBasic(t *testing.T) { + ctx := context.Background() + + // No quota → nil + if q := GetWebSearchQuota(ctx); q != nil { + t.Error("expected nil quota on bare context") + } + + ctx = WithWebSearchQuota(ctx, 3) + q := GetWebSearchQuota(ctx) + if q == nil { + t.Fatal("expected non-nil quota") + } + if q.Max() != 3 { + t.Errorf("max = %d, want 3", q.Max()) + } + if q.Remaining() != 3 { + t.Errorf("remaining = %d, want 3", q.Remaining()) + } + + // Consume 3 + for i := range 3 { + if !q.TryConsume() { + t.Errorf("consume %d should succeed", i+1) + } + } + + // 4th should fail + if q.TryConsume() { + t.Error("consume after exhaustion should fail") + } + if q.Remaining() != 0 { + t.Errorf("remaining = %d, want 0", q.Remaining()) + } +} + +func TestWebSearchQuotaConcurrent(t *testing.T) { + ctx := WithWebSearchQuota(context.Background(), 100) + q := GetWebSearchQuota(ctx) + + var wg sync.WaitGroup + consumed := make(chan bool, 200) + + for range 200 { + wg.Add(1) + go func() { + defer wg.Done() + consumed <- q.TryConsume() + }() + } + wg.Wait() + close(consumed) + + ok := 0 + for c := range consumed { + if c { + ok++ + } + } + if ok != 100 { + t.Errorf("consumed = %d, want exactly 100", ok) + } +} + +func TestHeartbeatContext(t *testing.T) { + ctx := context.Background() + if IsHeartbeatContext(ctx) { + t.Error("bare context should not be heartbeat") + } + ctx = WithHeartbeatContext(ctx) + if !IsHeartbeatContext(ctx) { + t.Error("should be heartbeat after WithHeartbeatContext") + } +} diff --git a/pkg/tools/research.go b/pkg/tools/research.go index e231263f7..f63a880ec 100644 --- a/pkg/tools/research.go +++ b/pkg/tools/research.go @@ -12,20 +12,27 @@ import ( ) // ResearchTool provides research task management for the agent. +// It implements StatusProvider to inject a lightweight research catalog +// and focus state into the system prompt. type ResearchTool struct { store *research.ResearchStore workspace string + focus *research.FocusTracker } // NewResearchTool creates a new ResearchTool. -func NewResearchTool(store *research.ResearchStore, workspace string) *ResearchTool { - return &ResearchTool{store: store, workspace: workspace} +func NewResearchTool(store *research.ResearchStore, workspace string, focus *research.FocusTracker) *ResearchTool { + return &ResearchTool{ + store: store, + workspace: workspace, + focus: focus, + } } func (t *ResearchTool) Name() string { return "research" } func (t *ResearchTool) Description() string { - return "Manage research tasks and findings. Use list_tasks to discover pending research, set_status to update task state, add_finding to record research results as markdown documents, and get_task to view task details." + return "Manage research tasks and findings. Actions: list_tasks, get_task, set_status, set_interval, add_finding, recall (load findings into context), forget (release focus when changing topics)." } func (t *ResearchTool) Parameters() map[string]any { @@ -33,13 +40,29 @@ func (t *ResearchTool) Parameters() map[string]any { "type": "object", "properties": map[string]any{ "action": map[string]any{ - "type": "string", - "enum": []string{"list_tasks", "get_task", "set_status", "add_finding"}, + "type": "string", + "enum": []string{ + "list_tasks", + "get_task", + "set_status", + "set_interval", + "add_finding", + "recall", + "forget", + }, "description": "Action to perform.", }, + "interval": map[string]any{ + "type": "string", + "description": "Research interval (for set_interval). Examples: '6h', '1d', '7d'.", + }, + "query": map[string]any{ + "type": "string", + "description": "Search query to find tasks by title/slug (for recall/forget). Alternative to task_id.", + }, "task_id": map[string]any{ "type": "string", - "description": "Task ID (required for get_task, set_status, add_finding).", + "description": "Task ID (required for get_task, set_status, add_finding; optional for recall/forget).", }, "status_filter": map[string]any{ "type": "string", @@ -76,14 +99,50 @@ func (t *ResearchTool) Execute(ctx context.Context, args map[string]any) *ToolRe case "get_task": return t.getTask(args) case "set_status": - return t.setStatus(args) + return t.setStatus(ctx, args) + case "set_interval": + return t.setInterval(args) case "add_finding": return t.addFinding(args) + case "recall": + return t.recall(args) + case "forget": + return t.forget(args) + // Keep backward compat for load_context + case "load_context": + return t.recall(args) default: return ErrorResult("unknown action: " + action) } } +// RuntimeStatus implements StatusProvider. It injects a lightweight research +// catalog (titles only) and current focus state into the system prompt. +func (t *ResearchTool) RuntimeStatus() string { + tasks, err := t.store.ListTasks("") + if err != nil || len(tasks) == 0 { + return "" + } + + var b strings.Builder + b.WriteString("# Research Knowledge Base\n\n") + b.WriteString("Available research topics:\n") + for _, task := range tasks { + docCount, _ := t.store.DocumentCount(task.ID) + b.WriteString(fmt.Sprintf("- \"%s\" [%s, %d docs] (id: %s)\n", + task.Title, task.Status, docCount, task.ID)) + } + + if focusID, focusTitle := t.focus.Current(); focusID != "" { + b.WriteString(fmt.Sprintf("\nCurrently focused: \"%s\" (id: %s)\n", focusTitle, focusID)) + } + + b.WriteString("\nUse `research recall` to load findings when the user asks about a research topic.\n") + b.WriteString("Use `research forget` to release focus when changing topics.\n") + + return b.String() +} + func (t *ResearchTool) listTasks(args map[string]any) *ToolResult { filter, _ := args["status_filter"].(string) tasks, err := t.store.ListTasks(research.TaskStatus(filter)) @@ -140,19 +199,46 @@ func (t *ResearchTool) getTask(args map[string]any) *ToolResult { return NewToolResult(b.String()) } -func (t *ResearchTool) setStatus(args map[string]any) *ToolResult { +func (t *ResearchTool) setStatus(ctx context.Context, args map[string]any) *ToolResult { taskID, _ := args["task_id"].(string) status, _ := args["status"].(string) if taskID == "" || status == "" { return ErrorResult("task_id and status are required") } + // Guard: during heartbeat, prevent completing tasks with too few findings + if IsHeartbeatContext(ctx) && research.TaskStatus(status) == research.StatusCompleted { + docCount, err := t.store.DocumentCount(taskID) + if err != nil { + return ErrorResult(fmt.Sprintf("check document count: %v", err)) + } + if docCount < research.MinFindingsForCompletion { + return ErrorResult(fmt.Sprintf( + "cannot mark as completed during heartbeat: task has %d findings (minimum %d required). Add more findings first.", + docCount, + research.MinFindingsForCompletion, + )) + } + } + if err := t.store.SetTaskStatus(taskID, research.TaskStatus(status)); err != nil { return ErrorResult(fmt.Sprintf("set status: %v", err)) } return NewToolResult(fmt.Sprintf("Task status updated to %s.", status)) } +func (t *ResearchTool) setInterval(args map[string]any) *ToolResult { + taskID, _ := args["task_id"].(string) + interval, _ := args["interval"].(string) + if taskID == "" || interval == "" { + return ErrorResult("task_id and interval are required") + } + if err := t.store.SetInterval(taskID, interval); err != nil { + return ErrorResult(fmt.Sprintf("set interval: %v", err)) + } + return NewToolResult(fmt.Sprintf("Research interval updated to %s.", interval)) +} + var sanitizeRe = regexp.MustCompile(`[^a-zA-Z0-9_-]+`) func (t *ResearchTool) addFinding(args map[string]any) *ToolResult { @@ -199,8 +285,133 @@ func (t *ResearchTool) addFinding(args map[string]any) *ToolResult { return ErrorResult(fmt.Sprintf("add document: %v", err)) } + // Touch last_researched_at for interval tracking + _ = t.store.TouchLastResearched(taskID) + return NewToolResult(fmt.Sprintf( "Finding recorded:\n- File: %s\n- Document ID: %s\n- Seq: %d", relPath, doc.ID, doc.Seq, )) } + +// maxContextBytes caps the total size of loaded research context. +const maxContextBytes = 100 * 1024 // 100KB + +// recall loads all findings for a task and marks it as focused in the system prompt. +func (t *ResearchTool) recall(args map[string]any) *ToolResult { + task, err := t.resolveTask(args) + if err != nil { + return ErrorResult(err.Error()) + } + + docs, err := t.store.ListDocuments(task.ID) + if err != nil { + return ErrorResult(fmt.Sprintf("list documents: %v", err)) + } + + // Mark as focused + t.focus.Focus(task.ID, task.Title) + + var b strings.Builder + b.WriteString(fmt.Sprintf("# Research Context: %s\n\n", task.Title)) + b.WriteString(fmt.Sprintf("**Status**: %s | **Documents**: %d\n", task.Status, len(docs))) + if task.Description != "" { + b.WriteString(fmt.Sprintf("**Description**: %s\n", task.Description)) + } + b.WriteString("\n---\n\n") + + if len(docs) == 0 { + b.WriteString("No findings recorded yet.\n") + return NewToolResult(b.String()) + } + + totalBytes := b.Len() + loaded := 0 + for _, d := range docs { + absPath := filepath.Join(t.workspace, d.FilePath) + data, readErr := os.ReadFile(absPath) + if readErr != nil { + b.WriteString(fmt.Sprintf("## [%d] %s\n\n(file not found: %s)\n\n", d.Seq, d.Title, d.FilePath)) + loaded++ + continue + } + + content := string(data) + entrySize := len(d.Title) + len(content) + 40 // rough overhead for headers + if totalBytes+entrySize > maxContextBytes { + remaining := len(docs) - loaded + b.WriteString( + fmt.Sprintf( + "\n---\n*Context limit reached. %d more finding(s) not shown. Use get_task to see the full list.*\n", + remaining, + ), + ) + break + } + + b.WriteString(fmt.Sprintf("## [%d] %s\n\n", d.Seq, d.Title)) + b.WriteString(content) + if !strings.HasSuffix(content, "\n") { + b.WriteByte('\n') + } + b.WriteString("\n---\n\n") + totalBytes += entrySize + loaded++ + } + + return NewToolResult(b.String()) +} + +// forget removes a task from the focus set. +func (t *ResearchTool) forget(args map[string]any) *ToolResult { + taskID, _ := args["task_id"].(string) + query, _ := args["query"].(string) + + // If no args, forget all + if taskID == "" && query == "" { + count := t.focus.UnfocusAll() + if count == 0 { + return NewToolResult("No research topics were focused.") + } + return NewToolResult(fmt.Sprintf("Released focus on %d research topic(s).", count)) + } + + task, err := t.resolveTask(args) + if err != nil { + return ErrorResult(err.Error()) + } + + if !t.focus.Unfocus(task.ID) { + return NewToolResult(fmt.Sprintf("Research topic \"%s\" was not focused.", task.Title)) + } + return NewToolResult( + fmt.Sprintf("Released focus on \"%s\". Research findings are no longer in active context.", task.Title), + ) +} + +// resolveTask finds a task by task_id or query. +func (t *ResearchTool) resolveTask(args map[string]any) (*research.Task, error) { + taskID, _ := args["task_id"].(string) + query, _ := args["query"].(string) + + if taskID == "" && query == "" { + return nil, fmt.Errorf("task_id or query is required") + } + + if taskID != "" { + task, err := t.store.GetTask(taskID) + if err != nil { + return nil, fmt.Errorf("get task: %w", err) + } + return task, nil + } + + tasks, err := t.store.SearchTasks(query) + if err != nil { + return nil, fmt.Errorf("search tasks: %w", err) + } + if len(tasks) == 0 { + return nil, fmt.Errorf("no research task found matching %q", query) + } + return tasks[0], nil +} diff --git a/pkg/tools/web.go b/pkg/tools/web.go index 4e5e31f47..25ec6050f 100644 --- a/pkg/tools/web.go +++ b/pkg/tools/web.go @@ -834,6 +834,15 @@ func (t *WebSearchTool) Parameters() map[string]any { } func (t *WebSearchTool) Execute(ctx context.Context, args map[string]any) *ToolResult { + // Enforce heartbeat web search quota + if quota := GetWebSearchQuota(ctx); quota != nil { + if !quota.TryConsume() { + return ErrorResult(fmt.Sprintf( + "web_search quota exhausted (%d/%d used this heartbeat). Work with findings you already have.", + quota.Max(), quota.Max())) + } + } + query, ok := args["query"].(string) if !ok { return ErrorResult("query is required") diff --git a/web/backend/api/research.go b/web/backend/api/research.go index 34bba7103..521c62ef4 100644 --- a/web/backend/api/research.go +++ b/web/backend/api/research.go @@ -29,16 +29,18 @@ func (h *Handler) openResearchStore() (*research.ResearchStore, error) { } type researchTaskJSON struct { - ID string `json:"id"` - Title string `json:"title"` - Slug string `json:"slug"` - Description string `json:"description"` - Status string `json:"status"` - OutputDir string `json:"output_dir"` - CreatedAt string `json:"created_at"` - UpdatedAt string `json:"updated_at"` - CompletedAt string `json:"completed_at,omitempty"` - DocumentCount int `json:"document_count"` + ID string `json:"id"` + Title string `json:"title"` + Slug string `json:"slug"` + Description string `json:"description"` + Status string `json:"status"` + OutputDir string `json:"output_dir"` + Interval string `json:"interval"` + LastResearchedAt string `json:"last_researched_at,omitempty"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` + CompletedAt string `json:"completed_at,omitempty"` + DocumentCount int `json:"document_count"` } type researchDocJSON struct { @@ -60,10 +62,14 @@ func taskToJSON(t *research.Task, docCount int) researchTaskJSON { Description: t.Description, Status: string(t.Status), OutputDir: t.OutputDir, + Interval: t.Interval, CreatedAt: t.CreatedAt.Format("2006-01-02T15:04:05Z"), UpdatedAt: t.UpdatedAt.Format("2006-01-02T15:04:05Z"), DocumentCount: docCount, } + if !t.LastResearchedAt.IsZero() { + r.LastResearchedAt = t.LastResearchedAt.Format("2006-01-02T15:04:05Z") + } if !t.CompletedAt.IsZero() { r.CompletedAt = t.CompletedAt.Format("2006-01-02T15:04:05Z") } @@ -103,6 +109,7 @@ func (h *Handler) handleResearch(w http.ResponseWriter, r *http.Request) { var req struct { Title string `json:"title"` Description string `json:"description"` + Interval string `json:"interval"` } if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { http.Error(w, `{"error":"invalid request body"}`, http.StatusBadRequest) @@ -112,7 +119,11 @@ func (h *Handler) handleResearch(w http.ResponseWriter, r *http.Request) { http.Error(w, `{"error":"title is required"}`, http.StatusBadRequest) return } - task, err := store.CreateTask(strings.TrimSpace(req.Title), strings.TrimSpace(req.Description)) + task, err := store.CreateTask( + strings.TrimSpace(req.Title), + strings.TrimSpace(req.Description), + strings.TrimSpace(req.Interval), + ) if err != nil { http.Error(w, `{"error":"`+err.Error()+`"}`, http.StatusInternalServerError) return diff --git a/web/frontend/src/api/research.ts b/web/frontend/src/api/research.ts index 236d93b12..f2cae551f 100644 --- a/web/frontend/src/api/research.ts +++ b/web/frontend/src/api/research.ts @@ -5,6 +5,8 @@ export interface ResearchTask { description: string status: "pending" | "active" | "completed" | "failed" | "canceled" output_dir: string + interval: string + last_researched_at?: string created_at: string updated_at: string completed_at?: string @@ -68,18 +70,19 @@ export async function getResearchTask( export async function createResearchTask( title: string, description: string, + interval?: string, ): Promise { return request("/api/research", { method: "POST", headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ title, description }), + body: JSON.stringify({ title, description, interval }), }) } export async function researchTaskAction( id: string, - action: "cancel" | "reopen" | "update", - data?: { title?: string; description?: string }, + action: "cancel" | "reopen" | "update" | "set_interval", + data?: { title?: string; description?: string; interval?: string }, ): Promise { return request( `/api/research/${encodeURIComponent(id)}`, diff --git a/web/frontend/src/components/research/research-page.tsx b/web/frontend/src/components/research/research-page.tsx index 5598ea1c9..b23d31545 100644 --- a/web/frontend/src/components/research/research-page.tsx +++ b/web/frontend/src/components/research/research-page.tsx @@ -218,9 +218,13 @@ function TaskCard({ task }: { task: ResearchTask }) { count: task.document_count, })} - - {new Date(task.created_at).toLocaleDateString()} - + ⏱ {task.interval || "24h"} + {task.last_researched_at ? ( + + {t("pages.research.last_researched")}:{" "} + {new Date(task.last_researched_at).toLocaleDateString()} + + ) : null} diff --git a/web/frontend/src/components/research/task-detail-page.tsx b/web/frontend/src/components/research/task-detail-page.tsx index 7ffbab5a7..ecbcdb507 100644 --- a/web/frontend/src/components/research/task-detail-page.tsx +++ b/web/frontend/src/components/research/task-detail-page.tsx @@ -171,6 +171,13 @@ export function TaskDetailPage({ taskId }: { taskId: string }) {
+
+ {t("pages.research.interval")}: {task.interval || "24h"} + {task.last_researched_at ? ( + <> · {t("pages.research.last_researched")}:{" "} + {new Date(task.last_researched_at).toLocaleString()} + ) : null} +
{t("pages.research.created_at")}:{" "} {new Date(task.created_at).toLocaleString()} diff --git a/web/frontend/src/i18n/locales/en.json b/web/frontend/src/i18n/locales/en.json index 80a40dab8..9dd4ab3de 100644 --- a/web/frontend/src/i18n/locales/en.json +++ b/web/frontend/src/i18n/locales/en.json @@ -474,6 +474,8 @@ "documents_title": "Documents ({{count}})", "documents_count": "{{count}} docs", "no_documents": "No documents yet.", + "interval": "Interval", + "last_researched": "Last researched", "created_at": "Created", "completed_at": "Completed", "output_dir": "Output", diff --git a/web/frontend/src/i18n/locales/zh.json b/web/frontend/src/i18n/locales/zh.json index e0f8b663a..2a729626f 100644 --- a/web/frontend/src/i18n/locales/zh.json +++ b/web/frontend/src/i18n/locales/zh.json @@ -474,6 +474,8 @@ "documents_title": "文档 ({{count}})", "documents_count": "{{count}} 篇文档", "no_documents": "暂无文档。", + "interval": "调查频率", + "last_researched": "最后调查", "created_at": "创建时间", "completed_at": "完成时间", "output_dir": "输出目录",