fix(sync): refresh inherited Omnigent metadata

Unsupported parser outcomes may be persisted as clean skip entries, so a cold tracker must preserve that result instead of turning restart validation into a failure.

Root workspace and branch changes also affect archived descendants even when their own source rows are unchanged. Resolving those virtual sources through the archive’s indexed parent relationships keeps incremental correctness proportional to the affected subagent tree.

Author: wesm

internal/db/sessions.go

@@ -2251,6 +2251,85 @@ func (db *DB) ListSessionIDsByFilePath(path, agent string) ([]string, error) {
 	return ids, nil
 }
 
+const descendantSessionRootBatchSize = 100
+
+// ListActiveDescendantSessionSourcePaths returns the source paths of active
+// descendants already linked beneath parentIDs. Both the seed and recursive
+// steps use idx_sessions_parent, so work scales with the affected subagent
+// trees rather than every archived session.
+func (db *DB) ListActiveDescendantSessionSourcePaths(
+	ctx context.Context,
+	machine, agent string,
+	parentIDs []string,
+) ([]string, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	seen := make(map[string]struct{})
+	var paths []string
+	for start := 0; start < len(parentIDs); start += descendantSessionRootBatchSize {
+		end := min(start+descendantSessionRootBatchSize, len(parentIDs))
+		batch := parentIDs[start:end]
+		placeholders := strings.TrimSuffix(strings.Repeat("?,", len(batch)), ",")
+		query := `
+			WITH RECURSIVE descendants(id, file_path) AS (
+				SELECT id, file_path
+				  FROM sessions INDEXED BY idx_sessions_parent
+				 WHERE parent_session_id IS NOT NULL
+				   AND parent_session_id IN (` + placeholders + `)
+				   AND machine = ? AND agent = ? AND deleted_at IS NULL
+				UNION
+				SELECT s.id, s.file_path
+				  FROM sessions AS s INDEXED BY idx_sessions_parent
+				  JOIN descendants AS d ON s.parent_session_id = d.id
+				 WHERE s.parent_session_id IS NOT NULL
+				   AND s.machine = ? AND s.agent = ? AND s.deleted_at IS NULL
+			)
+			SELECT file_path
+			  FROM descendants
+			 WHERE file_path IS NOT NULL AND file_path <> ''
+			 ORDER BY file_path`
+		args := make([]any, 0, len(batch)+4)
+		for _, id := range batch {
+			args = append(args, id)
+		}
+		args = append(args, machine, agent, machine, agent)
+		rows, err := db.getReader().QueryContext(ctx, query, args...)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"listing active descendant session sources: %w", err,
+			)
+		}
+		for rows.Next() {
+			var path string
+			if err := rows.Scan(&path); err != nil {
+				_ = rows.Close()
+				return nil, fmt.Errorf(
+					"scanning active descendant session source: %w", err,
+				)
+			}
+			if _, exists := seen[path]; exists {
+				continue
+			}
+			seen[path] = struct{}{}
+			paths = append(paths, path)
+		}
+		if err := rows.Err(); err != nil {
+			_ = rows.Close()
+			return nil, fmt.Errorf(
+				"iterating active descendant session sources: %w", err,
+			)
+		}
+		if err := rows.Close(); err != nil {
+			return nil, fmt.Errorf(
+				"closing active descendant session sources: %w", err,
+			)
+		}
+	}
+	sort.Strings(paths)
+	return paths, nil
+}
+
 const storedSourcePathHintRootBatchSize = 100
 
 // StoredSourcePathHintScope identifies one affected stored-source prefix.

internal/parser/omnigent_provider.go

@@ -245,6 +245,20 @@ func IsOmnigentContainerSource(source SourceRef) bool {
 	return ok && src.Container != "" && src.MemberID == ""
 }
 
+// OmnigentMemberSessionID returns the archived session identity addressed by a
+// virtual Omnigent member source. The sync engine uses it to look up already
+// archived descendants whose cwd and branch inherit from a changed root.
+func OmnigentMemberSessionID(source SourceRef) (string, bool) {
+	if source.Provider != AgentOmnigent {
+		return "", false
+	}
+	src, ok := source.Opaque.(multiSessionSource)
+	if !ok || src.MemberID == "" {
+		return "", false
+	}
+	return omnigentIDPrefix + src.MemberID, true
+}
+
 func omnigentLegacySessionIDs(
 	src multiSessionSource, results []ParseResult,
 ) []string {
@@ -1145,6 +1159,13 @@ func (t *omnigentChangeTracker) restoreCachedContainer(
 	defer conn.Close()
 	schema, err := detectOmnigentSchema(conn)
 	if err != nil {
+		if omnigentSchemaUnsupported(err) {
+			// Unsupported parse outcomes are intentionally skip-cached. A
+			// restart may validate that cache entry with a cold tracker; the
+			// known unsupported state still proves the cached skip, but cannot
+			// seed supported-schema change cursors.
+			return false, nil
+		}
 		return false, err
 	}
 	conversationRowID, conversationTail, err :=

internal/sync/engine.go

@@ -1103,6 +1103,21 @@ func (e *Engine) classifyProviderChangedPath(
 				}
 				continue
 			}
+			if agentType == parser.AgentOmnigent {
+				sources, err = e.expandOmnigentInheritedMetadataSources(
+					ctx, provider, sources,
+				)
+				if err != nil {
+					classificationErr = errors.Join(
+						classificationErr,
+						fmt.Errorf(
+							"%s provider dependent-source classification for %q: %w",
+							def.Type, path, err,
+						),
+					)
+					continue
+				}
+			}
 			for _, source := range sources {
 				sourcePath := providerDiscoveredPath(source)
 				if sourcePath == "" {
@@ -1148,6 +1163,60 @@ func (e *Engine) classifyProviderChangedPath(
 	return files, classificationErr
 }
 
+func (e *Engine) expandOmnigentInheritedMetadataSources(
+	ctx context.Context,
+	provider parser.Provider,
+	sources []parser.SourceRef,
+) ([]parser.SourceRef, error) {
+	resolver, ok := provider.(parser.ReconciliationSourceResolver)
+	if !ok || len(sources) == 0 {
+		return sources, nil
+	}
+	seenSources := make(map[string]struct{}, len(sources))
+	parentIDs := make([]string, 0, len(sources))
+	seenParents := make(map[string]struct{}, len(sources))
+	for _, source := range sources {
+		if path := providerDiscoveredPath(source); path != "" {
+			seenSources[path] = struct{}{}
+		}
+		id, member := parser.OmnigentMemberSessionID(source)
+		if !member {
+			continue
+		}
+		id = applyIDPrefixToID(e.idPrefix, id)
+		if _, exists := seenParents[id]; exists {
+			continue
+		}
+		seenParents[id] = struct{}{}
+		parentIDs = append(parentIDs, id)
+	}
+	paths, err := e.db.ListActiveDescendantSessionSourcePaths(
+		ctx, e.machine, string(parser.AgentOmnigent), parentIDs,
+	)
+	if err != nil {
+		return nil, err
+	}
+	for _, path := range paths {
+		if _, exists := seenSources[path]; exists {
+			continue
+		}
+		source, found, err := resolver.SourceForReconciliation(ctx, path, "")
+		if err != nil {
+			return nil, err
+		}
+		if !found {
+			continue
+		}
+		sourcePath := providerDiscoveredPath(source)
+		if sourcePath == "" {
+			continue
+		}
+		seenSources[sourcePath] = struct{}{}
+		sources = append(sources, source)
+	}
+	return sources, nil
+}
+
 func storedSourceDBHintScopes(
 	scopes []parser.StoredSourceHintScope,
 ) []db.StoredSourcePathHintScope {

internal/sync/omnigent_integration_test.go

@@ -1303,6 +1303,92 @@ func TestAuditOmnigentDetectsMultiWorkspaceMetadataOnlyEdit(t *testing.T) {
 		"authoritative reconciliation must refresh multi-workspace metadata")
 }
 
+func TestSyncPathsOmnigentRootMetadataRefreshesExistingSubagent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	observed := make(map[int]int64)
+	for _, archiveSize := range []int{130, 1030} {
+		t.Run(fmt.Sprintf("archive_%d", archiveSize), func(t *testing.T) {
+			root := t.TempDir()
+			dbPath := writeOmnigentSplitSyncDB(t, root, archiveSize)
+			writer, err := sql.Open("sqlite3", dbPath)
+			require.NoError(t, err)
+			_, err = writer.Exec(`
+				UPDATE conversations
+				   SET parent_conversation_id = 'conv_0000',
+				       root_conversation_id = 'conv_0000'
+				 WHERE workspace_id = 0 AND id = 'conv_0001'`)
+			require.NoError(t, err)
+			_, err = writer.Exec(`
+				UPDATE omnigent_conversation_metadata
+				   SET workspace = '/work/before', git_branch = 'main'
+				 WHERE workspace_id = 0 AND id = 'conv_0000'`)
+			require.NoError(t, err)
+			_, err = writer.Exec(`
+				UPDATE omnigent_conversation_metadata
+				   SET kind = 2, workspace = '', git_branch = ''
+				 WHERE workspace_id = 0 AND id = 'conv_0001'`)
+			require.NoError(t, err)
+			require.NoError(t, writer.Close())
+
+			archive := dbtest.OpenTestDB(t)
+			var resultCount atomic.Int64
+			factory := omnigentParseCountingFactory{
+				delegate: omnigentDefaultProviderFactory(t),
+				count:    new(atomic.Int64),
+				results:  &resultCount,
+			}
+			engine := sync.NewEngine(archive, sync.EngineConfig{
+				AgentDirs: map[parser.AgentType][]string{
+					parser.AgentOmnigent: {root},
+				},
+				Machine:           "local",
+				ProviderFactories: []parser.ProviderFactory{factory},
+			})
+			t.Cleanup(engine.Close)
+			syncOmnigentArchive(t, engine, archive, archiveSize)
+			childID := "omnigent:0:conv_0001"
+			before, err := archive.GetSession(t.Context(), childID)
+			require.NoError(t, err)
+			require.NotNil(t, before)
+			assert.Equal(t, "/work/before", before.Cwd)
+			assert.Equal(t, "before", before.Project)
+			assert.Equal(t, "main", before.GitBranch)
+
+			writer, err = sql.Open("sqlite3", dbPath)
+			require.NoError(t, err)
+			_, err = writer.Exec(`
+				UPDATE conversations
+				   SET updated_at = ?
+				 WHERE workspace_id = 0 AND id = 'conv_0000'`,
+				time.Now().Unix(),
+			)
+			require.NoError(t, err)
+			_, err = writer.Exec(`
+				UPDATE omnigent_conversation_metadata
+				   SET workspace = '/work/after', git_branch = 'review'
+				 WHERE workspace_id = 0 AND id = 'conv_0000'`)
+			require.NoError(t, err)
+			require.NoError(t, writer.Close())
+
+			resultCount.Store(0)
+			require.NoError(t, engine.SyncPathsContext(
+				t.Context(), []string{dbPath},
+			))
+			observed[archiveSize] = resultCount.Load()
+			after, err := archive.GetSession(t.Context(), childID)
+			require.NoError(t, err)
+			require.NotNil(t, after)
+			assert.Equal(t, "/work/after", after.Cwd)
+			assert.Equal(t, "after", after.Project)
+			assert.Equal(t, "review", after.GitBranch)
+		})
+	}
+	assert.Equal(t, observed[130], observed[1030],
+		"dependent metadata refresh work must not grow with unrelated conversations")
+}
+
 func TestScheduledOmnigentReconciliationIsBoundedByChangedMembers(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
@@ -1610,7 +1696,19 @@ func TestSyncOmnigentUnsupportedSchemaPreservesArchive(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, writer.Close())
 
-	engine.SyncAll(context.Background(), nil)
+	firstUnsupported := engine.SyncAll(context.Background(), nil)
+	assert.Zero(t, firstUnsupported.Failed)
+	engine.Close()
+	restarted := sync.NewEngine(archive, sync.EngineConfig{
+		AgentDirs: map[parser.AgentType][]string{
+			parser.AgentOmnigent: {root},
+		},
+		Machine: "local",
+	})
+	t.Cleanup(restarted.Close)
+	secondUnsupported := restarted.SyncAll(context.Background(), nil)
+	assert.Zero(t, secondUnsupported.Failed,
+		"a cached unsupported source must remain a clean skip")
 	after, err := archive.GetSession(context.Background(), "omnigent:conv_0000")
 	require.NoError(t, err)
 	require.NotNil(t, after, "unsupported source must not retire archived sessions")