From 2f742e0dd250735c728f9f29d9df08c903b98de0 Mon Sep 17 00:00:00 2001 From: cooper Date: Wed, 23 Sep 2026 06:31:00 -0700 Subject: [PATCH] refactor(agent): give each project one runtime owned by a registry The agent rebuilt its executor, store, shell manager and watchers on every project switch, and each switch started another goroutine reading the same fsnotify channel. The config watcher also took its state and gen directories from the agent's environment, which only describes the project the agent was started in. A runtime registry now owns one runtime per project root. The legacy current-project fields alias the current runtime, so switching back reuses it instead of starting duplicate watchers. Each runtime has its own config watcher with paths derived from its root, other projects' runtimes are evicted after ten idle minutes, and shutdown stops them all. Watchers still start only once the server is serving. This is the ownership model per-request project selection (ADR 0005) builds on: v1 handlers will resolve a project's runtime from the same registry. Refs: stackpanel-thq.8.1 --- .../internal/agent/server/runtime.go | 230 ++++++++++++++++++ .../internal/agent/server/runtime_test.go | 172 +++++++++++++ .../internal/agent/server/server.go | 143 ++++------- .../internal/agent/server/sse.go | 60 ----- 4 files changed, 448 insertions(+), 157 deletions(-) create mode 100644 apps/stackpanel-go/internal/agent/server/runtime.go create mode 100644 apps/stackpanel-go/internal/agent/server/runtime_test.go 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.