Add shutdown outcome telemetry to LLM queue worker

backend/src/Taskdeck.Api/Workers/LlmQueueToProposalWorker.cs

@@ -12,6 +12,11 @@ public class LlmQueueToProposalWorker : BackgroundService
 {
     private const string QueueNameLlm = "llm";
     private const string QueueNameCaptureTriage = "capture-triage";
+    // The claim was released back to Pending before the host stopped; nothing was abandoned.
+    private const string OutcomeReleasedOnShutdown = "released_on_shutdown";
+    // The release write failed, so the row stays Processing for the recovery sweep: the unhealthy path.
+    private const string OutcomeShutdownReleaseFailed = "shutdown_release_failed";
+    private const string OutcomeCancelledRequeued = "cancelled_requeued";
 
     // Floor on the stuck-recovery lease so an aggressively-low ProcessingLeaseSeconds can never sweep a
     // request that was claimed seconds ago and is still legitimately in flight. Mirrors the floor the
@@ -367,7 +372,11 @@ private async Task ProcessSingleItemAsync(Guid itemId, DateTimeOffset expectedUp
         // marked Failed and requeued purely because the host was stopping.
         catch (OperationCanceledException) when (ct.IsCancellationRequested)
         {
-            await ReleaseClaimOnShutdownAsync(item);
+            if (await ReleaseClaimOnShutdownAsync(item))
+            {
+                RecordWorkerOutcome(OutcomeReleasedOnShutdown);
+            }
+
             throw;
         }
         catch (Exception ex)
@@ -403,14 +412,14 @@ private async Task ProcessSingleItemAsync(Guid itemId, DateTimeOffset expectedUp
     /// Processing already is their queued state and returning one to Pending would drop it back to
     /// "untriaged in the inbox" — the wrong state, and one no worker drains.
     /// </remarks>
-    private async Task ReleaseClaimOnShutdownAsync(LlmRequest item)
+    private async Task<bool> ReleaseClaimOnShutdownAsync(LlmRequest item)
     {
         // The failure path may already have moved the row on (Failed, or Pending/Processing once
         // HandleFailureWithRetryAsync completed its own interrupted transition); only a still-claimed row
         // is ours to release.
         if (item.Status != RequestStatus.Processing)
         {
-            return;
+            return false;
         }
 
         try
@@ -423,7 +432,7 @@ private async Task ReleaseClaimOnShutdownAsync(LlmRequest item)
             var shutdownItem = await shutdownUnitOfWork.LlmQueue.GetByIdAsync(item.Id, CancellationToken.None);
             if (shutdownItem == null || shutdownItem.Status != RequestStatus.Processing)
             {
-                return;
+                return false;
             }
 
             shutdownItem.ReleaseClaim();
@@ -436,6 +445,7 @@ private async Task ReleaseClaimOnShutdownAsync(LlmRequest item)
             _logger.LogInformation(
                 "Queue item {ItemId} released back to Pending during shutdown; no retry charged",
                 item.Id);
+            return true;
         }
         catch (Exception ex)
         {
@@ -446,6 +456,10 @@ private async Task ReleaseClaimOnShutdownAsync(LlmRequest item)
                 "Failed to release queue item {ItemId} during shutdown; it stays Processing for the recovery sweep. {ExceptionSummary}",
                 item.Id,
                 SensitiveDataRedactor.SummarizeException(ex));
+            // Only a genuinely failed release write is the unhealthy outcome; a row that had already
+            // moved on (the early returns above) has nothing to release and emits nothing here.
+            RecordWorkerOutcome(OutcomeShutdownReleaseFailed);
+            return false;
         }
     }
 
@@ -741,11 +755,14 @@ private async Task<bool> HandleFailureWithRetryAsync(
         {
             try
             {
-                await CompleteRetryTransitionOnShutdownAsync(item.Id, retryAsProcessing);
-                _logger.LogInformation(
-                    "Queue item {ItemId} requeued for retry attempt {RetryCount} during shutdown",
-                    item.Id,
-                    item.RetryCount + 1);
+                if (await CompleteRetryTransitionOnShutdownAsync(item.Id, retryAsProcessing))
+                {
+                    RecordWorkerOutcome(OutcomeCancelledRequeued);
+                    _logger.LogInformation(
+                        "Queue item {ItemId} requeued for retry attempt {RetryCount} during shutdown",
+                        item.Id,
+                        item.RetryCount + 1);
+                }
             }
             catch (Exception ex)
             {
@@ -800,18 +817,19 @@ private async Task<bool> HandleFailureWithRetryInFreshScopeAsync(
             retryAsProcessing);
     }
 
-    private async Task CompleteRetryTransitionOnShutdownAsync(Guid itemId, bool retryAsProcessing)
+    private async Task<bool> CompleteRetryTransitionOnShutdownAsync(Guid itemId, bool retryAsProcessing)
     {
         using var scope = _scopeFactory.CreateScope();
         var shutdownUnitOfWork = scope.ServiceProvider.GetRequiredService<IUnitOfWork>();
         var shutdownItem = await shutdownUnitOfWork.LlmQueue.GetByIdAsync(itemId, CancellationToken.None);
         if (shutdownItem == null)
         {
-            return;
+            return false;
         }
 
         ApplyRetryTransition(shutdownItem, retryAsProcessing);
         await shutdownUnitOfWork.SaveChangesAsync(CancellationToken.None);
+        return true;
     }
 
     /// <summary>
@@ -868,5 +886,13 @@ private static void RecordWorkerProcessingMetrics(double durationMs, string outc
             new KeyValuePair<string, object?>(TaskdeckTelemetryTags.Outcome, outcome));
     }
 
+    private static void RecordWorkerOutcome(string outcome)
+    {
+        TaskdeckTelemetry.WorkerItemsProcessed.Add(
+            1,
+            new KeyValuePair<string, object?>(TaskdeckTelemetryTags.WorkerName, nameof(LlmQueueToProposalWorker)),
+            new KeyValuePair<string, object?>(TaskdeckTelemetryTags.Outcome, outcome));
+    }
+
     private readonly record struct WorkerBatchItem(Guid ItemId, bool IsCaptureTriage, DateTimeOffset? ExpectedUpdatedAt = null);
 }

backend/tests/Taskdeck.Api.Tests/LlmQueueToProposalWorkerTests.cs

@@ -1,9 +1,11 @@
+using System.Diagnostics.Metrics;
 using System.Reflection;
 using FluentAssertions;
 using Microsoft.Extensions.DependencyInjection;
 using Moq;
 using Microsoft.Extensions.Logging.Abstractions;
 using Taskdeck.Api.Workers;
+using Taskdeck.Api.Telemetry;
 using Taskdeck.Application.DTOs;
 using Taskdeck.Application.Interfaces;
 using Taskdeck.Application.Services;
@@ -15,6 +17,15 @@
 
 namespace Taskdeck.Api.Tests;
 
+// The outcome listener below observes the process-global WorkerItemsProcessed counter, which
+// the resilience and transcript-worker test classes also drive; this collection runs alone.
+[CollectionDefinition(Name, DisableParallelization = true)]
+public sealed class WorkerTelemetryCollection
+{
+    public const string Name = "Worker telemetry global state";
+}
+
+[Collection(WorkerTelemetryCollection.Name)]
 public class LlmQueueToProposalWorkerTests
 {
     #region Helper factories
@@ -121,6 +132,47 @@ private static async Task InvokeProcessBatchAsync(
         await (Task)task!;
     }
 
+    private static MeterListener ListenForWorkerOutcomes(List<string> outcomes)
+    {
+        var listener = new MeterListener
+        {
+            InstrumentPublished = (instrument, meterListener) =>
+            {
+                if (instrument == TaskdeckTelemetry.WorkerItemsProcessed)
+                {
+                    meterListener.EnableMeasurementEvents(instrument);
+                }
+            }
+        };
+        listener.SetMeasurementEventCallback<long>((instrument, _, tags, _) =>
+        {
+            string? outcome = null;
+            var thisWorker = false;
+            foreach (var tag in tags)
+            {
+                if (tag.Key == TaskdeckTelemetryTags.Outcome && tag.Value is string value)
+                {
+                    outcome = value;
+                }
+                else if (tag.Key == TaskdeckTelemetryTags.WorkerName && tag.Value is string name)
+                {
+                    thisWorker = name == nameof(LlmQueueToProposalWorker);
+                }
+            }
+
+            if (thisWorker && outcome is not null)
+            {
+                lock (outcomes)
+                {
+                    outcomes.Add(outcome);
+                }
+            }
+        });
+        listener.Start();
+        listener.EnableMeasurementEvents(TaskdeckTelemetry.WorkerItemsProcessed);
+        return listener;
+    }
+
     #endregion
 
     #region Happy path: pending item processed to completion
@@ -246,6 +298,8 @@ public async Task ProcessBatch_ProposalLaneCallerCancellation_PropagatesAndRelea
     {
         var item = CreatePendingItem();
         var queueRepo = new FakeLlmQueueRepository([item]);
+        var outcomes = new List<string>();
+        using var listener = ListenForWorkerOutcomes(outcomes);
         using var cts = new CancellationTokenSource();
         var planner = new FakeAutomationPlannerService
         {
@@ -270,6 +324,7 @@ await Assert.ThrowsAnyAsync<OperationCanceledException>(
         // the processing lease expires.
         item.Status.Should().Be(RequestStatus.Pending);
         item.RetryCount.Should().Be(0, "a shutdown mid-item must not consume the retry budget");
+        outcomes.Should().Equal("released_on_shutdown");
     }
 
     [Fact]
@@ -303,6 +358,35 @@ await Assert.ThrowsAnyAsync<OperationCanceledException>(
             "the release must be persisted with CancellationToken.None, not the cancelled caller token");
     }
 
+    [Fact]
+    public async Task ProcessBatch_ProposalLaneCallerCancellation_WhenReleaseFails_DoesNotEmitSuccessfulOutcome()
+    {
+        var item = CreatePendingItem();
+        var queueRepo = new FakeLlmQueueRepository([item]);
+        var unitOfWork = new FakeUnitOfWork(queueRepo)
+        {
+            OnSaveChanges = _ => throw new InvalidOperationException("release write failed")
+        };
+        var outcomes = new List<string>();
+        using var listener = ListenForWorkerOutcomes(outcomes);
+        using var cts = new CancellationTokenSource();
+        var planner = new FakeAutomationPlannerService
+        {
+            ResultFactory = _ =>
+            {
+                cts.Cancel();
+                throw new OperationCanceledException(cts.Token);
+            }
+        };
+        using var sp = BuildServiceProvider(queueRepo, planner, unitOfWork: unitOfWork);
+        var worker = CreateWorker(sp.GetRequiredService<IServiceScopeFactory>(), DefaultSettings(retryBackoff: [0]));
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            () => InvokeProcessBatchAsync(worker, cts.Token));
+
+        outcomes.Should().NotContain("released_on_shutdown");
+    }
+
     [Fact]
     public async Task ProcessBatch_ProposalLaneShutdownRelease_UsesFreshScopeWithoutFlushingPlannerState()
     {
@@ -605,6 +689,8 @@ public async Task ProcessBatch_CancellationDuringRetryBackoff_PersistsTheRequeue
         var item = CreatePendingItem();
         var queueRepo = new FakeLlmQueueRepository([item]);
         var unitOfWork = new FakeUnitOfWork(queueRepo);
+        var outcomes = new List<string>();
+        using var listener = ListenForWorkerOutcomes(outcomes);
         using var cts = new CancellationTokenSource();
         // Cancel from inside the FIRST write (the MarkAsFailed commit) so cancellation lands squarely in
         // the backoff wait that follows -- deterministic, unlike a timer race, and it keeps the token
@@ -632,6 +718,41 @@ await Assert.ThrowsAnyAsync<OperationCanceledException>(
         unitOfWork.SaveChangesTokens.Should().HaveCount(2, "the MarkAsFailed commit and the requeue commit");
         unitOfWork.SaveChangesTokens[1].IsCancellationRequested.Should().BeFalse(
             "the requeue must be persisted with CancellationToken.None; the caller token is already cancelled");
+        outcomes.Should().Equal("cancelled_requeued");
+    }
+
+    [Fact]
+    public async Task ProcessBatch_CancellationDuringRetryBackoff_WhenRequeueFails_DoesNotEmitSuccessfulOutcome()
+    {
+        var item = CreatePendingItem();
+        var queueRepo = new FakeLlmQueueRepository([item]);
+        var unitOfWork = new FakeUnitOfWork(queueRepo);
+        var outcomes = new List<string>();
+        using var listener = ListenForWorkerOutcomes(outcomes);
+        using var cts = new CancellationTokenSource();
+        unitOfWork.OnSaveChanges = saveNumber =>
+        {
+            if (saveNumber == 1)
+            {
+                cts.Cancel();
+            }
+            else if (saveNumber == 2)
+            {
+                throw new InvalidOperationException("requeue write failed");
+            }
+        };
+        var planner = new FakeAutomationPlannerService
+        {
+            ResultFactory = _ => Result.Failure<ProposalDto>(ErrorCodes.UnexpectedError, "transient failure")
+        };
+        using var sp = BuildServiceProvider(queueRepo, planner, unitOfWork: unitOfWork);
+        var worker = CreateWorker(sp.GetRequiredService<IServiceScopeFactory>(),
+            DefaultSettings(maxRetries: 3, retryBackoff: [5]));
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            () => InvokeProcessBatchAsync(worker, cts.Token));
+
+        outcomes.Should().NotContain("cancelled_requeued");
     }
 
     [Fact]