diff --git a/app_dart/bin/gae_server.dart b/app_dart/bin/gae_server.dart
index 7c65385fe..3f03cb9fb 100644
--- a/app_dart/bin/gae_server.dart
+++ b/app_dart/bin/gae_server.dart
@@ -43,7 +43,10 @@ Future<void> main() async {
     }
 
     final cache = CacheService.redis();
-    final firestore = await FirestoreService.from(const GoogleAuthProvider());
+    final firestore = await FirestoreService.from(
+      const GoogleAuthProvider(),
+      cache: cache,
+    );
     final bigQuery = await BigQueryService.from(const GoogleAuthProvider());
 
     // Start with a fresh copy of the DynamicConfig. If this throws, the server
diff --git a/app_dart/lib/src/model/firestore/task.dart b/app_dart/lib/src/model/firestore/task.dart
index e049e8e70..07873e2c4 100644
--- a/app_dart/lib/src/model/firestore/task.dart
+++ b/app_dart/lib/src/model/firestore/task.dart
@@ -5,6 +5,9 @@
 /// @docImport 'commit.dart';
 library;
 
+import 'dart:convert';
+import 'dart:typed_data';
+
 import 'package:buildbucket/buildbucket_pb.dart' as bbv2;
 import 'package:cocoon_common/task_status.dart';
 import 'package:googleapis/firestore/v1.dart' hide Status;
@@ -113,6 +116,7 @@ final class Task extends AppDocument<Task> {
   static const fieldStatus = 'status';
   static const fieldTestFlaky = 'testFlaky';
   static const fieldAttempt = 'attempt';
+  static const fieldRevisionId = 'revisionId';
 
   /// Returns a document ID for a task from the given parameters.
   static AppDocumentId<Task> documentIdFor({
@@ -136,6 +140,19 @@ final class Task extends AppDocument<Task> {
     fromDocument: Task.fromDocument,
   );
 
+  /// Serializes a [Task] document into bytes for caching.
+  static Uint8List serialize(Task task) {
+    return utf8.encode(
+      json.encode(Document(name: task.name, fields: task.fields).toJson()),
+    );
+  }
+
+  /// Deserializes bytes into a [Task] document.
+  static Task deserialize(Uint8List data) {
+    final jsonMap = json.decode(utf8.decode(data)) as Map<String, Object?>;
+    return Task.fromDocument(Document.fromJson(jsonMap));
+  }
+
   /// Lookup [Task] from Firestore.
   ///
   /// `documentName` follows `/projects/{project}/databases/{database}/documents/{document_path}`
@@ -143,10 +160,7 @@ final class Task extends AppDocument<Task> {
     FirestoreService firestoreService,
     AppDocumentId<Task> id,
   ) async {
-    final document = await firestoreService.getDocument(
-      p.posix.join(kDatabase, 'documents', kTaskCollectionId, id.documentId),
-    );
-    return Task.fromDocument(document);
+    return firestoreService.getTask(id);
   }
 
   factory Task({
@@ -178,6 +192,7 @@ final class Task extends AppDocument<Task> {
         fieldStatus: status.value.toValue(),
         fieldTestFlaky: testFlaky.toValue(),
         fieldAttempt: currentAttempt.toValue(),
+        fieldRevisionId: 1.toValue(),
       },
       name: p.posix.join(
         kDatabase,
@@ -214,27 +229,17 @@ final class Task extends AppDocument<Task> {
       ..name = name;
   }
 
-  /// Returns a Firestore [Write] that patches the [status] field for [id].
-  @useResult
-  static Write patchStatus(AppDocumentId<Task> id, TaskStatus status) {
-    return Write(
-      currentDocument: Precondition(exists: true),
-      update: Document(
-        name: p.posix.join(
-          kDatabase,
-          'documents',
-          kTaskCollectionId,
-          id.documentId,
-        ),
-        fields: {fieldStatus: Value(stringValue: status.value)},
-      ),
-      updateMask: DocumentMask(fieldPaths: [fieldStatus]),
-    );
-  }
-
   /// The task was run successfully.
   static const statusSucceeded = TaskStatus.succeeded;
 
+  int get revisionId => fields.containsKey(fieldRevisionId)
+      ? int.parse(fields[fieldRevisionId]!.integerValue!)
+      : 1;
+
+  void incrementRevisionId() {
+    fields[fieldRevisionId] = (revisionId + 1).toValue();
+  }
+
   /// The timestamp (in milliseconds since the Epoch) that this task was
   /// created.
   ///
@@ -308,14 +313,17 @@ final class Task extends AppDocument<Task> {
 
   void setStatus(TaskStatus status) {
     fields[fieldStatus] = status.value.toValue();
+    incrementRevisionId();
   }
 
   void setEndTimestamp(int endTimestamp) {
     fields[fieldEndTimestamp] = endTimestamp.toValue();
+    incrementRevisionId();
   }
 
   void setTestFlaky(bool testFlaky) {
     fields[fieldTestFlaky] = testFlaky.toValue();
+    incrementRevisionId();
   }
 
   void updateFromBuild(bbv2.Build build) {
@@ -335,6 +343,23 @@ final class Task extends AppDocument<Task> {
         .toValue();
 
     _setStatusFromLuciStatus(build);
+    incrementRevisionId();
+  }
+
+  Task createRetry({DateTime? now}) {
+    now ??= DateTime.now();
+    return Task(
+      builderName: taskName,
+      currentAttempt: currentAttempt + 1,
+      commitSha: commitSha,
+      bringup: bringup,
+      createTimestamp: now.millisecondsSinceEpoch,
+      startTimestamp: 0,
+      endTimestamp: 0,
+      status: TaskStatus.waitingForBackfill,
+      testFlaky: false,
+      buildNumber: null,
+    );
   }
 
   void resetAsRetry({int? attempt, DateTime? now}) {
@@ -360,11 +385,13 @@ final class Task extends AppDocument<Task> {
       fieldTestFlaky: false.toValue(),
       fieldCommitSha: commitSha.toValue(),
       fieldAttempt: attempt.toValue(),
+      fieldRevisionId: 1.toValue(),
     };
   }
 
   void setBuildNumber(int buildNumber) {
     fields[fieldBuildNumber] = buildNumber.toValue();
+    incrementRevisionId();
   }
 
   void _setStatusFromLuciStatus(bbv2.Build build) {
diff --git a/app_dart/lib/src/request_handlers/dart_internal_subscription.dart b/app_dart/lib/src/request_handlers/dart_internal_subscription.dart
index d861a080a..4f25a2269 100644
--- a/app_dart/lib/src/request_handlers/dart_internal_subscription.dart
+++ b/app_dart/lib/src/request_handlers/dart_internal_subscription.dart
@@ -7,7 +7,6 @@ import 'dart:convert';
 import 'package:buildbucket/buildbucket_pb.dart' as bbv2;
 import 'package:cocoon_common/is_dart_internal.dart';
 import 'package:cocoon_server/logging.dart';
-import 'package:googleapis/firestore/v1.dart';
 
 import '../../cocoon_service.dart';
 import '../model/bbv2_extension.dart';
@@ -89,10 +88,7 @@ final class DartInternalSubscription extends SubscriptionHandler {
     }
 
     log.info('Inserting Task into Firestore: ${fsTask.toString()}');
-    await _firestore.batchWriteDocuments(
-      BatchWriteRequest(writes: documentsToWrites([fsTask])),
-      kDatabase,
-    );
+    await _firestore.updateTasks([fsTask]);
 
     return Response.json(fsTask.toString());
   }
diff --git a/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart b/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart
index 1d305c0ff..fb6b1ad1e 100644
--- a/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart
+++ b/app_dart/lib/src/request_handlers/postsubmit_luci_subscription.dart
@@ -8,7 +8,6 @@ import 'package:archive/archive.dart';
 import 'package:buildbucket/buildbucket_pb.dart' as bbv2;
 import 'package:cocoon_common/is_release_branch.dart';
 import 'package:cocoon_server/logging.dart';
-import 'package:googleapis/firestore/v1.dart' hide Status;
 
 import '../../ci_yaml.dart';
 import '../model/firestore/commit.dart' as fs;
@@ -150,10 +149,7 @@ final class PostsubmitLuciSubscription extends SubscriptionHandler {
 
   Future<void> _updateFirestore(fs.Task fsTask, bbv2.Build build) async {
     fsTask.updateFromBuild(build);
-    await _firestore.batchWriteDocuments(
-      BatchWriteRequest(writes: documentsToWrites([fsTask], exists: true)),
-      kDatabase,
-    );
+    await _firestore.updateTasks([fsTask]);
   }
 
   // No need to update task in Firestore if
diff --git a/app_dart/lib/src/request_handlers/rerun_prod_task.dart b/app_dart/lib/src/request_handlers/rerun_prod_task.dart
index f458337e4..a30372814 100644
--- a/app_dart/lib/src/request_handlers/rerun_prod_task.dart
+++ b/app_dart/lib/src/request_handlers/rerun_prod_task.dart
@@ -5,7 +5,6 @@
 import 'package:cocoon_common/is_dart_internal.dart';
 import 'package:cocoon_common/task_status.dart';
 import 'package:cocoon_server/logging.dart';
-import 'package:googleapis/firestore/v1.dart' as g;
 import 'package:meta/meta.dart';
 
 import '../model/ci_yaml/ci_yaml.dart';
@@ -184,9 +183,9 @@ final class RerunProdTask extends ApiRequestHandler {
     ).withMostRecentTaskOnly().tasks;
 
     final wasMarkedNew = <String>[];
-    final documentWrites = <g.Write>[];
+    final tasksToUpdate = <fs.Task>[];
+    final createdTasks = <fs.Task>[];
 
-    // Wait for cancellations?
     final Future<void> cancelRunningTasks;
     if (statusesToRerun.contains(TaskStatus.inProgress)) {
       cancelRunningTasks = _luciBuildService.cancelBuildsBySha(
@@ -208,32 +207,28 @@ final class RerunProdTask extends ApiRequestHandler {
       }
 
       // If it appears the task was in progress, cancel any running builders
-      // and crease a _new_ task (to represent a new run).
+      // and create a _new_ task (to represent a new run).
       if (task.status == TaskStatus.inProgress) {
-        // Mark cancelled.
-        documentWrites.add(
-          fs.Task.patchStatus(
-            fs.TaskId(
-              commitSha: task.commitSha,
-              currentAttempt: task.currentAttempt,
-              taskName: task.taskName,
-            ),
-            TaskStatus.cancelled,
-          ),
-        );
+        // Mark current attempt cancelled with incremented revisionId
+        task.setStatus(TaskStatus.cancelled);
+        tasksToUpdate.add(task);
       }
 
-      // Start a new task.
-      task.resetAsRetry(now: _now());
-      documentWrites.add(
-        g.Write(currentDocument: g.Precondition(exists: false), update: task),
-      );
+      // Start a new task attempt
+      final retryTask = task.createRetry(now: _now());
+      tasksToUpdate.add(retryTask);
+      createdTasks.add(retryTask);
     }
 
+    final writes = documentsToWrites(tasksToUpdate);
     await Future.wait([
       cancelRunningTasks,
-      _firestore.commit(transaction, documentWrites),
+      _firestore.commit(transaction, writes),
     ]);
+    await _firestore.cacheTaskPayloads(tasksToUpdate);
+    if (createdTasks.isNotEmpty) {
+      await _firestore.updateCacheForCreatedTasks(createdTasks);
+    }
 
     return wasMarkedNew;
   }
diff --git a/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart b/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart
index da9394ab4..2fa4e5b44 100644
--- a/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart
+++ b/app_dart/lib/src/request_handlers/scheduler/batch_backfiller.dart
@@ -174,31 +174,23 @@ final class BatchBackfiller extends ApiRequestHandler {
     Iterable<BackfillTask> schedule,
     Iterable<SkippableTask> skip,
   ) async {
-    log.debug('Querying ${schedule.length} tasks in Firestore...');
-    await _firestore.writeViaTransaction([
-      ...schedule.map((toUpdate) {
-        final BackfillTask(:task) = toUpdate;
-        return fs.Task.patchStatus(
-          fs.TaskId(
-            commitSha: task.commitSha,
-            taskName: task.name,
-            currentAttempt: task.currentAttempt,
-          ),
-          TaskStatus.inProgress,
-        );
-      }),
-      ...skip.map((toSkip) {
-        final SkippableTask(:task) = toSkip;
-        return fs.Task.patchStatus(
-          fs.TaskId(
-            commitSha: task.commitSha,
-            taskName: task.name,
-            currentAttempt: task.currentAttempt,
-          ),
-          TaskStatus.skipped,
-        );
-      }),
-    ]);
+    final taskStatusMap = <fs.TaskId, TaskStatus>{
+      for (final item in schedule)
+        fs.TaskId(
+          commitSha: item.task.commitSha,
+          taskName: item.task.name,
+          currentAttempt: item.task.currentAttempt,
+        ): TaskStatus.inProgress,
+      for (final item in skip)
+        fs.TaskId(
+          commitSha: item.task.commitSha,
+          taskName: item.task.name,
+          currentAttempt: item.task.currentAttempt,
+        ): TaskStatus.skipped,
+    };
+    if (taskStatusMap.isEmpty) return;
+
+    await _firestore.updateTaskStatuses(taskStatusMap);
     log.debug('Wrote to Firestore for backfill');
   }
 
diff --git a/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart b/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart
index 10d767ca2..5d804585c 100644
--- a/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart
+++ b/app_dart/lib/src/request_handlers/scheduler/vacuum_stale_tasks.dart
@@ -9,7 +9,7 @@ import 'package:cocoon_common/is_release_branch.dart';
 import 'package:cocoon_common/task_status.dart';
 import 'package:cocoon_server/logging.dart';
 import 'package:github/github.dart' as gh;
-import 'package:googleapis/firestore/v1.dart';
+
 import 'package:meta/meta.dart';
 
 import '../../../cocoon_service.dart';
@@ -139,10 +139,7 @@ final class VacuumStaleTasks extends ApiRequestHandler {
       }
       tasks.add(task);
     }
-    await _firestore.batchWriteDocuments(
-      BatchWriteRequest(writes: documentsToWrites(tasks)),
-      kDatabase,
-    );
+    await _firestore.updateTasks(tasks);
   }
 }
 
diff --git a/app_dart/lib/src/service/cache_service.dart b/app_dart/lib/src/service/cache_service.dart
index d3263b0cc..8d3445203 100644
--- a/app_dart/lib/src/service/cache_service.dart
+++ b/app_dart/lib/src/service/cache_service.dart
@@ -12,6 +12,20 @@ import 'package:meta/meta.dart';
 import 'package:mutex/mutex.dart';
 import 'package:redis/redis.dart';
 
+class VersionedCacheEntry {
+  const VersionedCacheEntry({
+    required this.key,
+    required this.value,
+    required this.revisionId,
+    this.ttl = const Duration(hours: 12),
+  });
+
+  final String key;
+  final Uint8List value;
+  final int revisionId;
+  final Duration ttl;
+}
+
 /// Service for reading and writing values to a cache for quick access of data.
 abstract class CacheService {
   CacheService();
@@ -28,6 +42,9 @@ abstract class CacheService {
   /// Get value of [key] from the subcache [subcacheName].
   Future<Uint8List?> get(String subcacheName, String key);
 
+  /// Get values for multiple [keys] from the subcache [subcacheName] in a single batch API call.
+  Future<List<Uint8List?>> getMulti(String subcacheName, List<String> keys);
+
   /// Set [value] for [key] in the subcache [subcacheName] with [ttl].
   Future<Uint8List?> set(
     String subcacheName,
@@ -48,6 +65,31 @@ abstract class CacheService {
     Duration ttl = const Duration(minutes: 1),
   });
 
+  /// Atomically inserts multiple [entries] into [subcacheName] in a single batch API call
+  /// if and only if their [VersionedCacheEntry.revisionId] is strictly greater than any
+  /// existing cached revision for that key.
+  Future<void> insertVersioned(
+    String subcacheName,
+    List<VersionedCacheEntry> entries,
+  );
+
+  /// Get the set of string values for [key] from the subcache [subcacheName].
+  Future<Set<String>> getSet(String subcacheName, String key);
+
+  /// Atomically adds [values] to the set at [key] in [subcacheName].
+  /// If the set does not yet exist, it is created with [ttl].
+  /// If the set already exists, all [values] are added to it without altering its TTL.
+  Future<void> updateSet(
+    String subcacheName,
+    String key,
+    Set<String> values, {
+    Duration ttl = const Duration(hours: 12),
+  });
+
+  /// Atomically adds [value] to the set at [key] in [subcacheName] if and only if the set already exists.
+  /// Returns `true` if the set existed and [value] was added, or `false` if the set did not exist.
+  Future<bool> addToSetIfExists(String subcacheName, String key, String value);
+
   /// Get value of [key] from the subcache [subcacheName]. If the key has no
   /// value, call [createFn] to create a value for it, set it, and return it.
   Future<Uint8List?> getOrCreate(
@@ -183,6 +225,162 @@ class RedisCacheService extends CacheService {
     }
   }
 
+  @override
+  Future<List<Uint8List?>> getMulti(
+    String subcacheName,
+    List<String> keys,
+  ) async {
+    if (keys.isEmpty) return const [];
+    try {
+      final redisKeys = keys.map((k) => '$subcacheName/$k').toList();
+      final values = await _runCommand(
+        (client) => client.send_object(['MGET', ...redisKeys]),
+      );
+      if (values is! List) {
+        return List.filled(keys.length, null);
+      }
+      return values.map((value) {
+        if (value == null) return null;
+        return base64.decode(value as String);
+      }).toList();
+    } catch (e) {
+      log.warn('Unable to retrieve multi-values from cache.', e);
+      return List.filled(keys.length, null);
+    }
+  }
+
+  @override
+  Future<void> insertVersioned(
+    String subcacheName,
+    List<VersionedCacheEntry> entries,
+  ) async {
+    if (entries.isEmpty) return;
+    const insertVersionedScript = '''
+      local numKeys = tonumber(ARGV[1])
+      for i = 1, numKeys do
+        local key = KEYS[i]
+        local val = ARGV[1 + (i - 1) * 3 + 1]
+        local rev = tonumber(ARGV[1 + (i - 1) * 3 + 2])
+        local ttl = tonumber(ARGV[1 + (i - 1) * 3 + 3])
+
+        local revKey = "revisions/" .. key
+        local existingRev = tonumber(redis.call("get", revKey) or 0)
+        if rev > existingRev or redis.call("exists", key) == 0 then
+          redis.call("set", key, val, "PX", ttl)
+          redis.call("set", revKey, rev, "PX", ttl)
+        end
+      end
+      return 1
+    ''';
+    const batchSize = 20;
+    for (var i = 0; i < entries.length; i += batchSize) {
+      final chunk = entries.sublist(i, min(i + batchSize, entries.length));
+      try {
+        final keys = chunk.map((e) => '$subcacheName/${e.key}').toList();
+        final args = [
+          chunk.length.toString(),
+          for (final e in chunk) ...[
+            base64.encode(e.value),
+            e.revisionId.toString(),
+            e.ttl.inMilliseconds.toString(),
+          ],
+        ];
+        await _runCommand(
+          (client) => client.send_object([
+            'EVAL',
+            insertVersionedScript,
+            keys.length.toString(),
+            ...keys,
+            ...args,
+          ]),
+        );
+      } catch (e) {
+        log.warn('Unable to insert versioned entries into cache.', e);
+      }
+    }
+  }
+
+  @override
+  Future<Set<String>> getSet(String subcacheName, String key) async {
+    final redisKey = '$subcacheName/$key';
+    try {
+      final values = await _runCommand(
+        (client) => client.send_object(['SMEMBERS', redisKey]),
+      );
+      if (values is! List || values.isEmpty) {
+        return const {};
+      }
+      return values.map((e) => e.toString()).toSet();
+    } catch (e) {
+      log.warn('Unable to retrieve set for $key from cache.', e);
+      return const {};
+    }
+  }
+
+  @override
+  Future<void> updateSet(
+    String subcacheName,
+    String key,
+    Set<String> values, {
+    Duration ttl = const Duration(hours: 12),
+  }) async {
+    if (values.isEmpty) return;
+    const updateSetScript = '''
+      if redis.call("exists", KEYS[1]) == 1 then
+        for i = 1, #ARGV - 1 do
+          redis.call("sadd", KEYS[1], ARGV[i])
+        end
+      else
+        for i = 1, #ARGV - 1 do
+          redis.call("sadd", KEYS[1], ARGV[i])
+        end
+        redis.call("pexpire", KEYS[1], tonumber(ARGV[#ARGV]))
+      end
+      return 1
+    ''';
+    final redisKey = '$subcacheName/$key';
+    try {
+      final args = [...values, ttl.inMilliseconds.toString()];
+      await _runCommand(
+        (client) => client.send_object([
+          'EVAL',
+          updateSetScript,
+          '1',
+          redisKey,
+          ...args,
+        ]),
+      );
+    } catch (e) {
+      log.warn('Unable to update set for $key in cache.', e);
+    }
+  }
+
+  @override
+  Future<bool> addToSetIfExists(
+    String subcacheName,
+    String key,
+    String value,
+  ) async {
+    const addToSetScript = '''
+      if redis.call("exists", KEYS[1]) == 1 then
+        redis.call("sadd", KEYS[1], ARGV[1])
+        return 1
+      end
+      return 0
+    ''';
+    final redisKey = '$subcacheName/$key';
+    try {
+      final response = await _runCommand(
+        (client) =>
+            client.send_object(['EVAL', addToSetScript, '1', redisKey, value]),
+      );
+      return response == 1 || response == '1';
+    } catch (e) {
+      log.warn('Unable to add to set for $key in cache.', e);
+      return false;
+    }
+  }
+
   @override
   Future<Uint8List?> set(
     String subcacheName,
@@ -245,7 +443,10 @@ class RedisCacheService extends CacheService {
   Future<void> purge(String subcacheName, String key) async {
     final redisKey = '$subcacheName/$key';
     try {
-      await _runCommand((client) => client.send_object(['DEL', redisKey]));
+      await _runCommand(
+        (client) =>
+            client.send_object(['DEL', redisKey, 'revisions/$redisKey']),
+      );
     } catch (e) {
       log.warn('Unable to purge value for $key from cache.', e);
     }
@@ -308,9 +509,23 @@ class InMemoryCacheService extends CacheService {
 
   final int maxEntries;
   final Map<String, _InMemoryCacheEntry> _entries = {};
+  final Map<String, _InMemorySetEntry> _sets = {};
   final Map<String, _InMemoryLock> _locks = {};
   final _mutex = Mutex();
 
+  void _evictIfFull(String cacheKey) {
+    if (_entries.length >= maxEntries && !_entries.containsKey(cacheKey)) {
+      _entries.remove(_entries.keys.first);
+    }
+  }
+
+  void _touch(String cacheKey) {
+    final entry = _entries.remove(cacheKey);
+    if (entry != null) {
+      _entries[cacheKey] = entry;
+    }
+  }
+
   @override
   Future<Uint8List?> get(String subcacheName, String key) async {
     await _mutex.acquire();
@@ -322,12 +537,125 @@ class InMemoryCacheService extends CacheService {
         _entries.remove(cacheKey);
         return null;
       }
+      _touch(cacheKey);
       return entry.value;
     } finally {
       _mutex.release();
     }
   }
 
+  @override
+  Future<List<Uint8List?>> getMulti(
+    String subcacheName,
+    List<String> keys,
+  ) async {
+    await _mutex.acquire();
+    try {
+      return keys.map((key) {
+        final cacheKey = '$subcacheName/$key';
+        final entry = _entries[cacheKey];
+        if (entry == null || entry.isExpired) {
+          _entries.remove(cacheKey);
+          return null;
+        }
+        _touch(cacheKey);
+        return entry.value;
+      }).toList();
+    } finally {
+      _mutex.release();
+    }
+  }
+
+  @override
+  Future<void> insertVersioned(
+    String subcacheName,
+    List<VersionedCacheEntry> entries,
+  ) async {
+    if (entries.isEmpty) return;
+    await _mutex.acquire();
+    try {
+      _entries.removeWhere((k, v) => v.isExpired);
+      for (final entry in entries) {
+        final cacheKey = '$subcacheName/${entry.key}';
+        final existing = _entries[cacheKey];
+        if (existing == null ||
+            existing.isExpired ||
+            entry.revisionId > existing.revisionId) {
+          _evictIfFull(cacheKey);
+          _entries.remove(cacheKey);
+          _entries[cacheKey] = _InMemoryCacheEntry(
+            entry.value,
+            DateTime.now().add(entry.ttl),
+            revisionId: entry.revisionId,
+          );
+        }
+      }
+    } finally {
+      _mutex.release();
+    }
+  }
+
+  @override
+  Future<Set<String>> getSet(String subcacheName, String key) async {
+    await _mutex.acquire();
+    try {
+      final cacheKey = '$subcacheName/$key';
+      final entry = _sets[cacheKey];
+      if (entry == null || entry.isExpired) {
+        _sets.remove(cacheKey);
+        return const {};
+      }
+      return Set.of(entry.values);
+    } finally {
+      _mutex.release();
+    }
+  }
+
+  @override
+  Future<void> updateSet(
+    String subcacheName,
+    String key,
+    Set<String> values, {
+    Duration ttl = const Duration(hours: 12),
+  }) async {
+    if (values.isEmpty) return;
+    await _mutex.acquire();
+    try {
+      final cacheKey = '$subcacheName/$key';
+      final existing = _sets[cacheKey];
+      if (existing != null && !existing.isExpired) {
+        existing.values.addAll(values);
+      } else {
+        _sets[cacheKey] = _InMemorySetEntry(
+          Set.of(values),
+          DateTime.now().add(ttl),
+        );
+      }
+    } finally {
+      _mutex.release();
+    }
+  }
+
+  @override
+  Future<bool> addToSetIfExists(
+    String subcacheName,
+    String key,
+    String value,
+  ) async {
+    await _mutex.acquire();
+    try {
+      final cacheKey = '$subcacheName/$key';
+      final existing = _sets[cacheKey];
+      if (existing != null && !existing.isExpired) {
+        existing.values.add(value);
+        return true;
+      }
+      return false;
+    } finally {
+      _mutex.release();
+    }
+  }
+
   @override
   Future<Uint8List?> set(
     String subcacheName,
@@ -345,11 +673,9 @@ class InMemoryCacheService extends CacheService {
       final cacheKey = '$subcacheName/$key';
 
       _entries.removeWhere((k, v) => v.isExpired);
+      _evictIfFull(cacheKey);
 
-      if (_entries.length >= maxEntries && !_entries.containsKey(cacheKey)) {
-        _entries.remove(_entries.keys.first);
-      }
-
+      _entries.remove(cacheKey);
       _entries[cacheKey] = _InMemoryCacheEntry(value, DateTime.now().add(ttl));
       return value;
     } finally {
@@ -373,10 +699,7 @@ class InMemoryCacheService extends CacheService {
       }
 
       _entries.removeWhere((k, v) => v.isExpired);
-
-      if (_entries.length >= maxEntries && !_entries.containsKey(cacheKey)) {
-        _entries.remove(_entries.keys.first);
-      }
+      _evictIfFull(cacheKey);
 
       _entries[cacheKey] = _InMemoryCacheEntry(value, DateTime.now().add(ttl));
       return true;
@@ -391,6 +714,7 @@ class InMemoryCacheService extends CacheService {
     try {
       final cacheKey = '$subcacheName/$key';
       _entries.remove(cacheKey);
+      _sets.remove(cacheKey);
     } finally {
       _mutex.release();
     }
@@ -430,10 +754,20 @@ class InMemoryCacheService extends CacheService {
 }
 
 class _InMemoryCacheEntry {
-  _InMemoryCacheEntry(this.value, this.expiresAt);
+  _InMemoryCacheEntry(this.value, this.expiresAt, {this.revisionId = 0});
 
   final Uint8List value;
   final DateTime expiresAt;
+  final int revisionId;
+
+  bool get isExpired => DateTime.now().isAfter(expiresAt);
+}
+
+class _InMemorySetEntry {
+  _InMemorySetEntry(this.values, this.expiresAt);
+
+  final Set<String> values;
+  final DateTime expiresAt;
 
   bool get isExpired => DateTime.now().isAfter(expiresAt);
 }
diff --git a/app_dart/lib/src/service/config.dart b/app_dart/lib/src/service/config.dart
index 26a1d137c..cbb51d92b 100644
--- a/app_dart/lib/src/service/config.dart
+++ b/app_dart/lib/src/service/config.dart
@@ -9,7 +9,8 @@ import 'package:cocoon_server/generate_github_jws.dart';
 import 'package:cocoon_server/logging.dart';
 import 'package:cocoon_server/secret_manager.dart';
 import 'package:github/github.dart' as gh;
-import 'package:graphql/client.dart' hide JsonSerializable;
+import 'package:graphql/client.dart'
+    show AuthLink, GraphQLCache, GraphQLClient, HttpLink;
 import 'package:http/http.dart' as http;
 import 'package:meta/meta.dart';
 import 'package:retry/retry.dart';
diff --git a/app_dart/lib/src/service/firestore.dart b/app_dart/lib/src/service/firestore.dart
index 034ac2439..33c2528f2 100644
--- a/app_dart/lib/src/service/firestore.dart
+++ b/app_dart/lib/src/service/firestore.dart
@@ -11,12 +11,14 @@ import 'package:cocoon_server/google_auth_provider.dart';
 import 'package:github/github.dart';
 import 'package:googleapis/firestore/v1.dart';
 import 'package:meta/meta.dart';
+import 'package:path/path.dart' as p;
 
 import '../../cocoon_service.dart';
 import '../model/firestore/github_build_status.dart';
 import '../model/firestore/github_gold_status.dart';
 
 import 'firestore/commit_and_tasks.dart';
+import 'firestore/task_cache_service.dart';
 
 export '../model/common/firestore_extensions.dart';
 
@@ -49,7 +51,168 @@ final kFieldMapRegExp = RegExp(
 );
 
 @visibleForTesting
+/// Mixin that encapsulates Firestore query and caching operations across Cocoon.
+///
+/// Task caching orchestration is delegated to [TaskCacheService].
+/// See [TaskCacheService] for full architectural documentation and caching invariant flowcharts.
 mixin FirestoreQueries {
+  CacheService? get cache => null;
+  Config? get config => null;
+
+  /// Lazy accessor for [TaskCacheService].
+  TaskCacheService? get _taskCache {
+    if (cache == null || !(config?.flags.taskCachingEnabled ?? true)) {
+      return null;
+    }
+    return TaskCacheService(cache: cache!, config: config);
+  }
+
+  Future<Document> getDocument(String name, {Transaction? transaction});
+  @protected
+  @visibleForTesting
+  Future<List<Document>> batchGetDocuments(
+    List<String> names, {
+    Transaction? transaction,
+  });
+  Future<Transaction> beginTransaction();
+  Future<CommitResponse> commit(Transaction transaction, List<Write> writes);
+  Future<BatchWriteResponse> batchWriteDocuments(
+    BatchWriteRequest request,
+    String database,
+  );
+
+  /// Queries all tasks for [commitSha] from Firestore, updates their individual cache entries,
+  /// Retrieves a [Task] by its document ID, serving from versioned cache if available,
+  /// or fetching from Firestore DB on cache miss and populating the cache.
+  Future<Task> getTask(AppDocumentId<Task> id) async {
+    final docId = id.documentId;
+    if (_taskCache != null) {
+      final result = await _taskCache!.getTaskPayloads([docId]);
+      if (result.foundTasks.isNotEmpty) {
+        return result.foundTasks.first;
+      }
+    }
+
+    final docName = p.posix.join(
+      kDatabase,
+      'documents',
+      kTaskCollectionId,
+      docId,
+    );
+    final document = await getDocument(docName);
+    final task = Task.fromDocument(document);
+
+    await _taskCache?.cacheTaskPayloads([task]);
+    return task;
+  }
+
+  /// Caches payloads for modified [Task] entities.
+  Future<void> cacheTaskPayloads(Iterable<Task> tasks) async {
+    await _taskCache?.cacheTaskPayloads(tasks);
+  }
+
+  /// Queries all tasks for [commitSha] from Firestore, updates their individual cache entries,
+  /// and populates the `tasks_by_commit_ids` set.
+  Future<List<Task>> _fetchAndCacheCommitTasks(String commitSha) async {
+    final documents = await query(
+      kTaskCollectionId,
+      {'${Task.fieldCommitSha} =': commitSha},
+      orderMap: {Task.fieldCreateTimestamp: kQueryOrderDescending},
+    );
+    final tasks = documents.map(Task.fromDocument).toList();
+    if (_taskCache != null) {
+      final docIds = {for (final t in tasks) p.basename(t.name!)};
+      await [
+        _taskCache!.cacheTaskPayloads(tasks),
+        if (docIds.isNotEmpty)
+          () async {
+            final initialized = await _taskCache!.initializeCommitTaskSet(
+              commitSha,
+              docIds,
+            );
+            if (!initialized) {
+              await _taskCache!.addTasksToCommitSet(commitSha, docIds);
+            }
+          }(),
+      ].wait;
+    }
+    return tasks;
+  }
+
+  /// Explicitly updates the cache when new [Task]s are created.
+  Future<void> updateCacheForCreatedTasks(List<Task> tasks) async {
+    if (_taskCache == null || tasks.isEmpty) return;
+    await _taskCache!.cacheTaskPayloads(tasks);
+
+    final tasksByCommit = <String, Set<String>>{};
+    for (final task in tasks) {
+      if (task.commitSha.isNotEmpty) {
+        final docId = p.basename(task.name!);
+        tasksByCommit.putIfAbsent(task.commitSha, () => {}).add(docId);
+      }
+    }
+
+    for (final entry in tasksByCommit.entries) {
+      final commitSha = entry.key;
+      final docIds = entry.value;
+      final setExisted = await _taskCache!.addTasksToCommitSet(
+        commitSha,
+        docIds,
+      );
+      if (!setExisted) {
+        // Set was missing in Redis: fall back to full database query to populate set
+        await _fetchAndCacheCommitTasks(commitSha);
+      }
+    }
+  }
+
+  /// Saves or updates [Task] entities in Firestore and updates the versioned Redis cache.
+  Future<void> updateTasks(List<Task> tasks) async {
+    if (tasks.isEmpty) return;
+    final writes = documentsToWrites(tasks);
+    await batchWriteDocuments(BatchWriteRequest(writes: writes), kDatabase);
+    await _taskCache?.cacheTaskPayloads(tasks);
+  }
+
+  /// Updates the status of tasks identified by [taskStatusMap] within a single Firestore transaction
+  /// and updates the versioned task cache.
+  Future<void> updateTaskStatuses(Map<TaskId, TaskStatus> taskStatusMap) async {
+    if (taskStatusMap.isEmpty) return;
+
+    final docIds = taskStatusMap.keys
+        .map(
+          (id) => p.posix.join(
+            kDatabase,
+            'documents',
+            kTaskCollectionId,
+            id.documentId,
+          ),
+        )
+        .toList();
+
+    final transaction = await beginTransaction();
+    final docs = await batchGetDocuments(docIds, transaction: transaction);
+    final tasksToUpdate = <Task>[];
+
+    for (final doc in docs) {
+      final task = Task.fromDocument(doc);
+      final id = TaskId(
+        commitSha: task.commitSha,
+        taskName: task.taskName,
+        currentAttempt: task.currentAttempt,
+      );
+      final newStatus = taskStatusMap[id];
+      if (newStatus != null) {
+        task.setStatus(newStatus);
+        tasksToUpdate.add(task);
+      }
+    }
+
+    final writes = documentsToWrites(tasksToUpdate);
+    await commit(transaction, writes);
+    await _taskCache?.cacheTaskPayloads(tasksToUpdate);
+  }
+
   /// Wrapper to simplify Firestore query.
   ///
   /// The [filterMap] follows format:
@@ -130,21 +293,20 @@ mixin FirestoreQueries {
     return documents.isEmpty ? null : Commit.fromDocument(documents.first);
   }
 
-  Future<List<Task>> _queryTasks({
-    required int? limit,
+  Future<List<Task>> _queryTasksFromFirestore({
+    int? limit,
     String? name,
     TaskStatus? status,
     String? commitSha,
     Transaction? transaction,
+    bool cacheResults = true,
   }) async {
-    final filterMap = {
+    final filterMap = <String, Object>{
       '${Task.fieldName} =': ?name,
       '${Task.fieldStatus} =': ?status?.value,
       '${Task.fieldCommitSha} =': ?commitSha,
     };
 
-    // Avoid a full table-scan.
-    // TODO(matanlurey): Debatably this should be in the root query method.
     if (limit == null && filterMap.isEmpty) {
       throw ArgumentError.value(
         limit,
@@ -153,31 +315,130 @@ mixin FirestoreQueries {
       );
     }
 
-    // For tasks, therer is no reason to _not_ order this way.
     final orderMap = {Task.fieldCreateTimestamp: kQueryOrderDescending};
     final documents = await query(
       kTaskCollectionId,
       filterMap,
       orderMap: orderMap,
+      limit: limit,
       transaction: transaction,
     );
-    return [...documents.map(Task.fromDocument)];
+    final tasks = documents.map(Task.fromDocument).toList();
+    if (cacheResults && transaction == null && _taskCache != null) {
+      await _taskCache!.cacheTaskPayloads(tasks);
+    }
+    return tasks;
   }
 
-  /// Queries for recent [Task]s.
+  Future<List<Task>> _queryTasksByCommit({
+    required String commitSha,
+    String? name,
+    TaskStatus? status,
+    int? limit,
+    Transaction? transaction,
+  }) async {
+    if (transaction == null &&
+        cache != null &&
+        (config?.flags.taskCachingEnabled ?? true)) {
+      return await _queryTasksByCommitCached(
+        commitSha: commitSha,
+        name: name,
+        status: status,
+        limit: limit,
+      );
+    }
+    return await _queryTasksFromFirestore(
+      commitSha: commitSha,
+      name: name,
+      status: status,
+      limit: limit,
+      transaction: transaction,
+      cacheResults: false,
+    );
+  }
+
+  Future<List<Task>> _queryTasksByCommitCached({
+    required String commitSha,
+    String? name,
+    TaskStatus? status,
+    int? limit,
+  }) async {
+    List<Task>? tasks;
+    final docIds = await _taskCache!.getTaskIdsForCommit(commitSha);
+    if (docIds != null && docIds.isNotEmpty) {
+      final lookupResult = await _taskCache!.getTaskPayloads(docIds);
+      final foundTasks = lookupResult.foundTasks;
+
+      if (lookupResult.missingDocIds.isEmpty) {
+        tasks = foundTasks;
+      } else {
+        final missingNames = lookupResult.missingDocIds
+            .map(
+              (String missingId) => p.posix.join(
+                kDatabase,
+                'documents',
+                kTaskCollectionId,
+                missingId,
+              ),
+            )
+            .toList();
+        final documents = await batchGetDocuments(missingNames);
+        final fetchedMissingTasks = documents.map(Task.fromDocument).toList();
+        if (fetchedMissingTasks.isNotEmpty) {
+          await _taskCache!.cacheTaskPayloads(fetchedMissingTasks);
+          foundTasks.addAll(fetchedMissingTasks);
+        }
+        tasks = foundTasks;
+      }
+    }
+
+    tasks ??= await _fetchAndCacheCommitTasks(commitSha);
+
+    var result = tasks.toList();
+    if (name != null) {
+      result = result.where((t) => t.taskName == name).toList();
+    }
+    if (status != null) {
+      result = result.where((t) => t.status == status).toList();
+    }
+    result.sort((a, b) => b.createTimestamp.compareTo(a.createTimestamp));
+    if (limit != null) {
+      result = result.take(limit).toList();
+    }
+    return result;
+  }
+
+  /// Queries for recent [Task]s across commits matching [name].
   ///
-  /// If other named arguments are provided, they are used as a query filter.
-  Future<List<Task>> queryRecentTasks({
+  /// If [status] is provided, only tasks matching the status are returned.
+  Future<List<Task>> queryRecentTasksByName({
+    required String name,
     int limit = 100,
-    String? name,
     TaskStatus? status,
-    String? commitSha,
   }) async {
-    return await _queryTasks(
+    return await _queryTasksFromFirestore(
       limit: limit,
       name: name,
       status: status,
+    );
+  }
+
+  /// Queries [Task]s running against the specified [commitSha].
+  ///
+  /// If [name] is provided, only tasks matching the task name are returned.
+  /// If [status] is provided, only tasks matching the status are returned.
+  /// If [limit] is provided, at most [limit] tasks are returned.
+  Future<List<Task>> queryRecentTasksByCommit({
+    required String commitSha,
+    String? name,
+    TaskStatus? status,
+    int? limit,
+  }) async {
+    return await _queryTasksByCommit(
       commitSha: commitSha,
+      name: name,
+      status: status,
+      limit: limit,
     );
   }
 
@@ -188,8 +449,7 @@ mixin FirestoreQueries {
     String? name,
     Transaction? transaction,
   }) async {
-    return await _queryTasks(
-      limit: null,
+    return await _queryTasksByCommit(
       commitSha: commitSha,
       status: status,
       name: name,
@@ -270,17 +530,12 @@ mixin FirestoreQueries {
     required String commitSha,
     required String builderName,
   }) async {
-    final tasks = await query(
-      kTaskCollectionId,
-      {
-        '${Task.fieldCommitSha} =': commitSha,
-        '${Task.fieldName} =': builderName,
-      },
-      // Assumes the invariant where the newest task has the highest attempt #.
-      orderMap: {Task.fieldCreateTimestamp: kQueryOrderDescending},
+    final tasks = await _queryTasksByCommit(
+      commitSha: commitSha,
+      name: builderName,
       limit: 1,
     );
-    return tasks.isEmpty ? null : Task.fromDocument(tasks.first);
+    return tasks.isEmpty ? null : tasks.first;
   }
 
   /// Queries the last updated build status for the [slug], [prNumber], and [head].
@@ -331,7 +586,11 @@ mixin FirestoreQueries {
 
 class FirestoreService with FirestoreQueries {
   /// Creates a [FirestoreService] using Google API authentication.
-  static Future<FirestoreService> from(GoogleAuthProvider authProvider) async {
+  static Future<FirestoreService> from(
+    GoogleAuthProvider authProvider, {
+    CacheService? cache,
+    Config? config,
+  }) async {
     final client = await authProvider.createClient(
       scopes: const [FirestoreApi.datastoreScope],
       baseClient: FirestoreBaseClient(
@@ -339,13 +598,24 @@ class FirestoreService with FirestoreQueries {
         databaseId: Config.flutterGcpFirestoreDatabase,
       ),
     );
-    return FirestoreService._(FirestoreApi(client));
+    return FirestoreService._(
+      FirestoreApi(client),
+      cache: cache,
+      config: config,
+    );
   }
 
-  const FirestoreService._(this._api);
+  const FirestoreService._(this._api, {this.cache, this.config});
   final FirestoreApi _api;
 
+  @override
+  final CacheService? cache;
+
+  @override
+  final Config? config;
+
   /// Gets a document based on name.
+  @override
   Future<Document> getDocument(String name, {Transaction? transaction}) async {
     return _api.projects.databases.documents.get(
       name,
@@ -353,6 +623,27 @@ class FirestoreService with FirestoreQueries {
     );
   }
 
+  /// Gets multiple documents based on a list of names in a single batch request.
+  @override
+  Future<List<Document>> batchGetDocuments(
+    List<String> names, {
+    Transaction? transaction,
+  }) async {
+    if (names.isEmpty) return const [];
+    final request = BatchGetDocumentsRequest(
+      documents: names,
+      transaction: transaction?.identifier,
+    );
+    final response = await _api.projects.databases.documents.batchGet(
+      request,
+      kDatabase,
+    );
+    return response
+        .map((element) => element.found)
+        .whereType<Document>()
+        .toList();
+  }
+
   /// Creates a document.
   ///
   /// A document name is automatically generated if [documentId] is omitted.
@@ -363,12 +654,16 @@ class FirestoreService with FirestoreQueries {
     required String collectionId,
     String? documentId,
   }) async {
-    return _api.projects.databases.documents.createDocument(
+    final result = await _api.projects.databases.documents.createDocument(
       document,
       '$kDatabase/documents',
       collectionId,
       documentId: documentId,
     );
+    if (_taskCache != null && collectionId == kTaskCollectionId) {
+      await updateCacheForCreatedTasks([Task.fromDocument(result)]);
+    }
+    return result;
   }
 
   /// Batch writes documents to Firestore.
@@ -377,14 +672,19 @@ class FirestoreService with FirestoreQueries {
   /// Each write succeeds or fails independently.
   ///
   /// https://firebase.google.com/docs/firestore/reference/rest/v1/projects.databases.documents/batchWrite
+  @override
   Future<BatchWriteResponse> batchWriteDocuments(
     BatchWriteRequest request,
     String database,
   ) async {
-    return _api.projects.databases.documents.batchWrite(request, database);
+    return await _api.projects.databases.documents.batchWrite(
+      request,
+      database,
+    );
   }
 
   /// Begins a read-write transaction.
+  @override
   Future<Transaction> beginTransaction() async {
     final request = BeginTransactionRequest(
       options: TransactionOptions(readWrite: ReadWrite()),
@@ -397,6 +697,7 @@ class FirestoreService with FirestoreQueries {
   }
 
   /// Commits a transaction.
+  @override
   Future<CommitResponse> commit(
     Transaction transaction,
     List<Write> writes,
@@ -427,7 +728,10 @@ class FirestoreService with FirestoreQueries {
       transaction: beginTransactionResponse.transaction,
       writes: writes,
     );
-    return _api.projects.databases.documents.commit(commitRequest, kDatabase);
+    return await _api.projects.databases.documents.commit(
+      commitRequest,
+      kDatabase,
+    );
   }
 
   /// Returns Firestore [Value] based on corresponding object type.
diff --git a/app_dart/lib/src/service/firestore/task_cache_service.dart b/app_dart/lib/src/service/firestore/task_cache_service.dart
new file mode 100644
index 000000000..8b8b75581
--- /dev/null
+++ b/app_dart/lib/src/service/firestore/task_cache_service.dart
@@ -0,0 +1,214 @@
+// Copyright 2019 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:path/path.dart' as p;
+
+import '../../model/firestore/task.dart';
+import '../cache_service.dart';
+import '../config.dart';
+
+/// Result container for task payload batch lookups.
+final class TaskPayloadLookupResult {
+  TaskPayloadLookupResult({
+    required this.foundTasks,
+    required this.missingDocIds,
+  });
+
+  /// Tasks successfully retrieved and deserialized from cache.
+  final List<Task> foundTasks;
+
+  /// Document IDs whose payload was expired or missing in cache.
+  final List<String> missingDocIds;
+}
+
+/// Dedicated service encapsulating Redis caching logic for /tasks.
+///
+/// ## Two-Tier Lock-Free Task Caching Architecture
+///
+/// ### Invariant 1: Task Entity Payload Cache (`tasks/$docId`)
+/// - **Monotonically Increasing Revision ID**: Database write transactions establish a strict total order
+///   of document states, monotonically numbered by `revisionId`.
+/// - **Optimistic Version Guarding**: Write-backs call `insertVersioned(tasks/$docId, payload, revisionId)`.
+///   Redis atomically evaluates `newRev > existingRev` (or `EXISTS == 0`).
+///
+/// #### Scenario 1.1: Read Task Entry
+/// ```mermaid
+/// flowchart TD
+///     A["Query tasks/$docId Payload"] --> B{"Key Exists in Redis?"}
+///     B -- Cache Hit --> C["Return Serialized JSON Payload"]
+///     B -- Cache Miss --> D["Fetch Task from DB"]
+///     D --> E["cacheTaskPayloads"]
+///     E -- "!existingEntry || task.revisionId > existingEntry.revisionId" --> F[Cache write]
+///     E -- "task.revisionId <= existingEntry.revisionId" --> G[No cache write]
+///     F --> C
+///     G --> C
+/// ```
+///
+/// #### Scenario 1.2: Write Task Entry
+/// ```mermaid
+/// flowchart TD
+///     A["Write/create task"] --> B[Write task to database, increment revisionId]
+///     B --> C[cacheTaskPayloads]
+///     C -- "!existingEntry || task.revisionId > existingEntry.revisionId" --> D[Cache write]
+///     C -- "task.revisionId <= existingEntry.revisionId" --> E[No cache write]
+///     D --> F[Return task]
+///     E --> F[Return task]
+/// ```
+///
+/// ### Invariant 2: Commit Task Index Set Cache (`tasks_by_commit_ids/$commitSha`)
+/// - **Monotonically Growing Set & Immutable Commit Association**: A task's `commitSha` is permanent
+///   upon creation and never changes. Task status updates **never alter set membership** and **never touch
+///   `tasks_by_commit_ids`**. We only ever modify `tasks_by_commit_ids` when new tasks are created.
+///
+/// #### Scenario 2.1: Read `taskIdsByCommitId`
+/// ```mermaid
+/// flowchart TD
+///     A["getTaskIdsForCommit(commitSha)"] --> B["getTaskPayloads"]
+///     B -- Cache hit --> C["Return Set of Task IDs (SMEMBERS)"]
+///     B -- Cache miss --> D["initializeCommitTaskSet (setIfNotExists)"]
+///     D -- "Success (Set Initialized)" --> E["Set Created with Task IDs"]
+///     D -- "Failed (Intervening Initialization)" --> F[addTasksToCommitSet]
+///     F --> C
+///     E --> C
+/// ```
+///
+/// #### Scenario 2.2: Create Task
+/// ```mermaid
+/// flowchart TD
+///     A[Create task] --> B[addTasksToCommitSet]
+///     B -- Cache entry exists, taskId added --> C[Return]
+///     B -- Cache entry does not exist --> D[Fetch tasks for task.commitId]
+///     D --> E[initializeCommitTaskSet]
+///     E -- "Success (Set Initialized)" --> C
+///     E -- "Failed (Intervening Initialization)" --> F[addTasksToCommitSet]
+///     F --> C
+/// ```
+final class TaskCacheService {
+  TaskCacheService({required this.cache, this.config});
+
+  final CacheService cache;
+  final Config? config;
+
+  /// Returns whether caching is enabled in dynamic config.
+  bool get isEnabled => config?.flags.taskCachingEnabled ?? true;
+
+  /// Returns the default cache TTL based on dynamic configuration.
+  Duration get defaultTtl =>
+      Duration(hours: config?.flags.taskCacheTtlInHours ?? 12);
+
+  // --------------------------------------------------------------------------
+  // Invariant 1: Task Entity Payload Cache (`tasks/$docId`)
+  // --------------------------------------------------------------------------
+
+  /// Caches full [Task] document payloads using optimistic versioning (`revisionId`).
+  ///
+  /// The payload update only succeeds if `task.revisionId > existingRev` or the cache key is missing.
+  Future<void> cacheTaskPayloads(Iterable<Task> tasks, {Duration? ttl}) async {
+    if (tasks.isEmpty || !isEnabled) return;
+    final effectiveTtl = ttl ?? defaultTtl;
+    final entries = [
+      for (final t in tasks)
+        VersionedCacheEntry(
+          key: p.basename(t.name!),
+          value: Task.serialize(t),
+          revisionId: t.revisionId,
+          ttl: effectiveTtl,
+        ),
+    ];
+    await cache.insertVersioned('tasks', entries);
+  }
+
+  /// Evicts a deleted task payload from Redis.
+  Future<void> evictTaskPayload(String docId) async {
+    if (!isEnabled) return;
+    await cache.purge('tasks', docId);
+  }
+
+  /// Performs a batch lookup of task payloads by document IDs.
+  Future<TaskPayloadLookupResult> getTaskPayloads(
+    Iterable<String> docIds,
+  ) async {
+    final docIdList = docIds.toList();
+    if (docIdList.isEmpty || !isEnabled) {
+      return TaskPayloadLookupResult(foundTasks: [], missingDocIds: docIdList);
+    }
+
+    final cachedData = await cache.getMulti('tasks', docIdList);
+    final foundTasks = <Task>[];
+    final missingDocIds = <String>[];
+
+    for (var i = 0; i < docIdList.length; i++) {
+      final data = cachedData[i];
+      if (data != null) {
+        try {
+          foundTasks.add(Task.deserialize(data));
+        } catch (_) {
+          missingDocIds.add(docIdList[i]);
+        }
+      } else {
+        missingDocIds.add(docIdList[i]);
+      }
+    }
+
+    return TaskPayloadLookupResult(
+      foundTasks: foundTasks,
+      missingDocIds: missingDocIds,
+    );
+  }
+
+  // --------------------------------------------------------------------------
+  // Invariant 2: Commit Task Index Set Cache (`tasks_by_commit_ids/$commitSha`)
+  // --------------------------------------------------------------------------
+
+  /// Retrieves the set of task document IDs associated with [commitSha].
+  ///
+  /// Returns `null` if the set key does not exist in Redis (cache miss).
+  Future<Set<String>?> getTaskIdsForCommit(String commitSha) async {
+    if (!isEnabled) return null;
+    final set = await cache.getSet('tasks_by_commit_ids', commitSha);
+    if (set.isEmpty) return null;
+    return set;
+  }
+
+  /// Adds [taskIds] to the commit set if the set already exists in Redis (`addToSetIfExists`).
+  ///
+  /// Returns `true` if the set existed and task IDs were added; `false` if the set key is missing.
+  Future<bool> addTasksToCommitSet(
+    String commitSha,
+    Iterable<String> taskIds, {
+    Duration? ttl,
+  }) async {
+    if (taskIds.isEmpty || !isEnabled) return false;
+    var allAdded = true;
+    for (final taskId in taskIds) {
+      final added = await cache.addToSetIfExists(
+        'tasks_by_commit_ids',
+        commitSha,
+        taskId,
+      );
+      if (!added) {
+        allAdded = false;
+      }
+    }
+    return allAdded;
+  }
+
+  /// Initializes the commit task set if it does NOT already exist (`setIfNotExists`).
+  ///
+  /// Returns `true` if the set was successfully updated/initialized.
+  Future<bool> initializeCommitTaskSet(
+    String commitSha,
+    Iterable<String> taskIds, {
+    Duration? ttl,
+  }) async {
+    if (!isEnabled) return false;
+    await cache.updateSet(
+      'tasks_by_commit_ids',
+      commitSha,
+      taskIds.toSet(),
+      ttl: ttl ?? defaultTtl,
+    );
+    return true;
+  }
+}
diff --git a/app_dart/lib/src/service/flags/dynamic_config.dart b/app_dart/lib/src/service/flags/dynamic_config.dart
index 0e735074a..54e90f12a 100644
--- a/app_dart/lib/src/service/flags/dynamic_config.dart
+++ b/app_dart/lib/src/service/flags/dynamic_config.dart
@@ -43,6 +43,8 @@ final class DynamicConfig {
     orderedPresubmit: OrderedPresubmit.defaultInstance,
     dynamicTestSuppression: false,
     geminiModel: 'gemini-3-flash-preview',
+    taskCachingEnabled: true,
+    taskCacheTtlInHours: 12,
   );
 
   /// Upper limit of commit rows to be backfilled in API call.
@@ -85,6 +87,14 @@ final class DynamicConfig {
   @JsonKey()
   final String geminiModel;
 
+  /// Whether Redis caching for /tasks is enabled.
+  @JsonKey()
+  final bool taskCachingEnabled;
+
+  /// Cache TTL duration in hours for task cache entries.
+  @JsonKey()
+  final int taskCacheTtlInHours;
+
   const DynamicConfig._({
     required this.backfillerCommitLimit,
     required this.ciYaml,
@@ -95,6 +105,8 @@ final class DynamicConfig {
     required this.orderedPresubmit,
     required this.dynamicTestSuppression,
     required this.geminiModel,
+    required this.taskCachingEnabled,
+    required this.taskCacheTtlInHours,
   });
 
   /// Creates [DynamicConfig] flags from a [json] object.
@@ -110,6 +122,8 @@ final class DynamicConfig {
     OrderedPresubmit? orderedPresubmit,
     bool? dynamicTestSuppression,
     String? geminiModel,
+    bool? taskCachingEnabled,
+    int? taskCacheTtlInHours,
   }) {
     return DynamicConfig._(
       backfillerCommitLimit:
@@ -128,6 +142,10 @@ final class DynamicConfig {
       dynamicTestSuppression:
           dynamicTestSuppression ?? defaultInstance.dynamicTestSuppression,
       geminiModel: geminiModel ?? defaultInstance.geminiModel,
+      taskCachingEnabled:
+          taskCachingEnabled ?? defaultInstance.taskCachingEnabled,
+      taskCacheTtlInHours:
+          taskCacheTtlInHours ?? defaultInstance.taskCacheTtlInHours,
     );
   }
 
diff --git a/app_dart/lib/src/service/flags/dynamic_config.g.dart b/app_dart/lib/src/service/flags/dynamic_config.g.dart
index d958c97c1..51ca32175 100644
--- a/app_dart/lib/src/service/flags/dynamic_config.g.dart
+++ b/app_dart/lib/src/service/flags/dynamic_config.g.dart
@@ -33,6 +33,8 @@ DynamicConfig _$DynamicConfigFromJson(Map<String, dynamic> json) =>
             ),
       dynamicTestSuppression: json['dynamicTestSuppression'] as bool?,
       geminiModel: json['geminiModel'] as String?,
+      taskCachingEnabled: json['taskCachingEnabled'] as bool?,
+      taskCacheTtlInHours: (json['taskCacheTtlInHours'] as num?)?.toInt(),
     );
 
 Map<String, dynamic> _$DynamicConfigToJson(DynamicConfig instance) =>
@@ -46,4 +48,6 @@ Map<String, dynamic> _$DynamicConfigToJson(DynamicConfig instance) =>
       'orderedPresubmit': instance.orderedPresubmit.toJson(),
       'dynamicTestSuppression': instance.dynamicTestSuppression,
       'geminiModel': instance.geminiModel,
+      'taskCachingEnabled': instance.taskCachingEnabled,
+      'taskCacheTtlInHours': instance.taskCacheTtlInHours,
     };
diff --git a/app_dart/lib/src/service/luci_build_service.dart b/app_dart/lib/src/service/luci_build_service.dart
index b834078b9..b831cca15 100644
--- a/app_dart/lib/src/service/luci_build_service.dart
+++ b/app_dart/lib/src/service/luci_build_service.dart
@@ -12,7 +12,7 @@ import 'package:cocoon_common/task_status.dart';
 import 'package:cocoon_server/logging.dart';
 import 'package:fixnum/fixnum.dart';
 import 'package:github/github.dart';
-import 'package:googleapis/firestore/v1.dart' hide Status;
+
 import 'package:meta/meta.dart';
 
 import '../../cocoon_service.dart';
@@ -1169,10 +1169,7 @@ class LuciBuildService {
     task.resetAsRetry();
     task.setStatus(TaskStatus.inProgress);
 
-    await _firestore.batchWriteDocuments(
-      BatchWriteRequest(writes: documentsToWrites([task], exists: false)),
-      kDatabase,
-    );
+    await _firestore.updateTasks([task]);
 
     return task.currentAttempt;
   }
diff --git a/app_dart/lib/src/service/scheduler.dart b/app_dart/lib/src/service/scheduler.dart
index c7f793277..6f4f2199a 100644
--- a/app_dart/lib/src/service/scheduler.dart
+++ b/app_dart/lib/src/service/scheduler.dart
@@ -214,7 +214,9 @@ class Scheduler {
       final priority = await target.schedulerPolicy.triggerPriority(
         taskName: task.taskName,
         commitSha: commit.sha,
-        recentTasks: await _firestore.queryRecentTasks(name: task.taskName),
+        recentTasks: await _firestore.queryRecentTasksByName(
+          name: task.taskName,
+        ),
       );
       if (priority != null) {
         // Mark task as in progress to ensure it isn't scheduled over
@@ -255,6 +257,7 @@ class Scheduler {
     await _firestore.writeViaTransaction(
       documentsToWrites([...tasks, commit], exists: false),
     );
+    await _firestore.updateCacheForCreatedTasks(tasks);
   }
 
   /// Schedule all builds in batch requests instead of a single request.
@@ -1720,10 +1723,10 @@ $stacktrace
           final fs.Task fsTask;
 
           // Query the lastest run of the `checkName` againt commit `sha`.
-          final fsTasks = await _firestore.queryRecentTasks(
-            limit: 1,
+          final fsTasks = await _firestore.queryRecentTasksByCommit(
             commitSha: fsCommit.sha,
             name: checkName,
+            limit: 1,
           );
           if (fsTasks.isEmpty) {
             throw StateError('Expected 1+ tasks for $checkName');
diff --git a/app_dart/test/service/firestore_cache_test.dart b/app_dart/test/service/firestore_cache_test.dart
new file mode 100644
index 000000000..cf168861d
--- /dev/null
+++ b/app_dart/test/service/firestore_cache_test.dart
@@ -0,0 +1,236 @@
+// Copyright 2026 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:cocoon_common/task_status.dart';
+import 'package:cocoon_integration_test/testing.dart';
+import 'package:cocoon_server_test/test_logging.dart';
+import 'package:cocoon_service/src/model/firestore/task.dart';
+import 'package:cocoon_service/src/service/cache_service.dart';
+
+import 'package:googleapis/firestore/v1.dart';
+import 'package:path/path.dart' as p;
+import 'package:test/test.dart';
+
+void main() {
+  useTestLoggerPerTest();
+
+  group('Task Firestore Cache', () {
+    late FakeFirestoreService firestore;
+    late CacheService cache;
+    const commitSha = 'testSha';
+
+    setUp(() {
+      cache = CacheService.inMemory();
+      firestore = FakeFirestoreService(cache: cache);
+    });
+
+    test(
+      'Task.fromFirestore caches task and returns from cache on second call',
+      () async {
+        final task = generateFirestoreTask(
+          1,
+          name: 'Linux A',
+          attempts: 1,
+          status: TaskStatus.inProgress,
+        );
+        firestore.putDocument(task);
+
+        final fetched1 = await Task.fromFirestore(
+          firestore,
+          TaskId(
+            commitSha: commitSha,
+            taskName: task.taskName,
+            currentAttempt: task.currentAttempt,
+          ),
+        );
+        expect(fetched1.status, TaskStatus.inProgress);
+        expect(fetched1.revisionId, 1);
+
+        // Directly update firestore without using a service write so we know if cache is used on second read
+        final rawDoc = firestore.documents.firstWhere(
+          (d) => d.name == task.name,
+        );
+        rawDoc.fields![Task.fieldStatus] = Value(
+          stringValue: TaskStatus.succeeded.value,
+        );
+
+        final fetched2 = await Task.fromFirestore(
+          firestore,
+          TaskId(
+            commitSha: commitSha,
+            taskName: task.taskName,
+            currentAttempt: task.currentAttempt,
+          ),
+        );
+        // Because fetched2 comes from cache, it should still have status inProgress and rev 1
+        expect(fetched2.status, TaskStatus.inProgress);
+        expect(fetched2.revisionId, 1);
+      },
+    );
+
+    test(
+      'queryAllTasksForCommit caches commit task IDs and serves subsequent queries from versioned task cache',
+      () async {
+        final task1 = generateFirestoreTask(
+          1,
+          name: 'Linux A',
+          attempts: 1,
+          status: TaskStatus.inProgress,
+        );
+        final task2 = generateFirestoreTask(
+          2,
+          name: 'Mac B',
+          attempts: 1,
+          status: TaskStatus.succeeded,
+        );
+        firestore.putDocument(task1);
+        firestore.putDocument(task2);
+
+        final initialTasks = await firestore.queryAllTasksForCommit(
+          commitSha: commitSha,
+        );
+        expect(initialTasks.length, 2);
+        expect(
+          await cache.getSet('tasks_by_commit_ids', commitSha),
+          isNotEmpty,
+        );
+
+        // Verify cached task lookup without hitting firestore query
+        firestore.reset();
+        final cachedTasks = await firestore.queryAllTasksForCommit(
+          commitSha: commitSha,
+        );
+        expect(cachedTasks.length, 2);
+      },
+    );
+
+    test(
+      'queryAllTasksForCommit recovers missing individual task entries via partial query when commit set is populated',
+      () async {
+        final task1 = generateFirestoreTask(
+          1,
+          name: 'Linux A',
+          attempts: 1,
+          status: TaskStatus.inProgress,
+        );
+        final task2 = generateFirestoreTask(
+          2,
+          name: 'Mac B',
+          attempts: 1,
+          status: TaskStatus.succeeded,
+        );
+        firestore.putDocument(task1);
+        firestore.putDocument(task2);
+
+        await firestore.queryAllTasksForCommit(commitSha: commitSha);
+        await Future<void>.delayed(const Duration(milliseconds: 10));
+        await cache.purge('tasks', '${commitSha}_Mac B_1');
+        expect(await cache.get('tasks', '${commitSha}_Mac B_1'), isNull);
+
+        final recoveredTasks = await firestore.queryAllTasksForCommit(
+          commitSha: commitSha,
+        );
+        expect(recoveredTasks.length, 2);
+        expect(await cache.get('tasks', '${commitSha}_Mac B_1'), isNotNull);
+      },
+    );
+
+    test(
+      'batchWriteDocuments updates versioned task cache with full document and incremented revisionId',
+      () async {
+        final task1 = generateFirestoreTask(
+          1,
+          name: 'Linux A',
+          attempts: 1,
+          status: TaskStatus.inProgress,
+        );
+        firestore.putDocument(task1);
+
+        // Populate commit cache and task cache via queryAllTasksForCommit
+        final initialTasks = await firestore.queryAllTasksForCommit(
+          commitSha: commitSha,
+        );
+        expect(initialTasks.first.status, TaskStatus.inProgress);
+        expect(initialTasks.first.revisionId, 1);
+        expect(
+          await cache.getSet('tasks_by_commit_ids', commitSha),
+          isNotEmpty,
+        );
+
+        // Mutate task status and increment revisionId
+        task1.setStatus(TaskStatus.succeeded);
+        await firestore.updateTasks([task1]);
+
+        // Verify that tasks_by_commit_ids remains intact
+        expect(
+          await cache.getSet('tasks_by_commit_ids', commitSha),
+          isNotEmpty,
+        );
+
+        // Verify that single-task payload cache entry was updated in Redis with full document and rev 2
+        final cachedBytes = await cache.get('tasks', p.basename(task1.name!));
+        expect(cachedBytes, isNotNull);
+        final cachedTask = Task.deserialize(cachedBytes!);
+        expect(cachedTask.status, TaskStatus.succeeded);
+        expect(cachedTask.revisionId, 2);
+      },
+    );
+
+    test(
+      'updateTasks updates versioned task cache in-place lock-free',
+      () async {
+        final task = generateFirestoreTask(
+          1,
+          name: 'Linux B',
+          attempts: 1,
+          status: TaskStatus.inProgress,
+        );
+        firestore.putDocument(task);
+        await firestore.queryAllTasksForCommit(commitSha: commitSha);
+
+        task.setStatus(TaskStatus.succeeded);
+        await firestore.updateTasks([task]);
+
+        final updatedTask = await Task.fromFirestore(
+          firestore,
+          TaskId(
+            commitSha: commitSha,
+            taskName: task.taskName,
+            currentAttempt: task.currentAttempt,
+          ),
+        );
+        expect(updatedTask.status, TaskStatus.succeeded);
+        expect(updatedTask.revisionId, 2);
+      },
+    );
+
+    test('updateCacheForCreatedTasks groups tasks by commit SHA', () async {
+      final task1 = generateFirestoreTask(
+        1,
+        commitSha: 'shaA',
+        name: 'Linux A',
+      );
+      final task2 = generateFirestoreTask(
+        2,
+        commitSha: 'shaA',
+        name: 'Linux B',
+      );
+      final task3 = generateFirestoreTask(
+        3,
+        commitSha: 'shaB',
+        name: 'Linux C',
+      );
+
+      await firestore.updateCacheForCreatedTasks([task1, task2, task3]);
+
+      final cachedTask1 = await cache.get('tasks', p.basename(task1.name!));
+      final cachedTask2 = await cache.get('tasks', p.basename(task2.name!));
+      final cachedTask3 = await cache.get('tasks', p.basename(task3.name!));
+
+      expect(cachedTask1, isNotNull);
+      expect(cachedTask2, isNotNull);
+      expect(cachedTask3, isNotNull);
+    });
+  });
+}
diff --git a/app_dart/test/service/task_cache_service_test.dart b/app_dart/test/service/task_cache_service_test.dart
new file mode 100644
index 000000000..9e1076981
--- /dev/null
+++ b/app_dart/test/service/task_cache_service_test.dart
@@ -0,0 +1,149 @@
+// Copyright 2019 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:cocoon_common/task_status.dart';
+import 'package:cocoon_integration_test/testing.dart';
+import 'package:cocoon_server_test/test_logging.dart';
+import 'package:cocoon_service/src/model/firestore/task.dart';
+import 'package:cocoon_service/src/service/cache_service.dart';
+import 'package:cocoon_service/src/service/firestore/task_cache_service.dart';
+import 'package:cocoon_service/src/service/flags/dynamic_config.dart';
+import 'package:path/path.dart' as p;
+import 'package:test/test.dart';
+
+void main() {
+  useTestLoggerPerTest();
+
+  late InMemoryCacheService cache;
+  late FakeConfig config;
+  late TaskCacheService taskCache;
+
+  Task generateTestTask(
+    int attempt, {
+    String name = 'Linux A',
+    String commitSha = 'sha123',
+    TaskStatus status = TaskStatus.inProgress,
+  }) {
+    return Task(
+      builderName: name,
+      currentAttempt: attempt,
+      commitSha: commitSha,
+      bringup: false,
+      createTimestamp: 1000,
+      startTimestamp: 1000,
+      endTimestamp: 0,
+      status: status,
+      testFlaky: false,
+      buildNumber: null,
+    );
+  }
+
+  String docIdFor(Task task) => p.basename(task.name!);
+
+  setUp(() {
+    cache = InMemoryCacheService();
+    config = FakeConfig(dynamicConfig: DynamicConfig(taskCachingEnabled: true));
+    taskCache = TaskCacheService(cache: cache, config: config);
+  });
+
+  group('TaskCacheService Payload Caching', () {
+    test(
+      'cacheTaskPayloads stores versioned payload and getTaskPayloads retrieves it',
+      () async {
+        final task = generateTestTask(1);
+        await taskCache.cacheTaskPayloads([task]);
+
+        final result = await taskCache.getTaskPayloads([docIdFor(task)]);
+        expect(result.foundTasks.length, equals(1));
+        expect(result.foundTasks.first.taskName, equals('Linux A'));
+        expect(result.missingDocIds, isEmpty);
+      },
+    );
+
+    test('evictTaskPayload purges item from cache', () async {
+      final task = generateTestTask(1);
+      await taskCache.cacheTaskPayloads([task]);
+
+      await taskCache.evictTaskPayload(docIdFor(task));
+      final result = await taskCache.getTaskPayloads([docIdFor(task)]);
+      expect(result.foundTasks, isEmpty);
+      expect(result.missingDocIds, contains(docIdFor(task)));
+    });
+
+    test('getTaskPayloads correctly reports missing and found items', () async {
+      final task1 = generateTestTask(1, name: 'Linux A');
+      await taskCache.cacheTaskPayloads([task1]);
+
+      final result = await taskCache.getTaskPayloads([
+        docIdFor(task1),
+        'non_existent_doc',
+      ]);
+      expect(result.foundTasks.length, equals(1));
+      expect(result.foundTasks.first.taskName, equals('Linux A'));
+      expect(result.missingDocIds, equals(['non_existent_doc']));
+    });
+
+    test('cacheTaskPayloads respects taskCachingEnabled flag', () async {
+      config.dynamicConfig = DynamicConfig(taskCachingEnabled: false);
+      final task = generateTestTask(1);
+      await taskCache.cacheTaskPayloads([task]);
+
+      final result = await taskCache.getTaskPayloads([docIdFor(task)]);
+      expect(result.foundTasks, isEmpty);
+      expect(result.missingDocIds, contains(docIdFor(task)));
+    });
+  });
+
+  group('TaskCacheService Commit Set Caching', () {
+    test('getTaskIdsForCommit returns null on cache miss', () async {
+      final taskIds = await taskCache.getTaskIdsForCommit('sha123');
+      expect(taskIds, isNull);
+    });
+
+    test(
+      'initializeCommitTaskSet populates set and getTaskIdsForCommit retrieves it',
+      () async {
+        final created = await taskCache.initializeCommitTaskSet('sha123', [
+          'doc1',
+          'doc2',
+        ]);
+        expect(created, isTrue);
+
+        final taskIds = await taskCache.getTaskIdsForCommit('sha123');
+        expect(taskIds, equals({'doc1', 'doc2'}));
+      },
+    );
+
+    test('addTasksToCommitSet adds items only if set exists', () async {
+      // First attempt on missing set returns false
+      final addedMissing = await taskCache.addTasksToCommitSet('sha123', [
+        'doc1',
+      ]);
+      expect(addedMissing, isFalse);
+
+      // Initialize set
+      await taskCache.initializeCommitTaskSet('sha123', ['doc1']);
+
+      // Second attempt on existing set returns true and updates set
+      final addedExisting = await taskCache.addTasksToCommitSet('sha123', [
+        'doc2',
+      ]);
+      expect(addedExisting, isTrue);
+
+      final taskIds = await taskCache.getTaskIdsForCommit('sha123');
+      expect(taskIds, equals({'doc1', 'doc2'}));
+    });
+
+    test(
+      'initializeCommitTaskSet returns false if set flag is disabled',
+      () async {
+        config.dynamicConfig = DynamicConfig(taskCachingEnabled: false);
+        final first = await taskCache.initializeCommitTaskSet('sha123', [
+          'doc1',
+        ]);
+        expect(first, isFalse);
+      },
+    );
+  });
+}
diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart
index 4796cdab3..fd77960a6 100644
--- a/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart
+++ b/packages/cocoon_integration_test/lib/src/fakes/fake_cache_service.dart
@@ -14,6 +14,38 @@ class FakeCacheService extends CacheService {
   @override
   Future<Uint8List?> get(String subcacheName, String key) async => null;
 
+  @override
+  Future<List<Uint8List?>> getMulti(
+    String subcacheName,
+    List<String> keys,
+  ) async {
+    return List.filled(keys.length, null);
+  }
+
+  @override
+  Future<void> insertVersioned(
+    String subcacheName,
+    List<VersionedCacheEntry> entries,
+  ) async {}
+
+  @override
+  Future<Set<String>> getSet(String subcacheName, String key) async => const {};
+
+  @override
+  Future<void> updateSet(
+    String subcacheName,
+    String key,
+    Set<String> values, {
+    Duration ttl = const Duration(hours: 12),
+  }) async {}
+
+  @override
+  Future<bool> addToSetIfExists(
+    String subcacheName,
+    String key,
+    String value,
+  ) async => false;
+
   @override
   Future<void> dispose() async {}
 
diff --git a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart
index bcd6c4a1a..4d1e198a0 100644
--- a/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart
+++ b/packages/cocoon_integration_test/lib/src/fakes/fake_firestore_service.dart
@@ -2,10 +2,13 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
 import 'package:cocoon_service/src/model/firestore/base.dart';
+import 'package:cocoon_service/src/model/firestore/task.dart';
+import 'package:cocoon_service/src/service/cache_service.dart';
 import 'package:cocoon_service/src/service/config.dart';
 import 'package:cocoon_service/src/service/firestore.dart';
 import 'package:googleapis/firestore/v1.dart';
@@ -22,6 +25,11 @@ final queryKeyValidator = RegExp(r'(^[a-zA-Z_][a-zA-Z_0-9]*)$');
 abstract base class _FakeInMemoryFirestoreService
     with FirestoreQueries
     implements FirestoreService {
+  _FakeInMemoryFirestoreService({this.cache});
+
+  @override
+  final CacheService? cache;
+
   /// Every document currently stored in the fake.
   Iterable<Document> get documents => _documents.values;
   final _documents = <String, Document>{};
@@ -195,36 +203,31 @@ abstract base class _FakeInMemoryFirestoreService
 
     for (final transform in updateTransforms ?? const <FieldTransform>[]) {
       final field = fields[transform.fieldPath];
-      if (field == null) {
-        if (transform.appendMissingElements != null) {
-          // firestore: Append the given elements in order if they are not
-          // already present in the current field value. If the field is not an
-          // array, or if the field does not yet exist, it is first set to the
-          // empty array.
-          // https://firebase.google.com/docs/firestore/reference/rest/v1beta1/Write#FieldTransform
+      if (transform.appendMissingElements?.values case final elements?) {
+        if (field?.arrayValue?.values case final fieldArray?) {
+          for (final element in elements) {
+            if (fieldArray.contains(element)) continue;
+            fieldArray.add(element);
+          }
+        } else {
           fields[transform.fieldPath!] = Value(
-            arrayValue: transform.appendMissingElements!,
+            arrayValue: ArrayValue(values: elements),
           );
-          continue;
         }
-        // this is most certainly wrong as the real firestore probably(?) updates
-        // the field. We're not using it that way in the tests, so if you find
-        // yourself getting this error - congrats and welcome to the team.
+      } else if (transform.increment case final inc?) {
+        if (field?.integerValue case final oldVal?) {
+          final newVal = int.parse(oldVal) + int.parse(inc.integerValue!);
+          fields[transform.fieldPath!] = Value(integerValue: newVal.toString());
+        } else {
+          fields[transform.fieldPath!] = Value(integerValue: inc.integerValue!);
+        }
+      } else if (field == null) {
         throw ArgumentError.value(
           transform.fieldPath,
           'field',
           'not found, cannot patch',
         );
       }
-      // The following are "union" members; only one operation per transform.
-      if (transform.appendMissingElements?.values case final elements?) {
-        if (field.arrayValue?.values case final fieldArray?) {
-          for (final element in elements) {
-            if (fieldArray.contains(element)) continue;
-            fieldArray.add(element);
-          }
-        }
-      }
     }
 
     _checkDatabaseName(name);
@@ -237,6 +240,11 @@ abstract base class _FakeInMemoryFirestoreService
           _now().toUtc().toIso8601String(),
       updateTime: (updated ?? _now()).toUtc().toIso8601String(),
     );
+    if (cache != null && collection == kTaskCollectionId) {
+      unawaited(
+        updateCacheForCreatedTasks([Task.fromDocument(_documents[name]!)]),
+      );
+    }
     return _clone(_documents[name]!);
   }
 
@@ -377,6 +385,21 @@ abstract base class _FakeInMemoryFirestoreService
     return result;
   }
 
+  @override
+  Future<List<Document>> batchGetDocuments(
+    List<String> names, {
+    Transaction? transaction,
+  }) async {
+    final results = <Document>[];
+    for (final name in names) {
+      final doc = tryPeekDocumentByName(name);
+      if (doc != null) {
+        results.add(doc);
+      }
+    }
+    return results;
+  }
+
   @override
   Future<Document> createDocument(
     Document document, {
@@ -667,7 +690,9 @@ abstract base class _FakeInMemoryFirestoreService
 }
 
 /// A fake implementation of [FirestoreService].
-final class FakeFirestoreService extends _FakeInMemoryFirestoreService {}
+final class FakeFirestoreService extends _FakeInMemoryFirestoreService {
+  FakeFirestoreService({super.cache});
+}
 
 /// Checks that the models described by [metadata] match storage of [matcher].
 ///