fix(platform): panic-recovering supervisor for every background goroutine (#92)

platform/cmd/server/main.go

@@ -20,6 +20,7 @@ import (
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/registry"
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/router"
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/scheduler"
+	"github.com/Molecule-AI/molecule-monorepo/platform/internal/supervised"
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/ws"
 )
 
@@ -67,7 +68,12 @@ func main() {
 	// Start Redis pub/sub subscriber
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go broadcaster.Subscribe(ctx)
+	// Every long-running subsystem below is wrapped by supervised.RunWithRecover:
+	// a panic (e.g. from a single bad tenant row) is logged + the subsystem is
+	// restarted with exponential backoff instead of silently dying forever.
+	// Motivation: issue #85 (scheduler silent outage for 12+ hours) and #92
+	// (systemic — affects every background goroutine).
+	go supervised.RunWithRecover(ctx, "broadcaster", broadcaster.Subscribe)
 
 	// Activity log retention — configurable via env vars
 	retentionDays := envOr("ACTIVITY_RETENTION_DAYS", "7")
@@ -122,21 +128,25 @@ func main() {
 	}
 
 	// Start Liveness Monitor — Redis TTL expiry-based offline detection + auto-restart
-	go registry.StartLivenessMonitor(ctx, onWorkspaceOffline)
+	go supervised.RunWithRecover(ctx, "liveness-monitor", func(c context.Context) {
+		registry.StartLivenessMonitor(c, onWorkspaceOffline)
+	})
 
 	// Proactive container health sweep — detects dead containers faster than Redis TTL.
 	// Checks all "online" workspaces against Docker every 15 seconds.
 	if prov != nil {
-		go registry.StartHealthSweep(ctx, prov, 15*time.Second, onWorkspaceOffline)
+		go supervised.RunWithRecover(ctx, "health-sweep", func(c context.Context) {
+			registry.StartHealthSweep(c, prov, 15*time.Second, onWorkspaceOffline)
+		})
 	}
 
 	// Cron Scheduler — fires A2A messages to workspaces on user-defined schedules
 	cronSched := scheduler.New(wh, broadcaster)
-	go cronSched.Start(ctx)
+	go supervised.RunWithRecover(ctx, "scheduler", cronSched.Start)
 
 	// Channel Manager — social channel integrations (Telegram, Slack, etc.)
 	channelMgr := channels.NewManager(wh, broadcaster)
-	go channelMgr.Start(ctx)
+	go supervised.RunWithRecover(ctx, "channel-manager", channelMgr.Start)
 
 	// Router
 	r := router.Setup(hub, broadcaster, prov, platformURL, configsDir, wh, channelMgr)

platform/internal/router/router.go

@@ -15,6 +15,7 @@ import (
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/metrics"
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/middleware"
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/provisioner"
+	"github.com/Molecule-AI/molecule-monorepo/platform/internal/supervised"
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/ws"
 	"github.com/docker/docker/client"
 	"github.com/gin-contrib/cors"
@@ -63,6 +64,23 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
 		c.JSON(200, gin.H{"status": "ok"})
 	})
 
+	// /admin/liveness — per-subsystem last-tick timestamps. Operators read this
+	// to catch stuck-but-not-crashed goroutines (the failure mode that caused
+	// the 12h scheduler outage of 2026-04-14, issue #85). Any subsystem whose
+	// last tick is older than 2× its expected interval is stale.
+	r.GET("/admin/liveness", func(c *gin.Context) {
+		snap := supervised.Snapshot()
+		out := make(map[string]interface{}, len(snap))
+		now := time.Now()
+		for name, last := range snap {
+			out[name] = gin.H{
+				"last_tick_at":    last,
+				"seconds_ago":     int(now.Sub(last).Seconds()),
+			}
+		}
+		c.JSON(200, gin.H{"subsystems": out})
+	})
+
 	// Prometheus metrics — exempt from rate limiter via separate registration
 	// (registered before Use(limiter) takes effect on this specific route — the
 	// middleware.Middleware() still records it for observability).

platform/internal/scheduler/scheduler.go

@@ -12,6 +12,7 @@ import (
 	cronlib "github.com/robfig/cron/v3"
 
 	"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
+	"github.com/Molecule-AI/molecule-monorepo/platform/internal/supervised"
 )
 
 const (
@@ -105,8 +106,9 @@ func (s *Scheduler) Start(ctx context.Context) {
 		s.mu.Unlock()
 	}
 
-	// Mark a tick immediately on startup so Healthy() returns true before
-	// the first ticker fires (avoids spurious unhealthy on fresh start).
+	// Heartbeat + initial lastTickAt so /admin/liveness and Healthy() both
+	// pass during the first 30s interval after startup.
+	supervised.Heartbeat("scheduler")
 	s.mu.Lock()
 	s.lastTickAt = time.Now()
 	s.mu.Unlock()
@@ -118,14 +120,24 @@ func (s *Scheduler) Start(ctx context.Context) {
 			return
 		case <-ticker.C:
 			tickWithRecover()
+			supervised.Heartbeat("scheduler")
 		}
 	}
 }
 
 // tick queries all due schedules and fires each in a goroutine.
 // Waits for all goroutines to finish before returning so the next tick
 // doesn't re-fire schedules whose next_run_at hasn't been updated yet.
+//
+// Heartbeat is called at three points to keep /admin/liveness fresh during
+// long-running fires (some prompts take minutes — without these heartbeats
+// the scheduler looks "stale" the whole time it's working):
+//   - immediately on entering tick (proves we're past the ticker.C wait)
+//   - inside each per-fire goroutine (every fire bumps the heartbeat)
+//   - implicitly via the post-tick heartbeat in Start()
 func (s *Scheduler) tick(ctx context.Context) {
+	supervised.Heartbeat("scheduler")
+
 	rows, err := db.DB.QueryContext(ctx, `
 		SELECT id, workspace_id, name, cron_expr, timezone, prompt
 		FROM workspace_schedules
@@ -158,7 +170,9 @@ func (s *Scheduler) tick(ctx context.Context) {
 						s2.Name, s2.WorkspaceID, r)
 				}
 			}()
+			supervised.Heartbeat("scheduler")
 			s.fireSchedule(ctx, s2)
+			supervised.Heartbeat("scheduler")
 		}(sched)
 	}
 	if err := rows.Err(); err != nil {

platform/internal/supervised/supervised.go

@@ -0,0 +1,142 @@
+// Package supervised provides a panic-recovering supervisor for long-running
+// background goroutines on the platform. Every "go X.Start(ctx)" invocation
+// in main.go should go through [RunWithRecover] so a single panic from one
+// tenant's data cannot silently kill a subsystem that serves every tenant.
+//
+// Incident that motivated this (issue #85, 2026-04-14):
+//
+//   The cron scheduler goroutine died silently at 14:21 UTC and stayed dead
+//   for 12+ hours. Platform restart didn't recover it. Root cause: no
+//   defer recover() in the tick loop. Observable signals (HTTP 200, container
+//   healthy, DB healthy) all stayed green — only the subsystem was dead.
+//
+// In a multi-tenant SaaS deployment the blast radius is every tenant
+// simultaneously, which is exactly the class of failure we cannot afford.
+package supervised
+
+import (
+	"context"
+	"log"
+	"runtime/debug"
+	"sync"
+	"time"
+)
+
+// Default backoff bounds for RunWithRecover restarts.
+const (
+	initialBackoff = 1 * time.Second
+	maxBackoff     = 30 * time.Second
+)
+
+// RunWithRecover runs fn in a recover wrapper. If fn panics, the panic is
+// logged with its stack trace and fn is restarted after an exponential
+// backoff (capped at maxBackoff). The loop exits cleanly when ctx is done.
+//
+// fn is expected to be a long-running loop (e.g. "for { select { ticker ... } }").
+// If fn returns without panicking (e.g. ctx.Done), RunWithRecover returns.
+//
+//	go supervised.RunWithRecover(ctx, "scheduler", func(c context.Context) {
+//	    scheduler.Start(c)
+//	})
+//
+// name is used in log lines and by the liveness registry below.
+func RunWithRecover(ctx context.Context, name string, fn func(context.Context)) {
+	backoff := initialBackoff
+	for {
+		select {
+		case <-ctx.Done():
+			log.Printf("supervised[%s]: context done; stopping", name)
+			return
+		default:
+		}
+
+		panicked := runOnce(ctx, name, fn)
+
+		// Clean return → the goroutine decided to stop (likely ctx.Done inside fn).
+		// Don't restart.
+		if !panicked {
+			log.Printf("supervised[%s]: returned cleanly; not restarting", name)
+			return
+		}
+
+		// Panic → back off and restart.
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(backoff):
+		}
+		if backoff < maxBackoff {
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+		}
+	}
+}
+
+// runOnce invokes fn with recover. Returns true iff fn panicked.
+func runOnce(ctx context.Context, name string, fn func(context.Context)) (panicked bool) {
+	defer func() {
+		if r := recover(); r != nil {
+			panicked = true
+			log.Printf("supervised[%s]: PANIC recovered: %v\n%s", name, r, debug.Stack())
+		}
+	}()
+	fn(ctx)
+	return false
+}
+
+// --- Liveness registry -----------------------------------------------------
+//
+// Each subsystem calls Heartbeat(name) at the end of each tick / iteration.
+// Operators read the registry via /admin/liveness to detect stuck-but-not-
+// crashed subsystems (e.g. a tick that deadlocks without panicking).
+
+var (
+	livenessMu sync.RWMutex
+	lastTicks  = map[string]time.Time{}
+)
+
+// Heartbeat records that subsystem `name` is alive as of now.
+func Heartbeat(name string) {
+	livenessMu.Lock()
+	lastTicks[name] = time.Now()
+	livenessMu.Unlock()
+}
+
+// LastTick returns the wall-clock time of the most recent Heartbeat for
+// subsystem `name`. Returns the zero time if the subsystem has never
+// heartbeated.
+func LastTick(name string) time.Time {
+	livenessMu.RLock()
+	defer livenessMu.RUnlock()
+	return lastTicks[name]
+}
+
+// Snapshot returns a copy of every subsystem's last-tick time, for admin
+// endpoints.
+func Snapshot() map[string]time.Time {
+	livenessMu.RLock()
+	defer livenessMu.RUnlock()
+	out := make(map[string]time.Time, len(lastTicks))
+	for k, v := range lastTicks {
+		out[k] = v
+	}
+	return out
+}
+
+// IsHealthy returns true iff every subsystem in `expected` has tickled
+// within `staleThreshold` ago. Use from /health (or a strict variant of it)
+// to surface stuck subsystems to an external orchestrator.
+func IsHealthy(expected []string, staleThreshold time.Duration) (healthy bool, stale []string) {
+	livenessMu.RLock()
+	defer livenessMu.RUnlock()
+	now := time.Now()
+	for _, name := range expected {
+		last, ok := lastTicks[name]
+		if !ok || now.Sub(last) > staleThreshold {
+			stale = append(stale, name)
+		}
+	}
+	return len(stale) == 0, stale
+}