| 1 | // Heartbeat task engine — scheduled AI prompts that create or update topics. |
| 2 | // |
| 3 | // Each task is a prompt submitted to a dedicated topic on a schedule. |
| 4 | // The config file under the Reasonix user state directory is human- and |
| 5 | // AI-editable; the engine runs the schedule in a background goroutine and |
| 6 | // exposes Wails bindings on App for the frontend panel. |
| 7 | // |
| 8 | // Design goal: minimal upstream intrusion — one file, zero changes to existing |
| 9 | // Go code (App field + startup line + bindings are the only touch points). |
| 10 | |
| 11 | package main |
| 12 | |
| 13 | import ( |
| 14 | "crypto/sha256" |
| 15 | "encoding/hex" |
| 16 | "errors" |
| 17 | "log" |
| 18 | "math/rand" |
| 19 | "os" |
| 20 | "path/filepath" |
| 21 | "sync" |
| 22 | "time" |
| 23 | |
| 24 | "reasonix/internal/config" |
| 25 | "reasonix/internal/control" |
| 26 | "reasonix/internal/event" |
| 27 | filelock "reasonix/internal/identitylock" |
| 28 | "reasonix/internal/permissionpreset" |
| 29 | "reasonix/internal/secrets" |
| 30 | ) |
| 31 | |
| 32 | // ── Data model ────────────────────────────────────────────────────────────── |
| 33 | |
| 34 | // HeartbeatTask defines a single scheduled prompt. |
| 35 | type HeartbeatTask struct { |
| 36 | ID string `json:"id"` |
| 37 | Title string `json:"title"` // user-visible label |
| 38 | Prompt string `json:"prompt"` // the prompt to submit |
| 39 | Interval string `json:"interval"` // e.g. "5m", "1h", "30s" |
| 40 | Enabled bool `json:"enabled"` |
| 41 | Scope string `json:"scope,omitempty"` // "global" or "project" |
| 42 | WorkspaceRoot string `json:"workspaceRoot,omitempty"` // project root path when scope="project" |
| 43 | TopicID string `json:"topicId,omitempty"` // created topic, reused on re-run |
| 44 | LastRunAt int64 `json:"lastRunAt,omitempty"` // unix millis |
| 45 | NewConversationEachRun bool `json:"newConversationEachRun,omitempty"` // true = create new topic every run |
| 46 | RunHistory []HeartbeatRun `json:"runHistory,omitempty"` // recent executions (oldest first, capped) |
| 47 | CreatedAt int64 `json:"createdAt,omitempty"` |
| 48 | ApprovalMode string `json:"approvalMode"` // read-only | workspace-write | danger-full-access; empty defaults to workspace-write |
| 49 | TimeWindowStart string `json:"timeWindowStart,omitempty"` // "HH:MM" — interval tasks only run after this time (inclusive) |
| 50 | TimeWindowEnd string `json:"timeWindowEnd,omitempty"` // "HH:MM" — interval tasks only run before this time (exclusive) |
| 51 | NotifyChannels *bool `json:"notifyChannels,omitempty"` // true = push to bot channels; nil/false = skip |
| 52 | } |
| 53 | |
| 54 | // HeartbeatRun records a single successful execution of a heartbeat task. |
| 55 | // TopicID is the conversation created/reused by that run (may be empty if |
| 56 | // the run produced no topic). |
| 57 | type HeartbeatRun struct { |
| 58 | At int64 `json:"at"` // unix millis execution time |
| 59 | TopicID string `json:"topicId"` // topic used/created by this run |
| 60 | } |
| 61 | |
| 62 | // maxRunHistory caps how many recent executions are kept per task. |
| 63 | const maxRunHistory = 20 |
| 64 | |
| 65 | // heartbeatSchemaVersion is the current on-disk config schema version. |
| 66 | // v1 (schemaVersion absent/0): interval-only tasks, no runHistory. |
| 67 | // v2: adds runHistory per task (execution history, capped at maxRunHistory). |
| 68 | // |
| 69 | // Migration boundary: configs written by v2+ binaries are read fine by older |
| 70 | // binaries (unknown fields are ignored by json.Unmarshal), but an older |
| 71 | // binary doing a full-table save (ReplaceTasks/ReplaceConfig) will silently |
| 72 | // drop runHistory because it doesn't know the field. This is a one-way |
| 73 | // upgrade — once a v2+ binary has saved, do not run an older binary that |
| 74 | // writes the config. writeTasks refuses to overwrite a config with a |
| 75 | // schemaVersion newer than this binary understands (forward protection). |
| 76 | const heartbeatSchemaVersion = 2 |
| 77 | |
| 78 | // heartbeatConfig is the on-disk format. |
| 79 | type heartbeatConfig struct { |
| 80 | SchemaVersion int `json:"schemaVersion,omitempty"` |
| 81 | Revision uint64 `json:"revision,omitempty"` |
| 82 | Tasks []HeartbeatTask `json:"tasks"` |
| 83 | } |
| 84 | |
| 85 | // ErrHeartbeatConfigConflict means another writer changed the config after |
| 86 | // this engine last read it. Callers should reload before retrying the edit. |
| 87 | var ErrHeartbeatConfigConflict = errors.New("heartbeat config changed concurrently") |
| 88 | |
| 89 | type heartbeatConfigSnapshot struct { |
| 90 | cfg heartbeatConfig |
| 91 | digest [sha256.Size]byte |
| 92 | exists bool |
| 93 | } |
| 94 | |
| 95 | // HeartbeatConfigView is the revisioned Wails contract used by current |
| 96 | // frontends. ETag detects external editors that do not increment Revision. |
| 97 | type HeartbeatConfigView struct { |
| 98 | Revision uint64 `json:"revision"` |
| 99 | ETag string `json:"etag"` |
| 100 | Tasks []HeartbeatTask `json:"tasks"` |
| 101 | } |
| 102 | |
| 103 | type HeartbeatConfigUpdate struct { |
| 104 | Revision uint64 `json:"revision"` |
| 105 | ETag string `json:"etag"` |
| 106 | Tasks []HeartbeatTask `json:"tasks"` |
| 107 | } |
| 108 | |
| 109 | func (s heartbeatConfigSnapshot) view() HeartbeatConfigView { |
| 110 | tasks := s.cfg.Tasks |
| 111 | if tasks == nil { |
| 112 | tasks = []HeartbeatTask{} |
| 113 | } |
| 114 | etag := "" |
| 115 | if s.exists { |
| 116 | etag = hex.EncodeToString(s.digest[:]) |
| 117 | } |
| 118 | return HeartbeatConfigView{Revision: s.cfg.Revision, ETag: etag, Tasks: tasks} |
| 119 | } |
| 120 | |
| 121 | // ── Engine ────────────────────────────────────────────────────────────────── |
| 122 | |
| 123 | // HeartbeatEngine runs scheduled task execution in a background goroutine. |
| 124 | // It is owned by App and started during App.startup. |
| 125 | type HeartbeatEngine struct { |
| 126 | mu sync.Mutex |
| 127 | tasks []HeartbeatTask |
| 128 | cfgRevision uint64 // persisted config revision last observed |
| 129 | cfgDigest [sha256.Size]byte // decoded config bytes last observed |
| 130 | cfgKnown bool // cfgDigest describes an existing file |
| 131 | cfgInitialized bool // engine has observed existing or missing config state |
| 132 | cfgDeleted bool // an existing config was removed externally |
| 133 | pendingTopics map[string]heartbeatPendingTopic // in-memory retry/in-flight safety for NewConversationEachRun |
| 134 | runningTasks map[string]struct{} // task-level execution reservation shared by tick and TriggerNow |
| 135 | done chan struct{} |
| 136 | running bool |
| 137 | app *App // back-reference for topic creation, tab routing, and prompt submission |
| 138 | } |
| 139 | |
| 140 | type heartbeatPendingTopic struct { |
| 141 | TopicID string |
| 142 | Submitted bool |
| 143 | } |
| 144 | |
| 145 | func newHeartbeatEngine(app *App) *HeartbeatEngine { |
| 146 | return &HeartbeatEngine{ |
| 147 | app: app, |
| 148 | done: make(chan struct{}), |
| 149 | pendingTopics: make(map[string]heartbeatPendingTopic), |
| 150 | runningTasks: make(map[string]struct{}), |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // configPath returns the JSON file path. |
| 155 | func (e *HeartbeatEngine) configPath() string { |
| 156 | dir := config.MemoryUserDir() |
| 157 | if dir == "" { |
| 158 | dir = "." |
| 159 | } |
| 160 | return filepath.Join(dir, "heartbeat-tasks.json") |
| 161 | } |
| 162 | |
| 163 | // Start launches the scheduler goroutine. |
| 164 | func (e *HeartbeatEngine) Start() { |
| 165 | e.mu.Lock() |
| 166 | defer e.mu.Unlock() |
| 167 | if e.running { |
| 168 | return |
| 169 | } |
| 170 | snapshot, err := e.readConfigSnapshot() |
| 171 | if err != nil { |
| 172 | log.Printf("[heartbeat] invalid config: %v", err) |
| 173 | } else { |
| 174 | e.recordConfigSnapshotLocked(snapshot) |
| 175 | e.tasks = snapshot.cfg.Tasks |
| 176 | } |
| 177 | e.running = true |
| 178 | go e.loop() |
| 179 | log.Printf("[heartbeat] engine started (%d tasks)", len(e.tasks)) |
| 180 | } |
| 181 | |
| 182 | // Stop signals the scheduler goroutine to exit. |
| 183 | func (e *HeartbeatEngine) Stop() { |
| 184 | e.mu.Lock() |
| 185 | defer e.mu.Unlock() |
| 186 | if !e.running { |
| 187 | return |
| 188 | } |
| 189 | e.running = false |
| 190 | close(e.done) |
| 191 | } |
| 192 | |
| 193 | // loop is the main scheduler loop — tick every 30s and check each enabled task. |
| 194 | func (e *HeartbeatEngine) loop() { |
| 195 | ticker := time.NewTicker(30 * time.Second) |
| 196 | defer ticker.Stop() |
| 197 | for { |
| 198 | select { |
| 199 | case <-e.done: |
| 200 | return |
| 201 | case <-ticker.C: |
| 202 | e.tick() |
| 203 | } |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | // tick checks every enabled task and runs those whose interval has elapsed. |
| 208 | // It first adopts any external edit to the config file (human/AI-editable), |
| 209 | // then merges results (topicId, lastRunAt) rather than replacing the full |
| 210 | // list, so concurrent HeartbeatSaveTasks edits are not lost. |
| 211 | func (e *HeartbeatEngine) tick() { |
| 212 | e.mu.Lock() |
| 213 | e.adoptExternalEditsLocked() |
| 214 | tasks := append([]HeartbeatTask(nil), e.tasks...) |
| 215 | e.mu.Unlock() |
| 216 | |
| 217 | now := time.Now() |
| 218 | for _, t := range tasks { |
| 219 | if !t.Enabled { |
| 220 | continue |
| 221 | } |
| 222 | if !heartbeatTaskDueAt(t, now) { |
| 223 | continue |
| 224 | } |
| 225 | e.executeScheduledTask(t, now) |
| 226 | } |
| 227 | } |
| 228 | |
| 229 | // normalizeHeartbeatApprovalMode returns a valid approval mode for the task. |
| 230 | // Empty values default to workspace-write. Legacy values are conservatively |
| 231 | // migrated by the shared permission-preset normalizer. |
| 232 | func normalizeHeartbeatApprovalMode(mode string) string { |
| 233 | return string(permissionpreset.NormalizeDefault(mode)) |
| 234 | } |
| 235 | |
| 236 | type heartbeatRuntimeStatus interface { |
| 237 | RuntimeStatus() control.RuntimeStatus |
| 238 | } |
| 239 | |
| 240 | func heartbeatControllerBusy(ctrl heartbeatRuntimeStatus) bool { |
| 241 | status := ctrl.RuntimeStatus() |
| 242 | return status.Running || status.PendingPrompt |
| 243 | } |
| 244 | |
| 245 | func (e *HeartbeatEngine) executeScheduledTask(t HeartbeatTask, dueAt time.Time) HeartbeatTask { |
| 246 | return e.executeTaskWithLease(t, func(task HeartbeatTask) (HeartbeatTask, bool) { |
| 247 | snapshot, err := e.readConfigSnapshot() |
| 248 | if err != nil { |
| 249 | log.Printf("[heartbeat] cannot revalidate task %q before execution: %v", task.Title, err) |
| 250 | return task, false |
| 251 | } |
| 252 | for _, current := range snapshot.cfg.Tasks { |
| 253 | if current.ID == task.ID { |
| 254 | return current, current.Enabled && heartbeatTaskDueAt(current, dueAt) |
| 255 | } |
| 256 | } |
| 257 | return task, false |
| 258 | }) |
| 259 | } |
| 260 | |
| 261 | // executeTaskWithLease runs one heartbeat: creates/opens topic, submits prompt. |
| 262 | // Returns the updated task (topicId and LastRunAt may change). |
| 263 | // On controller failure the task is returned WITHOUT updating LastRunAt, |
| 264 | // so it will be retried on the next tick. |
| 265 | func (e *HeartbeatEngine) executeTaskWithLease(t HeartbeatTask, prepare func(HeartbeatTask) (HeartbeatTask, bool)) HeartbeatTask { |
| 266 | if !e.claimTask(t.ID) { |
| 267 | log.Printf("[heartbeat] task %q is already running, skipping overlapping trigger", t.Title) |
| 268 | return t |
| 269 | } |
| 270 | releaseLease, err := e.tryAcquireTaskLease(t.ID) |
| 271 | if err != nil { |
| 272 | log.Printf("[heartbeat] task %q is already owned by another runtime, skipping", t.Title) |
| 273 | e.releaseTask(t.ID) |
| 274 | return t |
| 275 | } |
| 276 | defer func() { |
| 277 | releaseLease() |
| 278 | e.releaseTask(t.ID) |
| 279 | }() |
| 280 | if prepare != nil { |
| 281 | var ready bool |
| 282 | t, ready = prepare(t) |
| 283 | if !ready { |
| 284 | return t |
| 285 | } |
| 286 | } |
| 287 | updated := e.executeTaskOwned(t) |
| 288 | e.mu.Lock() |
| 289 | e.mergeRunUpdatesLocked(map[string]HeartbeatTask{updated.ID: updated}) |
| 290 | e.mu.Unlock() |
| 291 | return updated |
| 292 | } |
| 293 | |
| 294 | // tryAcquireTaskLease extends the in-process reservation to other Reasonix |
| 295 | // processes. The lease is held from before topic creation through prompt |
| 296 | // submission, and the OS releases it automatically if the process is killed. |
| 297 | func (e *HeartbeatEngine) tryAcquireTaskLease(taskID string) (func(), error) { |
| 298 | path := e.heartbeatTaskLeasePath(taskID) |
| 299 | if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { |
| 300 | return nil, err |
| 301 | } |
| 302 | return filelock.TryAcquire(path) |
| 303 | } |
| 304 | |
| 305 | func (e *HeartbeatEngine) heartbeatTaskLeasePath(taskID string) string { |
| 306 | digest := sha256.Sum256([]byte(taskID)) |
| 307 | return e.configPath() + "." + hex.EncodeToString(digest[:8]) + ".run.lock" |
| 308 | } |
| 309 | |
| 310 | func (e *HeartbeatEngine) claimTask(id string) bool { |
| 311 | e.mu.Lock() |
| 312 | defer e.mu.Unlock() |
| 313 | if e.runningTasks == nil { |
| 314 | e.runningTasks = make(map[string]struct{}) |
| 315 | } |
| 316 | if _, exists := e.runningTasks[id]; exists { |
| 317 | return false |
| 318 | } |
| 319 | e.runningTasks[id] = struct{}{} |
| 320 | return true |
| 321 | } |
| 322 | |
| 323 | func (e *HeartbeatEngine) releaseTask(id string) { |
| 324 | e.mu.Lock() |
| 325 | delete(e.runningTasks, id) |
| 326 | e.mu.Unlock() |
| 327 | } |
| 328 | |
| 329 | // resolveHeartbeatTopic selects or creates the topic for one run. |
| 330 | // |
| 331 | // For NewConversationEachRun: |
| 332 | // - Reuse a pending topic from a failed pre-submit attempt. |
| 333 | // - Re-check a submitted topic until its controller is idle, so a long |
| 334 | // previous run cannot overlap with the next scheduled fresh topic. |
| 335 | // - Once the submitted topic is idle and due again, clear it and create a |
| 336 | // fresh topic. |
| 337 | // - topicId is always updated to the latest conversation so the task list |
| 338 | // always points to the most recent session regardless of mode switch. |
| 339 | // |
| 340 | // For the legacy mode: |
| 341 | // - Reuse the persisted topicID if available; create one on first run. |
| 342 | func (e *HeartbeatEngine) resolveHeartbeatTopic(t HeartbeatTask, scope, workspaceRoot, title string) (HeartbeatTask, string, bool, bool) { |
| 343 | var topicID string |
| 344 | var pendingSubmitted bool |
| 345 | if t.NewConversationEachRun { |
| 346 | e.mu.Lock() |
| 347 | pending := e.pendingTopics[t.ID] |
| 348 | e.mu.Unlock() |
| 349 | topicID = pending.TopicID |
| 350 | pendingSubmitted = pending.Submitted |
| 351 | if topicID == "" { |
| 352 | // No pending topic — create a fresh one. |
| 353 | meta, err := e.app.CreateTopic(scope, workspaceRoot, title) |
| 354 | if err != nil { |
| 355 | log.Printf("[heartbeat] CreateTopic(%q): %v", t.Title, err) |
| 356 | t.LastRunAt = time.Now().UnixMilli() |
| 357 | return t, "", false, false |
| 358 | } |
| 359 | topicID = meta.ID |
| 360 | t.TopicID = topicID // always persist the latest topic |
| 361 | // Save in-memory for retry safety (NOT persisted to disk). |
| 362 | e.mu.Lock() |
| 363 | if e.pendingTopics == nil { |
| 364 | e.pendingTopics = make(map[string]heartbeatPendingTopic) |
| 365 | } |
| 366 | e.pendingTopics[t.ID] = heartbeatPendingTopic{TopicID: topicID} |
| 367 | e.mu.Unlock() |
| 368 | } |
| 369 | } else { |
| 370 | topicID = t.TopicID |
| 371 | if topicID == "" { |
| 372 | meta, err := e.app.CreateTopic(scope, workspaceRoot, title) |
| 373 | if err != nil { |
| 374 | log.Printf("[heartbeat] CreateTopic(%q): %v", t.Title, err) |
| 375 | t.LastRunAt = time.Now().UnixMilli() |
| 376 | return t, "", false, false |
| 377 | } |
| 378 | topicID = meta.ID |
| 379 | t.TopicID = topicID |
| 380 | } |
| 381 | } |
| 382 | return t, topicID, pendingSubmitted, true |
| 383 | } |
| 384 | |
| 385 | func (e *HeartbeatEngine) executeTaskOwned(t HeartbeatTask) HeartbeatTask { |
| 386 | title := "Heartbeat: " + t.Title |
| 387 | scope := t.Scope |
| 388 | workspaceRoot := t.WorkspaceRoot |
| 389 | if scope == "" { |
| 390 | scope = "global" |
| 391 | } |
| 392 | t, topicID, pendingSubmitted, ok := e.resolveHeartbeatTopic(t, scope, workspaceRoot, title) |
| 393 | if !ok { |
| 394 | return t |
| 395 | } |
| 396 | |
| 397 | // Open the tab for the topic (creates one if needed) without changing the |
| 398 | // user's active tab or active workspace pointer. |
| 399 | var tabMeta TabMeta |
| 400 | var err error |
| 401 | if scope == "project" && workspaceRoot != "" { |
| 402 | tabMeta, err = e.app.openProjectTabInactive(workspaceRoot, topicID) |
| 403 | } else { |
| 404 | tabMeta, err = e.app.openGlobalTabInactive(topicID) |
| 405 | } |
| 406 | if err != nil { |
| 407 | log.Printf("[heartbeat] OpenTab(%q): %s", t.Title, secrets.RedactError(err)) |
| 408 | t.LastRunAt = time.Now().UnixMilli() |
| 409 | return t |
| 410 | } |
| 411 | |
| 412 | // Wait for the tab's controller to be built (it's started |
| 413 | // asynchronously in a goroutine by openTopicTab). |
| 414 | var ctrl heartbeatRuntimeStatus |
| 415 | for range 40 { |
| 416 | if candidate := e.app.ctrlByTabID(tabMeta.ID); candidate != nil { |
| 417 | ctrl = candidate |
| 418 | break |
| 419 | } |
| 420 | time.Sleep(250 * time.Millisecond) |
| 421 | } |
| 422 | if ctrl == nil { |
| 423 | log.Printf("[heartbeat] controller not ready for %q, skipping", t.Title) |
| 424 | return t // don't update LastRunAt — retry next tick |
| 425 | } |
| 426 | if heartbeatControllerBusy(ctrl) { |
| 427 | log.Printf("[heartbeat] controller busy for %q, skipping", t.Title) |
| 428 | return t // don't change approval mode for an existing turn — retry next tick |
| 429 | } |
| 430 | if t.NewConversationEachRun && pendingSubmitted { |
| 431 | e.mu.Lock() |
| 432 | if pending := e.pendingTopics[t.ID]; pending.TopicID == topicID && pending.Submitted { |
| 433 | delete(e.pendingTopics, t.ID) |
| 434 | } |
| 435 | e.mu.Unlock() |
| 436 | return e.executeTaskOwned(t) |
| 437 | } |
| 438 | |
| 439 | // Set the task's approval mode only after confirming the controller is idle. |
| 440 | // Applying the task preset rotates the permission revision before execution, |
| 441 | // so applying it to a busy reused topic would accidentally approve a previous |
| 442 | // turn instead of preparing this heartbeat prompt. |
| 443 | mode := normalizeHeartbeatApprovalMode(t.ApprovalMode) |
| 444 | t.ApprovalMode = mode |
| 445 | e.app.SetToolApprovalModeForTab(tabMeta.ID, mode) |
| 446 | |
| 447 | // Attach bot event forwarding if the bot runtime is active and has |
| 448 | // session-mapped targets. The forwarder is set on the tab's event sink |
| 449 | // so AI output events are streamed to connected bot channels in |
| 450 | // real-time alongside the desktop UI. |
| 451 | var botForwarder event.Sink |
| 452 | if t.NotifyChannels != nil && *t.NotifyChannels { |
| 453 | botForwarder = e.newBotForwarder(tabMeta.ID) |
| 454 | } |
| 455 | |
| 456 | // Submit as a plain user turn so scheduled prompts cannot invoke desktop |
| 457 | // shell or slash-command handlers such as "!cmd", "/clear", or "/compact". |
| 458 | if !e.app.submitUserTurnToTabWithSink(tabMeta.ID, t.Prompt, botForwarder) { |
| 459 | log.Printf("[heartbeat] submit skipped for %q", t.Title) |
| 460 | return t |
| 461 | } |
| 462 | |
| 463 | // After a successful submit, keep the topic as an in-flight guard. The next |
| 464 | // due run will busy-check this controller before creating a fresh topic. |
| 465 | if t.NewConversationEachRun { |
| 466 | e.mu.Lock() |
| 467 | if e.pendingTopics == nil { |
| 468 | e.pendingTopics = make(map[string]heartbeatPendingTopic) |
| 469 | } |
| 470 | e.pendingTopics[t.ID] = heartbeatPendingTopic{TopicID: topicID, Submitted: true} |
| 471 | e.mu.Unlock() |
| 472 | } |
| 473 | |
| 474 | t.LastRunAt = time.Now().UnixMilli() |
| 475 | if t.CreatedAt == 0 { |
| 476 | t.CreatedAt = t.LastRunAt |
| 477 | } |
| 478 | // 追加本次成功执行记录（最新追加到尾部，前端倒序展示；最多保留 20 条） |
| 479 | t.RunHistory = append(t.RunHistory, HeartbeatRun{At: t.LastRunAt, TopicID: topicID}) |
| 480 | if len(t.RunHistory) > maxRunHistory { |
| 481 | t.RunHistory = t.RunHistory[len(t.RunHistory)-maxRunHistory:] |
| 482 | } |
| 483 | return t |
| 484 | } |
| 485 | |
| 486 | // ListTasks returns a copy of the current tasks (in-memory). |
| 487 | func (e *HeartbeatEngine) ListTasks() []HeartbeatTask { |
| 488 | e.mu.Lock() |
| 489 | defer e.mu.Unlock() |
| 490 | out := make([]HeartbeatTask, len(e.tasks)) |
| 491 | copy(out, e.tasks) |
| 492 | return out |
| 493 | } |
| 494 | |
| 495 | // ReloadTasks reloads the task list from disk and replaces the in-memory copy. |
| 496 | func (e *HeartbeatEngine) ReloadTasks() []HeartbeatTask { |
| 497 | return e.ReloadConfig().Tasks |
| 498 | } |
| 499 | |
| 500 | func (e *HeartbeatEngine) ReloadConfig() HeartbeatConfigView { |
| 501 | e.mu.Lock() |
| 502 | defer e.mu.Unlock() |
| 503 | snapshot, err := e.readConfigSnapshot() |
| 504 | if err != nil { |
| 505 | log.Printf("[heartbeat] reload config: %v", err) |
| 506 | return heartbeatConfigSnapshot{cfg: heartbeatConfig{Tasks: []HeartbeatTask{}}}.view() |
| 507 | } |
| 508 | e.recordConfigSnapshotLocked(snapshot) |
| 509 | e.tasks = snapshot.cfg.Tasks |
| 510 | e.prunePendingTopicsLocked(e.tasks) |
| 511 | return snapshot.view() |
| 512 | } |
| 513 | |
| 514 | // ReplaceTasks atomically replaces the task list and persists it. |
| 515 | func (e *HeartbeatEngine) ReplaceTasks(tasks []HeartbeatTask) error { |
| 516 | e.mu.Lock() |
| 517 | defer e.mu.Unlock() |
| 518 | expected, err := e.readConfigSnapshot() |
| 519 | if err != nil { |
| 520 | return err |
| 521 | } |
| 522 | if e.cfgInitialized && (expected.exists != e.cfgKnown || expected.digest != e.cfgDigest || expected.cfg.Revision != e.cfgRevision) { |
| 523 | return ErrHeartbeatConfigConflict |
| 524 | } |
| 525 | // Protect run-state written by the engine since the frontend snapshot was |
| 526 | // loaded: a stale panel save (e.g. toggling enabled) must not clear the |
| 527 | // runHistory that a background execution persisted meanwhile. |
| 528 | tasks = mergeHeartbeatDiskRunHistory(tasks, expected.cfg.Tasks) |
| 529 | if err := e.writeTasks(tasks, expected, true); err != nil { |
| 530 | return err |
| 531 | } |
| 532 | latest, err := e.readConfigSnapshot() |
| 533 | if err != nil { |
| 534 | return err |
| 535 | } |
| 536 | e.recordConfigSnapshotLocked(latest) |
| 537 | e.tasks = tasks |
| 538 | e.prunePendingTopicsLocked(tasks) |
| 539 | return nil |
| 540 | } |
| 541 | |
| 542 | // ReplaceConfig applies a frontend edit only when its revision and ETag still |
| 543 | // identify the exact config the user edited. This prevents a stale panel from |
| 544 | // overwriting an external or second-process change. |
| 545 | func (e *HeartbeatEngine) ReplaceConfig(update HeartbeatConfigUpdate) (HeartbeatConfigView, error) { |
| 546 | e.mu.Lock() |
| 547 | defer e.mu.Unlock() |
| 548 | expected, err := e.readConfigSnapshot() |
| 549 | if err != nil { |
| 550 | return HeartbeatConfigView{}, err |
| 551 | } |
| 552 | if expected.cfg.Revision != update.Revision || expected.view().ETag != update.ETag { |
| 553 | return expected.view(), ErrHeartbeatConfigConflict |
| 554 | } |
| 555 | tasks := mergeHeartbeatDiskRunHistory(update.Tasks, expected.cfg.Tasks) |
| 556 | if err := e.writeTasks(tasks, expected, true); err != nil { |
| 557 | return expected.view(), err |
| 558 | } |
| 559 | latest, err := e.readConfigSnapshot() |
| 560 | if err != nil { |
| 561 | return HeartbeatConfigView{}, err |
| 562 | } |
| 563 | e.recordConfigSnapshotLocked(latest) |
| 564 | e.tasks = latest.cfg.Tasks |
| 565 | e.prunePendingTopicsLocked(e.tasks) |
| 566 | return latest.view(), nil |
| 567 | } |
| 568 | |
| 569 | func (e *HeartbeatEngine) prunePendingTopicsLocked(tasks []HeartbeatTask) { |
| 570 | if len(e.pendingTopics) == 0 { |
| 571 | return |
| 572 | } |
| 573 | keep := make(map[string]bool, len(tasks)) |
| 574 | for _, task := range tasks { |
| 575 | if task.NewConversationEachRun { |
| 576 | keep[task.ID] = true |
| 577 | } |
| 578 | } |
| 579 | for id := range e.pendingTopics { |
| 580 | if !keep[id] { |
| 581 | delete(e.pendingTopics, id) |
| 582 | } |
| 583 | } |
| 584 | } |
| 585 | |
| 586 | // TriggerNow runs a single task immediately by ID. |
| 587 | func (e *HeartbeatEngine) TriggerNow(id string) { |
| 588 | e.mu.Lock() |
| 589 | tasks := append([]HeartbeatTask(nil), e.tasks...) |
| 590 | e.mu.Unlock() |
| 591 | for _, t := range tasks { |
| 592 | if t.ID == id { |
| 593 | e.executeTaskWithLease(t, func(task HeartbeatTask) (HeartbeatTask, bool) { |
| 594 | snapshot, err := e.readConfigSnapshot() |
| 595 | if err != nil { |
| 596 | log.Printf("[heartbeat] cannot revalidate task %q before manual execution: %v", task.Title, err) |
| 597 | return task, false |
| 598 | } |
| 599 | for _, current := range snapshot.cfg.Tasks { |
| 600 | if current.ID == task.ID { |
| 601 | return current, true |
| 602 | } |
| 603 | } |
| 604 | return task, false |
| 605 | }) |
| 606 | return |
| 607 | } |
| 608 | } |
| 609 | } |
| 610 | |
| 611 | // ── Wails bindings on App ─────────────────────────────────────────────────── |
| 612 | |
| 613 | // HeartbeatListTasks returns all heartbeat tasks. |
| 614 | func (a *App) HeartbeatListTasks() []HeartbeatTask { |
| 615 | if a.heartbeat == nil { |
| 616 | return []HeartbeatTask{} |
| 617 | } |
| 618 | return a.heartbeat.ListTasks() |
| 619 | } |
| 620 | |
| 621 | // HeartbeatReloadTasks reloads tasks from disk and returns them. |
| 622 | func (a *App) HeartbeatReloadTasks() []HeartbeatTask { |
| 623 | if a.heartbeat == nil { |
| 624 | return []HeartbeatTask{} |
| 625 | } |
| 626 | return a.heartbeat.ReloadTasks() |
| 627 | } |
| 628 | |
| 629 | // HeartbeatReloadConfig returns tasks with the CAS token used by current UIs. |
| 630 | func (a *App) HeartbeatReloadConfig() HeartbeatConfigView { |
| 631 | if a.heartbeat == nil { |
| 632 | return HeartbeatConfigView{Tasks: []HeartbeatTask{}} |
| 633 | } |
| 634 | return a.heartbeat.ReloadConfig() |
| 635 | } |
| 636 | |
| 637 | // HeartbeatSaveTasks replaces the full task list and persists it. |
| 638 | func (a *App) HeartbeatSaveTasks(tasks []HeartbeatTask) error { |
| 639 | if a.heartbeat == nil { |
| 640 | return nil |
| 641 | } |
| 642 | return a.heartbeat.ReplaceTasks(tasks) |
| 643 | } |
| 644 | |
| 645 | // HeartbeatSaveConfig replaces tasks only when the frontend's revision and |
| 646 | // ETag still match the exact file it loaded. |
| 647 | func (a *App) HeartbeatSaveConfig(update HeartbeatConfigUpdate) (HeartbeatConfigView, error) { |
| 648 | if a.heartbeat == nil { |
| 649 | return HeartbeatConfigView{Tasks: []HeartbeatTask{}}, nil |
| 650 | } |
| 651 | return a.heartbeat.ReplaceConfig(update) |
| 652 | } |
| 653 | |
| 654 | // HeartbeatTriggerNow immediately executes the task with the given ID. |
| 655 | func (a *App) HeartbeatTriggerNow(id string) { |
| 656 | if a.heartbeat == nil { |
| 657 | return |
| 658 | } |
| 659 | a.heartbeat.TriggerNow(id) |
| 660 | } |
| 661 | |
| 662 | // HeartbeatGenerateID returns a random id for new tasks. |
| 663 | func (a *App) HeartbeatGenerateID() string { |
| 664 | const chars = "abcdefghijklmnopqrstuvwxyz0123456789" |
| 665 | b := make([]byte, 12) |
| 666 | for i := range b { |
| 667 | b[i] = chars[rand.Intn(len(chars))] |
| 668 | } |
| 669 | return string(b) |
| 670 | } |
| 671 | |
| 672 | // newBotForwarder builds event forwarding for a heartbeat turn. The caller |
| 673 | // attaches it only after acquiring the tab's turn-admission gate. |
| 674 | func (e *HeartbeatEngine) newBotForwarder(tabID string) event.Sink { |
| 675 | runtime := e.app.botRuntime |
| 676 | if runtime == nil || !runtime.Running() { |
| 677 | return nil |
| 678 | } |
| 679 | cfg, err := e.app.loadDesktopBotConfig() |
| 680 | if err != nil { |
| 681 | log.Printf("[heartbeat] load config for bot forward: %v", err) |
| 682 | return nil |
| 683 | } |
| 684 | targets := runtime.ForwardTargets(cfg) |
| 685 | if len(targets) == 0 { |
| 686 | return nil // no session-mapped channels to forward to |
| 687 | } |
| 688 | tab := e.app.tabByID(tabID) |
| 689 | if tab == nil || tab.sink == nil { |
| 690 | return nil |
| 691 | } |
| 692 | log.Printf("[heartbeat] bot forwarding attached: %d target(s) for tab %s", len(targets), tabID) |
| 693 | return newBotEventForwarder(runtime, targets) |
| 694 | } |
| 695 |