feat(cv)!: queue durable processing

CV upload now returns 202 with an owner-scoped operation instead of holding the request through parsing. Existing review approval remains required.

BREAKING CHANGE: profile-cv upload responses use the durable operation contract.

JobTrackerApi/Services/CvProcessingQueue.cs

@@ -1,97 +1,129 @@
-using System.Threading.Channels;
 using JobTrackerApi.Controllers;
 using JobTrackerApi.Data;
+using JobTrackerApi.Models;
 using Microsoft.EntityFrameworkCore;
 
 namespace JobTrackerApi.Services;
 
+public sealed record CvProcessingOutcome(
+    bool Succeeded,
+    string? FailureCategory = null,
+    string? FailureMessage = null,
+    bool Retryable = false,
+    string? Provider = null,
+    string? Model = null,
+    string? RouteReason = null);
+
 public interface ICvProcessingQueue
 {
-    ValueTask EnqueueAsync(int runId, CancellationToken cancellationToken = default);
-    IAsyncEnumerable<int> DequeueAllAsync(CancellationToken cancellationToken);
+    Task<AiOperationAdmissionResult?> EnqueueAsync(int runId, CancellationToken cancellationToken = default);
 }
 
-public sealed class CvProcessingQueue : ICvProcessingQueue
+/// <summary>
+/// Compatibility name for the CV producer bridge. Durable scheduling and execution are owned by
+/// the shared AI operation queue; this type does not keep an in-memory CV queue.
+/// </summary>
+public sealed class CvProcessingQueue(AiOperationAdmission admission) : ICvProcessingQueue
 {
-    private readonly Channel<int> _channel = Channel.CreateUnbounded<int>(new UnboundedChannelOptions
-    {
-        SingleReader = true,
-        SingleWriter = false,
-    });
+    public const string TaskType = "cv.process";
+    public const string SubjectType = "cv_extraction_run";
 
-    public ValueTask EnqueueAsync(int runId, CancellationToken cancellationToken = default)
-        => _channel.Writer.WriteAsync(runId, cancellationToken);
-
-    public IAsyncEnumerable<int> DequeueAllAsync(CancellationToken cancellationToken)
-        => _channel.Reader.ReadAllAsync(cancellationToken);
+    public async Task<AiOperationAdmissionResult?> EnqueueAsync(int runId, CancellationToken cancellationToken = default)
+        => await admission.EnqueueAsync(
+            TaskType,
+            $"run:{runId}",
+            SubjectType,
+            runId.ToString(System.Globalization.CultureInfo.InvariantCulture),
+            AiOperationPriorities.UserVisible,
+            cancellationToken);
 }
 
 public sealed class NoOpCvProcessingQueue : ICvProcessingQueue
 {
     public static readonly NoOpCvProcessingQueue Instance = new();
-    public ValueTask EnqueueAsync(int runId, CancellationToken cancellationToken = default) => ValueTask.CompletedTask;
-    public async IAsyncEnumerable<int> DequeueAllAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
-    {
-        await Task.CompletedTask;
-        yield break;
-    }
+    public Task<AiOperationAdmissionResult?> EnqueueAsync(int runId, CancellationToken cancellationToken = default)
+        => Task.FromResult<AiOperationAdmissionResult?>(null);
 }
 
-public sealed class CvProcessingHostedService : BackgroundService
+public sealed class CvProcessingOperationHandler : IAiOperationHandler
 {
-    private readonly IServiceScopeFactory _scopeFactory;
-    private readonly ICvProcessingQueue _queue;
-    private readonly ILogger<CvProcessingHostedService> _logger;
+    public string TaskType => CvProcessingQueue.TaskType;
 
-    public CvProcessingHostedService(IServiceScopeFactory scopeFactory, ICvProcessingQueue queue, ILogger<CvProcessingHostedService> logger)
+    public async Task<AiOperationExecutionResult> ExecuteAsync(
+        AiOperationExecutionContext context,
+        IServiceProvider services,
+        CancellationToken cancellationToken)
     {
-        _scopeFactory = scopeFactory;
-        _queue = queue;
-        _logger = logger;
-    }
-
-    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
-    {
-        await ProcessInterruptedRunsAsync(stoppingToken);
-
-        await foreach (var runId in _queue.DequeueAllAsync(stoppingToken))
+        if (!string.Equals(context.Lease.SubjectType, CvProcessingQueue.SubjectType, StringComparison.Ordinal) ||
+            !int.TryParse(context.Lease.SubjectId, out var runId) || runId <= 0)
         {
-            try
-            {
-                await ProcessRunAsync(runId, stoppingToken);
-            }
-            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
-            {
-                break;
-            }
-            catch (Exception ex)
-            {
-                _logger.LogError(ex, "Unhandled CV processing worker failure for run {RunId}", runId);
-            }
+            throw new AiOperationFailure("invalid_cv_run", "The CV processing operation has an invalid run reference.", retryable: false);
         }
-    }
 
-    private async Task ProcessInterruptedRunsAsync(CancellationToken cancellationToken)
-    {
-        await using var scope = _scopeFactory.CreateAsyncScope();
-        var db = scope.ServiceProvider.GetRequiredService<JobTrackerContext>();
-        var interruptedRuns = await db.CvExtractionRuns.IgnoreQueryFilters()
-            .Where(x => x.Status == "queued" || x.Status == "running")
-            .Select(x => new { x.Id, x.StartedAtUtc })
-            .ToListAsync(cancellationToken);
-
-        // ponytail: single-instance recovery; use row leasing if multiple workers are ever deployed.
-        // SQLite cannot ORDER BY DateTimeOffset, so the small interrupted-work set is ordered locally.
-        foreach (var run in interruptedRuns.OrderBy(x => x.StartedAtUtc))
+        CvProcessingOutcome? outcome;
+        try
         {
-            await ProcessRunAsync(run.Id, cancellationToken);
+            outcome = await services.GetRequiredService<ProfileCvController>()
+                .ProcessQueuedRunAsync(runId, cancellationToken);
         }
+        catch (OperationCanceledException)
+        {
+            var operation = await services.GetRequiredService<UserOperationStore>()
+                .GetAsync(context.Lease.OperationId, CancellationToken.None);
+            await SetRunStatusAsync(
+                services,
+                runId,
+                operation?.CancellationRequestedAtUtc is null ? "queued" : "cancelled",
+                operation?.CancellationRequestedAtUtc is null ? "CV processing timed out and may be retried." : "CV processing was cancelled.");
+            throw;
+        }
+        if (outcome is null)
+            throw new AiOperationFailure("cv_run_not_found", "The CV processing run is no longer available.", retryable: false);
+        if (!outcome.Succeeded)
+        {
+            if (outcome.Retryable)
+            {
+                var operation = await services.GetRequiredService<UserOperationStore>()
+                    .GetAsync(context.Lease.OperationId, cancellationToken);
+                var canRetry = operation is not null && operation.AttemptCount < operation.MaxAttempts &&
+                    (operation.DeadlineAtUtc is null || operation.DeadlineAtUtc > DateTime.UtcNow);
+                if (!canRetry)
+                    await SetRunStatusAsync(services, runId, "failed", outcome.FailureMessage ?? "CV processing failed.");
+            }
+
+            if (outcome.Provider is not null || outcome.Model is not null || outcome.RouteReason is not null)
+            {
+                throw new AiGenerationException(
+                    outcome.FailureCategory ?? "cv_processing_failed",
+                    outcome.FailureMessage ?? "CV processing failed.",
+                    outcome.Retryable,
+                    outcome.Provider,
+                    outcome.Model,
+                    outcome.RouteReason);
+            }
+
+            throw new AiOperationFailure(
+                outcome.FailureCategory ?? "cv_processing_failed",
+                outcome.FailureMessage ?? "CV processing failed.",
+                outcome.Retryable);
+        }
+
+        return new AiOperationExecutionResult(
+            $"/api/profile-cv/runs/{runId}/diff",
+            outcome.Provider,
+            outcome.Model,
+            outcome.RouteReason ?? "cv_pipeline");
     }
 
-    private async Task ProcessRunAsync(int runId, CancellationToken cancellationToken)
+    private static Task<int> SetRunStatusAsync(IServiceProvider services, int runId, string status, string message)
     {
-        await using var scope = _scopeFactory.CreateAsyncScope();
-        var controller = scope.ServiceProvider.GetRequiredService<ProfileCvController>();
-        await controller.ProcessQueuedRunAsync(runId, cancellationToken);
+        var completedAtUtc = status == "failed" || status == "cancelled" ? DateTimeOffset.UtcNow : (DateTimeOffset?)null;
+        return services.GetRequiredService<JobTrackerContext>().CvExtractionRuns
+            .Where(run => run.Id == runId)
+            .ExecuteUpdateAsync(setters => setters
+                .SetProperty(run => run.Status, status)
+                .SetProperty(run => run.ErrorMessage, message)
+                .SetProperty(run => run.CompletedAtUtc, completedAtUtc),
+                CancellationToken.None);
     }
 }