feat: resolve settled attachment uploads with retained receipts

Require identified admin evidence and locked reference checks before deleting bytes. Preserve immutable operational receipts, exclude them from retries, and expose the standard Review/Result table action with help.

Co-Authored-By: GPT-6 (Codex) <noreply@openai.com>

docs/ATTACHMENT-RECOVERY.md

@@ -10,18 +10,28 @@ Verification covers real PostgreSQL metadata rejection, outer upload/delete roll
 
 Failed upload rollback compensation now records its locator independently of the rolled-back transaction before attempting storage cleanup. Failed worker attempts persist a next-attempt time at least 60 seconds later; this lets healthy requests beyond a failing batch proceed. Tests verify later recovery once the failed requests become due.
 
-Upload I/O and runtime failures attempt durable compensation and preserve the original exception, adding cleanup failure as a suppressed exception. The tested ambiguous upload completes its write before throwing; a provider that completes a write after cleanup has already run remains outside this evidence.
+Upload I/O and runtime failures attempt durable compensation and preserve the original exception, adding cleanup failure as a suppressed exception. Deferred provider completion is covered by the settlement protocol below.
 
-Full Maven verification passes 196 tests, zero failures, errors or skips, including packaging, architecture guardrails and coverage checks.
+Full Maven verification including operator settlement passes 206 tests, zero failures, errors or skips, plus packaging, help coverage, architecture guardrails and JaCoCo checks. PostgreSQL tests cover the retained receipt, repeated requests, active owners, existing references, rejected receipt writes, provider failure and the real form/Review/Result flow with permission revocation. Coordinated host adoption and staging rollback still require separate verification.
 
 On backends supporting transactional FOR UPDATE SKIP LOCKED (including the PostgreSQL host), upload now commits a cleanup locator before calling storage and holds its row lock for the upload transaction. Metadata insertion and removal of that locator share the same transaction. Recovery workers skip active uploads and hold the same lock while deleting abandoned bytes and acknowledging cleanup. An existing caller transaction retains ownership until it ends; otherwise the service starts its own transaction after spooling the input. Backends without this locking capability retain callback-based compensation and are not covered by this ownership guarantee.
 
 Focused PostgreSQL tests verify the locator exists before provider I/O, cleanup skips an active upload, and recovery succeeds after an interrupted upload whose rollback callbacks were deliberately suppressed. That test simulates callback loss. A separate child JVM executes the actual upload service against the disposable PostgreSQL database, persists fixture bytes, then calls Runtime.halt before returning from the provider. The parent verifies exit code 73, absent metadata, retained intent and bytes, and successful idempotent recovery using the original locator. This proves abrupt JVM termination recovery, not power-loss durability or late remote-provider completion. Another test rejects intent insertion in PostgreSQL and verifies zero provider writes or deletes; a completed upload in an open outer transaction remains protected from cleanup until commit. Concurrent workers verify that only one worker enters provider deletion for a locator.
 
-Issue #1289 remains open. Outstanding acceptance includes late provider writes after timeout and host migration/rollback. Row locking does not make external object storage and the database one atomic transaction.
+Issue #1289 remains open for coordinated host adoption and application rollback acceptance. Row locking does not make external object storage and the database one atomic transaction.
 
 ## Unreleased late-upload settlement change
 
 A PostgreSQL regression reproduced a PUT finishing after timeout rollback had deleted the still-empty location and acknowledged its cleanup row. Upload intents (no attachment ID) now use AttachmentStorage.cleanupUpload, which removes current bytes and reports whether an earlier PUT can still complete. The generic adapter cannot prove settlement and returns false. Recovery retains the locator for later sweeps, with the existing 60-second backoff and 100-row batch. Ordinary committed deletion and successful upload acknowledgement remain unchanged.
 
-Adapters may report settlement only from authoritative completion/cancellation information, never merely object absence. A deterministic deferred-write fixture verifies the late-completion ordering; this does not establish remote S3 settlement. Generic uncertain-upload records currently remain pending indefinitely. Settlement/retention operations remain unfinished; there is no automatic expiry or operator acknowledgement in this revision. The earlier 196-test result describes the previous PR head, not this uncommitted change.
+Adapters may report settlement only from authoritative completion/cancellation information, never merely object absence. A deterministic deferred-write fixture verifies the late-completion ordering; this does not establish remote S3 settlement. Generic uncertain-upload records remain pending until the provider proves settlement or an administrator performs the explicit operation below. There is no age-based expiry.
+
+## Resolve a settled upload
+
+In the Attachment Cleanup table, select upload intents and run **Resolve settled uploads**. An identified user with `collaboration.admin` must supply provider completion or cancellation evidence covering every selected locator, then confirm the standard Review screen. A support case or request receipt establishing completion is suitable; object absence, an elapsed timeout, or an application process stopping does not establish that an external provider has stopped writing. If settlement cannot be established, leave the intent pending.
+
+Apply rechecks permission, locks each row, rejects active ownership and existing attachment references, and deletes current bytes before saving completion. Provider or database failures leave the intent unresolved for retry. Committed attachment deletions use the ordinary cleanup retry operation instead.
+
+The service-managed row retains `resolvedAt`, `resolvedBy` and `settlementEvidence`, together with the original locator. Repeated settlement does not overwrite that receipt; scheduled retries exclude resolved rows. Direct edits and deletes remain denied even to administrators. These fields provide the durable operational receipt without relying on the framework's best-effort audit writer. Store a reference to provider evidence, never credentials or private object contents.
+
+Apply the additive settlement columns and pending-row index before this code runs. Kof22's corresponding migration is `2026-10-26-collaboration-upload-settlement.sql`. Retain receipts and pending intents during rollback. Do not start an earlier cleanup worker that ignores `resolvedAt`: it could consume retained receipts. Roll back to the pre-cleanup application with cleanup processing disabled, or use a version that understands settlement receipts.

src/main/java/com/kingsrook/qbits/collaboration/CollaborationQBitProducer.java

@@ -188,6 +188,7 @@ public class CollaborationQBitProducer implements QBitMetaDataProducer<Collabora
       }
       output.add(DeleteAttachmentsProcess.metadata());
       output.add(com.kingsrook.qbits.collaboration.attachments.RetryAttachmentCleanupProcess.metadata(qBitConfig));
+      output.add(com.kingsrook.qbits.collaboration.attachments.ResolveAttachmentUploadProcess.metadata());
       output.add(FeatureSettingsDefaultsProcess.process());
       for(QPossibleValueSource source : CollaborationTables.possibleValueSources())
       {

src/main/java/com/kingsrook/qbits/collaboration/CollaborationSchema.java

@@ -96,10 +96,14 @@ public final class CollaborationSchema
       statements.add("CREATE INDEX IF NOT EXISTS idx_collaboration_assignment_assignee ON collaboration_record_assignment (assignee_user_id)");
       statements.add("ALTER TABLE collaboration_mention ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ");
       statements.add("ALTER TABLE collaboration_attachment_cleanup ADD COLUMN IF NOT EXISTS next_attempt_at TIMESTAMPTZ");
+      statements.add("ALTER TABLE collaboration_attachment_cleanup ADD COLUMN IF NOT EXISTS resolved_at TIMESTAMPTZ");
+      statements.add("ALTER TABLE collaboration_attachment_cleanup ADD COLUMN IF NOT EXISTS resolved_by VARCHAR(255)");
+      statements.add("ALTER TABLE collaboration_attachment_cleanup ADD COLUMN IF NOT EXISTS settlement_evidence VARCHAR(2000)");
       statements.add("ALTER TABLE collaboration_attachment_cleanup ALTER COLUMN attachment_id DROP NOT NULL");
       statements.add("DROP INDEX IF EXISTS uq_collaboration_attachment_cleanup_1");
       statements.add("CREATE INDEX IF NOT EXISTS idx_collaboration_attachment_storage_key ON collaboration_attachment USING HASH (storage_key)");
       statements.add("CREATE INDEX IF NOT EXISTS idx_collaboration_attachment_cleanup_due ON collaboration_attachment_cleanup (next_attempt_at, id)");
+      statements.add("CREATE INDEX IF NOT EXISTS idx_collaboration_attachment_cleanup_pending ON collaboration_attachment_cleanup (next_attempt_at, id) WHERE resolved_at IS NULL");
       statements.add("CREATE INDEX IF NOT EXISTS idx_collaboration_mention_unread ON collaboration_mention (mentioned_user_id) WHERE notified_at IS NOT NULL AND read_at IS NULL");
       return (String.join(";\n", statements) + ";\n");
    }

src/main/java/com/kingsrook/qbits/collaboration/attachments/AttachmentCleanupQueue.java

@@ -21,9 +21,11 @@
 package com.kingsrook.qbits.collaboration.attachments;
 
 
+import java.io.Serializable;
 import java.time.Instant;
 import java.util.List;
 import com.kingsrook.qbits.collaboration.CollaborationRecords;
+import com.kingsrook.qbits.collaboration.CollaborationSecurity;
 import com.kingsrook.qbits.collaboration.model.CollaborationTables;
 import com.kingsrook.qbits.collaboration.permissions.CollaborationPermissions;
 import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
@@ -300,6 +302,10 @@ public class AttachmentCleanupQueue implements CustomPermissionChecker
     *******************************************************************************/
    private CleanupOutcome finishUnlocked(QRecord pending) throws QException
    {
+      if(pending.getValue("resolvedAt") != null)
+      {
+         return (CleanupOutcome.COMPLETED);
+      }
       if(!AttachmentRecords.query(new QQueryFilter()
          .withCriteria("bucketName", QCriteriaOperator.EQUALS, pending.getValue("bucketName"))
          .withCriteria("storageKey", QCriteriaOperator.EQUALS, pending.getValue("storageKey"))).isEmpty())
@@ -361,9 +367,10 @@ public class AttachmentCleanupQueue implements CustomPermissionChecker
          throw (new QException("Attachment cleanup must run outside a caller transaction."));
       }
       QueryInput query = new QueryInput(CollaborationTables.ATTACHMENT_CLEANUP).withFilter(new QQueryFilter()
+         .withCriteria("resolvedAt", QCriteriaOperator.IS_BLANK).withSubFilter(new QQueryFilter()
          .withBooleanOperator(QQueryFilter.BooleanOperator.OR)
          .withCriteria("nextAttemptAt", QCriteriaOperator.IS_BLANK)
-         .withCriteria("nextAttemptAt", QCriteriaOperator.LESS_THAN_OR_EQUALS, Instant.now())
+         .withCriteria("nextAttemptAt", QCriteriaOperator.LESS_THAN_OR_EQUALS, Instant.now()))
          .withOrderBy(new QFilterOrderBy("id")).withLimit(100));
       query.withPermissionBypass("Authorized operator retries committed storage cleanup.");
       List<QRecord> pending = new QueryAction().execute(query).getRecords();
@@ -419,6 +426,93 @@ public class AttachmentCleanupQueue implements CustomPermissionChecker
    }
 
 
+   /*******************************************************************************
+    ** An identified administrator attests provider settlement; absence or age alone
+    ** is not evidence. Keep the receipt on the service-managed row permanently.
+    *******************************************************************************/
+   public void resolveUpload(Serializable id, String evidence) throws QException
+   {
+      String actor = requireSettlementOperator();
+      validateEvidence(evidence);
+      if(id == null || QContext.getQBackendTransaction() != null || !supportsUploadOwnership())
+      {
+         throw (new QException("Upload settlement requires a record ID and an independent locking transaction."));
+      }
+      try(QBackendTransaction transaction = QBackendTransaction.openFor(new UpdateInput(CollaborationTables.ATTACHMENT_CLEANUP)))
+      {
+         QContext.setTransaction(transaction);
+         try
+         {
+            QRecord pending = lock(new QRecord().withValue("id", id), transaction);
+            if(pending == null)
+            {
+               throw (new QException("Cleanup record is active or no longer exists. Retry after its owner finishes."));
+            }
+            if(pending.getValue("resolvedAt") != null)
+            {
+               transaction.commit();
+               return;
+            }
+            if(pending.getValue("attachmentId") != null)
+            {
+               throw (new QException("Use normal cleanup retry for committed attachment deletions."));
+            }
+            if(!AttachmentRecords.query(new QQueryFilter()
+               .withCriteria("bucketName", QCriteriaOperator.EQUALS, pending.getValue("bucketName"))
+               .withCriteria("storageKey", QCriteriaOperator.EQUALS, pending.getValue("storageKey"))).isEmpty())
+            {
+               throw (new QException("An attachment still references this object; settlement is not allowed."));
+            }
+            storage.delete(pending.getValueString("bucketName"), pending.getValueString("storageKey"));
+            UpdateInput update = new UpdateInput(CollaborationTables.ATTACHMENT_CLEANUP).withTransaction(transaction)
+               .withRecords(List.of(new QRecord().withValue("id", pending.getValue("id"))
+                  .withValue("resolvedAt", Instant.now()).withValue("resolvedBy", actor)
+                  .withValue("settlementEvidence", evidence.strip())));
+            update.withPermissionBypass("Identified collaboration administrator records confirmed upload settlement after storage cleanup.");
+            AttachmentRecords.checkErrors(new UpdateAction().execute(update).getRecords());
+            transaction.commit();
+         }
+         catch(QException | RuntimeException failure)
+         {
+            transaction.rollback();
+            throw (failure);
+         }
+         finally
+         {
+            QContext.clearTransaction();
+         }
+      }
+   }
+
+
+   /*******************************************************************************
+    ** The form and service both reject missing or oversized evidence.
+    *******************************************************************************/
+   static void validateEvidence(String evidence) throws QException
+   {
+      if(evidence == null || evidence.isBlank() || evidence.strip().length() > 2000)
+      {
+         throw (new QException("Record provider completion or cancellation evidence (1–2000 characters)."));
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Settlement is a deliberate, attributable admin action, never scheduled work.
+    *******************************************************************************/
+   static String requireSettlementOperator() throws QPermissionDeniedException
+   {
+      var session = QContext.getQSession();
+      String actor = CollaborationSecurity.userId(session);
+      if(session == null || session instanceof QSystemUserSession || !session.hasPermission(CollaborationPermissions.ADMIN)
+         || actor == null || actor.isBlank() || actor.length() > 255)
+      {
+         throw (new QPermissionDeniedException("An identified collaboration administrator must confirm upload settlement."));
+      }
+      return (actor);
+   }
+
+
    /*******************************************************************************
     ** Scheduled system work and explicit admin recovery share one authorization rule.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/collaboration/attachments/ResolveAttachmentUploadProcess.java

@@ -0,0 +1,227 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package com.kingsrook.qbits.collaboration.attachments;
+
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import com.kingsrook.qbits.collaboration.model.CollaborationTables;
+import com.kingsrook.qqq.backend.core.actions.permissions.CustomPermissionChecker;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException;
+import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.MetaDataWithPermissionRules;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.PermissionLevel;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QComponentType;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendComponentMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendStepMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ExtractViaQueryStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
+
+
+/*******************************************************************************
+ ** Explicit provider settlement through the standard table action and Review flow.
+ *******************************************************************************/
+public class ResolveAttachmentUploadProcess implements CustomPermissionChecker
+{
+   public static final String NAME = "collaborationResolveAttachmentUpload";
+   public static final String EVIDENCE = "settlementEvidence";
+
+
+   /*******************************************************************************
+    ** Administrative table action, never a navigation entry or scheduled process.
+    *******************************************************************************/
+   public static QProcessMetaData metadata()
+   {
+      Map<String, Serializable> values = new HashMap<>();
+      values.put(StreamedETLWithFrontendProcess.FIELD_SOURCE_TABLE, CollaborationTables.ATTACHMENT_CLEANUP);
+      values.put(StreamedETLWithFrontendProcess.FIELD_DESTINATION_TABLE, CollaborationTables.ATTACHMENT_CLEANUP);
+      values.put(StreamedETLWithFrontendProcess.FIELD_SUPPORTS_FULL_VALIDATION, true);
+      values.put(StreamedETLWithFrontendProcess.FIELD_DO_FULL_VALIDATION, true);
+      values.put(StreamedETLWithFrontendProcess.FIELD_PREVIEW_MESSAGE,
+         "Confirm that the provider has completed or cancelled every selected upload. Object absence or elapsed time alone is not proof. Current bytes will be deleted; the settlement receipt will be retained.");
+      return (StreamedETLWithFrontendProcess.defineProcessMetaData(ExtractViaQueryStep.class, Review.class, Apply.class, values)
+         .withName(NAME).withLabel("Resolve settled uploads").withTableName(CollaborationTables.ATTACHMENT_CLEANUP).withIcon(new QIcon("check"))
+         .withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.NOT_PROTECTED)
+            .withCustomPermissionChecker(new QCodeReference(ResolveAttachmentUploadProcess.class)))
+         .withStep(0, new QFrontendStepMetaData().withName("evidence").withLabel("Confirm provider settlement")
+            .withFormField(new QFieldMetaData(EVIDENCE, QFieldType.STRING).withLabel("Provider completion or cancellation evidence")
+               .withMaxLength(2000).withIsRequired(true))
+            .withComponent(new QFrontendComponentMetaData().withType(QComponentType.EDIT_FORM))));
+   }
+
+
+   /*******************************************************************************
+    ** Guard direct process invocation as well as menu visibility.
+    *******************************************************************************/
+   @Override
+   public void checkPermissionsThrowing(AbstractActionInput input, MetaDataWithPermissionRules metadata) throws QPermissionDeniedException
+   {
+      AttachmentCleanupQueue.requireSettlementOperator();
+   }
+
+
+   /*******************************************************************************
+    ** Shared standard summary language for Review and Result.
+    *******************************************************************************/
+   private static ProcessSummaryLine line(Status status, String future, String past)
+   {
+      ProcessSummaryLine line = new ProcessSummaryLine(status);
+      line.setSingularFutureMessage(future);
+      line.setPluralFutureMessage(future);
+      line.setSingularPastMessage(past);
+      line.setPluralPastMessage(past);
+      return (line);
+   }
+
+
+   /*******************************************************************************
+    ** Review never changes provider state or records a completed settlement.
+    *******************************************************************************/
+   public static class Review extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine selected = line(Status.OK, "uploads will be resolved", "uploads selected");
+      private final ProcessSummaryLine skipped = line(Status.WARNING, "resolved records or committed deletions will be skipped", "records skipped");
+
+
+      /*******************************************************************************
+       ** Require evidence before allowing the administrator to review selected rows.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         AttachmentCleanupQueue.requireSettlementOperator();
+         AttachmentCleanupQueue.validateEvidence(input.getValueString(EVIDENCE));
+         for(QRecord row : input.getRecords())
+         {
+            if(row.getValue("attachmentId") != null || row.getValue("resolvedAt") != null)
+            {
+               skipped.incrementCountAndAddPrimaryKey(row.getValue("id"));
+            }
+            else
+            {
+               output.getRecords().add(row);
+               selected.incrementCountAndAddPrimaryKey(row.getValue("id"));
+            }
+         }
+      }
+
+
+      /*******************************************************************************
+       ** Keep streamed totals across pages.
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         selected.addSelfToListIfAnyCount(lines);
+         skipped.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Apply checks current authorization, row ownership and references again.
+    *******************************************************************************/
+   public static class Apply extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private final AttachmentCleanupQueue queue;
+      private final ProcessSummaryLine done = line(Status.OK, "uploads will be resolved", "uploads resolved");
+      private final ProcessSummaryLine failed = line(Status.ERROR, "uploads cannot be resolved", "uploads unresolved; inspect and retry");
+
+
+      /*******************************************************************************
+       ** Construction is also used during context-free metadata validation.
+       *******************************************************************************/
+      public Apply()
+      {
+         this(new AttachmentCleanupQueue());
+      }
+
+
+      /*******************************************************************************
+       ** Use the same service with a fixture storage provider in process tests.
+       *******************************************************************************/
+      public Apply(AttachmentCleanupQueue queue)
+      {
+         this.queue = queue;
+      }
+
+
+      /*******************************************************************************
+       ** Each locator has its own transaction and immutable settlement receipt.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         AttachmentCleanupQueue.requireSettlementOperator();
+         AttachmentCleanupQueue.validateEvidence(input.getValueString(EVIDENCE));
+         for(QRecord row : input.getRecords())
+         {
+            try
+            {
+               queue.resolveUpload(row.getValue("id"), input.getValueString(EVIDENCE));
+               done.incrementCountAndAddPrimaryKey(row.getValue("id"));
+            }
+            catch(QException | RuntimeException failure)
+            {
+               failed.incrementCountAndAddPrimaryKey(row.getValue("id"));
+            }
+         }
+      }
+
+
+      /*******************************************************************************
+       ** Report actual application results and retain review exclusions.
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         done.addSelfToListIfAnyCount(lines);
+         failed.addSelfToListIfAnyCount(lines);
+         if(getTransformStep() instanceof Review review)
+         {
+            review.skipped.addSelfToListIfAnyCount(lines);
+         }
+         return (lines);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/collaboration/model/CollaborationTables.java

@@ -211,8 +211,13 @@ public final class CollaborationTables
          .withField(string("bucketName", "Bucket", 255).withIsRequired(true))
          .withField(string("storageKey", "Storage Key", 1024).withIsRequired(true))
          .withField(new QFieldMetaData("nextAttemptAt", QFieldType.DATE_TIME).withLabel("Next Attempt"))
+         .withField(new QFieldMetaData("resolvedAt", QFieldType.DATE_TIME).withLabel("Resolved At"))
+         .withField(string("resolvedBy", "Resolved By", 255))
+         .withField(string("settlementEvidence", "Provider Settlement Evidence", 2000))
          .withSection(new QFieldSection("cleanup", new QIcon("delete"), Tier.T1,
-            List.of("id", "attachmentId", "bucketName", "storageKey", "nextAttemptAt")).withLabel("Cleanup")));
+            List.of("id", "attachmentId", "bucketName", "storageKey", "nextAttemptAt")).withLabel("Cleanup"))
+         .withSection(new QFieldSection("settlement", new QIcon("check"), Tier.T2,
+            List.of("resolvedAt", "resolvedBy", "settlementEvidence")).withLabel("Operator Settlement")));
       for(QTableMetaData table : tables)
       {
          table.withSection(new QFieldSection("dates", new QIcon("calendar_today"), Tier.T3, List.of("createDate", "modifyDate")).withLabel("Dates"));

src/main/resources/help/qbit-collaboration.yaml

@@ -749,7 +749,7 @@ entries:
 - key: table:collaboration_attachment_cleanup;slot:help
   summary: "Review storage cleanup waiting to finish after an attachment was deleted."
   content: |
-    **How rows get created:** Authorized attachment deletion creates these records automatically. The attachment disappears from the record immediately; its file is removed after the database commits. Successful cleanup removes this entry. If storage is unavailable, the entry remains for the scheduled retry. Administrators can inspect pending work; these records cannot be edited directly.
+    **How rows get created:** Authorized attachment operations create these records automatically. Deleted files are removed after the database commits. Uncertain uploads remain pending until the provider has finished writing. Ordinary successful cleanup removes its entry; administrator-resolved uploads retain a receipt. Administrators can inspect the records but cannot edit them directly.
 - key: table:collaboration_attachment_cleanup;field:attachmentId
   summary: "The deleted attachment, when known; failed uploads may have no committed attachment ID."
 - key: table:collaboration_attachment_cleanup;field:bucketName
@@ -758,6 +758,18 @@ entries:
   summary: "The immutable file location retained until cleanup succeeds."
 - key: table:collaboration_attachment_cleanup;field:nextAttemptAt
   summary: "Failed cleanup waits at least one minute before another scheduled attempt."
+- key: table:collaboration_attachment_cleanup;field:resolvedAt
+  summary: "When an administrator resolved this upload. Resolved receipts are excluded from retries."
+- key: table:collaboration_attachment_cleanup;field:resolvedBy
+  summary: "The signed-in administrator who resolved this upload. The original identity is retained on repeat requests."
+- key: table:collaboration_attachment_cleanup;field:settlementEvidence
+  summary: "Evidence that the provider completed or cancelled the upload. Never include credentials or private file contents."
+- key: table:collaboration_attachment_cleanup;section:settlement
+  summary: "Retained evidence and identity for an administrator-resolved upload."
+- key: process:collaborationResolveAttachmentUpload;slot:help
+  summary: "Resolve uncertain uploads only after confirming that the provider has finished writing."
+  content: |
+    Select upload intents, record evidence of provider completion or cancellation for every selected location, and confirm the Review screen. Object absence or elapsed time alone is not proof. If completion cannot be established, leave the intent pending. The action requires an identified collaboration administrator, refuses active or referenced uploads, deletes current bytes, and retains a settlement receipt. Read Result to see which uploads remain unresolved. Existing attachment deletions use normal cleanup retry instead.
 - key: process:collaborationRetryAttachmentCleanup;slot:help
   summary: "Retry pending attachment file cleanup after storage or database recovery."
   content: |

src/test/java/com/kingsrook/qbits/collaboration/attachments/AttachmentTransactionsPostgresTest.java

@@ -31,6 +31,11 @@ import com.kingsrook.qbits.collaboration.CollaborationSchema;
 import com.kingsrook.qbits.collaboration.PersonalServiceFixture;
 import com.kingsrook.qbits.objectstorage.api.PutRequest;
 import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
 import com.kingsrook.qqq.backend.module.postgres.model.metadata.PostgreSQLBackendMetaData;
 import com.kingsrook.qqq.backend.module.rdbms.actions.RDBMSTransaction;
 import com.kingsrook.qqq.backend.module.rdbms.jdbc.ConnectionManager;
@@ -38,6 +43,8 @@ import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.wait.strategy.Wait;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -52,6 +59,7 @@ class AttachmentTransactionsPostgresTest extends PersonalServiceFixture
    private static GenericContainer<?> postgres;
    private RecordingStorage storage;
    private AttachmentsService service;
+   private static AttachmentCleanupQueue processQueue;
 
 
    /*******************************************************************************
@@ -673,6 +681,252 @@ class AttachmentTransactionsPostgresTest extends PersonalServiceFixture
    }
 
 
+   /*******************************************************************************
+    ** Manual settlement retains an attributable receipt and does not retry it.
+    *******************************************************************************/
+   @Test
+   void operatorSettlementRetainsReceiptAndStopsRetries() throws Exception
+   {
+      int id = settlementIntent();
+      admin();
+      new AttachmentCleanupQueue(storage).resolveUpload(id, " Provider case 123 confirms PUT completion. ");
+      assertThat(storage.objects).isEmpty();
+      try(Connection connection = connect(); var sql = connection.createStatement();
+         var row = sql.executeQuery("SELECT resolved_by, settlement_evidence, resolved_at FROM collaboration_attachment_cleanup WHERE id = " + id))
+      {
+         assertThat(row.next()).isTrue();
+         assertThat(row.getString(1)).isEqualTo("admin");
+         assertThat(row.getString(2)).isEqualTo("Provider case 123 confirms PUT completion.");
+         assertThat(row.getTimestamp(3)).isNotNull();
+      }
+      int attempts = storage.deleteAttempts.get();
+      new AttachmentCleanupQueue(storage).resolveUpload(id, "A repeat must not overwrite the original receipt.");
+      assertThat(new AttachmentCleanupQueue(storage).retryPending()).isZero();
+      assertThat(storage.deleteAttempts.get()).isEqualTo(attempts);
+      try(Connection connection = connect(); var sql = connection.createStatement())
+      {
+         sql.execute(CollaborationSchema.postgresScript(config));
+         try(var row = sql.executeQuery("SELECT settlement_evidence FROM collaboration_attachment_cleanup WHERE id = " + id))
+         {
+            assertThat(row.next()).isTrue();
+            assertThat(row.getString(1)).isEqualTo("Provider case 123 confirms PUT completion.");
+         }
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Deny unauthenticated, non-admin and scheduled callers before provider I/O.
+    *******************************************************************************/
+   @Test
+   void operatorSettlementRequiresIdentityPermissionAndEvidence() throws Exception
+   {
+      int id = settlementIntent();
+      AttachmentCleanupQueue queue = new AttachmentCleanupQueue(storage);
+      assertThatThrownBy(() -> queue.resolveUpload(id, "confirmed")).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+      QContext.init(instance, new com.kingsrook.qqq.backend.core.model.session.QSystemUserSession());
+      assertThatThrownBy(() -> queue.resolveUpload(id, "confirmed")).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+      admin();
+      assertThatThrownBy(() -> queue.resolveUpload(id, " ")).hasMessageContaining("evidence");
+      assertThatThrownBy(() -> queue.resolveUpload(id, "x".repeat(2001))).hasMessageContaining("evidence");
+      assertThatThrownBy(() -> queue.resolveUpload(null, "confirmed")).hasMessageContaining("record ID");
+      assertThat(storage.deleteAttempts.get()).isZero();
+      assertThat(storage.objects).hasSize(1);
+   }
+
+
+   /*******************************************************************************
+    ** A locked upload cannot be settled by an administrator on another connection.
+    *******************************************************************************/
+   @Test
+   void operatorSettlementSkipsActiveOwner() throws Exception
+   {
+      int id = settlementIntent();
+      admin();
+      try(Connection connection = connect(); var sql = connection.createStatement())
+      {
+         connection.setAutoCommit(false);
+         sql.executeQuery("SELECT id FROM collaboration_attachment_cleanup WHERE id = " + id + " FOR UPDATE").close();
+         assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed"))
+            .hasMessageContaining("active or no longer exists");
+         connection.rollback();
+      }
+      assertThat(storage.deleteAttempts.get()).isZero();
+      assertThat(QContext.getQBackendTransaction()).isNull();
+   }
+
+
+   /*******************************************************************************
+    ** Even an admin cannot delete a referenced locator or settle normal deletions.
+    *******************************************************************************/
+   @Test
+   void operatorSettlementProtectsReferencedObjects() throws Exception
+   {
+      var attachment = upload();
+      try(Connection connection = connect(); var sql = connection.createStatement())
+      {
+         sql.executeUpdate("INSERT INTO collaboration_attachment_cleanup (bucket_name, storage_key) SELECT bucket_name, storage_key FROM collaboration_attachment");
+      }
+      admin();
+      int id;
+      try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery("SELECT id FROM collaboration_attachment_cleanup"))
+      {
+         row.next();
+         id = row.getInt(1);
+      }
+      assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed"))
+         .hasMessageContaining("still references");
+      try(Connection connection = connect(); var sql = connection.createStatement())
+      {
+         sql.executeUpdate("UPDATE collaboration_attachment_cleanup SET attachment_id = " + attachment.id());
+      }
+      assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed"))
+         .hasMessageContaining("normal cleanup retry");
+      assertThat(storage.deleteAttempts.get()).isZero();
+      assertThat(storage.objects).hasSize(1);
+   }
+
+
+   /*******************************************************************************
+    ** Provider failure or rejected receipt persistence leaves a retryable locator.
+    *******************************************************************************/
+   @Test
+   void operatorSettlementFailureRetainsIntent() throws Exception
+   {
+      int id = settlementIntent();
+      admin();
+      storage.failDelete = true;
+      assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")).isInstanceOf(Exception.class);
+      assertThat(storage.objects).hasSize(1);
+      storage.failDelete = false;
+      try(Connection connection = connect(); var sql = connection.createStatement())
+      {
+         sql.execute("CREATE FUNCTION reject_receipt() RETURNS trigger LANGUAGE plpgsql AS 'BEGIN RAISE EXCEPTION ''receipt failure''; END'; CREATE TRIGGER reject_receipt BEFORE UPDATE ON collaboration_attachment_cleanup FOR EACH ROW EXECUTE FUNCTION reject_receipt()");
+      }
+      assertThatThrownBy(() -> new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed")).isInstanceOf(Exception.class);
+      assertThat(storage.objects).isEmpty();
+      try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery("SELECT resolved_at FROM collaboration_attachment_cleanup WHERE id = " + id))
+      {
+         assertThat(row.next()).isTrue();
+         assertThat(row.getTimestamp(1)).isNull();
+      }
+      try(Connection connection = connect(); var sql = connection.createStatement())
+      {
+         sql.execute("DROP TRIGGER reject_receipt ON collaboration_attachment_cleanup");
+      }
+      new AttachmentCleanupQueue(storage).resolveUpload(id, "confirmed after database recovery");
+      assertThat(pendingCount()).isEqualTo(1);
+      assertThat(QContext.getQBackendTransaction()).isNull();
+   }
+
+
+   /*******************************************************************************
+    ** Fixture-only orphan, after its writer has definitively stopped.
+    *******************************************************************************/
+   private int settlementIntent() throws Exception
+   {
+      storage.objects.put("fixture/settled", new byte[] { 1, 2, 3 });
+      try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery(
+         "INSERT INTO collaboration_attachment_cleanup (bucket_name, storage_key) VALUES ('fixture', 'settled') RETURNING id"))
+      {
+         row.next();
+         return (row.getInt(1));
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Genuine form, Review and Result boundaries; revoked access is checked again.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = { "success", "revoked", "storageFailure" })
+   @SuppressWarnings("unchecked")
+   void operatorSettlementProcess(String scenario) throws Exception
+   {
+      int id = settlementIntent();
+      int deletionId;
+      try(Connection connection = connect(); var sql = connection.createStatement(); var row = sql.executeQuery(
+         "INSERT INTO collaboration_attachment_cleanup (attachment_id, bucket_name, storage_key) VALUES (99, 'fixture', 'normal-deletion') RETURNING id"))
+      {
+         row.next();
+         deletionId = row.getInt(1);
+      }
+      var process = instance.getProcess(ResolveAttachmentUploadProcess.NAME);
+      process.getBackendStep(StreamedETLWithFrontendProcess.STEP_NAME_EXECUTE).getInputMetaData()
+         .getFieldThrowing(StreamedETLWithFrontendProcess.FIELD_LOAD_CODE).setDefaultValue(new QCodeReference(FixtureSettlement.class));
+      processQueue = new AttachmentCleanupQueue(storage);
+      try
+      {
+         RunProcessInput start = new RunProcessInput();
+         start.setProcessName(ResolveAttachmentUploadProcess.NAME);
+         start.addValue("recordsParam", "recordIds");
+         start.addValue("recordIds", id + "," + deletionId);
+         start.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.BREAK);
+         assertThatThrownBy(() -> new RunProcessAction().execute(start))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+         admin();
+         var form = new RunProcessAction().execute(start);
+         assertThat(form.getException()).isEmpty();
+         start.setProcessUUID(form.getProcessUUID());
+         start.setStartAfterStep("evidence");
+         start.addValue(ResolveAttachmentUploadProcess.EVIDENCE, "Provider request 123 completed; writer stopped.");
+         var review = new RunProcessAction().execute(start);
+         assertThat(review.getException()).isEmpty();
+         List<ProcessSummaryLine> reviewLines = (List<ProcessSummaryLine>) review.getValues().get(StreamedETLWithFrontendProcess.FIELD_VALIDATION_SUMMARY);
+         assertThat(reviewLines).hasSize(2);
+         assertThat(storage.deleteAttempts.get()).isZero();
+         if(scenario.equals("revoked"))
+         {
+            login("admin");
+         }
+         storage.failDelete = scenario.equals("storageFailure");
+         RunProcessInput go = new RunProcessInput();
+         go.setProcessName(ResolveAttachmentUploadProcess.NAME);
+         go.setProcessUUID(review.getProcessUUID());
+         go.setStartAfterStep(StreamedETLWithFrontendProcess.STEP_NAME_REVIEW);
+         go.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.SKIP);
+         if(scenario.equals("revoked"))
+         {
+            assertThatThrownBy(() -> new RunProcessAction().execute(go))
+               .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException.class);
+            assertThat(storage.deleteAttempts.get()).isZero();
+         }
+         else
+         {
+            var result = new RunProcessAction().execute(go);
+            assertThat(result.getException()).isEmpty();
+            List<ProcessSummaryLine> lines = (List<ProcessSummaryLine>) result.getValues().get(StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY);
+            assertThat(lines).anySatisfy(line ->
+            {
+               line.pickMessage(true);
+               assertThat(line.getMessage()).isEqualTo(scenario.equals("success") ? "uploads resolved" : "uploads unresolved; inspect and retry");
+               assertThat(line.getCount()).isEqualTo(1);
+            });
+         }
+         assertThat(storage.objects).hasSize(scenario.equals("success") ? 0 : 1);
+      }
+      finally
+      {
+         processQueue = null;
+      }
+   }
+
+
+   /*******************************************************************************
+    ** The real process uses the real queue with deterministic fixture storage.
+    *******************************************************************************/
+   public static class FixtureSettlement extends ResolveAttachmentUploadProcess.Apply
+   {
+      /*******************************************************************************
+       ** Instantiated by the standard process loader.
+       *******************************************************************************/
+      public FixtureSettlement()
+      {
+         super(processQueue);
+      }
+   }
+
+
    /*******************************************************************************
     ** A timed-out remote PUT can finish after rollback has already attempted DELETE.
     *******************************************************************************/