Перевести FakePipelineStore на NSubstitute

Хелпер Support/TestPipelineStore: очередь/отсев/дедуп с полной семантикой
(upsert с сохранением Returned, LIKE-поиск, атомарный claim, purge), сценарий
claim-гонки параметром (claimRace/claimResult), счётчики ClaimCalls/
DeleteClaimCalls через When/Do. Потребители (12 файлов) перетипизированы на
.Store, фейк удалён, тесты 1340 зелёные.

src/core/tests/Deal.Tests.Unit/Contracts/PipelineWorkerGrpcAiTests.cs

@@ -154,7 +154,7 @@ public sealed class PipelineWorkerGrpcAiTests
 
     private sealed record Context(
         PipelineWorkerService Worker,
-        FakePipelineStore PipelineStore,
+        TestPipelineStore PipelineStore,
         FakeKanjStore KanjStore,
         FakeSettingsStore Settings,
         FakeTenantLimitStore Limits);
@@ -165,7 +165,7 @@ public sealed class PipelineWorkerGrpcAiTests
         bool budgeted = false)
     {
         var settings = new FakeSettingsStore();
-        var pipelineStore = new FakePipelineStore();
+        var pipelineStore = new TestPipelineStore();
         var kanjStore = new FakeKanjStore();
         var mlClient = new FakeMlClient { Predict = NotReadyPrediction() };
         var rules = new IncomingRules(settings);
@@ -192,11 +192,11 @@ public sealed class PipelineWorkerGrpcAiTests
                 NullLogger<BudgetedAiClassifier>.Instance)
             : grpcClassifier;
 
-        var processing = new PipelineProcessingService(pipelineStore, mlClient, new PipelineIngestService(pipelineStore));
+        var processing = new PipelineProcessingService(pipelineStore.Store, mlClient, new PipelineIngestService(pipelineStore.Store));
         var composer = new CardComposer(kanjStore, settings);
-        var writer = new PipelineCardWriter(kanjStore, pipelineStore, composer);
+        var writer = new PipelineCardWriter(kanjStore, pipelineStore.Store, composer);
         var worker = new PipelineWorkerService(
-            pipelineStore, settings, rules, kanjStore, mlClient, aiClassifier, processing, writer, fieldsParser);
+            pipelineStore.Store, settings, rules, kanjStore, mlClient, aiClassifier, processing, writer, fieldsParser);
         return new Context(worker, pipelineStore, kanjStore, settings, limits);
     }
 

src/core/tests/Deal.Tests.Unit/Modules/Kanban/MlReviewServiceTests.cs

@@ -19,15 +19,15 @@ public sealed class MlReviewServiceTests
     private const string Dialog = "d_1";
 
     private static MlReviewService Create(
-        FakePipelineStore pipeline,
+        TestPipelineStore pipeline,
         FakeKanjStore kanj,
         FakeMlClient ml,
         out CardsService cards)
     {
         var settings = new FakeSettingsStore();
         cards = new CardsService(kanj, settings, ml, new TestFileStorage().Storage);
-        var processing = new PipelineProcessingService(pipeline, ml, new PipelineIngestService(pipeline));
-        return new MlReviewService(pipeline, kanj, cards, processing, ml);
+        var processing = new PipelineProcessingService(pipeline.Store, ml, new PipelineIngestService(pipeline.Store));
+        return new MlReviewService(pipeline.Store, kanj, cards, processing, ml);
     }
 
     private static QueueItemDto QueueRow(
@@ -73,7 +73,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Candidates_MergesQueueRejectedAndCardsWithVerdicts()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         pipeline.SeedQueue(QueueRow(101, "из очереди"));
         pipeline.SeedRejected(RejectedRow(102, "из отсева"));
         var kanj = new FakeKanjStore();
@@ -94,7 +94,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Candidates_FiltersByDialog()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         pipeline.SeedQueue(QueueRow(101, "нужный", Dialog));
         pipeline.SeedQueue(QueueRow(201, "другой", "d_2"));
         MlReviewService service = Create(pipeline, new FakeKanjStore(), new FakeMlClient(), out _);
@@ -108,7 +108,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Candidates_NoDialog_ReturnsAllSources()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         pipeline.SeedQueue(QueueRow(101, "a", "d_1"));
         pipeline.SeedQueue(QueueRow(201, "b", "d_2"));
         MlReviewService service = Create(pipeline, new FakeKanjStore(), new FakeMlClient(), out _);
@@ -121,7 +121,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Candidates_ClampsLimitToMax()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         for (int i = 0; i < 5; i++)
         {
             pipeline.SeedQueue(QueueRow(100 + i, $"текст {i}"));
@@ -137,7 +137,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Apply_Skip_DoesNotLearn()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         pipeline.SeedQueue(QueueRow(101, "текст"));
         var ml = new FakeMlClient();
         MlReviewService service = Create(pipeline, new FakeKanjStore(), ml, out _);
@@ -157,7 +157,7 @@ public sealed class MlReviewServiceTests
         var kanj = new FakeKanjStore();
         kanj.SeedCard(Card(101, "спамный текст"));
         var ml = new FakeMlClient();
-        MlReviewService service = Create(new FakePipelineStore(), kanj, ml, out _);
+        MlReviewService service = Create(new TestPipelineStore(), kanj, ml, out _);
 
         MlApplyResult? result = await service.ApplyAsync(Dialog, 101, MlReviewService.ActionSpam, CancellationToken.None);
 
@@ -172,7 +172,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Apply_Spam_QueuedMessage_RejectsAndRemovesFromQueue()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         pipeline.SeedQueue(QueueRow(101, "рекламный текст"));
         var ml = new FakeMlClient();
         MlReviewService service = Create(pipeline, new FakeKanjStore(), ml, out _);
@@ -202,7 +202,7 @@ public sealed class MlReviewServiceTests
         });
         kanj.SeedCard(Card(101, "python разработчик"));
         var ml = new FakeMlClient();
-        MlReviewService service = Create(new FakePipelineStore(), kanj, ml, out _);
+        MlReviewService service = Create(new TestPipelineStore(), kanj, ml, out _);
 
         MlApplyResult? result = await service.ApplyAsync(Dialog, 101, "board:b_py", CancellationToken.None);
 
@@ -218,7 +218,7 @@ public sealed class MlReviewServiceTests
     {
         var kanj = new FakeKanjStore();
         kanj.SeedCard(Card(101, "текст"));
-        MlReviewService service = Create(new FakePipelineStore(), kanj, new FakeMlClient(), out _);
+        MlReviewService service = Create(new TestPipelineStore(), kanj, new FakeMlClient(), out _);
 
         MlApplyResult? result = await service.ApplyAsync(Dialog, 101, "board:missing", CancellationToken.None);
 
@@ -230,7 +230,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Apply_UnknownAction_ReturnsError()
     {
-        var pipeline = new FakePipelineStore();
+        var pipeline = new TestPipelineStore();
         pipeline.SeedQueue(QueueRow(101, "текст"));
         MlReviewService service = Create(pipeline, new FakeKanjStore(), new FakeMlClient(), out _);
 
@@ -244,7 +244,7 @@ public sealed class MlReviewServiceTests
     [Fact]
     public async Task Apply_MessageNotFound_ReturnsNull()
     {
-        MlReviewService service = Create(new FakePipelineStore(), new FakeKanjStore(), new FakeMlClient(), out _);
+        MlReviewService service = Create(new TestPipelineStore(), new FakeKanjStore(), new FakeMlClient(), out _);
 
         MlApplyResult? result = await service.ApplyAsync(Dialog, 999, MlReviewService.ActionSpam, CancellationToken.None);
 

src/core/tests/Deal.Tests.Unit/Support/AdminTickOrchestratorTests.cs

@@ -12,6 +12,7 @@ using Deal.Modules.Settings.Application.Models;
 using Deal.Tests.Unit.Contracts;
 using Deal.Tests.Unit.Modules.Cards;
 using Deal.Tests.Unit.Modules.Kanban;
+using NSubstitute;
 using Deal.Tests.Unit.Modules.Settings;
 using Microsoft.Extensions.Logging.Abstractions;
 
@@ -28,7 +29,7 @@ public sealed class AdminTickOrchestratorTests
     // Контекст теста: оркестратор на общих фейках + канал подписки.
     private sealed record Context(
         AdminTickOrchestrator Orchestrator,
-        FakePipelineStore PipelineStore,
+        TestPipelineStore PipelineStore,
         FakeKanjStore KanjStore,
         FakeSettingsStore Settings,
         TestAiClassifier AiClassifier,
@@ -202,18 +203,24 @@ public sealed class AdminTickOrchestratorTests
     {
         var settings = new FakeSettingsStore();
         // При сбое чтения очереди подменяем хранилище целиком (тот же объект идёт во все сервисы и в контекст).
-        FakePipelineStore store = withThrowingQueueRead ? new ThrowingQueueReadPipelineStore() : new FakePipelineStore();
+        TestPipelineStore store = new();
+        if (withThrowingQueueRead)
+        {
+            store.Store.ListAsync(Arg.Any<string?>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+                .Returns<IReadOnlyList<QueueItemDto>>(_ =>
+                    throw new InvalidOperationException("Тестовый сбой чтения очереди (ListAsync)."));
+        }
         var kanjStore = new FakeKanjStore();
         var mlClient = new FakeMlClient();
         var aiClassifier = new TestAiClassifier();
         var rules = new IncomingRules(settings);
         var fieldsParser = new LocalFieldsParser(settings);
-        var ingest = new PipelineIngestService(store);
-        var processing = new PipelineProcessingService(store, mlClient, ingest);
+        var ingest = new PipelineIngestService(store.Store);
+        var processing = new PipelineProcessingService(store.Store, mlClient, ingest);
         var composer = new CardComposer(kanjStore, settings);
-        var writer = new PipelineCardWriter(kanjStore, store, composer);
+        var writer = new PipelineCardWriter(kanjStore, store.Store, composer);
         var worker = new PipelineWorkerService(
-            store, settings, rules, kanjStore, mlClient, aiClassifier.Classifier, processing, writer, fieldsParser);
+            store.Store, settings, rules, kanjStore, mlClient, aiClassifier.Classifier, processing, writer, fieldsParser);
 
         var broker = new SseBroker();
         var tickService = new StorageTickService(kanjStore, settings);
@@ -261,19 +268,6 @@ public sealed class AdminTickOrchestratorTests
         };
     }
 
-    // Хранилище со сбоем чтения очереди: ListAsync бросает (сценарий «БД/схема недоступны» на pump).
-    private sealed class ThrowingQueueReadPipelineStore : FakePipelineStore
-    {
-        /// <inheritdoc />
-        public override Task<IReadOnlyList<QueueItemDto>> ListAsync(
-            string? status,
-            int limit,
-            CancellationToken ct)
-        {
-            throw new InvalidOperationException("Тестовый сбой чтения очереди (ListAsync).");
-        }
-    }
-
     // Строка очереди (status new; сценарий перекрывает текст/время).
     private static QueueItemDto QueueRow(string id, string text) => SourceItemFactory.Queue(
         id,

src/core/tests/Deal.Tests.Unit/Support/PipelineWorkerServiceTests.cs

@@ -24,7 +24,7 @@ public sealed class PipelineWorkerServiceTests
     // Контекст теста: воркер поверх in-memory фейков хранилищ/портов.
     private sealed record Context(
         PipelineWorkerService Worker,
-        FakePipelineStore PipelineStore,
+        TestPipelineStore PipelineStore,
         FakeKanjStore KanjStore,
         FakeSettingsStore Settings,
         FakeMlClient MlClient,
@@ -33,20 +33,20 @@ public sealed class PipelineWorkerServiceTests
     // Собирает контекст: дефолты настроек/досок, ML «не готов» (как LocalMlClient), ИИ-фильтр пропускает.
     // pipelineStore: Хранилище очереди (по умолчанию — обычный фейк; сценарий гонки claim'а — RacingClaimPipelineStore).
     // Возвращает: Воркер и фейки (сценарий до-настраивает строки очереди/ответы портов).
-    private static Context CreateContext(FakePipelineStore? pipelineStore = null)
+    private static Context CreateContext(TestPipelineStore? pipelineStore = null)
     {
         var settings = new FakeSettingsStore();
-        var store = pipelineStore ?? new FakePipelineStore();
+        var store = pipelineStore ?? new TestPipelineStore();
         var kanjStore = new FakeKanjStore();
         var mlClient = new FakeMlClient { Predict = NotReadyPrediction() };
         var aiClassifier = new TestAiClassifier();
         var rules = new IncomingRules(settings);
         var fieldsParser = new LocalFieldsParser(settings);
-        var processing = new PipelineProcessingService(store, mlClient, new PipelineIngestService(store));
+        var processing = new PipelineProcessingService(store.Store, mlClient, new PipelineIngestService(store.Store));
         var composer = new CardComposer(kanjStore, settings);
-        var writer = new PipelineCardWriter(kanjStore, store, composer);
+        var writer = new PipelineCardWriter(kanjStore, store.Store, composer);
         var worker = new PipelineWorkerService(
-            store, settings, rules, kanjStore, mlClient, aiClassifier.Classifier, processing, writer, fieldsParser);
+            store.Store, settings, rules, kanjStore, mlClient, aiClassifier.Classifier, processing, writer, fieldsParser);
         return new Context(worker, store, kanjStore, settings, mlClient, aiClassifier);
     }
 
@@ -54,38 +54,7 @@ public sealed class PipelineWorkerServiceTests
     // видит заявку, но ClaimAsync (INSERT … ON CONFLICT DO NOTHING в адаптере) возвращает false — хэш успел
     // заявить другой проход. DeleteClaimAsync записывается — проверяем, что чужая заявка НЕ снимается.
     // claimResult: Результат ClaimAsync (по умолчанию false — «победил другой проход»).
-    private sealed class RacingClaimPipelineStore(bool claimResult = false) : FakePipelineStore
-    {
-        /// <summary>
-        /// Сколько раз вызван ClaimAsync
-        /// </summary>
-        public int ClaimCalls { get; private set; }
 
-        /// <summary>
-        /// Сколько раз вызван DeleteClaimAsync
-        /// </summary>
-        public int DeleteClaimCalls { get; private set; }
-
-        /// <inheritdoc />
-        public override Task<bool> ExistsAsync(string hash, CancellationToken ct)
-        {
-            return Task.FromResult(false); // параллельный проход ещё не вставил строку на момент проверки
-        }
-
-        /// <inheritdoc />
-        public override Task<bool> ClaimAsync(string hash, CancellationToken ct)
-        {
-            ClaimCalls++;
-            return Task.FromResult(claimResult); // ON CONFLICT DO NOTHING: вставил другой проход → false
-        }
-
-        /// <inheritdoc />
-        public override Task DeleteClaimAsync(string hash, CancellationToken ct)
-        {
-            DeleteClaimCalls++;
-            return Task.CompletedTask;
-        }
-    }
 
     // Сериализует значение настройки в JSON-строку строки settings (как пишет SettingsStore).
     // value: Значение (число/булево/строка/список).
@@ -284,13 +253,13 @@ public sealed class PipelineWorkerServiceTests
     [Fact]
     public async Task Pump_ClaimLostToParallelPass_RejectsDuplicateAndDoesNotCreateCard()
     {
-        Context ctx = CreateContext(new RacingClaimPipelineStore());
+        Context ctx = CreateContext(new TestPipelineStore(claimRace: true));
         const string text = "Вакансия: Python-разработчик в команду, удалённая работа, оплата 2000$ в месяц";
         ctx.PipelineStore.SeedQueue(QueueRow("p_1", text));
 
         PipelinePumpResult result = await ctx.Worker.PumpOnceAsync(default);
 
-        RacingClaimPipelineStore store = Assert.IsType<RacingClaimPipelineStore>(ctx.PipelineStore);
+        TestPipelineStore store = ctx.PipelineStore;
         Assert.Equal(1, store.ClaimCalls); // конфликт claim'а обработан, а не пропущен молча
         RejectedItemDto rejected = Assert.Single(ctx.PipelineStore.Rejected);
         Assert.Equal("dup", rejected.Stage);
@@ -377,7 +346,7 @@ public sealed class PipelineWorkerServiceTests
         Assert.Empty(ctx.KanjStore.CardDtos);
         Assert.Equal(1, result.NoBudget);
         Assert.Equal(0, result.AiStored); // отсев фильтром не считается решением ИИ (python L1143–1147)
-        Assert.False(await ctx.PipelineStore.ExistsAsync(DedupHasher.Hash(text), default));
+        Assert.False(await ctx.PipelineStore.Store.ExistsAsync(DedupHasher.Hash(text), default));
         Assert.Null(ctx.Settings.GetStoredJson(SettingsKeys.AiDecisions));
     }
 
@@ -615,7 +584,7 @@ public sealed class PipelineWorkerServiceTests
 
         QueueItemDto row = Assert.Single(ctx.PipelineStore.Queue);
         Assert.Equal(PipelineQueueStatuses.Filtered, row.Status); // прошла «new»-проход, ИИ-шаг не завершился
-        Assert.True(await ctx.PipelineStore.ExistsAsync(DedupHasher.Hash(text), default)); // claim на месте
+        Assert.True(await ctx.PipelineStore.Store.ExistsAsync(DedupHasher.Hash(text), default)); // claim на месте
         Assert.Empty(ctx.PipelineStore.Rejected);
         Assert.Empty(ctx.KanjStore.CardDtos);
     }

src/core/tests/Deal.Tests.Unit/Support/StorageTickSchedulerTests.cs

@@ -153,7 +153,7 @@ public sealed class StorageTickSchedulerTests
         var kanjStore = new FakeKanjStore();
         var settings = new FakeSettingsStore();
         var tenantContext = new TenantContext();
-        var pipelineStoreA = new FakePipelineStore();
+        var pipelineStoreA = new TestPipelineStore();
         long nowMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
         pipelineStoreA.SeedRejected(Rejected("r_old_a", (long)(nowMs - TimeSpan.FromDays(4).TotalMilliseconds)));
         pipelineStoreA.SeedRejected(Rejected("r_fresh", (long)(nowMs - TimeSpan.FromDays(1).TotalMilliseconds)));
@@ -162,7 +162,7 @@ public sealed class StorageTickSchedulerTests
             tenantContext,
             new Dictionary<Guid, FakeKanjStore> { [TenantA] = kanjStore, [TenantB] = new FakeKanjStore() },
             new Dictionary<Guid, ISettingsStore> { [TenantA] = settings, [TenantB] = settings },
-            new Dictionary<Guid, FakePipelineStore> { [TenantA] = pipelineStoreA, [TenantB] = new FakePipelineStore() });
+            new Dictionary<Guid, TestPipelineStore> { [TenantA] = pipelineStoreA, [TenantB] = new TestPipelineStore() });
 
         SseBroker broker = provider.GetRequiredService<SseBroker>();
         SseSubscription subscriptionA = broker.Subscribe(TenantA);
@@ -192,7 +192,7 @@ public sealed class StorageTickSchedulerTests
             tenantContext,
             new Dictionary<Guid, FakeKanjStore> { [TenantA] = cardStoreA, [TenantB] = cardStoreB },
             new Dictionary<Guid, ISettingsStore> { [TenantA] = new FakeSettingsStore(), [TenantB] = new FakeSettingsStore() },
-            new Dictionary<Guid, FakePipelineStore> { [TenantA] = new(), [TenantB] = new() });
+            new Dictionary<Guid, TestPipelineStore> { [TenantA] = new(), [TenantB] = new() });
 
         SseBroker broker = provider.GetRequiredService<SseBroker>();
         SseSubscription subscriptionA = broker.Subscribe(TenantA);
@@ -227,7 +227,7 @@ public sealed class StorageTickSchedulerTests
             tenantContext,
             new Dictionary<Guid, FakeKanjStore> { [TenantA] = cardStoreA },
             new Dictionary<Guid, ISettingsStore> { [TenantA] = settingsA },
-            new Dictionary<Guid, FakePipelineStore> { [TenantA] = new() });
+            new Dictionary<Guid, TestPipelineStore> { [TenantA] = new() });
 
         SseBroker broker = provider.GetRequiredService<SseBroker>();
         SseSubscription subscriptionA = broker.Subscribe(TenantA);
@@ -253,7 +253,7 @@ public sealed class StorageTickSchedulerTests
             tenantContext,
             new Dictionary<Guid, FakeKanjStore> { [TenantA] = cardStoreA, [TenantB] = storeB },
             new Dictionary<Guid, ISettingsStore> { [TenantA] = new FakeSettingsStore(), [TenantB] = new FakeSettingsStore() },
-            new Dictionary<Guid, FakePipelineStore> { [TenantA] = new(), [TenantB] = new() });
+            new Dictionary<Guid, TestPipelineStore> { [TenantA] = new(), [TenantB] = new() });
 
         SseBroker broker = provider.GetRequiredService<SseBroker>();
         SseSubscription subscriptionA = broker.Subscribe(TenantA);
@@ -304,12 +304,12 @@ public sealed class StorageTickSchedulerTests
         TenantContext tenantContext,
         Dictionary<Guid, FakeKanjStore> storesByTenant,
         Dictionary<Guid, ISettingsStore> settingsByTenant,
-        Dictionary<Guid, FakePipelineStore>? pipelineStoresByTenant = null)
+        Dictionary<Guid, TestPipelineStore>? pipelineStoresByTenant = null)
     {
-        pipelineStoresByTenant ??= new Dictionary<Guid, FakePipelineStore>
+        pipelineStoresByTenant ??= new Dictionary<Guid, TestPipelineStore>
         {
-            [TenantA] = new FakePipelineStore(),
-            [TenantB] = new FakePipelineStore(),
+            [TenantA] = new TestPipelineStore(),
+            [TenantB] = new TestPipelineStore(),
         };
 
         var services = new ServiceCollection();
@@ -321,7 +321,7 @@ public sealed class StorageTickSchedulerTests
         // выбирает хранилище по тому же ITenantContext, который планировщик заполняет SetTenant.
         services.AddScoped<ICardStore>(provider => storesByTenant[TenantOf(provider)]);
         services.AddScoped<ISettingsStore>(provider => settingsByTenant[TenantOf(provider)]);
-        services.AddScoped<IPipelineStore>(provider => pipelineStoresByTenant[TenantOf(provider)]);
+        services.AddScoped<IPipelineStore>(provider => pipelineStoresByTenant[TenantOf(provider)].Store);
         services.AddSingleton<IMlClient>(new FakeMlClient());
         services.AddSingleton<IFileStorage>(new TestFileStorage().Storage);
         services.AddScoped<PipelineIngestService>();

src/core/tests/Deal.Tests.Unit/Support/TestPipelineStore.cs

@@ -0,0 +1,278 @@
+using Deal.Modules.Cards.Application.Sources;
+using Deal.Modules.Kanban.Application.Models;
+using Deal.Modules.Pipeline.Application.Abstractions;
+using Deal.Modules.Pipeline.Application.Models;
+using NSubstitute;
+
+namespace Deal.Tests.Unit.Support;
+
+/// <summary>
+/// Подставка <see cref="IPipelineStore"/> на списках: сервисы получают NSubstitute-подставку
+/// (<see cref="Store"/>), тесты сеют/проверяют очередь и отсеянных через <see cref="Queue"/>,
+/// <see cref="Rejected"/>, <see cref="SeedQueue"/>, <see cref="SeedRejected"/>.
+/// </summary>
+public sealed class TestPipelineStore
+{
+    private readonly List<QueueItemDto> _queue = [];
+    private readonly List<RejectedItemDto> _rejected = [];
+    private readonly Dictionary<string, string?> _dedup = new(StringComparer.Ordinal);
+
+    /// <summary>
+    /// Подставка порта пайплайна (создаётся в конструкторе).
+    /// </summary>
+    public IPipelineStore Store { get; }
+
+    /// <summary>
+    /// Строки очереди подставки
+    /// </summary>
+    public IReadOnlyList<QueueItemDto> Queue => _queue.ToList();
+
+    /// <summary>
+    /// Записи отсева подставки
+    /// </summary>
+    public IReadOnlyList<RejectedItemDto> Rejected => _rejected.ToList();
+
+    private readonly bool _claimRace;
+    private readonly bool _claimResult;
+
+    /// <summary>
+    /// Сколько раз вызван ClaimAsync
+    /// </summary>
+    public int ClaimCalls { get; private set; }
+
+    /// <summary>
+    /// Сколько раз вызван DeleteClaimAsync
+    /// </summary>
+    public int DeleteClaimCalls { get; private set; }
+
+    /// <summary>
+    /// Создаёт подставку с пустыми списками.
+    /// </summary>
+    /// <param name="claimRace">Сценарий claim-гонки: ExistsAsync=false, ClaimAsync возвращает claimResult.</param>
+    /// <param name="claimResult">Результат ClaimAsync в сценарии гонки (false — хэш уже заявлен другим проходом).</param>
+    public TestPipelineStore(bool claimRace = false, bool claimResult = false)
+    {
+        _claimRace = claimRace;
+        _claimResult = claimResult;
+        Store = Substitute.For<IPipelineStore>();
+        Store.ExistsDuplicateAsync(Arg.Any<SourceRef>(), Arg.Any<CancellationToken>())
+            .Returns(ci => ExistsDuplicate(ci.Arg<SourceRef>()));
+        Store.When(s => s.AddAsync(Arg.Any<QueueItemDto>(), Arg.Any<CancellationToken>()))
+            .Do(ci => _queue.Add(ci.Arg<QueueItemDto>()));
+        Store.ListAsync(Arg.Any<string?>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(ci => List(ci.ArgAt<string?>(0), ci.ArgAt<int>(1)));
+        Store.CountByStatusAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(ci => CountByStatus(ci.ArgAt<string>(0)));
+        Store.When(s => s.SetStatusAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(ci => SetStatus(ci.ArgAt<string>(0), ci.ArgAt<string>(1)));
+        Store.When(s => s.RemoveAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(ci => Remove(ci.ArgAt<string>(0)));
+        Store.When(s => s.UpsertAsync(Arg.Any<RejectRecord>(), Arg.Any<CancellationToken>()))
+            .Do(ci => Upsert(ci.Arg<RejectRecord>()));
+        Store.ListPageAsync(Arg.Any<int>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(ci => ListPage(ci.ArgAt<int>(0), ci.ArgAt<int>(1)));
+        Store.SearchAsync(Arg.Any<string>(), Arg.Any<int>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(ci => Search(ci.ArgAt<string>(0), ci.ArgAt<int>(2)));
+        Store.CountAsync(Arg.Any<CancellationToken>()).Returns(_rejected.Count);
+        Store.GetAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(ci => _rejected.FirstOrDefault(item => item.Id == ci.ArgAt<string>(0)));
+        Store.When(s => s.DeleteAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(ci => _rejected.RemoveAll(item => item.Id == ci.ArgAt<string>(0)));
+        Store.ClearAsync(Arg.Any<CancellationToken>())
+            .Returns(ci =>
+            {
+                int cleared = _rejected.Count;
+                _rejected.Clear();
+                return cleared;
+            });
+        Store.PurgeExpiredAsync(Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
+            .Returns(ci => PurgeExpired(ci.ArgAt<DateTimeOffset>(0)));
+        Store.When(s => s.MarkReturnedAsync(
+                Arg.Any<string>(), Arg.Any<string>(), Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>()))
+            .Do(ci => MarkReturned(ci.ArgAt<string>(0), ci.ArgAt<string>(1), ci.ArgAt<DateTimeOffset>(2)));
+        Store.When(s => s.LinkAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(ci => _dedup[ci.ArgAt<string>(0)] = ci.ArgAt<string>(1));
+        Store.When(s => s.DeleteByCardAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(ci => DeleteByCard(ci.ArgAt<string>(0)));
+        Store.ExistsAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(ci => !_claimRace && _dedup.ContainsKey(ci.ArgAt<string>(0)));
+        Store.When(s => s.ClaimAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(_ => ClaimCalls++);
+        Store.ClaimAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
+            .Returns(ci => _claimRace
+                ? _claimResult
+                : _dedup.TryAdd(ci.ArgAt<string>(0), null));
+        Store.When(s => s.DeleteClaimAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(_ => DeleteClaimCalls++);
+        Store.When(s => s.DeleteClaimAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))
+            .Do(ci =>
+            {
+                if (!_claimRace && _dedup.TryGetValue(ci.ArgAt<string>(0), out string? leadId) && leadId is null)
+                {
+                    _dedup.Remove(ci.ArgAt<string>(0));
+                }
+            });
+    }
+
+    /// <summary>
+    /// Кладёт строку очереди напрямую
+    /// </summary>
+    /// <param name="item">Строка как если бы была сохранена в БД.</param>
+    public void SeedQueue(QueueItemDto item)
+    {
+        _queue.Add(item);
+    }
+
+    /// <summary>
+    /// Кладёт запись отсева напрямую
+    /// </summary>
+    /// <param name="item">Запись как если бы была сохранена в БД.</param>
+    public void SeedRejected(RejectedItemDto item)
+    {
+        _rejected.Add(item);
+    }
+
+    /// <summary>
+    /// Id карточки, связанной с хэшем дедупа
+    /// </summary>
+    /// <param name="hash">SHA1-hex нормализованного текста.</param>
+    public string? DedupLeadId(string hash)
+    {
+        return _dedup.TryGetValue(hash, out string? leadId) ? leadId : null;
+    }
+
+    private bool ExistsDuplicate(SourceRef source)
+    {
+        string key = source.DedupeKey();
+        return _queue.Any(item => item.Source.DedupeKey() == key);
+    }
+
+    private IReadOnlyList<QueueItemDto> List(string? status, int limit)
+    {
+        return _queue
+            .Where(item => status is null || item.Status == status)
+            .OrderBy(item => item.QueuedAtMs)
+            .Take(limit)
+            .ToList();
+    }
+
+    private int CountByStatus(string status) => _queue.Count(item => item.Status == status);
+
+    private void SetStatus(string id, string status)
+    {
+        int index = _queue.FindIndex(item => item.Id == id);
+        if (index >= 0)
+        {
+            _queue[index] = _queue[index] with { Status = status };
+        }
+    }
+
+    private void Remove(string id) => _queue.RemoveAll(item => item.Id == id);
+
+    private void Upsert(RejectRecord record)
+    {
+        if (string.IsNullOrWhiteSpace(record.Text))
+        {
+            return;
+        }
+
+        string id = record.DeterministicId ?? PrefixId.New(PipelineIdPrefixes.Rejected);
+        long nowMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+
+        int index = _rejected.FindIndex(item => item.Id == id);
+        RejectedItemDto row = ToItem(record, id, nowMs);
+        if (index >= 0)
+        {
+            RejectedItemDto existing = _rejected[index];
+            row = row with
+            {
+                Returned = existing.Returned,
+                ReturnedAtMs = existing.ReturnedAtMs,
+                ReturnReason = existing.ReturnReason,
+            };
+            _rejected[index] = row;
+        }
+        else
+        {
+            _rejected.Add(row);
+        }
+    }
+
+    private IReadOnlyList<RejectedItemDto> ListPage(int offset, int limit)
+    {
+        return _rejected
+            .OrderByDescending(item => item.RejectedAtMs)
+            .Skip(offset)
+            .Take(limit)
+            .ToList();
+    }
+
+    private IReadOnlyList<RejectedItemDto> Search(string q, int limitLike)
+    {
+        string query = q.Trim();
+        return _rejected
+            .Where(item => LikeField(item.Text, query)
+                || LikeField(item.Reason, query)
+                || LikeField(item.Kw, query)
+                || LikeField(item.Source.DisplayName ?? string.Empty, query))
+            .OrderByDescending(item => item.RejectedAtMs)
+            .Take(limitLike)
+            .ToList();
+    }
+
+    private int PurgeExpired(DateTimeOffset olderThan)
+    {
+        long cutMs = olderThan.ToUnixTimeMilliseconds();
+        return _rejected.RemoveAll(item => item.RejectedAtMs < cutMs);
+    }
+
+    private void MarkReturned(string id, string reason, DateTimeOffset returnedAt)
+    {
+        int index = _rejected.FindIndex(item => item.Id == id);
+        if (index >= 0)
+        {
+            _rejected[index] = _rejected[index] with
+            {
+                Returned = true,
+                ReturnedAtMs = returnedAt.ToUnixTimeMilliseconds(),
+                ReturnReason = reason,
+            };
+        }
+    }
+
+    private void DeleteByCard(string cardId)
+    {
+        foreach (string hash in _dedup.Where(pair => pair.Value == cardId).Select(pair => pair.Key).ToList())
+        {
+            _dedup.Remove(hash);
+        }
+    }
+
+    // Подстрока q в lower(поле) — LIKE-половина адаптера (поле колонки SQL приводится к нижнему регистру).
+    // field: Поле записи отсева (text/reason/kw/ch_name).
+    // query: Поисковый запрос, нормализованный сервисом (trim+lowercase).
+    // Возвращает: True — поле содержит запрос без учёта регистра поля.
+    private static bool LikeField(string field, string query)
+    {
+        return field.ToLowerInvariant().Contains(query, StringComparison.Ordinal);
+    }
+
+    private static RejectedItemDto ToItem(
+        RejectRecord record,
+        string id,
+        long rejectedAtMs) => new()
+    {
+        Id = id,
+        Source = record.Source,
+        Content = record.Content,
+        Text = record.Text,
+        Stage = record.Stage,
+        StageLabel = PipelineRejectConstants.StageLabel(record.Stage),
+        Reason = record.Reason,
+        Kw = record.Kw,
+        DecidedBy = record.DecidedBy,
+        DecidedByLabel = PipelineRejectConstants.SourceLabel(record.DecidedBy),
+        MsgAtMs = record.MsgAtMs,
+        RejectedAtMs = rejectedAtMs,
+    };
+}