feat(bulk-upload): gate finalise on unresolved addresses + overlay corrections (ADR-0057)

dispatchFinaliser now:
- blocks (before the CAS claim) when any flagged combiner row is still
  unresolved -> new "unresolved_addresses" outcome; finalize route returns 409.
- overlays the user's confirmed UPRNs onto the combiner CSV (buildConfirmedCombinerUri
  writes a sibling .confirmed.csv) and points the finaliser at it, so a confirmed
  UPRN drives identity + property_overrides in one pass. No corrections -> original
  URI unchanged.

tsc --noEmit clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/finalize/route.ts

@@ -35,6 +35,14 @@ export async function POST(
       return NextResponse.json({ error: "Combiner not finished" }, { status: 409 });
     case "missing_task":
       return NextResponse.json({ error: "Upload has no task to finalise" }, { status: 409 });
+    case "unresolved_addresses":
+      return NextResponse.json(
+        {
+          error: `${result.count} address${result.count === 1 ? "" : "es"} still need a UPRN. Resolve them on the Addresses tab before finalising.`,
+          unresolvedAddresses: result.count,
+        },
+        { status: 409 },
+      );
     case "wrong_state":
       return NextResponse.json(
         { error: `Upload not ready to finalize (state: ${result.current})` },

src/lib/bulkUpload/addressMatches.ts

@@ -121,6 +121,75 @@ export async function countUnresolvedFlaggedAddresses(uploadId: string): Promise
   return flagged.filter((f) => !f.resolved).length;
 }
 
+const ADDRESS_COL = "address2uprn_address";
+const OVERLAY_STATUS_CONFIRMED = "user_confirmed";
+const OVERLAY_STATUS_NO_UPRN = "user_no_uprn";
+
+/**
+ * Overlay the user's saved corrections onto the combiner CSV and write the
+ * result to a sibling `…​.confirmed.csv`, returning its S3 URI (ADR-0057). The
+ * finaliser reads this so a confirmed UPRN drives both the identity insert and
+ * property_overrides in one pass. No corrections ⇒ the original URI is returned
+ * unchanged (no rewrite). A `markedNoUprn` row keeps an empty UPRN (accepted
+ * terminal state — the finaliser leaves it as a no-UPRN property).
+ */
+export async function buildConfirmedCombinerUri(
+  uploadId: string,
+  combinedOutputS3Uri: string,
+): Promise<string> {
+  const corrections = await db
+    .select()
+    .from(bulkUploadUprnCorrections)
+    .where(eq(bulkUploadUprnCorrections.uploadId, uploadId));
+  if (corrections.length === 0) return combinedOutputS3Uri;
+  const byRow = new Map(corrections.map((c) => [c.sourceRowId, c]));
+
+  const { bucket, key } = parseS3Uri(combinedOutputS3Uri);
+  const s3 = createRetrofitDataS3Client();
+  const result = await s3.getObject({ Bucket: bucket, Key: key }).promise();
+  const body = result.Body?.toString("utf-8");
+  if (!body) return combinedOutputS3Uri;
+
+  const wb = XLSX.read(body, { type: "string" });
+  const sheet = wb.Sheets[wb.SheetNames[0]];
+  // Array-of-arrays so every column + its order is preserved exactly; we patch
+  // only the address2uprn cells of corrected rows.
+  const aoa = XLSX.utils.sheet_to_json<string[]>(sheet, {
+    header: 1,
+    defval: "",
+    raw: false,
+    blankrows: false,
+  });
+  const headers = (aoa[0] ?? []).map((h) => String(h));
+  const iRow = headers.indexOf(SOURCE_ROW_ID_COLUMN);
+  const iUprn = headers.indexOf(UPRN_COL);
+  const iStatus = headers.indexOf(STATUS_COL);
+  const iAddr = headers.indexOf(ADDRESS_COL);
+  if (iRow < 0 || iUprn < 0) return combinedOutputS3Uri; // not an address2uprn CSV
+
+  for (let r = 1; r < aoa.length; r++) {
+    const line = aoa[r];
+    const srid = String(line[iRow] ?? "").trim();
+    const c = byRow.get(srid);
+    if (!c) continue;
+    if (c.markedNoUprn) {
+      line[iUprn] = "";
+      if (iStatus >= 0) line[iStatus] = OVERLAY_STATUS_NO_UPRN;
+    } else {
+      line[iUprn] = c.uprn ?? "";
+      if (iStatus >= 0) line[iStatus] = OVERLAY_STATUS_CONFIRMED;
+      if (iAddr >= 0 && c.address) line[iAddr] = c.address;
+    }
+  }
+
+  const csv = XLSX.utils.sheet_to_csv(XLSX.utils.aoa_to_sheet(aoa));
+  const confirmedKey = `${key.replace(/\.csv$/i, "")}.confirmed.csv`;
+  await s3
+    .putObject({ Bucket: bucket, Key: confirmedKey, Body: csv, ContentType: "text/csv" })
+    .promise();
+  return `s3://${bucket}/${confirmedKey}`;
+}
+
 export type UprnCorrectionInput =
   | {
       sourceRowId: string;

src/lib/bulkUpload/server.ts

@@ -20,6 +20,10 @@ import { retrofitDataS3Bucket } from "@/app/utils/s3";
 import { SUBTASK_SERVICE } from "./types";
 import type { MultiEntrySummary } from "./multiEntry";
 import { isPermutation } from "./multiEntry";
+import {
+  buildConfirmedCombinerUri,
+  countUnresolvedFlaggedAddresses,
+} from "./addressMatches";
 
 const REMAP_ALLOWED: ReadonlySet<BulkUploadStatus> = new Set([
   "ready_for_processing",
@@ -677,6 +681,7 @@ export type DispatchFinaliserOutcome =
   | { kind: "not_yet_combined" }
   | { kind: "wrong_state"; current: string }
   | { kind: "missing_task" }
+  | { kind: "unresolved_addresses"; count: number }
   | { kind: "trigger_failed"; status: number; message: string };
 
 // Dispatch the async bulk_upload_finaliser (ADR-0005). Replaces the old
@@ -705,6 +710,14 @@ export async function dispatchFinaliser(args: {
   const upload = guarded.upload;
   if (!upload.taskId) return { kind: "missing_task" };
 
+  // ADR-0057: every combiner row address2uprn could not confidently match must
+  // be resolved on the Addresses tab (a chosen UPRN or an explicit no-UPRN)
+  // before finalise — otherwise a withheld row would land as an unintended
+  // no-UPRN property. Gate before the CAS claim so a blocked upload stays in
+  // awaiting_review.
+  const unresolved = await countUnresolvedFlaggedAddresses(args.uploadId);
+  if (unresolved > 0) return { kind: "unresolved_addresses", count: unresolved };
+
   // CAS: atomically claim the dispatch. Only the request that flips
   // awaiting_review → finalising proceeds; a concurrent one updates 0 rows.
   const claimed = await db
@@ -748,10 +761,18 @@ export async function dispatchFinaliser(args: {
       ? `s3://${retrofitDataS3Bucket()}/${classifierCsvKey(upload.portfolioId, args.uploadId)}`
       : null;
 
+  // Overlay the user's confirmed UPRNs onto the combiner CSV and point the
+  // finaliser at the corrected copy (ADR-0057). No corrections ⇒ the original
+  // URI is returned unchanged.
+  const confirmedS3Uri = await buildConfirmedCombinerUri(
+    args.uploadId,
+    upload.combinedOutputS3Uri!,
+  );
+
   const payload = {
     task_id: upload.taskId,
     sub_task_id: subTask.id,
-    s3_uri: upload.combinedOutputS3Uri,
+    s3_uri: confirmedS3Uri,
     portfolio_id: Number(upload.portfolioId),
     bulk_upload_id: args.uploadId,
     classifier_s3_uri: classifierS3Uri,