feat(cv-import): add source-grounded review and explicit corrections

2026-10-26 15:56:54 +01:00
parent 9e1726c61d
commit e941b3ccb7
24 changed files with 943 additions and 48 deletions
@@ -42,11 +42,17 @@ public sealed partial class ProfileCvController
        var candidate = CvDocumentInterpreter.Parse(blocks);
        await Stage("cv_validating_details");
        var profile = CvDocumentReview.Ground(candidate, document);
+        foreach (var field in profile.Metadata.Fields.Values)
+        {
+            field.SourceArtifactId = artifact.Id;
+            field.SourceExtractionRunId = run.Id;
+            field.SourceFileSha256 = artifact.Sha256;
+        }
        await Stage("cv_preparing_review");
        await Stage("cv_review_ready");
        cancellationToken.ThrowIfCancellationRequested();
        var documentJson = JsonSerializer.Serialize(document);
-        var profileJson = StructuredCvProfileJson.Serialize(profile);
+        var profileJson = StructuredCvProfileJson.SerializePersisted(profile);
        if (context is null)
        {
            run.DocumentJson = documentJson;
@@ -54,8 +60,9 @@ public sealed partial class ProfileCvController
            return;
        }
        var currentLease = context.Lease;
+        var completedAt = DateTimeOffset.UtcNow;
        var published = await _db.CvExtractionRuns.Where(r => r.Id == run.Id && r.OwnerUserId == run.OwnerUserId && r.Status == "running" && _db.UserOperations.Any(o => o.Id == currentLease.OperationId && o.LeaseToken == currentLease.LeaseToken && o.Status == OperationStatuses.Running && o.CancellationRequestedAtUtc == null && o.LeaseExpiresAtUtc > DateTime.UtcNow))
-            .ExecuteUpdateAsync(s => s.SetProperty(r => r.DocumentJson, documentJson).SetProperty(r => r.RawExtractedText, document.Text).SetProperty(r => r.NormalizedText, document.Text).SetProperty(r => r.StructuredProfileJson, profileJson).SetProperty(r => r.Status, "pending_review").SetProperty(r => r.CompletedAtUtc, DateTimeOffset.UtcNow), cancellationToken);
+            .ExecuteUpdateAsync(s => s.SetProperty(r => r.DocumentJson, documentJson).SetProperty(r => r.RawExtractedText, document.Text).SetProperty(r => r.NormalizedText, document.Text).SetProperty(r => r.StructuredProfileJson, profileJson).SetProperty(r => r.Status, "pending_review").SetProperty(r => r.CompletedAtUtc, completedAt), cancellationToken);
        if (published != 1) throw new OperationCanceledException("The document processing lease was lost before publication.", cancellationToken);
        await _db.Entry(run).ReloadAsync(cancellationToken);
        await PruneExtractionRunsAsync(run.OwnerUserId, cancellationToken);
@@ -126,7 +133,8 @@ public sealed partial class ProfileCvController
                    using var reader = new StreamReader(entry.Open());
                    var relationships = await reader.ReadToEndAsync(cancellationToken);
                    // Hyperlinks remain inert text; linked templates/OLE/media are refused.
-                    var xml = System.Xml.Linq.XDocument.Parse(relationships);
+                    using var xmlReader = System.Xml.XmlReader.Create(new StringReader(relationships), new System.Xml.XmlReaderSettings { DtdProcessing = System.Xml.DtdProcessing.Prohibit, XmlResolver = null, MaxCharactersInDocument = 2 * 1024 * 1024 });
+                    var xml = System.Xml.Linq.XDocument.Load(xmlReader);
                    if (xml.Descendants().Any(e => string.Equals((string?)e.Attribute("TargetMode"), "External", StringComparison.OrdinalIgnoreCase) && !((string?)e.Attribute("Type") ?? "").EndsWith("/hyperlink", StringComparison.Ordinal))) throw new InvalidDataException();
                }
            }
@@ -151,7 +159,7 @@ public sealed partial class ProfileCvController
        if (run.ParserVersion != DocumentParserVersion || string.IsNullOrWhiteSpace(run.StructuredProfileJson)) return Conflict("This import is not ready for review.");
        var snapshot = await _careerProfileService.LoadSnapshotAsync(user.Id, HttpContext.RequestAborted);
        var document = JsonSerializer.Deserialize<CvImportDocument>(run.DocumentJson ?? "null");
-        return Ok(new CvImportReview(run.Id, run.Status, snapshot.Revision, run.Artifact?.OriginalFileName, $"/api/profile-cv/runs/{run.Id}/source", run.RawExtractedText, document?.Warnings ?? Array.Empty<string>(), CvDocumentReview.Build(snapshot.Profile, StructuredCvProfileJson.Deserialize(run.StructuredProfileJson))));
+        return Ok(new CvImportReview(run.Id, run.Status, snapshot.Revision, run.Artifact?.OriginalFileName, $"/api/profile-cv/runs/{run.Id}/source", run.RawExtractedText, document?.Warnings ?? Array.Empty<string>(), CvDocumentReview.Build(snapshot.Profile, StructuredCvProfileJson.DeserializePersisted(run.StructuredProfileJson))));
    }

    [HttpGet("runs/{id:int}/source")]
@@ -189,11 +197,11 @@ public sealed partial class ProfileCvController
                if (run.Status != "pending_review" || string.IsNullOrWhiteSpace(run.StructuredProfileJson)) return Conflict("This import is not awaiting review.");
                var snapshot = await _careerProfileService.LoadSnapshotAsync(user.Id, HttpContext.RequestAborted);
                if (snapshot.Revision != request.ExpectedVersion) throw new DbUpdateConcurrencyException();
-                var merged = CvDocumentReview.Apply(snapshot.Profile, StructuredCvProfileJson.Deserialize(run.StructuredProfileJson), request.Decisions);
+                var merged = CvDocumentReview.Apply(snapshot.Profile, StructuredCvProfileJson.DeserializePersisted(run.StructuredProfileJson), request.Decisions);
                if (!request.Decisions.Any(d => d.Action != "keep")) return BadRequest("No imported fields were selected. Discard this import to keep your current profile.");
                var validation = CareerProfileValidator.Validate(merged);
                if (validation is not null || StructuredCvProfileJson.SerializePersisted(merged).Length > 500000) return BadRequest(validation ?? "The reviewed profile exceeds its size limit.");
-                merged.Metadata.ProfileVersion = (user.CurrentCvProfileVersion ?? 0) + 1;
+                merged.Metadata.ProfileVersion = snapshot.Revision + 1;
                merged.Metadata.AppliedExtractionRunId = run.Id;
                merged.Metadata.UpdatedAtUtc = DateTimeOffset.UtcNow;
                await _careerProfileService.SaveVersionAsync(user.Id, merged, "document_import:reviewed", request.ExpectedVersion.Value, HttpContext.RequestAborted);
@@ -411,9 +411,19 @@ public sealed partial class ProfileCvController : ControllerBase
            ? await _db.CvExtractionRuns.IgnoreQueryFilters().FirstOrDefaultAsync(x => x.Id == runId, cancellationToken)
            : await _db.CvExtractionRuns.FirstOrDefaultAsync(x => x.Id == runId && x.OwnerUserId == ownerUserId, cancellationToken);
        if (run is null) return null;
+        var guardedDocument = run.ParserVersion == DocumentParserVersion && operationContext is not null;
+        if (guardedDocument)
+        {
+            var lease = operationContext!.Lease;
+            var claimed = await _db.CvExtractionRuns.Where(r => r.Id == run.Id && r.OwnerUserId == run.OwnerUserId && (r.Status == "queued" || r.Status == "running") && _db.UserOperations.Any(o => o.Id == lease.OperationId && o.LeaseToken == lease.LeaseToken && o.Status == OperationStatuses.Running && o.CancellationRequestedAtUtc == null && o.LeaseExpiresAtUtc > DateTime.UtcNow))
+                .ExecuteUpdateAsync(s => s.SetProperty(r => r.Status, "running").SetProperty(r => r.ErrorMessage, (string?)null), cancellationToken);
+            if (claimed != 1) { _db.Entry(run).State = EntityState.Detached; throw new OperationCanceledException("The import worker lease is no longer active."); }
+            await _db.Entry(run).ReloadAsync(cancellationToken);
+        }
        var user = await _users.FindByIdAsync(run.OwnerUserId);
        if (user is null)
        {
+            if (guardedDocument) throw new AiOperationFailure("cv_user_not_found", "The import account is no longer available.", false);
            run.Status = "failed";
            run.ErrorMessage = "CV processing user was not found.";
            run.CompletedAtUtc = DateTimeOffset.UtcNow;
@@ -423,6 +433,7 @@ public sealed partial class ProfileCvController : ControllerBase

        if (!user.AiEnabled || !AccountPlans.ForRoles(await _users.GetRolesAsync(user)).Ai)
        {
+            if (guardedDocument) throw new AiOperationFailure("entitlement_changed", "CV import access changed before processing began.", false);
            run.Status = "failed";
            run.ErrorMessage = user.AiEnabled
                ? "This AI feature requires Pro."
@@ -432,9 +443,12 @@ public sealed partial class ProfileCvController : ControllerBase
            return new CvProcessingOutcome(false, "entitlement_changed", run.ErrorMessage);
        }

-        run.Status = "running";
-        run.ErrorMessage = null;
-        await _db.SaveChangesAsync(cancellationToken);
+        if (!guardedDocument)
+        {
+            run.Status = "running";
+            run.ErrorMessage = null;
+            await _db.SaveChangesAsync(cancellationToken);
+        }

        try
        {
@@ -535,20 +549,36 @@ public sealed partial class ProfileCvController : ControllerBase
            }
            var generationFailure = ex as AiGenerationException;
            var retryable = generationFailure?.Retryable == true;
-            var failureMessage = generationFailure?.Message
+            var failureMessage = run.ParserVersion == DocumentParserVersion
+                ? "The document could not be processed safely. Retry or upload a supported PDF or DOCX CV."
+                : generationFailure?.Message
                ?? (ex is InvalidOperationException
                    ? ex.Message
                    : "CV processing failed unexpectedly. Please try again.");
-            run.Status = retryable ? "queued" : "failed";
-            run.ErrorMessage = failureMessage;
-            run.CompletedAtUtc = retryable ? null : DateTimeOffset.UtcNow;
-            await _db.SaveChangesAsync(cancellationToken);
+            if (guardedDocument)
+            {
+                var lease = operationContext!.Lease;
+                var failedAt = retryable ? (DateTimeOffset?)null : DateTimeOffset.UtcNow;
+                var failedStatus = retryable ? "queued" : "failed";
+                var updated = await _db.CvExtractionRuns.Where(r => r.Id == run.Id && (r.Status == "queued" || r.Status == "running") && _db.UserOperations.Any(o => o.Id == lease.OperationId && o.LeaseToken == lease.LeaseToken && o.Status == OperationStatuses.Running && o.CancellationRequestedAtUtc == null && o.LeaseExpiresAtUtc > DateTime.UtcNow))
+                    .ExecuteUpdateAsync(s => s.SetProperty(r => r.Status, failedStatus).SetProperty(r => r.ErrorMessage, failureMessage).SetProperty(r => r.CompletedAtUtc, failedAt), cancellationToken);
+                _db.Entry(run).State = EntityState.Detached;
+                if (updated != 1) throw new OperationCanceledException("The import worker lease is no longer active.");
+            }
+            else
+            {
+                run.Status = retryable ? "queued" : "failed";
+                run.ErrorMessage = failureMessage;
+                run.CompletedAtUtc = retryable ? null : DateTimeOffset.UtcNow;
+                await _db.SaveChangesAsync(cancellationToken);
+            }
            if (!retryable)
            {
                await PruneExtractionRunsAsync(user.Id, cancellationToken);
                await SendRunCompletionEmailAsync(user, run, false, cancellationToken);
            }
-            _logger.LogWarning(ex, "CV processing run {RunId} failed for user {UserId}", run.Id, user.Id);
+            if (run.ParserVersion == DocumentParserVersion) _logger.LogWarning("CV document processing run {RunId} failed ({Category}).", run.Id, generationFailure?.Category ?? "cv_processing_failed");
+            else _logger.LogWarning(ex, "CV processing run {RunId} failed for user {UserId}", run.Id, user.Id);
            return new CvProcessingOutcome(
                false,
                generationFailure?.Category ?? "cv_processing_failed",
@@ -564,7 +594,7 @@ public sealed partial class ProfileCvController : ControllerBase
    {
        run.RawExtractedText = rawText;
        run.NormalizedText = normalizedText;
-        run.StructuredProfileJson = StructuredCvProfileJson.Serialize(structuredCv);
+        run.StructuredProfileJson = run.ParserVersion == DocumentParserVersion ? StructuredCvProfileJson.SerializePersisted(structuredCv) : StructuredCvProfileJson.Serialize(structuredCv);
        run.Status = "pending_review";
        run.CompletedAtUtc = DateTimeOffset.UtcNow;
        await _db.SaveChangesAsync(cancellationToken);
@@ -219,7 +219,8 @@ public sealed partial class ProfileCvController : ControllerBase
            .Where(x => x.OwnerUserId == user.Id);
        var isSqlite = _db.Database.IsSqlite();
        if (!isSqlite)
-            runsQuery = runsQuery.OrderByDescending(x => x.StartedAtUtc).Take(10);
+            runsQuery = runsQuery.OrderByDescending(x => x.StartedAtUtc).Take(10)
+                .Union(_db.CvExtractionRuns.AsNoTracking().Where(x => x.OwnerUserId == user.Id && x.Status == "pending_review"));
        var runs = await runsQuery.Select(x => new CvExtractionRunListItem(
                x.Id,
                x.Trigger,
@@ -233,8 +234,8 @@ public sealed partial class ProfileCvController : ControllerBase
                x.LlmPromptVersion,
                x.ErrorMessage,
                null)).ToListAsync(HttpContext.RequestAborted);
-        if (isSqlite)
-            runs = runs.OrderByDescending(x => x.StartedAtUtc).Take(10).ToList();
+        runs = runs.OrderByDescending(x => x.StartedAtUtc).Take(10).Concat(runs.Where(x => x.Status == "pending_review"))
+            .DistinctBy(x => x.Id).OrderByDescending(x => x.StartedAtUtc).ToList();

        var runIds = runs.Select(run => run.Id.ToString(System.Globalization.CultureInfo.InvariantCulture)).ToList();
        var operations = await _db.UserOperations.AsNoTracking()
@@ -36,6 +36,9 @@ public sealed class StructuredCvFieldMetadata
    public string? SourceSnippet { get; set; }
    public int? SourcePage { get; set; }
    public string? SourceBlockId { get; set; }
+    public int? SourceArtifactId { get; set; }
+    public int? SourceExtractionRunId { get; set; }
+    public string? SourceFileSha256 { get; set; }
    public string? ReviewState { get; set; }
    public DateTimeOffset? LastUpdatedAtUtc { get; set; }
 }
@@ -22,7 +22,7 @@ public static class CvDocumentReview

    public static StructuredCvProfile Ground(StructuredCvProfile candidate, CvImportDocument document)
    {
-        var node = Node(candidate);
+        var node = Node(StructuredCvProfileJson.NormalizeForPersistence(candidate));
        node["sections"] = new JsonArray();
        var metadata = new Dictionary<string, StructuredCvFieldMetadata>();
        void Walk(JsonNode? value, string path)
@@ -162,6 +162,7 @@ public static class CvDocumentReview
        var selected = decisions.Where(d => d.Action != "keep").ToDictionary(d => d.Id, StringComparer.Ordinal);
        var result = Node(current);
        var linkTargets = new Dictionary<int, int>();
+        var newTargets = new Dictionary<string, int>(StringComparer.Ordinal);
        foreach (var map in maps)
        {
            var choices = map.Review.Fields.Where(f => selected.ContainsKey(f.Id)).ToList();
@@ -171,8 +172,14 @@ public static class CvDocumentReview
            if (map.IsArray)
            {
                var list = result[map.Category]!.AsArray();
-                var index = map.ExistingIndex ?? list.Count;
-                if (!map.ExistingIndex.HasValue) list.Add(map.Incoming is JsonObject ? new JsonObject() : JsonValue.Create(""));
+                var identity = map.Category + "|" + map.Incoming.ToJsonString();
+                var scalarText = selected[choices[0].Id].Action == "correct" ? selected[choices[0].Id].Value! : choices[0].ExtractedValue;
+                var matches = Enumerable.Range(0, list.Count).Where(i => map.Incoming is JsonValue
+                    ? string.Equals(Clean(list[i]?.ToString() ?? ""), Clean(scalarText), StringComparison.OrdinalIgnoreCase)
+                    : Same(map.Category, list[i], map.Incoming)).ToList();
+                var index = map.ExistingIndex ?? (newTargets.TryGetValue(identity, out var prior) ? prior : matches.Count == 1 ? matches[0] : list.Count);
+                if (index == list.Count) list.Add(map.Incoming is JsonObject ? new JsonObject() : JsonValue.Create(""));
+                newTargets[identity] = index;
                target = list[index]!;
                targetPrefix = map.Category + "." + index;
            }
@@ -205,17 +212,21 @@ public static class CvDocumentReview
                    var list = result[map.Category]!.AsArray();
                    list[int.Parse(targetPrefix.Split('.').Last())] = value;
                }
-                else Set(target, suffix.Split('.'), value);
+                else suffix = Set(target, suffix.Split('.'), value);
                var path = targetPrefix + (suffix.Length == 0 ? "" : "." + suffix);
                incoming.Metadata.Fields.TryGetValue(field.Id, out var source);
                var metadataNode = result["metadata"]!["fields"]!.AsObject();
-                metadataNode[path] = JsonSerializer.SerializeToNode(new StructuredCvFieldMetadata { Method = decision.Action == "correct" ? "user-correction" : source?.Method ?? "deterministic-source", SourceSnippet = source?.SourceSnippet, SourcePage = source?.SourcePage, SourceBlockId = source?.SourceBlockId, ReviewState = decision.Action == "correct" ? "corrected" : "accepted", LastUpdatedAtUtc = DateTimeOffset.UtcNow }, Json);
+                metadataNode[path] = JsonSerializer.SerializeToNode(new StructuredCvFieldMetadata { Method = decision.Action == "correct" ? "user-correction" : source?.Method ?? "deterministic-source", SourceSnippet = source?.SourceSnippet, SourcePage = source?.SourcePage, SourceBlockId = source?.SourceBlockId, SourceArtifactId = source?.SourceArtifactId, SourceExtractionRunId = source?.SourceExtractionRunId, SourceFileSha256 = source?.SourceFileSha256, ReviewState = decision.Action == "correct" ? "corrected" : "accepted", LastUpdatedAtUtc = DateTimeOffset.UtcNow }, Json);
            }
        }
-        return result.Deserialize<StructuredCvProfile>(Json)!;
+        var profile = result.Deserialize<StructuredCvProfile>(Json)!;
+        var linkCount = profile.Contact.Links.Count;
+        profile = StructuredCvProfileJson.NormalizeForPersistence(profile);
+        if (profile.Contact.Links.Count != linkCount) throw new ArgumentException("A reviewed contact link is incomplete, duplicated or exceeds the supported limit. Keep the existing link or correct its URL.");
+        return profile;
    }

-    private static void Set(JsonNode target, string[] path, JsonNode value)
+    private static string Set(JsonNode target, string[] path, JsonNode value)
    {
        for (var i = 0; i < path.Length - 1; i++)
        {
@@ -232,10 +243,12 @@ public static class CvDocumentReview
        }
        if (target is JsonArray array)
        {
-            var index = int.Parse(path[^1]);
            // Individual approved list entries append instead of erasing user-only entries.
-            if (!array.Any(v => v?.ToString() == value.ToString())) array.Add(value);
+            var index = Enumerable.Range(0, array.Count).FirstOrDefault(i => string.Equals(array[i]?.ToString(), value.ToString(), StringComparison.OrdinalIgnoreCase), -1);
+            if (index < 0) { index = array.Count; array.Add(value); }
+            path[^1] = index.ToString(System.Globalization.CultureInfo.InvariantCulture);
        }
        else target[path[^1]] = value;
+        return string.Join('.', path);
    }
 }
@@ -74,7 +74,7 @@ public sealed class CvProcessingOperationHandler : IAiOperationHandler
                services,
                runId,
                operation?.CancellationRequestedAtUtc is null ? "queued" : "cancelled",
-                operation?.CancellationRequestedAtUtc is null ? "CV processing timed out and may be retried." : "CV processing was cancelled.");
+                operation?.CancellationRequestedAtUtc is null ? "CV processing timed out and may be retried." : "CV processing was cancelled.", context.Lease);
            throw;
        }
        if (outcome is null)
@@ -88,7 +88,7 @@ public sealed class CvProcessingOperationHandler : IAiOperationHandler
                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.");
+                    await SetRunStatusAsync(services, runId, "failed", outcome.FailureMessage ?? "CV processing failed.", context.Lease);
            }

            if (outcome.Provider is not null || outcome.Model is not null || outcome.RouteReason is not null)
@@ -108,18 +108,20 @@ public sealed class CvProcessingOperationHandler : IAiOperationHandler
                outcome.Retryable);
        }

+        var isDocumentImport = await services.GetRequiredService<JobTrackerContext>().CvExtractionRuns.Where(r => r.Id == runId).Select(r => r.ParserVersion).FirstAsync(cancellationToken) == ProfileCvController.DocumentParserVersion;
        return new AiOperationExecutionResult(
-            $"/api/profile-cv/runs/{runId}/{(await services.GetRequiredService<JobTrackerContext>().CvExtractionRuns.Where(r => r.Id == runId).Select(r => r.ParserVersion).FirstAsync(cancellationToken) == ProfileCvController.DocumentParserVersion ? "review" : "diff")}",
+            $"/api/profile-cv/runs/{runId}/{(isDocumentImport ? "review" : "diff")}",
            outcome.Provider,
            outcome.Model,
-            outcome.RouteReason ?? "cv_pipeline");
+            outcome.RouteReason ?? (isDocumentImport ? "cv_review_ready" : "cv_pipeline"));
    }

-    private static Task<int> SetRunStatusAsync(IServiceProvider services, int runId, string status, string message)
+    private static Task<int> SetRunStatusAsync(IServiceProvider services, int runId, string status, string message, UserOperationLease lease)
    {
        var completedAtUtc = status == "failed" || status == "cancelled" ? DateTimeOffset.UtcNow : (DateTimeOffset?)null;
-        return services.GetRequiredService<JobTrackerContext>().CvExtractionRuns
-            .Where(run => run.Id == runId && run.Status != "expired" && run.Status != "applied")
+        var db = services.GetRequiredService<JobTrackerContext>();
+        return db.CvExtractionRuns
+            .Where(run => run.Id == runId && (run.Status == "running" || run.Status == "queued") && db.UserOperations.Any(o => o.Id == lease.OperationId && o.LeaseToken == lease.LeaseToken && o.Status == OperationStatuses.Running))
            .ExecuteUpdateAsync(setters => setters
                .SetProperty(run => run.Status, status)
                .SetProperty(run => run.ErrorMessage, message)