feat(api): add shared node-state stream foundation (#746)

* feat(api): add shared node-state stream

* fix(desktop): decode streamed state domains

apps/sisyphus/dev/shared-state-domain.test.ts

@@ -0,0 +1,11 @@
+import assert from 'node:assert/strict'
+import { test } from 'node:test'
+import { sharedStateDomainCode } from '../src/renderer/src/lib/shared-state-domain.ts'
+
+test('shared-state domains accept grpc-js enum names and browser numeric values', () => {
+  assert.equal(sharedStateDomainCode('SHARED_STATE_DOMAIN_WORKERS'), 'workers')
+  assert.equal(sharedStateDomainCode('SHARED_STATE_DOMAIN_JOINED_POOL'), 'joinedPool')
+  assert.equal(sharedStateDomainCode(3), 'workers')
+  assert.equal(sharedStateDomainCode(5), 'joinedPool')
+  assert.equal(sharedStateDomainCode('SHARED_STATE_DOMAIN_PEERS'), undefined)
+})

apps/sisyphus/src/main/index.ts

@@ -310,7 +310,7 @@ const unaryRpcMethods = new Set([
   'getModelConfig', 'setModelConfig', 'listModelProfiles', 'saveModelProfile', 'deleteModelProfile', 'listProviders', 'listModels', 'removeModel', 'getWebAccessConfig', 'setWebAccessConfig', 'listNetworkInterfaces', 'getModelLending', 'setModelLending', 'inspectOllama', 'getLocationDiscoverySettings', 'setLocationDiscoverySettings', 'getNodeLocation', 'setNodeLocation', 'getCityTable', 'downloadCityTable', 'removeCityTable', 'getPoolLimits', 'setPoolLimits', 'listChats', 'getChat', 'deleteChat',
   'listFiles', 'removeFile', 'createInvitation', 'listMembers', 'removeMember', 'joinPool', 'getJoinedPool', 'leavePool', 'resolveActionConfirmation',
 ])
-const streamingRpcMethods = new Set(['watchJobs', 'watchJobEvents', 'watchPairedBrowsers', 'watchPendingActionConfirmations', 'watchWorkers', 'watchJoinedPool', 'ask', 'pullModel'])
+const streamingRpcMethods = new Set(['watchJobs', 'watchJobEvents', 'watchPairedBrowsers', 'watchPendingActionConfirmations', 'watchWorkers', 'watchJoinedPool', 'watchSharedState', 'ask', 'pullModel'])
 
 function callUnary(method: string, request: Record<string, unknown>): Promise<unknown> {
   return new Promise((resolve, reject) => {

apps/sisyphus/src/preload/index.ts

@@ -80,7 +80,7 @@ export type NodeRpcMethod =
   | 'listFiles' | 'removeFile'
   | 'createInvitation' | 'listMembers' | 'removeMember' | 'joinPool' | 'resolveActionConfirmation'
 
-export type NodeStreamMethod = 'watchJobs' | 'watchJobEvents' | 'watchPairedBrowsers' | 'watchPendingActionConfirmations' | 'watchWorkers' | 'watchJoinedPool' | 'ask' | 'pullModel'
+export type NodeStreamMethod = 'watchJobs' | 'watchJobEvents' | 'watchPairedBrowsers' | 'watchPendingActionConfirmations' | 'watchWorkers' | 'watchJoinedPool' | 'watchSharedState' | 'ask' | 'pullModel'
 
 const api = {
   getSnapshot: (): Promise<NodeSnapshot> => ipcRenderer.invoke('node:get-snapshot'),

apps/sisyphus/src/renderer/src/lib/shared-state-domain.ts

@@ -0,0 +1,13 @@
+// grpc-js with proto-loader's `enums: String` option returns the protobuf
+// enum name, while browser Connect clients return the numeric enum value.
+export type SharedStateKind = 'workers' | 'joinedPool'
+
+export function sharedStateDomainCode(domain: number | string): SharedStateKind | undefined {
+  switch (domain) {
+    case 3:
+    case 'SHARED_STATE_DOMAIN_WORKERS': return 'workers'
+    case 5:
+    case 'SHARED_STATE_DOMAIN_JOINED_POOL': return 'joinedPool'
+    default: return undefined
+  }
+}

apps/sisyphus/src/renderer/src/pages/NodeDetailsPage.tsx

@@ -15,6 +15,8 @@ import { displayDaemonVersion } from '@/lib/daemon-version'
 import { ResponsiveActionDialog } from '@/components/ui/responsive-action-dialog'
 import { PoolHostingOverview } from '@/components/pool-hosting-overview'
 import { offeredWorkloads } from '@/lib/worker-workloads'
+import { SharedStateDomain } from '@sisyphus/client'
+import { sharedStateDomainCode } from '@/lib/shared-state-domain'
 
 type Messages = ReturnType<typeof getMessages>
 function shortId(id: string) { return id.length > 22 ? `${id.slice(0, 12)}…${id.slice(-8)}` : id }
@@ -34,6 +36,20 @@ function formatBytes(value: string | number) {
   return `${(bytes / 1024 ** power).toFixed(power ? 1 : 0)} ${units[power]}`
 }
 
+type SharedStatePacket = {
+  domain: number | string
+  error?: string
+  state?: { case?: string; value?: unknown } | string
+  workers?: unknown
+  joinedPool?: JoinedPool
+}
+
+function sharedStatePayload<T>(event: SharedStatePacket, key: 'workers' | 'joinedPool'): T | undefined {
+  if (event.state && typeof event.state === 'object' && event.state.case === key) return event.state.value as T
+  const direct = event[key]
+  return direct === undefined ? undefined : direct as T
+}
+
 export function NodeDetailsPage({ snapshot, messages }: { snapshot: NodeSnapshot; messages: Messages }) {
   const [refreshing, setRefreshing] = useState(false)
   const [peerAddress, setPeerAddress] = useState('')
@@ -49,80 +65,69 @@ export function NodeDetailsPage({ snapshot, messages }: { snapshot: NodeSnapshot
   const connectedPeers = useMemo(() => snapshot.peers.filter(peerIsConnected).length, [snapshot.peers])
   const statusLabel = snapshot.status === 'connected' ? messages.connected : snapshot.status === 'connecting' ? messages.connecting : messages.offline
   useEffect(() => {
-    if (snapshot.status !== 'connected') { setWorkers([]); return }
+    if (snapshot.status !== 'connected') {
+      setWorkers([])
+      setWorkersState('loading')
+      setJoinedPool(null)
+      setJoinedPoolState('loading')
+      return
+    }
     let active = true
     let cancel: (() => void) | undefined
     let retry: number | undefined
     let attempt = 0
     setWorkersState('loading')
+    setJoinedPoolState('loading')
     const connect = () => {
       if (!active) return
-      cancel = getNodeApi().stream<{ workers?: typeof workers }>('watchWorkers', {}, (result) => {
+      cancel = getNodeApi().stream<SharedStatePacket>('watchSharedState', {
+        domains: [SharedStateDomain.WORKERS, SharedStateDomain.JOINED_POOL],
+      }, (event) => {
         attempt = 0
-        setWorkers(result.workers ?? [])
+        const domain = sharedStateDomainCode(event.domain)
-        setWorkersState('ready')
+        if (domain === 'workers') {
-      }, () => {
+          if (event.error) {
+            setWorkersState('error')
+          } else {
+            setWorkers(sharedStatePayload<{ workers?: typeof workers }>(event, 'workers')?.workers ?? [])
+            setWorkersState('ready')
+          }
+        } else if (domain === 'joinedPool') {
+          if (event.error) {
+            if (/has joined no other/i.test(event.error)) {
+              setJoinedPool(null)
+              setJoinedPoolState('none')
+            } else {
+              setJoinedPoolState('unavailable')
+            }
+          } else {
+            const pool = sharedStatePayload<JoinedPool>(event, 'joinedPool')
+            if (!pool || (!pool.coordinatorId && !pool.coordinatorName && !pool.address)) {
+              setJoinedPool(null)
+              setJoinedPoolState('none')
+            } else {
+              setJoinedPool(pool)
+              setJoinedPoolState('ready')
+            }
+          }
+        }
+      }, (error) => {
         cancel = undefined
         if (active) {
           setWorkersState('error')
-          retry = window.setTimeout(connect, Math.min(30_000, 500 * (2 ** Math.min(attempt++, 6))))
-        }
-      }, () => {
-        cancel = undefined
-        if (active) retry = window.setTimeout(connect, Math.min(30_000, 500 * (2 ** Math.min(attempt++, 6))))
-      })
-    }
-    connect()
-    return () => {
-      active = false
-      if (retry !== undefined) window.clearTimeout(retry)
-      cancel?.()
-    }
-  }, [snapshot.status, snapshot.endpoint])
-  useEffect(() => {
-    if (snapshot.status !== 'connected') { setJoinedPoolState((current) => current === 'ready' ? 'unavailable' : current); return }
-    let active = true
-    let cancel: (() => void) | undefined
-    let retry: number | undefined
-    let attempt = 0
-    const connect = () => {
-      if (!active) return
-      cancel = getNodeApi().stream<JoinedPool>('watchJoinedPool', {}, (event) => {
-        attempt = 0
-        setJoinedPool(event)
-        setJoinedPoolState('ready')
-      }, (error, code) => {
-        cancel = undefined
-        if (active) {
-          if (code === 9 || /has joined no other/i.test(error)) { setJoinedPool(null); setJoinedPoolState('none'); return }
           setJoinedPoolState('unavailable')
           retry = window.setTimeout(connect, Math.min(30_000, 500 * (2 ** Math.min(attempt++, 6))))
         }
       }, () => {
         cancel = undefined
-        if (active) retry = window.setTimeout(connect, Math.min(30_000, 500 * (2 ** Math.min(attempt++, 6))))
+        if (active) {
+          setWorkersState('error')
+          setJoinedPoolState('unavailable')
+          retry = window.setTimeout(connect, Math.min(30_000, 500 * (2 ** Math.min(attempt++, 6))))
+        }
       })
     }
-    // Fetch the baseline before subscribing. WatchJoinedPool sends the current
+    connect()
-    // state as its first event too, closing the gap without allowing an older
-    // fetch response to overwrite a newer stream update.
-    void getNodeApi().call<JoinedPool>('getJoinedPool').then((initial) => {
-      if (!active) return
-      setJoinedPool(initial)
-      setJoinedPoolState('ready')
-      connect()
-    }).catch((error: unknown) => {
-      if (!active) return
-      // A coordinating node has joined no other pool; don't render a false
-      // error card for the normal standalone-node case.
-      if (/has joined no other/i.test(error instanceof Error ? error.message : String(error))) {
-        setJoinedPool(null)
-        setJoinedPoolState('none')
-        return
-      }
-      setJoinedPoolState('unavailable')
-      connect()
-    })
     return () => { active = false; if (retry !== undefined) window.clearTimeout(retry); cancel?.() }
   }, [snapshot.status, snapshot.endpoint])
   // Leaving has the pool let go of what it kept for this node; the node then

apps/sisyphusd/README.md

@@ -947,6 +947,8 @@ Country markers ignore addresses inside this machine's directly connected interf
 | owner's location | Where the owner says the node is, in degrees, and how much of it other members of a pool may see. See below. |
 | city | The city the node's public address is in, from DB-IP's free city table, which is downloaded only when the owner asks for it. See below. |
 
+`WatchSharedState` is the authenticated stream for clients connected to the same node. It sends an initial full snapshot for each requested domain, then complete typed snapshots when that state changes; each event has a per-stream revision. A slow client may receive coalesced changes, but it reads the latest committed snapshot, and reconnecting starts from a fresh baseline. A domain that is unavailable (for example, workers on a node with no coordinator) reports that condition without ending other subscriptions. The stream currently covers node details, peers, workers, paired browsers and joined-pool state; additional shared settings and resources are being migrated incrementally. Tokens and provider credentials are not part of its payloads.
+
 Location discovery flags take precedence over the node database only when explicitly supplied (for example, `--locate-country=false`); flags omitted at startup use the saved setting, then the flag's default. This lets an operator pin an opt-out on the command line while allowing paired browsers and Electron to manage saved settings through the local API. `GetLocationDiscoverySettings` reports the effective running settings and whether a restart is needed.
 
 The limits a coordinator holds each member to, `--max-jobs-per-client`, `--max-tasks-per-client`, `--max-task-timeout`, `--max-store-bytes-per-client` and `--max-upload-bytes-per-attempt`, can be saved in the node database the same way, with `SetPoolLimits`, and take effect when the node restarts. A flag supplied explicitly (for example, `--max-jobs-per-client=0`) overrides its saved limit; an omitted flag uses the saved limit, then the flag's default of no limit. `GetPoolLimits` reports the saved and running limits, whether a restart is needed, and whether the node runs the coordinator role, without which the limits hold no one to anything. The task timeout is saved in whole seconds. So that what is reported is what is enforced, the node refuses to start with a job or task count past 2147483647 or a task timeout that is not a whole number of seconds, whether it comes from a flag or was saved.

apps/sisyphusd/api/local.go

@@ -384,8 +384,9 @@ func (s *localService) nodeInfo(ctx context.Context) *nodepb.GetNodeInfoResponse
 	return info
 }
 
-// WatchNodeState streams complete node-and-peer snapshots, triggered by node
+// WatchNodeState streams complete node-and-peer snapshots when their sources
-// discovery and peer events, with a slow reconciliation check as a safeguard.
+// notify a change. It does not poll; callers that need several shared domains
+// can use WatchSharedState.
 func (s *localService) WatchNodeState(_ *nodepb.GetNodeInfoRequest, stream grpc.ServerStreamingServer[nodepb.NodeState]) error {
 	var changes <-chan struct{}
 	var unsubscribe func()
@@ -396,6 +397,129 @@ func (s *localService) WatchNodeState(_ *nodepb.GetNodeInfoRequest, stream grpc.
 	return watchNodeState(stream.Context(), changes, func() *nodepb.NodeState { return s.nodeState(stream.Context()) }, stream.Send)
 }
 
+// WatchSharedState streams typed, authoritative snapshots for shared domains.
+// The token is required because some domains contain owner-only state. Changes
+// are coalesced by the notifier; each wake reads the latest committed state,
+// so a slow subscriber cannot miss the final value. A reconnect starts with a
+// fresh baseline rather than replaying an unbounded event log.
+func (s *localService) WatchSharedState(req *nodepb.WatchSharedStateRequest, stream grpc.ServerStreamingServer[nodepb.SharedStateEvent]) error {
+	if err := s.authorize(stream.Context()); err != nil {
+		return err
+	}
+	domains, err := sharedStateDomains(req.GetDomains())
+	if err != nil {
+		return err
+	}
+	var changes <-chan struct{}
+	var unsubscribe func()
+	if s.cfg.NodeInfoChanges != nil {
+		changes, unsubscribe = s.cfg.NodeInfoChanges.subscribe()
+		defer unsubscribe()
+	}
+
+	last := make(map[nodepb.SharedStateDomain]*nodepb.SharedStateEvent, len(domains))
+	var revision uint64
+	sendCurrent := func(domain nodepb.SharedStateDomain, snapshot bool) error {
+		current := s.sharedStateSnapshot(stream.Context(), domain)
+		if previous, ok := last[domain]; ok && sharedStateEqual(previous, current) {
+			return nil
+		}
+		revision++
+		current.Revision = revision
+		current.Snapshot = snapshot
+		if err := stream.Send(current); err != nil {
+			return err
+		}
+		last[domain] = proto.Clone(current).(*nodepb.SharedStateEvent)
+		return nil
+	}
+	for _, domain := range domains {
+		if err := sendCurrent(domain, true); err != nil {
+			return err
+		}
+	}
+	for {
+		select {
+		case <-changes:
+			for _, domain := range domains {
+				if err := sendCurrent(domain, false); err != nil {
+					return err
+				}
+			}
+		case <-stream.Context().Done():
+			return nil
+		}
+	}
+}
+
+func sharedStateDomains(requested []nodepb.SharedStateDomain) ([]nodepb.SharedStateDomain, error) {
+	if len(requested) == 0 {
+		return []nodepb.SharedStateDomain{
+			nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE,
+			nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS,
+			nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_WORKERS,
+			nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PAIRED_BROWSERS,
+			nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_JOINED_POOL,
+		}, nil
+	}
+	seen := make(map[nodepb.SharedStateDomain]struct{}, len(requested))
+	domains := make([]nodepb.SharedStateDomain, 0, len(requested))
+	for _, domain := range requested {
+		if domain < nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE || domain > nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_JOINED_POOL {
+			return nil, status.Errorf(codes.InvalidArgument, "unsupported shared-state domain %s", domain)
+		}
+		if _, ok := seen[domain]; ok {
+			continue
+		}
+		seen[domain] = struct{}{}
+		domains = append(domains, domain)
+	}
+	sort.Slice(domains, func(i, j int) bool { return domains[i] < domains[j] })
+	return domains, nil
+}
+
+func (s *localService) sharedStateSnapshot(ctx context.Context, domain nodepb.SharedStateDomain) *nodepb.SharedStateEvent {
+	event := &nodepb.SharedStateEvent{Domain: domain}
+	switch domain {
+	case nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE:
+		event.State = &nodepb.SharedStateEvent_Node{Node: s.nodeState(ctx)}
+	case nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS:
+		var peers []*nodepb.Peer
+		if s.cfg.Peers != nil {
+			peers = s.cfg.Peers()
+		}
+		event.State = &nodepb.SharedStateEvent_Peers{Peers: &nodepb.ListPeersResponse{Peers: peers}}
+	case nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_WORKERS:
+		if err := s.jobsAvailable(); err != nil {
+			event.Error = err.Error()
+		} else {
+			event.State = &nodepb.SharedStateEvent_Workers{Workers: s.workerList(true)}
+		}
+	case nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PAIRED_BROWSERS:
+		state, err := s.pairedBrowsers()
+		if err != nil {
+			event.Error = err.Error()
+		} else {
+			event.State = &nodepb.SharedStateEvent_PairedBrowsers{PairedBrowsers: state}
+		}
+	case nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_JOINED_POOL:
+		state := &nodepb.JoinedPool{}
+		if s.cfg.JoinedPool != nil {
+			state = s.cfg.JoinedPool()
+		}
+		event.State = &nodepb.SharedStateEvent_JoinedPool{JoinedPool: state}
+	}
+	return event
+}
+
+func sharedStateEqual(left, right *nodepb.SharedStateEvent) bool {
+	left = proto.Clone(left).(*nodepb.SharedStateEvent)
+	right = proto.Clone(right).(*nodepb.SharedStateEvent)
+	left.Revision, right.Revision = 0, 0
+	left.Snapshot, right.Snapshot = false, false
+	return proto.Equal(left, right)
+}
+
 // WatchPairedBrowsers streams the authorized paired-device list on changes.
 func (s *localService) WatchPairedBrowsers(_ *nodepb.WatchPairedBrowsersRequest, stream grpc.ServerStreamingServer[nodepb.PairedBrowsersState]) error {
 	if err := s.authorize(stream.Context()); err != nil {

apps/sisyphusd/api/local_test.go

@@ -341,6 +341,203 @@ func TestWatchNodeStatePublishesNodeAndPeerChanges(t *testing.T) {
 	}
 }
 
+func TestWatchSharedStateSendsBaselinesUpdatesAndRequiresToken(t *testing.T) {
+	var mu sync.Mutex
+	country := ""
+	var peers []*nodepb.Peer
+	changes := NewNodeInfoChanges()
+	client := serveLocal(t, LocalConfig{
+		Token: "the-token", NodeInfoChanges: changes,
+		Peers: func() []*nodepb.Peer {
+			mu.Lock()
+			defer mu.Unlock()
+			return peers
+		},
+		Country: func() (string, string) {
+			mu.Lock()
+			defer mu.Unlock()
+			return country, ""
+		},
+	})
+	request := &nodepb.WatchSharedStateRequest{Domains: []nodepb.SharedStateDomain{
+		nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS,
+		nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE,
+	}}
+	ctx, cancel := context.WithCancel(withToken("the-token"))
+	defer cancel()
+	stream, err := client.WatchSharedState(ctx, request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	initialNode, err := stream.Recv()
+	if err != nil || initialNode.GetRevision() != 1 || !initialNode.GetSnapshot() || initialNode.GetDomain() != nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE || initialNode.GetNode().GetInfo().GetCountryCode() != "" {
+		t.Fatalf("initial node snapshot = %v, %v", initialNode, err)
+	}
+	initialPeers, err := stream.Recv()
+	if err != nil || initialPeers.GetRevision() != 2 || !initialPeers.GetSnapshot() || initialPeers.GetDomain() != nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS {
+		t.Fatalf("initial peers snapshot = %v, %v", initialPeers, err)
+	}
+
+	// A second client sees the same baseline independently; neither subscriber
+	// consumes the other's change notification.
+	secondCtx, cancelSecond := context.WithCancel(withToken("the-token"))
+	defer cancelSecond()
+	second, err := client.WatchSharedState(secondCtx, request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if event, err := second.Recv(); err != nil || event.GetRevision() != 1 || event.GetNode().GetInfo().GetCountryCode() != "" {
+		t.Fatalf("second client's node baseline = %v, %v", event, err)
+	}
+	if event, err := second.Recv(); err != nil || event.GetRevision() != 2 || event.GetPeers() == nil {
+		t.Fatalf("second client's peers baseline = %v, %v", event, err)
+	}
+
+	mu.Lock()
+	country = "IL"
+	peers = []*nodepb.Peer{peerEntry("12D3KooWa", true, true)}
+	mu.Unlock()
+	changes.Notify()
+	for _, watcher := range []nodepb.NodeService_WatchSharedStateClient{stream, second} {
+		nodeUpdate, err := watcher.Recv()
+		if err != nil || nodeUpdate.GetRevision() != 3 || nodeUpdate.GetSnapshot() || nodeUpdate.GetNode().GetInfo().GetCountryCode() != "IL" {
+			t.Fatalf("node update = %v, %v", nodeUpdate, err)
+		}
+		peerUpdate, err := watcher.Recv()
+		if err != nil || peerUpdate.GetRevision() != 4 || peerUpdate.GetSnapshot() || len(peerUpdate.GetPeers().GetPeers()) != 1 {
+			t.Fatalf("peer update = %v, %v", peerUpdate, err)
+		}
+	}
+
+	unauthorized, err := client.WatchSharedState(context.Background(), request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := unauthorized.Recv(); status.Code(err) != codes.PermissionDenied {
+		t.Fatalf("unauthorized shared-state watcher error = %v, want permission denied", err)
+	}
+}
+
+func TestSharedStateDomainsAndSnapshots(t *testing.T) {
+	all, err := sharedStateDomains(nil)
+	if err != nil || len(all) != 5 || all[0] != nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE || all[4] != nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_JOINED_POOL {
+		t.Fatalf("default shared-state domains = %v, %v", all, err)
+	}
+	unique, err := sharedStateDomains([]nodepb.SharedStateDomain{
+		nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS,
+		nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE,
+		nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS,
+	})
+	if err != nil || len(unique) != 2 || unique[0] != nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE || unique[1] != nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PEERS {
+		t.Fatalf("deduplicated shared-state domains = %v, %v", unique, err)
+	}
+	if _, err := sharedStateDomains([]nodepb.SharedStateDomain{nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_UNSPECIFIED}); status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("unspecified shared-state domain error = %v, want InvalidArgument", err)
+	}
+
+	store := &failingBrowserAccessStore{devices: []nodedb.BrowserDevice{{ID: "device", Name: "Phone"}}}
+	service := &localService{cfg: LocalConfig{
+		NodeID: "node", Token: "the-token", Peers: func() []*nodepb.Peer { return []*nodepb.Peer{peerEntry("peer", true, true)} },
+		Jobs: &office{nodes: []*pb.NodeInfo{{NodeId: "worker"}}}, BrowserAccess: store,
+		JoinedPool: func() *nodepb.JoinedPool { return &nodepb.JoinedPool{CoordinatorId: "coordinator"} },
+	}}
+	for _, domain := range all {
+		event := service.sharedStateSnapshot(withToken("the-token"), domain)
+		if event.GetError() != "" || event.GetState() == nil {
+			t.Fatalf("snapshot for %s = %v; want typed state", domain, event)
+		}
+	}
+	store.listErr = errors.New("device store unavailable")
+	failedDevices := (&localService{cfg: LocalConfig{BrowserAccess: store}}).sharedStateSnapshot(context.Background(), nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_PAIRED_BROWSERS)
+	if failedDevices.GetError() == "" || failedDevices.GetState() != nil {
+		t.Fatalf("failed paired-device snapshot = %v; want per-domain error", failedDevices)
+	}
+	unavailableWorkers := (&localService{}).sharedStateSnapshot(context.Background(), nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_WORKERS)
+	if unavailableWorkers.GetError() == "" || unavailableWorkers.GetState() != nil {
+		t.Fatalf("unavailable worker snapshot = %v; want per-domain error", unavailableWorkers)
+	}
+	noJoinedPool := (&localService{}).sharedStateSnapshot(context.Background(), nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_JOINED_POOL)
+	if noJoinedPool.GetJoinedPool() == nil {
+		t.Fatalf("standalone joined-pool snapshot = %v; want an empty typed state", noJoinedPool)
+	}
+}
+
+type sharedStateTestStream struct {
+	grpc.ServerStream
+	ctx       context.Context
+	sendErr   error
+	afterSend func(int)
+	sent      []*nodepb.SharedStateEvent
+}
+
+func (s *sharedStateTestStream) Context() context.Context { return s.ctx }
+
+func (s *sharedStateTestStream) Send(event *nodepb.SharedStateEvent) error {
+	if s.sendErr != nil {
+		return s.sendErr
+	}
+	s.sent = append(s.sent, event)
+	if s.afterSend != nil {
+		s.afterSend(len(s.sent))
+	}
+	return nil
+}
+
+func TestWatchSharedStateStopsOnSendFailureOrCancellation(t *testing.T) {
+	request := &nodepb.WatchSharedStateRequest{Domains: []nodepb.SharedStateDomain{nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_NODE}}
+	sendFailure := errors.New("shared state send failed")
+	failed := &sharedStateTestStream{ctx: incomingToken("the-token"), sendErr: sendFailure}
+	if err := (&localService{cfg: LocalConfig{Token: "the-token"}}).WatchSharedState(request, failed); !errors.Is(err, sendFailure) {
+		t.Fatalf("initial shared-state send error = %v; want %v", err, sendFailure)
+	}
+	invalid := &sharedStateTestStream{ctx: incomingToken("the-token")}
+	invalidRequest := &nodepb.WatchSharedStateRequest{Domains: []nodepb.SharedStateDomain{nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_UNSPECIFIED}}
+	if err := (&localService{cfg: LocalConfig{Token: "the-token"}}).WatchSharedState(invalidRequest, invalid); status.Code(err) != codes.InvalidArgument {
+		t.Fatalf("invalid shared-state request error = %v; want InvalidArgument", err)
+	}
+
+	changes := NewNodeInfoChanges()
+	joined := false
+	var updateFailure *sharedStateTestStream
+	updateFailure = &sharedStateTestStream{ctx: incomingToken("the-token"), afterSend: func(sent int) {
+		if sent == 1 {
+			updateFailure.sendErr = sendFailure
+			joined = true
+			changes.Notify()
+		}
+	}}
+	joinedPoolRequest := &nodepb.WatchSharedStateRequest{Domains: []nodepb.SharedStateDomain{nodepb.SharedStateDomain_SHARED_STATE_DOMAIN_JOINED_POOL}}
+	if err := (&localService{cfg: LocalConfig{Token: "the-token", NodeInfoChanges: changes, JoinedPool: func() *nodepb.JoinedPool {
+		if joined {
+			return &nodepb.JoinedPool{CoordinatorId: "coordinator"}
+		}
+		return &nodepb.JoinedPool{}
+	}}}).WatchSharedState(joinedPoolRequest, updateFailure); !errors.Is(err, sendFailure) || len(updateFailure.sent) != 1 {
+		t.Fatalf("updated shared-state send error = %v after %d events; want %v after baseline", err, len(updateFailure.sent), sendFailure)
+	}
+
+	unchangedCtx, cancelUnchanged := context.WithCancel(incomingToken("the-token"))
+	unchangedChanges := NewNodeInfoChanges()
+	unchanged := &sharedStateTestStream{ctx: unchangedCtx, afterSend: func(sent int) {
+		if sent == 1 {
+			unchangedChanges.Notify()
+			go func() {
+				time.Sleep(50 * time.Millisecond)
+				cancelUnchanged()
+			}()
+		}
+	}}
+	if err := (&localService{cfg: LocalConfig{Token: "the-token", NodeInfoChanges: unchangedChanges}}).WatchSharedState(request, unchanged); err != nil || len(unchanged.sent) != 1 {
+		t.Fatalf("unchanged shared-state notification = %v after %d events; want only the baseline", err, len(unchanged.sent))
+	}
+
+	ctx, cancel := context.WithCancel(incomingToken("the-token"))
+	stream := &sharedStateTestStream{ctx: ctx, afterSend: func(int) { cancel() }}
+	if err := (&localService{cfg: LocalConfig{Token: "the-token"}}).WatchSharedState(request, stream); err != nil || len(stream.sent) != 1 {
+		t.Fatalf("cancelled shared-state watch = %v, sent %d events; want clean stop after baseline", err, len(stream.sent))
+	}
+}
+
 func TestWatchPairedBrowsersPublishesChanges(t *testing.T) {
 	store := &mutableBrowserAccess{volatileBrowserAccess: volatileBrowserAccess{devices: map[string]nodedb.BrowserDevice{
 		"device-b": {ID: "device-b", Name: "Second"},

apps/sisyphusd/daemon.go

@@ -313,7 +313,11 @@ func runDaemon(ctx context.Context, args []string) error {
 	stopped := make(chan error, 2)
 
 	// local is what the desktop client on this machine is told and may do.
-	local := api.LocalConfig{NodeID: ident.ID(), Version: version, Workloads: workloads, NodeInfoChanges: api.NewNodeInfoChanges(), BrowserDeviceChanges: api.NewNodeInfoChanges()}
+	sharedStateChanges := api.NewNodeInfoChanges()
+	local := api.LocalConfig{
+		NodeID: ident.ID(), Version: version, Workloads: workloads,
+		NodeInfoChanges: sharedStateChanges, BrowserDeviceChanges: sharedStateChanges, JoinedPoolChanges: sharedStateChanges,
+	}
 
 	// coordinatorID is the ID of the node this one's worker answers to: the
 	// node itself, or for a worker-only node the coordinator it joined.
@@ -971,7 +975,6 @@ func runDaemon(ctx context.Context, args []string) error {
 			// A worker's pool, as far as it knows, is its coordinator.
 			coordinatorAddr := *join
 			local.JoinedPool = func() *nodepb.JoinedPool { return joinedPool(w.JoinedPool(), coordinatorAddr) }
-			local.JoinedPoolChanges = api.NewNodeInfoChanges()
 			jobs := newPoolJobs(ctx, pb.NewNodeServiceClient(poolConn), db, w.JoinedPool, local.NodeInfoChanges.Notify)
 			local.Jobs = jobs
 			// Its user's files go to that pool too, kept there for this

apps/sisyphusd/mcpserver/kinds.go

@@ -14,6 +14,7 @@ var rpcKinds = map[string]kind{
 	// The node, its peers and its settings, as they stand.
 	"GetNodeInfo":                     reads,
 	"WatchNodeState":                  reads,
+	"WatchSharedState":                reads,
 	"WatchPairedBrowsers":             reads,
 	"WatchPendingActionConfirmations": reads,
 	"GetLocationDiscoverySettings":    reads,

packages/client/src/gen/sisyphus/node/v1/node_pb.ts

@@ -10,7 +10,7 @@ import type { Message } from "@bufbuild/protobuf";
  * Describes the file sisyphus/node/v1/node.proto.
  */
 export const file_sisyphus_node_v1_node: GenFile = /*@__PURE__*/
-  fileDesc("ChtzaXN5cGh1cy9ub2RlL3YxL25vZGUucHJvdG8SEHNpc3lwaHVzLm5vZGUudjEiFAoSR2V0Tm9kZUluZm9SZXF1ZXN0Ip4CChNHZXROb2RlSW5mb1Jlc3BvbnNlEg8KB3BlZXJfaWQYASABKAkSFgoOZGFlbW9uX3ZlcnNpb24YAiABKAkSGAoQbGlzdGVuX2FkZHJlc3NlcxgDIAMoCRIUCgxjb3VudHJ5X2NvZGUYBCABKAkSEQoJd29ya2xvYWRzGAUgAygJEhQKDGNhcGFiaWxpdGllcxgGIAMoCRIWCg5jb3VudHJ5X3NvdXJjZRgHIAEoCRI2Cg5vd25lcl9sb2NhdGlvbhgIIAEoCzIeLnNpc3lwaHVzLm5vZGUudjEuTm9kZUxvY2F0aW9uEjUKDWNpdHlfZXN0aW1hdGUYCSABKAsyHi5zaXN5cGh1cy5ub2RlLnYxLkNpdHlFc3RpbWF0ZSKlAQoMQ2l0eUVzdGltYXRlEgwKBGNpdHkYASABKAkSDgoGcmVnaW9uGAIgASgJEhQKDGNvdW50cnlfY29kZRgDIAEoCRIQCghsYXRpdHVkZRgEIAEoARIRCglsb25naXR1ZGUYBSABKAESDgoGc291cmNlGAYgASgJEhMKC2F0dHJpYnV0aW9uGAcgASgJEhcKD2F0dHJpYnV0aW9uX3VybBgIIAEoCSIbChlHZXRXZWJBY2Nlc3NDb25maWdSZXF1ZXN0IqMBCg9XZWJBY2Nlc3NDb25maWcSDwoHZW5hYmxlZBgBIAEoCBIWCg5saXN0ZW5fYWRkcmVzcxgCIAEoCRIfChdjZXJ0aWZpY2F0ZV9maW5nZXJwcmludBgDIAEoCRIYChByZXN0YXJ0X3JlcXVpcmVkGAQgASgIEhIKCnB1YmxpY191cmwYBSABKAkSGAoQbGlzdGVuX2FkZHJlc3NlcxgGIAMoCSJyChlTZXRXZWJBY2Nlc3NDb25maWdSZXF1ZXN0Eg8KB2VuYWJsZWQYASABKAgSFgoObGlzdGVuX2FkZHJlc3MYAiABKAkSEgoKcHVibGljX3VybBgDIAEoCRIYChBsaXN0ZW5fYWRkcmVzc2VzGAQgAygJIh4KHExpc3ROZXR3b3JrSW50ZXJmYWNlc1JlcXVlc3QiVwodTGlzdE5ldHdvcmtJbnRlcmZhY2VzUmVzcG9uc2USNgoKaW50ZXJmYWNlcxgBIAMoCzIiLnNpc3lwaHVzLm5vZGUudjEuTmV0d29ya0ludGVyZmFjZSIxChBOZXR3b3JrSW50ZXJmYWNlEgwKBG5hbWUYASABKAkSDwoHYWRkcmVzcxgCIAEoCSISChBMaXN0UGVlcnNSZXF1ZXN0IhMKEVdhdGNoUGVlcnNSZXF1ZXN0IhoKGEdldEJvb3RzdHJhcFBlZXJzUmVxdWVzdCJLChlHZXRCb290c3RyYXBQZWVyc1Jlc3BvbnNlEi4KBXBlZXJzGAEgAygLMh8uc2lzeXBodXMubm9kZS52MS5Cb290c3RyYXBQZWVyIkoKGFNldEJvb3RzdHJhcFBlZXJzUmVxdWVzdBIuCgVwZWVycxgBIAMoCzIfLnNpc3lwaHVzLm5vZGUudjEuQm9vdHN0cmFwUGVlciJLChlTZXRCb290c3RyYXBQZWVyc1Jlc3BvbnNlEi4KBXBlZXJzGAEgAygLMh8uc2lzeXBodXMubm9kZS52MS5Cb290c3RyYXBQZWVyIiUKEkNvbm5lY3RQZWVyUmVxdWVzdBIPCgdhZGRyZXNzGAEgASgJIiYKE0Nvbm5lY3RQZWVyUmVzcG9uc2USDwoHcGVlcl9pZBgBIAEoCSI+ChpTZXRQZWVyQ29tcHV0ZVRydXN0UmVxdWVzdBIPCgdwZWVyX2lkGAEgASgJEg8KB3RydXN0ZWQYAiABKAgiLgobU2V0UGVlckNvbXB1dGVUcnVzdFJlc3BvbnNlEg8KB3RydXN0ZWQYASABKAgiWwogU2V0UGVlckNvbXB1dGVQZXJtaXNzaW9uc1JlcXVlc3QSDwoHcGVlcl9pZBgBIAEoCRISCgpnaXZlc193b3JrGAIgASgIEhIKCnRha2VzX3dvcmsYAyABKAgiSwohU2V0UGVlckNvbXB1dGVQZXJtaXNzaW9uc1Jlc3BvbnNlEhIKCmdpdmVzX3dvcmsYASABKAgSEgoKdGFrZXNfd29yaxgCIAEoCCIxCg1Cb290c3RyYXBQZWVyEg8KB3BlZXJfaWQYASABKAkSDwoHYWRkcmVzcxgCIAEoCSJMChFMaXN0UGVlcnNSZXNwb25zZRIlCgVwZWVycxgBIAMoCzIWLnNpc3lwaHVzLm5vZGUudjEuUGVlchIQCghyZXZpc2lvbhgCIAEoBCKGAgoEUGVlchIPCgdwZWVyX2lkGAEgASgJEj8KEGNvbm5lY3Rpb25fc3RhdGUYAiABKA4yJS5zaXN5cGh1cy5ub2RlLnYxLlBlZXJDb25uZWN0aW9uU3RhdGUSFwoPa25vd25fYWRkcmVzc2VzGAMgAygJEhsKE3RydXN0ZWRfZm9yX2NvbXB1dGUYBCABKAgSGwoTd29ya3NfZm9yX3RoaXNfbm9kZRgFIAEoCBIbChN0aGlzX25vZGVfd29ya3NfZm9yGAYgASgIEhIKCmdpdmVzX3dvcmsYByABKAgSEgoKdGFrZXNfd29yaxgIIAEoCBIUCgxjb3VudHJ5X2NvZGUYCSABKAkiFAoSTGlzdFdvcmtlcnNSZXF1ZXN0IlIKE0xpc3RXb3JrZXJzUmVzcG9uc2USKQoHd29ya2VycxgBIAMoCzIYLnNpc3lwaHVzLm5vZGUudjEuV29ya2VyEhAKCHJldmlzaW9uGAIgASgEIoQECgZXb3JrZXISDwoHcGVlcl9pZBgBIAEoCRIMCgRuYW1lGAIgASgJEhAKCGhvc3RuYW1lGAMgASgJEgoKAm9zGAQgASgJEgwKBGFyY2gYBSABKAkSEQoJY3B1X2NvcmVzGAYgASgNEhIKCnRhc2tfc2xvdHMYByABKA0SFQoNcnVubmluZ190YXNrcxgIIAEoDRIRCgl3b3JrbG9hZHMYCSADKAkSDgoGcmVsYXlzGAogASgIEhsKE3JlbGF5ZWRfY29ubmVjdGlvbnMYCyABKAQSFQoNcmVsYXllZF9ieXRlcxgMIAEoBBIRCgljcHVfbW9kZWwYDSABKAkSFAoMbWVtb3J5X2J5dGVzGA4gASgEEikKBGdwdXMYDyADKAsyGy5zaXN5cGh1cy5ub2RlLnYxLldvcmtlckdwdRIOCgZtb2RlbHMYECADKAkSNAoNbG9hZGVkX21vZGVscxgUIAMoCzIdLnNpc3lwaHVzLm5vZGUudjEuTG9hZGVkTW9kZWwSFwoPdmVyaWZpZWRfYWdyZWVkGBEgASgEEhkKEXZlcmlmaWVkX291dHZvdGVkGBIgASgEEhEKCXByb2JhdGlvbhgTIAEoDRI5Cg9zaGFyZWRfbG9jYXRpb24YFSABKAsyIC5zaXN5cGh1cy5ub2RlLnYxLlNoYXJlZExvY2F0aW9uIjUKDlNoYXJlZExvY2F0aW9uEhAKCGxhdGl0dWRlGAEgASgBEhEKCWxvbmdpdHVkZRgCIAEoASIWChRMaXN0V29ya2xvYWRzUmVxdWVzdCJGChVMaXN0V29ya2xvYWRzUmVzcG9uc2USLQoJd29ya2xvYWRzGAEgAygLMhouc2lzeXBodXMubm9kZS52MS5Xb3JrbG9hZCKNAQoIV29ya2xvYWQSDAoEbmFtZRgBIAEoCRITCgtkZXNjcmlwdGlvbhgCIAEoCRIvCgZmaWVsZHMYAyADKAsyHy5zaXN5cGh1cy5ub2RlLnYxLldvcmtsb2FkRmllbGQSGgoSd29ya2Vyc19ydW5uaW5nX2l0GAQgASgNEhEKCWNvbXBvc2l0ZRgFIAEoCCL4AQoNV29ya2xvYWRGaWVsZBIMCgRuYW1lGAEgASgJEjEKBGtpbmQYAiABKA4yIy5zaXN5cGh1cy5ub2RlLnYxLldvcmtsb2FkRmllbGRLaW5kEg0KBWxhYmVsGAMgASgJEgwKBGhlbHAYBCABKAkSFAoMZGVmYXVsdF9qc29uGAUgASgJEhQKB21pbmltdW0YBiABKAFIAIgBARIUCgdtYXhpbXVtGAcgASgBSAGIAQESDAoEc3RlcBgIIAEoARIPCgdjaG9pY2VzGAkgAygJEhAKCHJlcXVpcmVkGAogASgIQgoKCF9taW5pbXVtQgoKCF9tYXhpbXVtIi8KCVdvcmtlckdwdRIMCgRuYW1lGAEgASgJEhQKDG1lbW9yeV9ieXRlcxgCIAEoBCI0CgtMb2FkZWRNb2RlbBIMCgRuYW1lGAEgASgJEhcKD29uX2NhcmRfcGVyY2VudBgCIAEoDSKPBAoDSm9iEg4KBmpvYl9pZBgBIAEoCRIQCgh3b3JrbG9hZBgCIAEoCRIOCgZwYXJhbXMYAyABKAwSJwoEbW9kZRgEIAEoDjIZLnNpc3lwaHVzLm5vZGUudjEuSm9iTW9kZRIpCgVzdGF0ZRgFIAEoDjIaLnNpc3lwaHVzLm5vZGUudjEuSm9iU3RhdGUSKAoFdGFza3MYBiADKAsyGS5zaXN5cGh1cy5ub2RlLnYxLkpvYlRhc2sSDgoGcmVzdWx0GAcgASgMEg0KBWVycm9yGAggASgJEhUKDWNyZWF0ZWRfYXRfbXMYCSABKAMSFgoOZmluaXNoZWRfYXRfbXMYCiABKAMSEwoLaW5wdXRfYmxvYnMYCyADKAkSFAoMb3V0cHV0X2Jsb2JzGAwgAygJEhAKCHByb2dyZXNzGA0gASgBEhwKFHRhc2tfdGltZW91dF9zZWNvbmRzGA4gASgNEhgKEG1pbl9tZW1vcnlfYnl0ZXMYDyABKAQSEAoIbWluX2dwdXMYECABKA0SDwoHcHJLine truncated
+  fileDesc("ChtzaXN5cGh1cy9ub2RlL3YxL25vZGUucHJvdG8SEHNpc3lwaHVzLm5vZGUudjEiFAoSR2V0Tm9kZUluZm9SZXF1ZXN0Ip4CChNHZXROb2RlSW5mb1Jlc3BvbnNlEg8KB3BlZXJfaWQYASABKAkSFgoOZGFlbW9uX3ZlcnNpb24YAiABKAkSGAoQbGlzdGVuX2FkZHJlc3NlcxgDIAMoCRIUCgxjb3VudHJ5X2NvZGUYBCABKAkSEQoJd29ya2xvYWRzGAUgAygJEhQKDGNhcGFiaWxpdGllcxgGIAMoCRIWCg5jb3VudHJ5X3NvdXJjZRgHIAEoCRI2Cg5vd25lcl9sb2NhdGlvbhgIIAEoCzIeLnNpc3lwaHVzLm5vZGUudjEuTm9kZUxvY2F0aW9uEjUKDWNpdHlfZXN0aW1hdGUYCSABKAsyHi5zaXN5cGh1cy5ub2RlLnYxLkNpdHlFc3RpbWF0ZSKlAQoMQ2l0eUVzdGltYXRlEgwKBGNpdHkYASABKAkSDgoGcmVnaW9uGAIgASgJEhQKDGNvdW50cnlfY29kZRgDIAEoCRIQCghsYXRpdHVkZRgEIAEoARIRCglsb25naXR1ZGUYBSABKAESDgoGc291cmNlGAYgASgJEhMKC2F0dHJpYnV0aW9uGAcgASgJEhcKD2F0dHJpYnV0aW9uX3VybBgIIAEoCSIbChlHZXRXZWJBY2Nlc3NDb25maWdSZXF1ZXN0IqMBCg9XZWJBY2Nlc3NDb25maWcSDwoHZW5hYmxlZBgBIAEoCBIWCg5saXN0ZW5fYWRkcmVzcxgCIAEoCRIfChdjZXJ0aWZpY2F0ZV9maW5nZXJwcmludBgDIAEoCRIYChByZXN0YXJ0X3JlcXVpcmVkGAQgASgIEhIKCnB1YmxpY191cmwYBSABKAkSGAoQbGlzdGVuX2FkZHJlc3NlcxgGIAMoCSJyChlTZXRXZWJBY2Nlc3NDb25maWdSZXF1ZXN0Eg8KB2VuYWJsZWQYASABKAgSFgoObGlzdGVuX2FkZHJlc3MYAiABKAkSEgoKcHVibGljX3VybBgDIAEoCRIYChBsaXN0ZW5fYWRkcmVzc2VzGAQgAygJIh4KHExpc3ROZXR3b3JrSW50ZXJmYWNlc1JlcXVlc3QiVwodTGlzdE5ldHdvcmtJbnRlcmZhY2VzUmVzcG9uc2USNgoKaW50ZXJmYWNlcxgBIAMoCzIiLnNpc3lwaHVzLm5vZGUudjEuTmV0d29ya0ludGVyZmFjZSIxChBOZXR3b3JrSW50ZXJmYWNlEgwKBG5hbWUYASABKAkSDwoHYWRkcmVzcxgCIAEoCSISChBMaXN0UGVlcnNSZXF1ZXN0IhMKEVdhdGNoUGVlcnNSZXF1ZXN0IhoKGEdldEJvb3RzdHJhcFBlZXJzUmVxdWVzdCJLChlHZXRCb290c3RyYXBQZWVyc1Jlc3BvbnNlEi4KBXBlZXJzGAEgAygLMh8uc2lzeXBodXMubm9kZS52MS5Cb290c3RyYXBQZWVyIkoKGFNldEJvb3RzdHJhcFBlZXJzUmVxdWVzdBIuCgVwZWVycxgBIAMoCzIfLnNpc3lwaHVzLm5vZGUudjEuQm9vdHN0cmFwUGVlciJLChlTZXRCb290c3RyYXBQZWVyc1Jlc3BvbnNlEi4KBXBlZXJzGAEgAygLMh8uc2lzeXBodXMubm9kZS52MS5Cb290c3RyYXBQZWVyIiUKEkNvbm5lY3RQZWVyUmVxdWVzdBIPCgdhZGRyZXNzGAEgASgJIiYKE0Nvbm5lY3RQZWVyUmVzcG9uc2USDwoHcGVlcl9pZBgBIAEoCSI+ChpTZXRQZWVyQ29tcHV0ZVRydXN0UmVxdWVzdBIPCgdwZWVyX2lkGAEgASgJEg8KB3RydXN0ZWQYAiABKAgiLgobU2V0UGVlckNvbXB1dGVUcnVzdFJlc3BvbnNlEg8KB3RydXN0ZWQYASABKAgiWwogU2V0UGVlckNvbXB1dGVQZXJtaXNzaW9uc1JlcXVlc3QSDwoHcGVlcl9pZBgBIAEoCRISCgpnaXZlc193b3JrGAIgASgIEhIKCnRha2VzX3dvcmsYAyABKAgiSwohU2V0UGVlckNvbXB1dGVQZXJtaXNzaW9uc1Jlc3BvbnNlEhIKCmdpdmVzX3dvcmsYASABKAgSEgoKdGFrZXNfd29yaxgCIAEoCCIxCg1Cb290c3RyYXBQZWVyEg8KB3BlZXJfaWQYASABKAkSDwoHYWRkcmVzcxgCIAEoCSJMChFMaXN0UGVlcnNSZXNwb25zZRIlCgVwZWVycxgBIAMoCzIWLnNpc3lwaHVzLm5vZGUudjEuUGVlchIQCghyZXZpc2lvbhgCIAEoBCKGAgoEUGVlchIPCgdwZWVyX2lkGAEgASgJEj8KEGNvbm5lY3Rpb25fc3RhdGUYAiABKA4yJS5zaXN5cGh1cy5ub2RlLnYxLlBlZXJDb25uZWN0aW9uU3RhdGUSFwoPa25vd25fYWRkcmVzc2VzGAMgAygJEhsKE3RydXN0ZWRfZm9yX2NvbXB1dGUYBCABKAgSGwoTd29ya3NfZm9yX3RoaXNfbm9kZRgFIAEoCBIbChN0aGlzX25vZGVfd29ya3NfZm9yGAYgASgIEhIKCmdpdmVzX3dvcmsYByABKAgSEgoKdGFrZXNfd29yaxgIIAEoCBIUCgxjb3VudHJ5X2NvZGUYCSABKAkiFAoSTGlzdFdvcmtlcnNSZXF1ZXN0IlIKE0xpc3RXb3JrZXJzUmVzcG9uc2USKQoHd29ya2VycxgBIAMoCzIYLnNpc3lwaHVzLm5vZGUudjEuV29ya2VyEhAKCHJldmlzaW9uGAIgASgEIoQECgZXb3JrZXISDwoHcGVlcl9pZBgBIAEoCRIMCgRuYW1lGAIgASgJEhAKCGhvc3RuYW1lGAMgASgJEgoKAm9zGAQgASgJEgwKBGFyY2gYBSABKAkSEQoJY3B1X2NvcmVzGAYgASgNEhIKCnRhc2tfc2xvdHMYByABKA0SFQoNcnVubmluZ190YXNrcxgIIAEoDRIRCgl3b3JrbG9hZHMYCSADKAkSDgoGcmVsYXlzGAogASgIEhsKE3JlbGF5ZWRfY29ubmVjdGlvbnMYCyABKAQSFQoNcmVsYXllZF9ieXRlcxgMIAEoBBIRCgljcHVfbW9kZWwYDSABKAkSFAoMbWVtb3J5X2J5dGVzGA4gASgEEikKBGdwdXMYDyADKAsyGy5zaXN5cGh1cy5ub2RlLnYxLldvcmtlckdwdRIOCgZtb2RlbHMYECADKAkSNAoNbG9hZGVkX21vZGVscxgUIAMoCzIdLnNpc3lwaHVzLm5vZGUudjEuTG9hZGVkTW9kZWwSFwoPdmVyaWZpZWRfYWdyZWVkGBEgASgEEhkKEXZlcmlmaWVkX291dHZvdGVkGBIgASgEEhEKCXByb2JhdGlvbhgTIAEoDRI5Cg9zaGFyZWRfbG9jYXRpb24YFSABKAsyIC5zaXN5cGh1cy5ub2RlLnYxLlNoYXJlZExvY2F0aW9uIjUKDlNoYXJlZExvY2F0aW9uEhAKCGxhdGl0dWRlGAEgASgBEhEKCWxvbmdpdHVkZRgCIAEoASIWChRMaXN0V29ya2xvYWRzUmVxdWVzdCJGChVMaXN0V29ya2xvYWRzUmVzcG9uc2USLQoJd29ya2xvYWRzGAEgAygLMhouc2lzeXBodXMubm9kZS52MS5Xb3JrbG9hZCKNAQoIV29ya2xvYWQSDAoEbmFtZRgBIAEoCRITCgtkZXNjcmlwdGlvbhgCIAEoCRIvCgZmaWVsZHMYAyADKAsyHy5zaXN5cGh1cy5ub2RlLnYxLldvcmtsb2FkRmllbGQSGgoSd29ya2Vyc19ydW5uaW5nX2l0GAQgASgNEhEKCWNvbXBvc2l0ZRgFIAEoCCL4AQoNV29ya2xvYWRGaWVsZBIMCgRuYW1lGAEgASgJEjEKBGtpbmQYAiABKA4yIy5zaXN5cGh1cy5ub2RlLnYxLldvcmtsb2FkRmllbGRLaW5kEg0KBWxhYmVsGAMgASgJEgwKBGhlbHAYBCABKAkSFAoMZGVmYXVsdF9qc29uGAUgASgJEhQKB21pbmltdW0YBiABKAFIAIgBARIUCgdtYXhpbXVtGAcgASgBSAGIAQESDAoEc3RlcBgIIAEoARIPCgdjaG9pY2VzGAkgAygJEhAKCHJlcXVpcmVkGAogASgIQgoKCF9taW5pbXVtQgoKCF9tYXhpbXVtIi8KCVdvcmtlckdwdRIMCgRuYW1lGAEgASgJEhQKDG1lbW9yeV9ieXRlcxgCIAEoBCI0CgtMb2FkZWRNb2RlbBIMCgRuYW1lGAEgASgJEhcKD29uX2NhcmRfcGVyY2VudBgCIAEoDSKPBAoDSm9iEg4KBmpvYl9pZBgBIAEoCRIQCgh3b3JrbG9hZBgCIAEoCRIOCgZwYXJhbXMYAyABKAwSJwoEbW9kZRgEIAEoDjIZLnNpc3lwaHVzLm5vZGUudjEuSm9iTW9kZRIpCgVzdGF0ZRgFIAEoDjIaLnNpc3lwaHVzLm5vZGUudjEuSm9iU3RhdGUSKAoFdGFza3MYBiADKAsyGS5zaXN5cGh1cy5ub2RlLnYxLkpvYlRhc2sSDgoGcmVzdWx0GAcgASgMEg0KBWVycm9yGAggASgJEhUKDWNyZWF0ZWRfYXRfbXMYCSABKAMSFgoOZmluaXNoZWRfYXRfbXMYCiABKAMSEwoLaW5wdXRfYmxvYnMYCyADKAkSFAoMb3V0cHV0X2Jsb2JzGAwgAygJEhAKCHByb2dyZXNzGA0gASgBEhwKFHRhc2tfdGltZW91dF9zZWNvbmRzGA4gASgNEhgKEG1pbl9tZW1vcnlfYnl0ZXMYDyABKAQSEAoIbWluX2dwdXMYECABKA0SDwoHcHJLine truncated
 
 /**
  * @generated from message sisyphus.node.v1.GetNodeInfoRequest
@@ -3447,6 +3447,99 @@ export type NodeState = Message<"sisyphus.node.v1.NodeState"> & {
 export const NodeStateSchema: GenMessage<NodeState> = /*@__PURE__*/
   messageDesc(file_sisyphus_node_v1_node, 121);
 
+/**
+ * @generated from message sisyphus.node.v1.WatchSharedStateRequest
+ */
+export type WatchSharedStateRequest = Message<"sisyphus.node.v1.WatchSharedStateRequest"> & {
+  /**
+   * Empty requests every domain supported by this daemon. Repeated values are
+   * harmless and are treated as one subscription.
+   *
+   * @generated from field: repeated sisyphus.node.v1.SharedStateDomain domains = 1;
+   */
+  domains: SharedStateDomain[];
+};
+
+/**
+ * Describes the message sisyphus.node.v1.WatchSharedStateRequest.
+ * Use `create(WatchSharedStateRequestSchema)` to create a new message.
+ */
+export const WatchSharedStateRequestSchema: GenMessage<WatchSharedStateRequest> = /*@__PURE__*/
+  messageDesc(file_sisyphus_node_v1_node, 122);
+
+/**
+ * @generated from message sisyphus.node.v1.SharedStateEvent
+ */
+export type SharedStateEvent = Message<"sisyphus.node.v1.SharedStateEvent"> & {
+  /**
+   * Increases for each event on this stream. A reconnect starts with a new
+   * authoritative baseline and a new revision sequence.
+   *
+   * @generated from field: uint64 revision = 1;
+   */
+  revision: bigint;
+
+  /**
+   * @generated from field: sisyphus.node.v1.SharedStateDomain domain = 2;
+   */
+  domain: SharedStateDomain;
+
+  /**
+   * @generated from field: bool snapshot = 3;
+   */
+  snapshot: boolean;
+
+  /**
+   * A domain can temporarily be unavailable (for example, workers on a node
+   * that coordinates no pool) without terminating other domain subscriptions.
+   *
+   * @generated from field: string error = 4;
+   */
+  error: string;
+
+  /**
+   * @generated from oneof sisyphus.node.v1.SharedStateEvent.state
+   */
+  state: {
+    /**
+     * @generated from field: sisyphus.node.v1.NodeState node = 10;
+     */
+    value: NodeState;
+    case: "node";
+  } | {
+    /**
+     * @generated from field: sisyphus.node.v1.ListPeersResponse peers = 11;
+     */
+    value: ListPeersResponse;
+    case: "peers";
+  } | {
+    /**
+     * @generated from field: sisyphus.node.v1.ListWorkersResponse workers = 12;
+     */
+    value: ListWorkersResponse;
+    case: "workers";
+  } | {
+    /**
+     * @generated from field: sisyphus.node.v1.PairedBrowsersState paired_browsers = 13;
+     */
+    value: PairedBrowsersState;
+    case: "pairedBrowsers";
+  } | {
+    /**
+     * @generated from field: sisyphus.node.v1.JoinedPool joined_pool = 14;
+     */
+    value: JoinedPool;
+    case: "joinedPool";
+  } | { case: undefined; value?: undefined };
+};
+
+/**
+ * Describes the message sisyphus.node.v1.SharedStateEvent.
+ * Use `create(SharedStateEventSchema)` to create a new message.
+ */
+export const SharedStateEventSchema: GenMessage<SharedStateEvent> = /*@__PURE__*/
+  messageDesc(file_sisyphus_node_v1_node, 123);
+
 /**
  * @generated from message sisyphus.node.v1.WatchPairedBrowsersRequest
  */
@@ -3458,7 +3551,7 @@ export type WatchPairedBrowsersRequest = Message<"sisyphus.node.v1.WatchPairedBr
  * Use `create(WatchPairedBrowsersRequestSchema)` to create a new message.
  */
 export const WatchPairedBrowsersRequestSchema: GenMessage<WatchPairedBrowsersRequest> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 122);
+  messageDesc(file_sisyphus_node_v1_node, 124);
 
 /**
  * @generated from message sisyphus.node.v1.RequestActionConfirmationRequest
@@ -3475,7 +3568,7 @@ export type RequestActionConfirmationRequest = Message<"sisyphus.node.v1.Request
  * Use `create(RequestActionConfirmationRequestSchema)` to create a new message.
  */
 export const RequestActionConfirmationRequestSchema: GenMessage<RequestActionConfirmationRequest> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 123);
+  messageDesc(file_sisyphus_node_v1_node, 125);
 
 /**
  * @generated from message sisyphus.node.v1.RequestActionConfirmationResponse
@@ -3492,7 +3585,7 @@ export type RequestActionConfirmationResponse = Message<"sisyphus.node.v1.Reques
  * Use `create(RequestActionConfirmationResponseSchema)` to create a new message.
  */
 export const RequestActionConfirmationResponseSchema: GenMessage<RequestActionConfirmationResponse> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 124);
+  messageDesc(file_sisyphus_node_v1_node, 126);
 
 /**
  * @generated from message sisyphus.node.v1.WatchPendingActionConfirmationsRequest
@@ -3505,7 +3598,7 @@ export type WatchPendingActionConfirmationsRequest = Message<"sisyphus.node.v1.W
  * Use `create(WatchPendingActionConfirmationsRequestSchema)` to create a new message.
  */
 export const WatchPendingActionConfirmationsRequestSchema: GenMessage<WatchPendingActionConfirmationsRequest> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 125);
+  messageDesc(file_sisyphus_node_v1_node, 127);
 
 /**
  * @generated from message sisyphus.node.v1.PendingActionConfirmation
@@ -3537,7 +3630,7 @@ export type PendingActionConfirmation = Message<"sisyphus.node.v1.PendingActionC
  * Use `create(PendingActionConfirmationSchema)` to create a new message.
  */
 export const PendingActionConfirmationSchema: GenMessage<PendingActionConfirmation> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 126);
+  messageDesc(file_sisyphus_node_v1_node, 128);
 
 /**
  * @generated from message sisyphus.node.v1.PendingActionConfirmationsState
@@ -3559,7 +3652,7 @@ export type PendingActionConfirmationsState = Message<"sisyphus.node.v1.PendingA
  * Use `create(PendingActionConfirmationsStateSchema)` to create a new message.
  */
 export const PendingActionConfirmationsStateSchema: GenMessage<PendingActionConfirmationsState> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 127);
+  messageDesc(file_sisyphus_node_v1_node, 129);
 
 /**
  * @generated from message sisyphus.node.v1.ResolveActionConfirmationRequest
@@ -3581,7 +3674,7 @@ export type ResolveActionConfirmationRequest = Message<"sisyphus.node.v1.Resolve
  * Use `create(ResolveActionConfirmationRequestSchema)` to create a new message.
  */
 export const ResolveActionConfirmationRequestSchema: GenMessage<ResolveActionConfirmationRequest> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 128);
+  messageDesc(file_sisyphus_node_v1_node, 130);
 
 /**
  * @generated from message sisyphus.node.v1.ResolveActionConfirmationResponse
@@ -3594,7 +3687,7 @@ export type ResolveActionConfirmationResponse = Message<"sisyphus.node.v1.Resolv
  * Use `create(ResolveActionConfirmationResponseSchema)` to create a new message.
  */
 export const ResolveActionConfirmationResponseSchema: GenMessage<ResolveActionConfirmationResponse> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 129);
+  messageDesc(file_sisyphus_node_v1_node, 131);
 
 /**
  * @generated from message sisyphus.node.v1.PairedBrowser
@@ -3621,7 +3714,7 @@ export type PairedBrowser = Message<"sisyphus.node.v1.PairedBrowser"> & {
  * Use `create(PairedBrowserSchema)` to create a new message.
  */
 export const PairedBrowserSchema: GenMessage<PairedBrowser> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 130);
+  messageDesc(file_sisyphus_node_v1_node, 132);
 
 /**
  * @generated from message sisyphus.node.v1.PairedBrowsersState
@@ -3643,7 +3736,7 @@ export type PairedBrowsersState = Message<"sisyphus.node.v1.PairedBrowsersState"
  * Use `create(PairedBrowsersStateSchema)` to create a new message.
  */
 export const PairedBrowsersStateSchema: GenMessage<PairedBrowsersState> = /*@__PURE__*/
-  messageDesc(file_sisyphus_node_v1_node, 131);
+  messageDesc(file_sisyphus_node_v1_node, 133);
 
 /**
  * @generated from enum sisyphus.node.v1.PeerConnectionState
@@ -3954,6 +4047,47 @@ export enum LocationShare {
 export const LocationShareSchema: GenEnum<LocationShare> = /*@__PURE__*/
   enumDesc(file_sisyphus_node_v1_node, 7);
 
+/**
+ * @generated from enum sisyphus.node.v1.SharedStateDomain
+ */
+export enum SharedStateDomain {
+  /**
+   * @generated from enum value: SHARED_STATE_DOMAIN_UNSPECIFIED = 0;
+   */
+  UNSPECIFIED = 0,
+
+  /**
+   * @generated from enum value: SHARED_STATE_DOMAIN_NODE = 1;
+   */
+  NODE = 1,
+
+  /**
+   * @generated from enum value: SHARED_STATE_DOMAIN_PEERS = 2;
+   */
+  PEERS = 2,
+
+  /**
+   * @generated from enum value: SHARED_STATE_DOMAIN_WORKERS = 3;
+   */
+  WORKERS = 3,
+
+  /**
+   * @generated from enum value: SHARED_STATE_DOMAIN_PAIRED_BROWSERS = 4;
+   */
+  PAIRED_BROWSERS = 4,
+
+  /**
+   * @generated from enum value: SHARED_STATE_DOMAIN_JOINED_POOL = 5;
+   */
+  JOINED_POOL = 5,
+}
+
+/**
+ * Describes the enum sisyphus.node.v1.SharedStateDomain.
+ */
+export const SharedStateDomainSchema: GenEnum<SharedStateDomain> = /*@__PURE__*/
+  enumDesc(file_sisyphus_node_v1_node, 8);
+
 /**
  * @generated from service sisyphus.node.v1.NodeService
  */
@@ -3977,6 +4111,18 @@ export const NodeService: GenService<{
     input: typeof GetNodeInfoRequestSchema;
     output: typeof NodeStateSchema;
   },
+  /**
+   * WatchSharedState sends one initial snapshot per requested domain, then
+   * complete domain snapshots whenever shared node state changes. Reconnect
+   * starts a new stream with a fresh baseline.
+   *
+   * @generated from rpc sisyphus.node.v1.NodeService.WatchSharedState
+   */
+  watchSharedState: {
+    methodKind: "server_streaming";
+    input: typeof WatchSharedStateRequestSchema;
+    output: typeof SharedStateEventSchema;
+  },
   /**
    * WatchPairedBrowsers sends the authorized device list at once and whenever
    * a browser is paired or revoked.

packages/protocol/sisyphus/node/v1/node_grpc.pb.go

@@ -21,6 +21,7 @@ const _ = grpc.SupportPackageIsVersion9
 const (
 	NodeService_GetNodeInfo_FullMethodName                     = "/sisyphus.node.v1.NodeService/GetNodeInfo"
 	NodeService_WatchNodeState_FullMethodName                  = "/sisyphus.node.v1.NodeService/WatchNodeState"
+	NodeService_WatchSharedState_FullMethodName                = "/sisyphus.node.v1.NodeService/WatchSharedState"
 	NodeService_WatchPairedBrowsers_FullMethodName             = "/sisyphus.node.v1.NodeService/WatchPairedBrowsers"
 	NodeService_RequestActionConfirmation_FullMethodName       = "/sisyphus.node.v1.NodeService/RequestActionConfirmation"
 	NodeService_WatchPendingActionConfirmations_FullMethodName = "/sisyphus.node.v1.NodeService/WatchPendingActionConfirmations"
@@ -90,6 +91,10 @@ type NodeServiceClient interface {
 	// WatchNodeState sends a complete node-and-peer snapshot at once and
 	// again whenever node details or peers change.
 	WatchNodeState(ctx context.Context, in *GetNodeInfoRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[NodeState], error)
+	// WatchSharedState sends one initial snapshot per requested domain, then
+	// complete domain snapshots whenever shared node state changes. Reconnect
+	// starts a new stream with a fresh baseline.
+	WatchSharedState(ctx context.Context, in *WatchSharedStateRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SharedStateEvent], error)
 	// WatchPairedBrowsers sends the authorized device list at once and whenever
 	// a browser is paired or revoked.
 	WatchPairedBrowsers(ctx context.Context, in *WatchPairedBrowsersRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[PairedBrowsersState], error)
@@ -289,9 +294,28 @@ func (c *nodeServiceClient) WatchNodeState(ctx context.Context, in *GetNodeInfoR
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type NodeService_WatchNodeStateClient = grpc.ServerStreamingClient[NodeState]
 
+func (c *nodeServiceClient) WatchSharedState(ctx context.Context, in *WatchSharedStateRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[SharedStateEvent], error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[1], NodeService_WatchSharedState_FullMethodName, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &grpc.GenericClientStream[WatchSharedStateRequest, SharedStateEvent]{ClientStream: stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type NodeService_WatchSharedStateClient = grpc.ServerStreamingClient[SharedStateEvent]
+
 func (c *nodeServiceClient) WatchPairedBrowsers(ctx context.Context, in *WatchPairedBrowsersRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[PairedBrowsersState], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[1], NodeService_WatchPairedBrowsers_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[2], NodeService_WatchPairedBrowsers_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -320,7 +344,7 @@ func (c *nodeServiceClient) RequestActionConfirmation(ctx context.Context, in *R
 
 func (c *nodeServiceClient) WatchPendingActionConfirmations(ctx context.Context, in *WatchPendingActionConfirmationsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[PendingActionConfirmationsState], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[2], NodeService_WatchPendingActionConfirmations_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[3], NodeService_WatchPendingActionConfirmations_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -479,7 +503,7 @@ func (c *nodeServiceClient) ListPeers(ctx context.Context, in *ListPeersRequest,
 
 func (c *nodeServiceClient) WatchPeers(ctx context.Context, in *WatchPeersRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListPeersResponse], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[3], NodeService_WatchPeers_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[4], NodeService_WatchPeers_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -558,7 +582,7 @@ func (c *nodeServiceClient) ListWorkers(ctx context.Context, in *ListWorkersRequ
 
 func (c *nodeServiceClient) WatchWorkers(ctx context.Context, in *ListWorkersRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListWorkersResponse], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[4], NodeService_WatchWorkers_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[5], NodeService_WatchWorkers_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -617,7 +641,7 @@ func (c *nodeServiceClient) ListJobs(ctx context.Context, in *ListJobsRequest, o
 
 func (c *nodeServiceClient) WatchJobs(ctx context.Context, in *WatchJobsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[ListJobsResponse], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[5], NodeService_WatchJobs_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[6], NodeService_WatchJobs_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -646,7 +670,7 @@ func (c *nodeServiceClient) CancelJob(ctx context.Context, in *CancelJobRequest,
 
 func (c *nodeServiceClient) WatchJobEvents(ctx context.Context, in *WatchJobEventsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JobEvent], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[6], NodeService_WatchJobEvents_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[7], NodeService_WatchJobEvents_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -765,7 +789,7 @@ func (c *nodeServiceClient) ListModels(ctx context.Context, in *ListModelsReques
 
 func (c *nodeServiceClient) PullModel(ctx context.Context, in *PullModelRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[PullModelProgress], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[7], NodeService_PullModel_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[8], NodeService_PullModel_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -794,7 +818,7 @@ func (c *nodeServiceClient) RemoveModel(ctx context.Context, in *RemoveModelRequ
 
 func (c *nodeServiceClient) Ask(ctx context.Context, in *AskRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[AskEvent], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[8], NodeService_Ask_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[9], NodeService_Ask_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -843,7 +867,7 @@ func (c *nodeServiceClient) DeleteChat(ctx context.Context, in *DeleteChatReques
 
 func (c *nodeServiceClient) StoreFile(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[StoreFileRequest, File], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[9], NodeService_StoreFile_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[10], NodeService_StoreFile_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -856,7 +880,7 @@ type NodeService_StoreFileClient = grpc.ClientStreamingClient[StoreFileRequest,
 
 func (c *nodeServiceClient) FetchFile(ctx context.Context, in *FetchFileRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[FetchFileResponse], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[10], NodeService_FetchFile_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[11], NodeService_FetchFile_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -955,7 +979,7 @@ func (c *nodeServiceClient) GetJoinedPool(ctx context.Context, in *GetJoinedPool
 
 func (c *nodeServiceClient) WatchJoinedPool(ctx context.Context, in *GetJoinedPoolRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JoinedPool], error) {
 	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
-	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[11], NodeService_WatchJoinedPool_FullMethodName, cOpts...)
+	stream, err := c.cc.NewStream(ctx, &NodeService_ServiceDesc.Streams[12], NodeService_WatchJoinedPool_FullMethodName, cOpts...)
 	if err != nil {
 		return nil, err
 	}
@@ -980,6 +1004,10 @@ type NodeServiceServer interface {
 	// WatchNodeState sends a complete node-and-peer snapshot at once and
 	// again whenever node details or peers change.
 	WatchNodeState(*GetNodeInfoRequest, grpc.ServerStreamingServer[NodeState]) error
+	// WatchSharedState sends one initial snapshot per requested domain, then
+	// complete domain snapshots whenever shared node state changes. Reconnect
+	// starts a new stream with a fresh baseline.
+	WatchSharedState(*WatchSharedStateRequest, grpc.ServerStreamingServer[SharedStateEvent]) error
 	// WatchPairedBrowsers sends the authorized device list at once and whenever
 	// a browser is paired or revoked.
 	WatchPairedBrowsers(*WatchPairedBrowsersRequest, grpc.ServerStreamingServer[PairedBrowsersState]) error
@@ -1156,6 +1184,9 @@ func (UnimplementedNodeServiceServer) GetNodeInfo(context.Context, *GetNodeInfoR
 func (UnimplementedNodeServiceServer) WatchNodeState(*GetNodeInfoRequest, grpc.ServerStreamingServer[NodeState]) error {
 	return status.Error(codes.Unimplemented, "method WatchNodeState not implemented")
 }
+func (UnimplementedNodeServiceServer) WatchSharedState(*WatchSharedStateRequest, grpc.ServerStreamingServer[SharedStateEvent]) error {
+	return status.Error(codes.Unimplemented, "method WatchSharedState not implemented")
+}
 func (UnimplementedNodeServiceServer) WatchPairedBrowsers(*WatchPairedBrowsersRequest, grpc.ServerStreamingServer[PairedBrowsersState]) error {
 	return status.Error(codes.Unimplemented, "method WatchPairedBrowsers not implemented")
 }
@@ -1383,6 +1414,17 @@ func _NodeService_WatchNodeState_Handler(srv interface{}, stream grpc.ServerStre
 // This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
 type NodeService_WatchNodeStateServer = grpc.ServerStreamingServer[NodeState]
 
+func _NodeService_WatchSharedState_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(WatchSharedStateRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(NodeServiceServer).WatchSharedState(m, &grpc.GenericServerStream[WatchSharedStateRequest, SharedStateEvent]{ServerStream: stream})
+}
+
+// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
+type NodeService_WatchSharedStateServer = grpc.ServerStreamingServer[SharedStateEvent]
+
 func _NodeService_WatchPairedBrowsers_Handler(srv interface{}, stream grpc.ServerStream) error {
 	m := new(WatchPairedBrowsersRequest)
 	if err := stream.RecvMsg(m); err != nil {
@@ -2574,6 +2616,11 @@ var NodeService_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _NodeService_WatchNodeState_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "WatchSharedState",
+			Handler:       _NodeService_WatchSharedState_Handler,
+			ServerStreams: true,
+		},
 		{
 			StreamName:    "WatchPairedBrowsers",
 			Handler:       _NodeService_WatchPairedBrowsers_Handler,

proto/sisyphus/node/v1/node.proto

@@ -9,6 +9,10 @@ service NodeService {
   // WatchNodeState sends a complete node-and-peer snapshot at once and
   // again whenever node details or peers change.
   rpc WatchNodeState(GetNodeInfoRequest) returns (stream NodeState);
+  // WatchSharedState sends one initial snapshot per requested domain, then
+  // complete domain snapshots whenever shared node state changes. Reconnect
+  // starts a new stream with a fresh baseline.
+  rpc WatchSharedState(WatchSharedStateRequest) returns (stream SharedStateEvent);
   // WatchPairedBrowsers sends the authorized device list at once and whenever
   // a browser is paired or revoked.
   rpc WatchPairedBrowsers(WatchPairedBrowsersRequest) returns (stream PairedBrowsersState);
@@ -1088,6 +1092,39 @@ message NodeState {
   uint64 revision = 3;
 }
 
+enum SharedStateDomain {
+  SHARED_STATE_DOMAIN_UNSPECIFIED = 0;
+  SHARED_STATE_DOMAIN_NODE = 1;
+  SHARED_STATE_DOMAIN_PEERS = 2;
+  SHARED_STATE_DOMAIN_WORKERS = 3;
+  SHARED_STATE_DOMAIN_PAIRED_BROWSERS = 4;
+  SHARED_STATE_DOMAIN_JOINED_POOL = 5;
+}
+
+message WatchSharedStateRequest {
+  // Empty requests every domain supported by this daemon. Repeated values are
+  // harmless and are treated as one subscription.
+  repeated SharedStateDomain domains = 1;
+}
+
+message SharedStateEvent {
+  // Increases for each event on this stream. A reconnect starts with a new
+  // authoritative baseline and a new revision sequence.
+  uint64 revision = 1;
+  SharedStateDomain domain = 2;
+  bool snapshot = 3;
+  // A domain can temporarily be unavailable (for example, workers on a node
+  // that coordinates no pool) without terminating other domain subscriptions.
+  string error = 4;
+  oneof state {
+    NodeState node = 10;
+    ListPeersResponse peers = 11;
+    ListWorkersResponse workers = 12;
+    PairedBrowsersState paired_browsers = 13;
+    JoinedPool joined_pool = 14;
+  }
+}
+
 message WatchPairedBrowsersRequest {}
 
 message RequestActionConfirmationRequest {