Keep degraded polling provider scoped through fallback

Author: rodboev

cmd/agentsview/archive_write_backend.go

@@ -925,6 +925,7 @@ func (b *localArchiveWriteBackend) PGPushWatch(
 			OnPollingRequired: func(obligation syncpkg.PollingObligation) error {
 				return poller.AddObligation(pollingObligation{
 					Key:           obligation.Key,
+					Agent:         obligation.Agent,
 					Roots:         obligation.Roots,
 					Probe:         obligation.Probe,
 					DegradedProbe: obligation.DegradedProbe,

cmd/agentsview/main.go

@@ -331,6 +331,7 @@ func runServe(cfg config.Config, opts serveOptions) {
 				OnPollingRequired: func(obligation sync.PollingObligation) error {
 					return unwatchedPoller.AddObligation(pollingObligation{
 						Key:           obligation.Key,
+						Agent:         obligation.Agent,
 						Roots:         obligation.Roots,
 						Probe:         obligation.Probe,
 						DegradedProbe: obligation.DegradedProbe,
@@ -2582,6 +2583,11 @@ func collectProviderWatchRoots(
 					def.Type, root, probeErr)
 				degradedProbe = nil
 			}
+		} else if def.Type == parser.AgentOpenCode {
+			degradedProbe = lateBoundDegradedPollProbe{
+				provider: provider,
+				root:     root,
+			}
 		}
 		_, err := os.Stat(root)
 		exists := err == nil

cmd/agentsview/main_test.go

@@ -77,6 +77,32 @@ func TestLateBoundDegradedPollProbeActivatesWhenOpenCodeRootBecomesSQLite(t *tes
 	assert.NotEmpty(t, state)
 }
 
+func TestLateBoundDegradedPollProbeFallsBackWhenOpenCodeRootBecomesHybrid(t *testing.T) {
+	root := t.TempDir()
+	provider, ok := parser.NewProvider(parser.AgentOpenCode, parser.ProviderConfig{
+		Roots: []string{root},
+	})
+	require.True(t, ok)
+	probe := lateBoundDegradedPollProbe{provider: provider, root: root}
+
+	db, err := sql.Open("sqlite3", filepath.Join(root, "opencode.db"))
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, db.Close()) })
+	require.NoError(t, db.Ping())
+	_, err = db.Exec(`CREATE TABLE probe_state (id INTEGER PRIMARY KEY)`)
+	require.NoError(t, err)
+
+	state, err := probe.DegradedPollingState(t.Context())
+	require.NoError(t, err)
+	assert.NotEmpty(t, state)
+
+	require.NoError(t,
+		os.MkdirAll(filepath.Join(root, "storage", "session"), 0o755))
+
+	_, err = probe.DegradedPollingState(t.Context())
+	require.ErrorIs(t, err, parser.ErrUnsupportedProviderFeature)
+}
+
 func TestServeRuntimeRecordWriteFailureWarnsVisibleAfterSlowStartup(t *testing.T) {
 	out, err := runServeRuntimeWarningHelper(t, true, 1200*time.Millisecond)
 	require.NoError(t, err, string(out))
@@ -567,12 +593,24 @@ func (f *fakeUnwatchedPollSyncer) ReconcileWatchRoots(
 	return nil
 }
 
+func (f *fakeUnwatchedPollSyncer) ReconcileProviderRoots(
+	ctx context.Context, _ parser.AgentType, roots []string,
+) error {
+	return f.ReconcileWatchRoots(ctx, roots, false)
+}
+
 func TestPollUnwatchedRootsOnceUsesScopedAuthoritativeReconciliation(t *testing.T) {
 	fake := &fakeUnwatchedPollSyncer{}
 	roots := []string{"/tmp/claude", "/tmp/codex"}
 
-	pollUnwatchedRootsOnce(t.Context(), fake, roots)
-	pollUnwatchedRootsOnce(t.Context(), fake, roots)
+	pollUnwatchedRootsOnce(t.Context(), fake, []pollingObligation{{
+		Key: "roots",
+		Roots: roots,
+	}})
+	pollUnwatchedRootsOnce(t.Context(), fake, []pollingObligation{{
+		Key: "roots",
+		Roots: roots,
+	}})
 
 	require.Equal(t, 2, fake.calls)
 	assert.Equal(t, roots, fake.callRoots[0])

cmd/agentsview/unwatched_poll.go

@@ -20,6 +20,7 @@ var errUnwatchedPollStopped = errors.New("unwatched poll coordinator stopped")
 
 type unwatchedPollSyncer interface {
 	ReconcileWatchRoots(context.Context, []string, bool) error
+	ReconcileProviderRoots(context.Context, parser.AgentType, []string) error
 }
 
 type unwatchedPollAdd struct {
@@ -30,6 +31,7 @@ type unwatchedPollAdd struct {
 
 type pollingObligation struct {
 	Key   string
+	Agent parser.AgentType
 	Roots []string
 	// Probe mirrors sync.PollingObligation.Probe: the physical watcher path
 	// whose availability gates this obligation's reconciliation Roots. When
@@ -173,8 +175,9 @@ func (c *sharedUnwatchedPollCoordinator) run() {
 			} else {
 				obligations[request.obligation.Key] = request.obligation
 			}
-			c.setPollObligations(obligations)
+			snapshot := sortedPollObligations(obligations)
 			c.pollMu.Lock()
+			c.pollObligations = snapshot
 			c.pollRevisions[request.obligation.Key]++
 			delete(c.pollStates, request.obligation.Key)
 			c.pollMu.Unlock()
@@ -188,19 +191,17 @@ func (c *sharedUnwatchedPollCoordinator) run() {
 	}
 }
 
-func (c *sharedUnwatchedPollCoordinator) setPollObligations(
+func sortedPollObligations(
 	obligations map[string]pollingObligation,
-) {
+) []pollingObligation {
 	snapshot := make([]pollingObligation, 0, len(obligations))
 	for _, obligation := range obligations {
 		snapshot = append(snapshot, obligation)
 	}
 	slices.SortFunc(snapshot, func(a, b pollingObligation) int {
 		return strings.Compare(a.Key, b.Key)
 	})
-	c.pollMu.Lock()
-	c.pollObligations = snapshot
-	c.pollMu.Unlock()
+	return snapshot
 }
 
 func (c *sharedUnwatchedPollCoordinator) currentPollObligations() []pollingObligation {
@@ -232,18 +233,18 @@ func (c *sharedUnwatchedPollCoordinator) runPollWorker() {
 				return
 			}
 			obligations, priorStates, priorRevisions := c.currentPollSnapshot()
-			roots, nextStates := c.preparePollRun(
+			selected, nextStates := c.preparePollRun(
 				c.workerCtx, obligations, priorStates,
 			)
-			if len(roots) == 0 {
+			if len(selected) == 0 {
 				continue
 			}
-			log.Printf("polling %d unwatched root(s)", len(roots))
+			log.Printf("polling %d unwatched root(s)", len(unwatchedPollObligationSliceRoots(selected)))
 			c.doWork(func() {
 				if c.workerCtx.Err() != nil {
 					return
 				}
-				if err := pollUnwatchedRootsOnce(c.workerCtx, c.engine, roots); err != nil {
+				if err := pollUnwatchedRootsOnce(c.workerCtx, c.engine, selected); err != nil {
 					return
 				}
 				c.commitPollStates(nextStates, priorRevisions)
@@ -270,10 +271,10 @@ func (c *sharedUnwatchedPollCoordinator) runPollWorker() {
 // deferred scope as an authoritative empty discovery and tombstone its
 // sessions.
 func availableUnwatchedPollRoots(obligations []pollingObligation) []string {
-	roots, _ := availableUnwatchedPollRootsWithState(
+	selected, _ := availableUnwatchedPollObligationsWithState(
 		context.Background(), obligations, nil,
 	)
-	return roots
+	return unwatchedPollObligationSliceRoots(selected)
 }
 
 func unwatchedPollObligationRoots(obligations map[string]pollingObligation) []string {
@@ -288,6 +289,18 @@ func unwatchedPollObligationRoots(obligations map[string]pollingObligation) []st
 	return unwatchedPollRoots(owned)
 }
 
+func unwatchedPollObligationSliceRoots(obligations []pollingObligation) []string {
+	owned := make(map[string]struct{})
+	for _, obligation := range obligations {
+		for _, root := range obligation.Roots {
+			if root != "" {
+				owned[root] = struct{}{}
+			}
+		}
+	}
+	return unwatchedPollRoots(owned)
+}
+
 func unwatchedPollRoots(owned map[string]struct{}) []string {
 	roots := make([]string, 0, len(owned))
 	for root := range owned {
@@ -298,14 +311,47 @@ func unwatchedPollRoots(owned map[string]struct{}) []string {
 }
 
 func pollUnwatchedRootsOnce(
-	ctx context.Context, engine unwatchedPollSyncer, roots []string,
+	ctx context.Context, engine unwatchedPollSyncer, obligations []pollingObligation,
 ) error {
-	if len(roots) == 0 {
+	if len(obligations) == 0 {
 		return nil
 	}
-	if err := engine.ReconcileWatchRoots(ctx, roots, false); err != nil {
-		log.Printf("polling unwatched roots: %v", err)
-		return err
+	watchRoots := make(map[string]struct{})
+	byAgent := make(map[parser.AgentType]map[string]struct{})
+	for _, obligation := range obligations {
+		target := watchRoots
+		if obligation.Agent != "" {
+			if byAgent[obligation.Agent] == nil {
+				byAgent[obligation.Agent] = make(map[string]struct{})
+			}
+			target = byAgent[obligation.Agent]
+		}
+		for _, root := range obligation.Roots {
+			if root != "" {
+				target[root] = struct{}{}
+			}
+		}
+	}
+	if roots := unwatchedPollRoots(watchRoots); len(roots) > 0 {
+		if err := engine.ReconcileWatchRoots(ctx, roots, false); err != nil {
+			log.Printf("polling unwatched roots: %v", err)
+			return err
+		}
+	}
+	agents := make([]parser.AgentType, 0, len(byAgent))
+	for agent := range byAgent {
+		agents = append(agents, agent)
+	}
+	slices.Sort(agents)
+	for _, agent := range agents {
+		roots := unwatchedPollRoots(byAgent[agent])
+		if len(roots) == 0 {
+			continue
+		}
+		if err := engine.ReconcileProviderRoots(ctx, agent, roots); err != nil {
+			log.Printf("polling unwatched %s roots: %v", agent, err)
+			return err
+		}
 	}
 	return nil
 }
@@ -326,8 +372,8 @@ func (c *sharedUnwatchedPollCoordinator) preparePollRun(
 	ctx context.Context,
 	obligations []pollingObligation,
 	prior map[string]string,
-) ([]string, map[string]string) {
-	return availableUnwatchedPollRootsWithState(
+) ([]pollingObligation, map[string]string) {
+	return availableUnwatchedPollObligationsWithState(
 		ctx, obligations, prior,
 	)
 }
@@ -349,15 +395,14 @@ func (c *sharedUnwatchedPollCoordinator) commitPollStates(
 	}
 }
 
-func availableUnwatchedPollRootsWithState(
+func availableUnwatchedPollObligationsWithState(
 	ctx context.Context,
 	obligations []pollingObligation,
 	prior map[string]string,
-) ([]string, map[string]string) {
-	candidates := make(map[string]struct{})
+) ([]pollingObligation, map[string]string) {
 	blocked := make(map[string]struct{})
-	candidateOwners := make(map[string][]string)
 	pendingStates := make(map[string]string)
+	selected := make([]pollingObligation, 0, len(obligations))
 	for _, obligation := range obligations {
 		probeMissing := false
 		if obligation.Probe != "" {
@@ -382,33 +427,42 @@ func availableUnwatchedPollRootsWithState(
 				pendingStates[obligation.Key] = state
 			}
 		}
+		roots := make([]string, 0, len(obligation.Roots))
 		for _, root := range obligation.Roots {
 			if root == "" {
 				continue
 			}
 			if _, err := os.Stat(root); err == nil {
-				candidates[root] = struct{}{}
-				candidateOwners[root] = append(candidateOwners[root], obligation.Key)
+				roots = append(roots, root)
 			}
 		}
-	}
-	for root := range candidates {
-		if overlapsDeferredScope(filepath.Clean(root), blocked) {
-			delete(candidates, root)
+		if len(roots) == 0 {
+			continue
 		}
+		selected = append(selected, pollingObligation{
+			Key:           obligation.Key,
+			Agent:         obligation.Agent,
+			Roots:         roots,
+			Probe:         obligation.Probe,
+			DegradedProbe: obligation.DegradedProbe,
+		})
 	}
-	if len(pendingStates) == 0 {
-		return unwatchedPollRoots(candidates), nil
-	}
-	committed := make(map[string]string)
-	for root := range candidates {
-		for _, key := range candidateOwners[root] {
-			if state, ok := pendingStates[key]; ok {
-				committed[key] = state
+	filtered := make([]pollingObligation, 0, len(selected))
+	for _, obligation := range selected {
+		deferred := false
+		for _, root := range obligation.Roots {
+			if overlapsDeferredScope(filepath.Clean(root), blocked) {
+				deferred = true
+				break
 			}
 		}
+		if deferred {
+			delete(pendingStates, obligation.Key)
+			continue
+		}
+		filtered = append(filtered, obligation)
 	}
-	return unwatchedPollRoots(candidates), committed
+	return filtered, pendingStates
 }
 
 func clonePollStates(states map[string]string) map[string]string {