diff --git a/apps/stackpanel-go/internal/agent/server/runtime.go b/apps/stackpanel-go/internal/agent/server/runtime.go new file mode 100644 index 00000000..b95aa729 --- /dev/null +++ b/apps/stackpanel-go/internal/agent/server/runtime.go @@ -0,0 +1,230 @@ +package server + +import ( + "fmt" + "path/filepath" + "sync" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/rs/zerolog/log" + + sharedexec "github.com/darkmatter/stackpanel/stackpanel-go/pkg/exec" + "github.com/darkmatter/stackpanel/stackpanel-go/pkg/nixdata" +) + +// runtimeIdleTimeout is how long a project's runtime is kept after its last +// use before its watchers are stopped. The agent's current project is never +// evicted. +const runtimeIdleTimeout = 10 * time.Minute + +// projectRuntime owns everything bound to one project: the command executor, +// the config store, the shell manager and the file watchers. +// +// The registry creates one per project root, so the legacy current-project +// handlers and the v1 per-request handlers share the same instances instead +// of racing each other with duplicates (ADR 0005). +type projectRuntime struct { + root string + exec *sharedexec.Executor + store *nixdata.Store + shell *ShellManager + + // Watchers run only while the server is serving; see startWatching. + flake *FlakeWatcher + files *fsnotify.Watcher + + lastUsed time.Time // guarded by runtimeRegistry.mu +} + +// runtimeRegistry owns the per-project runtimes. +type runtimeRegistry struct { + mu sync.Mutex + runtimes map[string]*projectRuntime + watching bool // set once the server starts serving + + // Lifecycle hooks, replaced in tests so they need no executor or Nix. + create func(root string) (*projectRuntime, error) + watch func(rt *projectRuntime) error + unwatch func(rt *projectRuntime) + now func() time.Time +} + +func newRuntimeRegistry(s *Server) *runtimeRegistry { + return &runtimeRegistry{ + runtimes: map[string]*projectRuntime{}, + create: s.createRuntime, + watch: s.watchRuntime, + unwatch: unwatchRuntime, + now: time.Now, + } +} + +// get returns the runtime for root, creating it on first use. Runtimes created +// while the server is serving start their watchers immediately. +func (r *runtimeRegistry) get(root string) (*projectRuntime, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if rt, ok := r.runtimes[root]; ok { + rt.lastUsed = r.now() + return rt, nil + } + rt, err := r.create(root) + if err != nil { + return nil, err + } + if r.watching { + r.startWatchers(rt) + } + rt.lastUsed = r.now() + r.runtimes[root] = rt + return rt, nil +} + +// startWatching starts the watchers of existing runtimes and makes get start +// them for runtimes created later. It runs once, when the server starts. +func (r *runtimeRegistry) startWatching() { + r.mu.Lock() + defer r.mu.Unlock() + + if r.watching { + return + } + r.watching = true + for _, rt := range r.runtimes { + r.startWatchers(rt) + } +} + +func (r *runtimeRegistry) startWatchers(rt *projectRuntime) { + if err := r.watch(rt); err != nil { + // The runtime stays usable; only live config updates are lost. + log.Warn().Err(err).Str("project_root", rt.root).Msg("project watchers not started") + } +} + +// evictIdle stops and forgets runtimes unused for longer than idle, except the +// one for keep (the agent's current project). +func (r *runtimeRegistry) evictIdle(idle time.Duration, keep string) { + r.mu.Lock() + defer r.mu.Unlock() + + cutoff := r.now().Add(-idle) + for root, rt := range r.runtimes { + if root == keep || rt.lastUsed.After(cutoff) { + continue + } + r.unwatch(rt) + delete(r.runtimes, root) + log.Info().Str("project_root", root).Msg("evicted idle project runtime") + } +} + +// stopAll stops every runtime's watchers at shutdown. +func (r *runtimeRegistry) stopAll() { + r.mu.Lock() + defer r.mu.Unlock() + + for root, rt := range r.runtimes { + r.unwatch(rt) + delete(r.runtimes, root) + } + r.watching = false +} + +func (s *Server) createRuntime(root string) (*projectRuntime, error) { + exec, err := sharedexec.New(root, s.config.AllowedCommands) + if err != nil { + return nil, fmt.Errorf("failed to create executor: %w", err) + } + log.Info(). + Str("project_root", root). + Bool("in_devshell", exec.InDevshell()). + Bool("has_devshell_env", exec.HasDevshellEnv()). + Msg("Executor initialized with devshell support") + + return &projectRuntime{ + root: root, + exec: exec, + store: nixdata.NewStore(root, exec), + shell: NewShellManager(root, s), + }, nil +} + +// watchRuntime starts live evaluation of the project's flake and the +// config-file watcher that drives config.changed events. +func (s *Server) watchRuntime(rt *projectRuntime) error { + fw, err := NewFlakeWatcher(FlakeWatcherConfig{ProjectRoot: rt.root, Server: s}) + if err != nil { + return fmt.Errorf("create flake watcher: %w", err) + } + if err := fw.Start(); err != nil { + return fmt.Errorf("start flake watcher: %w", err) + } + rt.flake = fw + + files, err := fsnotify.NewWatcher() + if err != nil { + return fmt.Errorf("create config file watcher: %w", err) + } + rt.files = files + go watchConfigFiles(files, rt.root, s.broadcastSSE) + return nil +} + +func unwatchRuntime(rt *projectRuntime) { + if rt.flake != nil { + _ = rt.flake.Stop() + rt.flake = nil + } + if rt.files != nil { + _ = rt.files.Close() + rt.files = nil + } +} + +// watchConfigFiles broadcasts config.changed when files are written in the +// project's .stack state, generated or data directories. Paths derive from +// the project root rather than the agent's environment, which only describes +// the project the agent was started in. It returns when w is closed. +func watchConfigFiles(w *fsnotify.Watcher, root string, broadcast func(SSEEvent)) { + // Runtime state is split between profile/ and state/ until the .stack + // layout is consolidated (ADR 0006); watch whichever exists. + for _, dir := range []string{"profile", "state", "gen", "data"} { + path := filepath.Join(root, ".stack", dir) + if err := w.Add(path); err != nil { + log.Debug().Err(err).Str("path", path).Msg("not watching config directory (may not exist yet)") + } + } + + const debounce = 100 * time.Millisecond + var timer *time.Timer + for { + select { + case event, ok := <-w.Events: + if !ok { + return + } + if event.Op&(fsnotify.Write|fsnotify.Create) == 0 { + continue + } + name := event.Name + if timer != nil { + timer.Stop() + } + timer = time.AfterFunc(debounce, func() { + log.Debug().Str("file", name).Msg("config file changed, broadcasting") + broadcast(SSEEvent{ + Event: "config.changed", + Data: map[string]string{"file": name}, + }) + }) + case err, ok := <-w.Errors: + if !ok { + return + } + log.Warn().Err(err).Msg("file watcher error") + } + } +} diff --git a/apps/stackpanel-go/internal/agent/server/runtime_test.go b/apps/stackpanel-go/internal/agent/server/runtime_test.go new file mode 100644 index 00000000..4c94fc57 --- /dev/null +++ b/apps/stackpanel-go/internal/agent/server/runtime_test.go @@ -0,0 +1,172 @@ +package server + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/fsnotify/fsnotify" +) + +// fakeRuntimes records lifecycle calls so registry behavior can be tested +// without real executors, flake evaluation or file watchers. +type fakeRuntimes struct { + created map[string]int + watched map[string]int + unwatched map[string]int + clock time.Time +} + +func newTestRegistry() (*runtimeRegistry, *fakeRuntimes) { + f := &fakeRuntimes{ + created: map[string]int{}, + watched: map[string]int{}, + unwatched: map[string]int{}, + clock: time.Unix(1_700_000_000, 0), + } + r := &runtimeRegistry{ + runtimes: map[string]*projectRuntime{}, + create: func(root string) (*projectRuntime, error) { + f.created[root]++ + return &projectRuntime{root: root}, nil + }, + watch: func(rt *projectRuntime) error { f.watched[rt.root]++; return nil }, + unwatch: func(rt *projectRuntime) { f.unwatched[rt.root]++ }, + now: func() time.Time { return f.clock }, + } + return r, f +} + +func mustGet(t *testing.T, r *runtimeRegistry, root string) *projectRuntime { + t.Helper() + rt, err := r.get(root) + if err != nil { + t.Fatalf("get(%q): %v", root, err) + } + return rt +} + +func TestRuntimeRegistryReusesOneRuntimePerProject(t *testing.T) { + r, f := newTestRegistry() + + a1 := mustGet(t, r, "/p/a") + a2 := mustGet(t, r, "/p/a") + b := mustGet(t, r, "/p/b") + + if a1 != a2 { + t.Error("the same project returned different runtimes") + } + if a1 == b { + t.Error("different projects shared a runtime") + } + if f.created["/p/a"] != 1 || f.created["/p/b"] != 1 { + t.Errorf("created = %v, want one per project", f.created) + } +} + +func TestRuntimeRegistryStartsWatchersOnceAndOnlyWhileServing(t *testing.T) { + r, f := newTestRegistry() + + mustGet(t, r, "/p/a") + if f.watched["/p/a"] != 0 { + t.Fatal("watchers started before the server was serving") + } + + r.startWatching() + r.startWatching() + mustGet(t, r, "/p/a") // switching back to a known project + mustGet(t, r, "/p/b") // a project first used while serving + + if f.watched["/p/a"] != 1 || f.watched["/p/b"] != 1 { + t.Errorf("watched = %v, want exactly once per project", f.watched) + } +} + +func TestRuntimeRegistryEvictsIdleRuntimesExceptCurrent(t *testing.T) { + r, f := newTestRegistry() + r.startWatching() + + mustGet(t, r, "/p/current") + mustGet(t, r, "/p/idle") + f.clock = f.clock.Add(runtimeIdleTimeout + time.Second) + mustGet(t, r, "/p/recent") + + r.evictIdle(runtimeIdleTimeout, "/p/current") + + if f.unwatched["/p/idle"] != 1 { + t.Errorf("idle runtime not stopped: unwatched = %v", f.unwatched) + } + if f.unwatched["/p/current"] != 0 || f.unwatched["/p/recent"] != 0 { + t.Errorf("stopped a runtime that should stay: unwatched = %v", f.unwatched) + } + + // An evicted project gets a fresh runtime (and watchers) when used again. + mustGet(t, r, "/p/idle") + if f.created["/p/idle"] != 2 || f.watched["/p/idle"] != 2 { + t.Errorf("re-used evicted project: created=%d watched=%d, want 2 and 2", + f.created["/p/idle"], f.watched["/p/idle"]) + } +} + +func TestRuntimeRegistryStopAllStopsEveryRuntime(t *testing.T) { + r, f := newTestRegistry() + r.startWatching() + mustGet(t, r, "/p/a") + mustGet(t, r, "/p/b") + + r.stopAll() + + if f.unwatched["/p/a"] != 1 || f.unwatched["/p/b"] != 1 { + t.Errorf("unwatched = %v, want every runtime stopped", f.unwatched) + } + if len(r.runtimes) != 0 { + t.Errorf("%d runtimes left after stopAll", len(r.runtimes)) + } +} + +func TestWatchConfigFilesBroadcastsChangesUnderProjectRoot(t *testing.T) { + root := t.TempDir() + dataDir := filepath.Join(root, ".stack", "data") + if err := os.MkdirAll(dataDir, 0o755); err != nil { + t.Fatal(err) + } + + w, err := fsnotify.NewWatcher() + if err != nil { + t.Fatal(err) + } + events := make(chan SSEEvent, 4) + done := make(chan struct{}) + go func() { + watchConfigFiles(w, root, func(e SSEEvent) { events <- e }) + close(done) + }() + + // Let the watcher register its directories before writing. + time.Sleep(50 * time.Millisecond) + changed := filepath.Join(dataDir, "packages.json") + if err := os.WriteFile(changed, []byte("[]"), 0o644); err != nil { + t.Fatal(err) + } + + select { + case e := <-events: + if e.Event != "config.changed" { + t.Errorf("event = %q, want config.changed", e.Event) + } + data, _ := e.Data.(map[string]string) + if got := data["file"]; filepath.Base(got) != "packages.json" { + t.Errorf("file = %q, want the changed file", got) + } + case <-time.After(3 * time.Second): + t.Fatal("no config.changed event for a write under .stack/data") + } + + _ = w.Close() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("watchConfigFiles did not return after the watcher was closed") + } +} diff --git a/apps/stackpanel-go/internal/agent/server/server.go b/apps/stackpanel-go/internal/agent/server/server.go index c3269486..a536f079 100644 --- a/apps/stackpanel-go/internal/agent/server/server.go +++ b/apps/stackpanel-go/internal/agent/server/server.go @@ -13,13 +13,13 @@ import ( "html/template" "net/http" "sync" + "time" "github.com/darkmatter/stackpanel/packages/proto/gen/gopb/gopbconnect" "github.com/darkmatter/stackpanel/stackpanel-go/internal/agent/config" "github.com/darkmatter/stackpanel/stackpanel-go/internal/agent/project" sharedexec "github.com/darkmatter/stackpanel/stackpanel-go/pkg/exec" "github.com/darkmatter/stackpanel/stackpanel-go/pkg/nixdata" - "github.com/fsnotify/fsnotify" "github.com/rs/zerolog/log" ) @@ -45,9 +45,11 @@ type Server struct { sseSubscribers map[chan SSEEvent]struct{} sseSubscribersMu sync.RWMutex - // watcher monitors .stack/state, .stack/gen, and .stack/data for filesystem - // changes, broadcasting "config.changed" SSE events to trigger UI refreshes. - watcher *fsnotify.Watcher + // runtimes owns every project's executor, store, shell manager and file + // watchers. exec, store, flakeWatcher and shellManager are aliases for the + // current project's runtime, kept for handlers that don't yet resolve + // their project per request (ADR 0005). + runtimes *runtimeRegistry // flakeWatcher runs `nix eval` on .nix file changes to keep the cached // stackpanelConfig and stackpanelPackages up to date without a full rebuild. @@ -56,6 +58,9 @@ type Server struct { // shellManager tracks whether the devshell is stale (nix files changed // since last build) and handles rebuild orchestration. shellManager *ShellManager + + done chan struct{} // closed by Stop + stopOnce sync.Once } // New creates a new server instance, wiring up project detection, JWT auth, @@ -115,29 +120,11 @@ func New(cfg *config.Config) (*Server, error) { } } - // Initialize executor (may have empty project root if no project selected) - var exec *sharedexec.Executor - if cfg.ProjectRoot != "" { - exec, err = sharedexec.New(cfg.ProjectRoot, cfg.AllowedCommands) - if err != nil { - return nil, fmt.Errorf("failed to create executor: %w", err) - } - log.Info(). - Bool("in_devshell", exec.InDevshell()). - Bool("has_devshell_env", exec.HasDevshellEnv()). - Msg("Executor initialized with devshell support") - } - pairTmpl, err := template.ParseFS(templatesFS, "templates/pair.html") if err != nil { return nil, fmt.Errorf("failed to parse pair template: %w", err) } - watcher, err := fsnotify.NewWatcher() - if err != nil { - return nil, fmt.Errorf("failed to create file watcher: %w", err) - } - // Initialize JWT manager for token auth jwtMgr, err := NewJWTManagerWithOptions(JWTManagerOptions{ TestPairingToken: cfg.TestPairingToken, @@ -146,27 +133,21 @@ func New(cfg *config.Config) (*Server, error) { return nil, fmt.Errorf("failed to create JWT manager: %w", err) } - // Initialize the shared Nix data store (nil-safe if no executor yet) - var store *nixdata.Store - if exec != nil && cfg.ProjectRoot != "" { - store = nixdata.NewStore(cfg.ProjectRoot, exec) - } - s := &Server{ config: cfg, httpServer: nil, - exec: exec, - store: store, projectMgr: projectMgr, jwtManager: jwtMgr, pairTemplate: pairTmpl, sseSubscribers: make(map[chan SSEEvent]struct{}), - watcher: watcher, + done: make(chan struct{}), } + s.runtimes = newRuntimeRegistry(s) - // Initialize shell manager for tracking devshell state - if cfg.ProjectRoot != "" { - s.shellManager = NewShellManager(cfg.ProjectRoot, s) + // Create the current project's runtime (executor, store, shell manager); + // its watchers start with the server. + if err := s.reinitializeExecutor(); err != nil { + return nil, err } mux := http.NewServeMux() @@ -503,45 +484,42 @@ func New(cfg *config.Config) (*Server, error) { return s, nil } -// Start begins serving requests and starts the config file watcher. +// Start starts the project runtimes' watchers and begins serving requests. func (s *Server) Start() error { - // Start watching config files for changes (only if we have a project) - if s.config.ProjectRoot != "" { - go s.watchConfigFiles() - - // Start the FlakeWatcher for live .#stackpanelConfig and .#stackpanelPackages evaluation - fw, err := NewFlakeWatcher(FlakeWatcherConfig{ - ProjectRoot: s.config.ProjectRoot, - Server: s, - }) - if err != nil { - log.Warn(). - Err(err). - Msg("Failed to create FlakeWatcher, config/packages watching disabled") - } else { - s.flakeWatcher = fw - if err := fw.Start(); err != nil { - log.Warn().Err(err).Msg("Failed to start FlakeWatcher") - } - } + s.runtimes.startWatching() + // Refresh the current-project aliases now that its watchers exist. + if err := s.reinitializeExecutor(); err != nil { + log.Warn().Err(err).Msg("Failed to prepare the current project's runtime") } + go s.evictIdleRuntimes() return s.httpServer.ListenAndServe() } // Stop gracefully shuts down the server. func (s *Server) Stop() { - if s.flakeWatcher != nil { - _ = s.flakeWatcher.Stop() - } - if s.watcher != nil { - _ = s.watcher.Close() - } + s.stopOnce.Do(func() { close(s.done) }) + s.runtimes.stopAll() if s.httpServer != nil { _ = s.httpServer.Close() } } +// evictIdleRuntimes periodically stops the watchers of projects that are no +// longer used, keeping the current project's runtime. +func (s *Server) evictIdleRuntimes() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case <-ticker.C: + s.runtimes.evictIdle(runtimeIdleTimeout, s.config.ProjectRoot) + case <-s.done: + return + } + } +} + // requireProject is middleware that ensures a project is open before handling the request. // Returns 428 Precondition Required so the web UI can detect "no project" state // and show the project picker instead of a generic error. @@ -558,49 +536,20 @@ func (s *Server) requireProject(next http.HandlerFunc) http.HandlerFunc { } } -// reinitializeExecutor tears down and recreates the executor, store, shell manager, -// and all file watchers for the current project root. Called when switching projects. +// reinitializeExecutor points the current-project aliases (exec, store, +// shellManager, flakeWatcher) at the registry's runtime for +// s.config.ProjectRoot, creating it on first use. Switching back to a recently +// used project reuses its runtime rather than starting duplicate watchers. func (s *Server) reinitializeExecutor() error { if s.config.ProjectRoot == "" { - s.exec = nil - s.shellManager = nil + s.exec, s.store, s.shellManager, s.flakeWatcher = nil, nil, nil, nil return nil } - exec, err := sharedexec.New(s.config.ProjectRoot, s.config.AllowedCommands) - if err != nil { - return fmt.Errorf("failed to create executor: %w", err) - } - - s.exec = exec - s.store = nixdata.NewStore(s.config.ProjectRoot, exec) - s.shellManager = NewShellManager(s.config.ProjectRoot, s) - - log.Info(). - Str("project_root", s.config.ProjectRoot). - Bool("in_devshell", exec.InDevshell()). - Bool("has_devshell_env", exec.HasDevshellEnv()). - Msg("Executor reinitialized for new project") - - // Restart file watcher for new project - go s.watchConfigFiles() - - // Restart FlakeWatcher for new project - if s.flakeWatcher != nil { - _ = s.flakeWatcher.Stop() - } - fw, err := NewFlakeWatcher(FlakeWatcherConfig{ - ProjectRoot: s.config.ProjectRoot, - Server: s, - }) + rt, err := s.runtimes.get(s.config.ProjectRoot) if err != nil { - log.Warn().Err(err).Msg("Failed to create FlakeWatcher for new project") - } else { - s.flakeWatcher = fw - if err := fw.Start(); err != nil { - log.Warn().Err(err).Msg("Failed to start FlakeWatcher for new project") - } + return err } - + s.exec, s.store, s.shellManager, s.flakeWatcher = rt.exec, rt.store, rt.shell, rt.flake return nil } diff --git a/apps/stackpanel-go/internal/agent/server/sse.go b/apps/stackpanel-go/internal/agent/server/sse.go index 8335a904..270e4089 100644 --- a/apps/stackpanel-go/internal/agent/server/sse.go +++ b/apps/stackpanel-go/internal/agent/server/sse.go @@ -6,12 +6,9 @@ import ( "encoding/json" "fmt" "net/http" - "path/filepath" "time" "github.com/darkmatter/stackpanel/stackpanel-go/internal/setupsession" - envvars "github.com/darkmatter/stackpanel/stackpanel-go/pkg/envvars" - "github.com/fsnotify/fsnotify" "github.com/rs/zerolog/log" ) @@ -28,63 +25,6 @@ type SSEEvent struct { Data any `json:"data"` } -// watchConfigFiles uses fsnotify to watch .stack/state, .stack/gen, and .stack/data -// for changes, broadcasting "config.changed" SSE events. This is separate from -// FlakeWatcher — it handles simple file change notifications without re-evaluating Nix. -// Changes are debounced at 100ms to coalesce rapid file writes (e.g., during codegen). -func (s *Server) watchConfigFiles() { - stateFile := envvars.StackpanelStateFile.Get() - if err := s.watcher.Add(filepath.Dir(stateFile)); err != nil { - log.Warn().Err(err).Str("path", stateFile).Msg("failed to watch state directory") - } - - genDir := envvars.StackpanelGenDir.Get() - if err := s.watcher.Add(genDir); err != nil { - log.Warn().Err(err).Str("path", genDir).Msg("failed to watch gen directory") - } - - dataDir := filepath.Join(s.config.ProjectRoot, ".stack", "data") - if err := s.watcher.Add(dataDir); err != nil { - log.Debug(). - Err(err). - Str("path", dataDir). - Msg("failed to watch data directory (may not exist yet)") - } - - var debounceTimer *time.Timer - debounceDuration := 100 * time.Millisecond - - for { - select { - case event, ok := <-s.watcher.Events: - if !ok { - return - } - if event.Op&(fsnotify.Write|fsnotify.Create) != 0 { - name := event.Name - // Debounce: reset timer on each event - if debounceTimer != nil { - debounceTimer.Stop() - } - debounceTimer = time.AfterFunc(debounceDuration, func() { - log.Debug().Str("file", name).Msg("config file changed, broadcasting") - s.broadcastSSE(SSEEvent{ - Event: "config.changed", - Data: map[string]string{ - "file": name, - }, - }) - }) - } - case err, ok := <-s.watcher.Errors: - if !ok { - return - } - log.Warn().Err(err).Msg("file watcher error") - } - } -} - // handleSSE handles Server-Sent Events connections for real-time updates. // The web UI's AgentSSEProvider maintains a persistent connection to this endpoint, // using received events to invalidate TanStack Query caches and trigger refetches.