feat: improve population of sequence log (#2752)

Author: matthiasn

CHANGELOG.md

@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   lowered from 384 to 256 estimated tokens to leave sufficient headroom for
   content where the word-based heuristic undercounts.
 
+### Added
+- Sequence log backfill now includes agent entities and agent links alongside
+  journal entries and entry links, closing sync gaps for agent data.
+
 ## [0.9.903] - 2026-03-06
 ### Fixed
 - AI task agent no longer repeatedly proposes checklist items that already exist

flatpak/com.matthiasn.lotti.metainfo.xml

@@ -34,6 +34,7 @@
     <release version="0.9.904" date="2026-03-06">
       <description>
           <p>Fixed embedding backfill failing for entries with code, URLs, or long unpunctuated text by lowering the chunk target to 256 tokens for safer headroom against the model's 512-token context window.</p>
+          <p>Sequence log backfill now includes agent entities and agent links alongside journal entries and entry links, closing sync gaps for agent data.</p>
       </description>
     </release>
     <release version="0.9.903" date="2026-03-06">

lib/features/agents/database/agent_database.dart

@@ -1,3 +1,4 @@
+import 'dart:convert';
 import 'dart:io';
 
 import 'package:drift/drift.dart';
@@ -69,4 +70,96 @@ class AgentDatabase extends _$AgentDatabase {
       },
     );
   }
+
+  /// Stream agent entities with their vector clocks for populating the
+  /// sequence log. Yields batches of records with entity ID and vector
+  /// clock map. Uses lightweight SQL + JSON extraction to avoid full
+  /// deserialization.
+  Stream<List<({String id, Map<String, int>? vectorClock})>>
+      streamAgentEntitiesWithVectorClock({int batchSize = 1000}) async* {
+    var offset = 0;
+
+    while (true) {
+      final rows = await customSelect(
+        'SELECT id, serialized FROM agent_entities '
+        'ORDER BY id LIMIT ? OFFSET ?',
+        variables: [Variable(batchSize), Variable(offset)],
+      ).get();
+
+      if (rows.isEmpty) break;
+
+      yield rows
+          .map(
+            (row) => (
+              id: row.read<String>('id'),
+              vectorClock: _extractVectorClock(row.read<String>('serialized')),
+            ),
+          )
+          .toList();
+      offset += batchSize;
+    }
+  }
+
+  /// Stream agent links with their vector clocks for populating the
+  /// sequence log. Yields batches of records with link ID and vector
+  /// clock map.
+  Stream<List<({String id, Map<String, int>? vectorClock})>>
+      streamAgentLinksWithVectorClock({int batchSize = 1000}) async* {
+    var offset = 0;
+
+    while (true) {
+      final rows = await customSelect(
+        'SELECT id, serialized FROM agent_links '
+        'ORDER BY id LIMIT ? OFFSET ?',
+        variables: [Variable(batchSize), Variable(offset)],
+      ).get();
+
+      if (rows.isEmpty) break;
+
+      yield rows
+          .map(
+            (row) => (
+              id: row.read<String>('id'),
+              vectorClock: _extractVectorClock(row.read<String>('serialized')),
+            ),
+          )
+          .toList();
+      offset += batchSize;
+    }
+  }
+
+  /// Count total agent entities for progress reporting.
+  Future<int> countAllAgentEntities() => _countAll('agent_entities');
+
+  /// Count total agent links for progress reporting.
+  Future<int> countAllAgentLinks() => _countAll('agent_links');
+
+  Future<int> _countAll(String tableName) async {
+    final result = await customSelect(
+      'SELECT COUNT(*) AS cnt FROM $tableName',
+    ).getSingle();
+    return result.read<int>('cnt');
+  }
+
+  /// Lightweight extraction of vector clock from serialized JSON.
+  /// Agent entities and links store vectorClock at JSON root level.
+  /// Returns null for any malformed data rather than throwing.
+  static Map<String, int>? _extractVectorClock(String serialized) {
+    try {
+      final decoded = jsonDecode(serialized);
+      if (decoded is! Map<String, dynamic>) return null;
+
+      final vc = decoded['vectorClock'];
+      if (vc is! Map<String, dynamic>) return null;
+
+      final result = <String, int>{};
+      for (final entry in vc.entries) {
+        if (entry.value is! num) return null;
+        result[entry.key] = (entry.value as num).toInt();
+      }
+      return result;
+    } on Object catch (_) {
+      return null;
+    }
+  }
 }

lib/features/agents/state/agent_providers.dart

@@ -110,12 +110,10 @@ DomainLogger domainLogger(Ref ref) {
   return logger;
 }
 
-/// The agent database instance (lazy singleton).
+/// The agent database instance (singleton via GetIt).
 @Riverpod(keepAlive: true)
 AgentDatabase agentDatabase(Ref ref) {
-  final db = AgentDatabase();
-  ref.onDispose(db.close);
-  return db;
+  return getIt<AgentDatabase>();
 }
 
 /// The agent repository wrapping the database.

lib/features/sync/sequence/sync_sequence_log_service.dart

@@ -729,150 +729,102 @@ class SyncSequenceLogService {
   }
 
   /// Populate the sequence log from existing journal entries.
-  /// This is used to backfill the sequence log for entries that were
-  /// created before the sequence log feature was added.
-  ///
-  /// This method streams journal entries in batches and records their vector
-  /// clocks in the sequence log. Records entries for ALL hosts in each entry's
-  /// vector clock so any device with the entry can respond to backfill requests.
-  ///
-  /// [onProgress] is called with progress from 0.0 to 1.0 as entries are
-  /// processed.
-  ///
-  /// Returns the number of entries populated.
+  /// Returns the number of sequence log entries populated.
   Future<int> populateFromJournal({
     required Stream<List<({String id, Map<String, int>? vectorClock})>>
         entryStream,
     required Future<int> Function() getTotalCount,
     void Function(double progress)? onProgress,
-  }) async {
-    final total = await getTotalCount();
-    var processed = 0;
-    var populated = 0;
-    final now = DateTime.now();
-
-    // Cache of existing (hostId, counter) pairs to avoid duplicates
-    // We'll populate this lazily per-host as we encounter them
-    final existingByHost = <String, Set<int>>{};
-
-    await for (final batch in entryStream) {
-      final toInsert = <SyncSequenceLogCompanion>[];
-
-      for (final entry in batch) {
-        processed++;
-
-        final vc = entry.vectorClock;
-        if (vc == null || vc.isEmpty) continue;
-
-        // Find the originating host (the one with the highest counter,
-        // which is typically the creator of this specific entry version)
-        String? originatingHost;
-        var maxCounter = 0;
-        for (final e in vc.entries) {
-          if (e.value > maxCounter) {
-            maxCounter = e.value;
-            originatingHost = e.key;
-          }
-        }
-
-        // Record entry for each host in the vector clock
-        for (final vcEntry in vc.entries) {
-          final hostId = vcEntry.key;
-          final counter = vcEntry.value;
-
-          // Lazily load existing counters for this host
-          if (!existingByHost.containsKey(hostId)) {
-            existingByHost[hostId] =
-                await _syncDatabase.getCountersForHost(hostId);
-          }
-
-          final existing = existingByHost[hostId]!;
-
-          // Skip if already exists
-          if (existing.contains(counter)) continue;
-
-          // Mark as existing to avoid duplicates within this run
-          existing.add(counter);
-
-          toInsert.add(
-            SyncSequenceLogCompanion(
-              hostId: Value(hostId),
-              counter: Value(counter),
-              entryId: Value(entry.id),
-              originatingHostId: Value(originatingHost ?? hostId),
-              status: Value(SyncSequenceStatus.received.index),
-              createdAt: Value(now),
-              updatedAt: Value(now),
-            ),
-          );
-        }
-      }
-
-      // Batch insert
-      if (toInsert.isNotEmpty) {
-        await _syncDatabase.batchInsertSequenceEntries(toInsert);
-        populated += toInsert.length;
-      }
-
-      // Report progress after each batch
-      if (onProgress != null && total > 0) {
-        onProgress(processed / total);
-      }
-    }
-
-    if (populated > 0) {
-      _loggingService.captureEvent(
-        'populateFromJournal: added $populated sequence log entries',
-        domain: 'SYNC_SEQUENCE',
-        subDomain: 'populate',
-      );
-    }
-
-    return populated;
+  }) {
+    return _populateFromStream(
+      dataStream: entryStream,
+      getTotalCount: getTotalCount,
+      onProgress: onProgress,
+      payloadType: SyncSequencePayloadType.journalEntity,
+      label: 'populateFromJournal',
+    );
   }
 
   /// Populate the sequence log from existing entry links.
-  /// This is used to backfill the sequence log for entry links that were
-  /// created before the sequence log feature was added, or to resolve
-  /// "ghost missing" counters that correspond to EntryLink operations.
-  ///
-  /// This method streams entry links in batches and records their vector
-  /// clocks in the sequence log with [SyncSequencePayloadType.entryLink].
-  /// Records entries for ALL hosts in each link's vector clock so any device
-  /// with the link can respond to backfill requests.
-  ///
-  /// [onProgress] is called with progress from 0.0 to 1.0 as links are
-  /// processed.
-  ///
-  /// Returns the number of entries populated.
+  /// Returns the number of sequence log entries populated.
   Future<int> populateFromEntryLinks({
     required Stream<List<({String id, Map<String, int>? vectorClock})>>
         linkStream,
     required Future<int> Function() getTotalCount,
     void Function(double progress)? onProgress,
+  }) {
+    return _populateFromStream(
+      dataStream: linkStream,
+      getTotalCount: getTotalCount,
+      onProgress: onProgress,
+      payloadType: SyncSequencePayloadType.entryLink,
+      label: 'populateFromEntryLinks',
+    );
+  }
+
+  /// Populate the sequence log from existing agent entities.
+  /// Returns the number of sequence log entries populated.
+  Future<int> populateFromAgentEntities({
+    required Stream<List<({String id, Map<String, int>? vectorClock})>>
+        entityStream,
+    required Future<int> Function() getTotalCount,
+    void Function(double progress)? onProgress,
+  }) {
+    return _populateFromStream(
+      dataStream: entityStream,
+      getTotalCount: getTotalCount,
+      onProgress: onProgress,
+      payloadType: SyncSequencePayloadType.agentEntity,
+      label: 'populateFromAgentEntities',
+    );
+  }
+
+  /// Populate the sequence log from existing agent links.
+  /// Returns the number of sequence log entries populated.
+  Future<int> populateFromAgentLinks({
+    required Stream<List<({String id, Map<String, int>? vectorClock})>>
+        linkStream,
+    required Future<int> Function() getTotalCount,
+    void Function(double progress)? onProgress,
+  }) {
+    return _populateFromStream(
+      dataStream: linkStream,
+      getTotalCount: getTotalCount,
+      onProgress: onProgress,
+      payloadType: SyncSequencePayloadType.agentLink,
+      label: 'populateFromAgentLinks',
+    );
+  }
+
+  /// Shared implementation for populating the sequence log from a paginated
+  /// stream of records with vector clocks. Used by all four populate methods.
+  Future<int> _populateFromStream({
+    required Stream<List<({String id, Map<String, int>? vectorClock})>>
+        dataStream,
+    required Future<int> Function() getTotalCount,
+    required SyncSequencePayloadType payloadType,
+    required String label,
+    void Function(double progress)? onProgress,
   }) async {
     final total = await getTotalCount();
     var processed = 0;
     var populated = 0;
     final now = DateTime.now();
 
     // Cache of existing (hostId, counter) pairs to avoid duplicates
-    // We'll populate this lazily per-host as we encounter them
     final existingByHost = <String, Set<int>>{};
 
-    await for (final batch in linkStream) {
+    await for (final batch in dataStream) {
       final toInsert = <SyncSequenceLogCompanion>[];
 
-      for (final link in batch) {
+      for (final record in batch) {
         processed++;
 
-        final vc = link.vectorClock;
+        final vc = record.vectorClock;
         if (vc == null || vc.isEmpty) continue;
 
-        // Find the originating host (the one with the highest counter,
-        // which is typically the creator of this specific link version).
-        // Sort entries by host ID first to ensure deterministic tie-breaking
-        // when multiple hosts have the same max counter.
+        // Find the originating host (the one with the highest counter).
+        // Sort entries by host ID for deterministic tie-breaking.
         String? originatingHost;
         var maxCounter = -1;
         final sortedEntries = vc.entries.toList()
@@ -907,8 +859,8 @@ class SyncSequenceLogService {
             SyncSequenceLogCompanion(
               hostId: Value(hostId),
               counter: Value(counter),
-              entryId: Value(link.id),
-              payloadType: Value(SyncSequencePayloadType.entryLink.index),
+              entryId: Value(record.id),
+              payloadType: Value(payloadType.index),
               originatingHostId: Value(originatingHost ?? hostId),
               status: Value(SyncSequenceStatus.received.index),
               createdAt: Value(now),
@@ -932,7 +884,7 @@ class SyncSequenceLogService {
 
     if (populated > 0) {
       _loggingService.captureEvent(
-        'populateFromEntryLinks: added $populated sequence log entries',
+        '$label: added $populated sequence log entries',
         domain: 'SYNC_SEQUENCE',
         subDomain: 'populate',
       );