Add large-payload tombstone fetch and purge-result RPCs for blob auto-purge

protos/orchestrator_service.proto

@@ -786,6 +786,19 @@ service TaskHubSidecarService {
     rpc PurgeInstances(PurgeInstancesRequest) returns (PurgeInstancesResponse);
 
     rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem);
+
+    // Returns a bounded, deterministically ordered batch of due large-payload tombstones whose
+    // external blobs the worker must delete. Scoped to the caller's authenticated task hub.
+    // Only rows that are pending and whose next attempt time has arrived are returned; a row stays
+    // pending until its outcome is reported, so this is safe under retries and duplicate callers.
+    rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);
+
+    // Reports the outcome of each attempted blob deletion. The backend owns retry scheduling and
+    // branches solely on `disposition`: it deletes rows reported as DELETED, reschedules RETRY on
+    // its own backoff, and moves QUARANTINED rows out of the active fetch while preserving their
+    // evidence. The worker never computes a retry delay.
+    rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse);
+
     rpc CompleteActivityTask(ActivityResponse) returns (CompleteTaskResponse);
     rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse);
     rpc CompleteEntityTask(EntityBatchResult) returns (CompleteTaskResponse);
@@ -825,13 +838,102 @@ service TaskHubSidecarService {
     rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse);
 }
 
+// server -> client: one tombstoned large-payload row whose external blob the worker must delete.
+message LargePayloadTombstone {
+    int32 partitionId = 1;
+    int64 instanceKey = 2;
+    int64 payloadId = 3;
+
+    // A self-describing SDK v2 token: "blob:v2:{fullBlobUrl}".
+    // Legacy v1 tokens are never tombstoned: v1 carries a container name but not the storage
+    // account, so a delete against the configured account cannot be verified. The backend
+    // hard-deletes v1 payload rows instead.
+    string token = 4;
+
+    // Optimistic-concurrency guard. The worker echoes this value back unmodified in the result so
+    // the backend can reject duplicate or stale reports without taking a per-row lease.
+    int64 revision = 5;
+}
+
+// The outcome of a single blob deletion attempt. The split is by whether a failure can self-heal.
+enum LargePayloadPurgeDisposition {
+    // Required: proto3 reserves 0 as the first value, and scalars have no field presence, so an
+    // unset field arrives as 0. Keeping 0 meaningless is load-bearing here: if 0 meant DELETED, a
+    // client that failed to set this field would make the backend delete tombstones and orphan the
+    // blobs permanently. The backend must reject a result carrying this value.
+    LARGE_PAYLOAD_PURGE_DISPOSITION_UNSPECIFIED = 0;
+
+    // Terminal success: the tombstone is resolved and the backend deletes it. Covers the blob being
+    // deleted, the blob already being absent, and the blob being deliberately left in place because
+    // the payload store does not own it. All three are terminal because none of them can be
+    // improved by trying again.
+    LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED = 1;
+
+    // The failure may self-heal, so the row stays pending and the backend sets the next attempt.
+    LARGE_PAYLOAD_PURGE_DISPOSITION_RETRY = 2;
+
+    // A deterministic failure or protocol violation that retrying can never fix. The row leaves the
+    // polling set but is never deleted or expired: it keeps the token, which after the payload row is
+    // gone is the only durable record of the blob, so discarding it would orphan the blob silently.
+    // Resolving a quarantined row is a deliberate operator action. Why it failed is not recorded here
+    // and is not meant to be; that detail lives in the worker's telemetry at full fidelity.
+    LARGE_PAYLOAD_PURGE_DISPOSITION_QUARANTINED = 3;
+}
+
+// client -> server: the outcome of exactly one tombstoned row.
+message LargePayloadPurgeResult {
+    // Row identity, echoed from the corresponding LargePayloadTombstone.
+    int32 partitionId = 1;
+    int64 instanceKey = 2;
+    int64 payloadId = 3;
+
+    // Echoed unmodified from the fetched tombstone; used as a compare-and-swap guard.
+    int64 revision = 4;
+
+    // The only field the backend acts on. Deliberately the only outcome field on this message:
+    // anything finer would be write-only. Failure detail stays in the worker's own telemetry, which
+    // holds the full exception rather than a lossy classification, and a row is correlated to it by
+    // (partitionId, instanceKey, payloadId).
+    LargePayloadPurgeDisposition disposition = 5;
+}
+
+// client -> server: request up to `limit` due tombstones for the caller's task hub.
+message GetLargePayloadTombstonesRequest {
+    // The maximum number of rows to return. The service clamps this to its own maximum.
+    int32 limit = 1;
+}
+
+// server -> client: the due tombstones whose blobs the worker must delete.
+message GetLargePayloadTombstonesResponse {
+    repeated LargePayloadTombstone tombstones = 1;
+}
+
+// client -> server: a bounded batch of purge outcomes.
+message ReportLargePayloadPurgeResultsRequest {
+    repeated LargePayloadPurgeResult results = 1;
+}
+
+// server -> client: acknowledgement that the reported outcomes were recorded.
+message ReportLargePayloadPurgeResultsResponse {
+}
+
 message GetWorkItemsRequest {
     int32 maxConcurrentOrchestrationWorkItems = 1;
     int32 maxConcurrentActivityWorkItems = 2;
     int32 maxConcurrentEntityWorkItems = 3;
 
     repeated WorkerCapability capabilities = 10;
     WorkItemFilters workItemFilters = 11;
+
+    // Task-hub scoped opt-in for large-payload blob auto-purge.
+    //
+    // This is a setting, not a WorkerCapability: WORKER_CAPABILITY_LARGE_PAYLOADS means a worker
+    // *can resolve* externalized payloads, not that the customer opted into *deleting* them.
+    // A capability list is presence-only and cannot express an explicit false, so a three-state
+    // wrapper is used instead:
+    //   absent      -> the worker has no opinion (older SDK); the persisted setting is unchanged.
+    //   true/false  -> an explicit customer choice; persisted only when the value actually differs.
+    google.protobuf.BoolValue large_payload_auto_purge_enabled = 12;
 }
 
 enum WorkerCapability {