feat(artifact): add wait_for_upload_headroom producer-side backpressure hook

crates/prover-types/src/artifacts.rs

@@ -152,6 +152,20 @@ pub trait ArtifactClient: Send + Sync + Clone + 'static {
         artifact_type: ArtifactType,
     ) -> impl Future<Output = Result<()>> + Send;
 
+    /// Block until the store has room for an upload of this artifact.
+    ///
+    /// Default is a no-op; memory-bounded stores (Redis) override to re-introduce
+    /// the producer/consumer backpressure lost across a distributed split. The
+    /// return type is deliberately `()` — this call is advisory and must never
+    /// fail a proof. Implementations that can't determine store state should
+    /// return silently and let the upload proceed.
+    fn wait_for_upload_headroom(
+        &self,
+        _artifact: &impl ArtifactId,
+    ) -> impl Future<Output = ()> + Send {
+        async {}
+    }
+
     fn try_delete(
         &self,
         artifact: &impl ArtifactId,

crates/prover/src/worker/controller/core.rs

@@ -566,6 +566,11 @@ pub(super) async fn create_core_proving_task<A: ArtifactClient, W: WorkerClient>
         TraceData::Memory(_) | TraceData::Precompile(_, _) => None,
     };
 
+    // Block if the artifact store is near capacity — the distributed analog of the
+    // local node's `ProverSemaphore` backpressure. Advisory (returns `()`, cannot
+    // fail the proof). No-op for S3 / in-memory stores.
+    artifact_client.wait_for_upload_headroom(&record_artifact).await;
+
     artifact_client
         .upload(&record_artifact, trace_data)
         .await