diff --git a/gen/pb-cpp/flyteidl/admin/agent.pb.cc b/gen/pb-cpp/flyteidl/admin/agent.pb.cc
index 3948427bc..2bd0789f2 100644
--- a/gen/pb-cpp/flyteidl/admin/agent.pb.cc
+++ b/gen/pb-cpp/flyteidl/admin/agent.pb.cc
@@ -324,6 +324,7 @@ const ::google::protobuf::uint32 TableStruct_flyteidl_2fadmin_2fagent_2eproto::o
   ~0u,  // no _weak_field_map_
   PROTOBUF_FIELD_OFFSET(::flyteidl::admin::Resource, state_),
   PROTOBUF_FIELD_OFFSET(::flyteidl::admin::Resource, outputs_),
+  PROTOBUF_FIELD_OFFSET(::flyteidl::admin::Resource, message_),
   ~0u,  // no _has_bits_
   PROTOBUF_FIELD_OFFSET(::flyteidl::admin::DeleteTaskRequest, _internal_metadata_),
   ~0u,  // no _extensions_
@@ -347,8 +348,8 @@ static const ::google::protobuf::internal::MigrationSchema schemas[] PROTOBUF_SE
   { 53, -1, sizeof(::flyteidl::admin::GetTaskRequest)},
   { 60, -1, sizeof(::flyteidl::admin::GetTaskResponse)},
   { 66, -1, sizeof(::flyteidl::admin::Resource)},
-  { 73, -1, sizeof(::flyteidl::admin::DeleteTaskRequest)},
-  { 80, -1, sizeof(::flyteidl::admin::DeleteTaskResponse)},
+  { 74, -1, sizeof(::flyteidl::admin::DeleteTaskRequest)},
+  { 81, -1, sizeof(::flyteidl::admin::DeleteTaskResponse)},
 };
 
 static ::google::protobuf::Message const * const file_default_instances[] = {
@@ -399,20 +400,20 @@ const char descriptor_table_protodef_flyteidl_2fadmin_2fagent_2eproto[] =
   "\016GetTaskRequest\022\021\n\ttask_type\030\001 \001(\t\022\025\n\rre"
   "source_meta\030\002 \001(\014\"=\n\017GetTaskResponse\022*\n\010"
   "resource\030\001 \001(\0132\030.flyteidl.admin.Resource"
-  "\"\\\n\010Resource\022$\n\005state\030\001 \001(\0162\025.flyteidl.a"
+  "\"m\n\010Resource\022$\n\005state\030\001 \001(\0162\025.flyteidl.a"
   "dmin.State\022*\n\007outputs\030\002 \001(\0132\031.flyteidl.c"
-  "ore.LiteralMap\"=\n\021DeleteTaskRequest\022\021\n\tt"
-  "ask_type\030\001 \001(\t\022\025\n\rresource_meta\030\002 \001(\014\"\024\n"
-  "\022DeleteTaskResponse*^\n\005State\022\025\n\021RETRYABL"
-  "E_FAILURE\020\000\022\025\n\021PERMANENT_FAILURE\020\001\022\013\n\007PE"
-  "NDING\020\002\022\013\n\007RUNNING\020\003\022\r\n\tSUCCEEDED\020\004B7Z5g"
-  "ithub.com/flyteorg/flyteidl/gen/pb-go/fl"
-  "yteidl/adminb\006proto3"
+  "ore.LiteralMap\022\017\n\007message\030\003 \001(\t\"=\n\021Delet"
+  "eTaskRequest\022\021\n\ttask_type\030\001 \001(\t\022\025\n\rresou"
+  "rce_meta\030\002 \001(\014\"\024\n\022DeleteTaskResponse*^\n\005"
+  "State\022\025\n\021RETRYABLE_FAILURE\020\000\022\025\n\021PERMANEN"
+  "T_FAILURE\020\001\022\013\n\007PENDING\020\002\022\013\n\007RUNNING\020\003\022\r\n"
+  "\tSUCCEEDED\020\004B7Z5github.com/flyteorg/flyt"
+  "eidl/gen/pb-go/flyteidl/adminb\006proto3"
   ;
 ::google::protobuf::internal::DescriptorTable descriptor_table_flyteidl_2fadmin_2fagent_2eproto = {
   false, InitDefaults_flyteidl_2fadmin_2fagent_2eproto, 
   descriptor_table_protodef_flyteidl_2fadmin_2fagent_2eproto,
-  "flyteidl/admin/agent.proto", &assign_descriptors_table_flyteidl_2fadmin_2fagent_2eproto, 1420,
+  "flyteidl/admin/agent.proto", &assign_descriptors_table_flyteidl_2fadmin_2fagent_2eproto, 1437,
 };
 
 void AddDescriptors_flyteidl_2fadmin_2fagent_2eproto() {
@@ -2962,6 +2963,7 @@ void Resource::clear_outputs() {
 #if !defined(_MSC_VER) || _MSC_VER >= 1900
 const int Resource::kStateFieldNumber;
 const int Resource::kOutputsFieldNumber;
+const int Resource::kMessageFieldNumber;
 #endif  // !defined(_MSC_VER) || _MSC_VER >= 1900
 
 Resource::Resource()
@@ -2973,6 +2975,10 @@ Resource::Resource(const Resource& from)
   : ::google::protobuf::Message(),
       _internal_metadata_(nullptr) {
   _internal_metadata_.MergeFrom(from._internal_metadata_);
+  message_.UnsafeSetDefault(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
+  if (from.message().size() > 0) {
+    message_.AssignWithDefault(&::google::protobuf::internal::GetEmptyStringAlreadyInited(), from.message_);
+  }
   if (from.has_outputs()) {
     outputs_ = new ::flyteidl::core::LiteralMap(*from.outputs_);
   } else {
@@ -2985,6 +2991,7 @@ Resource::Resource(const Resource& from)
 void Resource::SharedCtor() {
   ::google::protobuf::internal::InitSCC(
       &scc_info_Resource_flyteidl_2fadmin_2fagent_2eproto.base);
+  message_.UnsafeSetDefault(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
   ::memset(&outputs_, 0, static_cast<size_t>(
       reinterpret_cast<char*>(&state_) -
       reinterpret_cast<char*>(&outputs_)) + sizeof(state_));
@@ -2996,6 +3003,7 @@ Resource::~Resource() {
 }
 
 void Resource::SharedDtor() {
+  message_.DestroyNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
   if (this != internal_default_instance()) delete outputs_;
 }
 
@@ -3014,6 +3022,7 @@ void Resource::Clear() {
   // Prevent compiler warnings about cached_has_bits being unused
   (void) cached_has_bits;
 
+  message_.ClearToEmptyNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
   if (GetArenaNoVirtual() == nullptr && outputs_ != nullptr) {
     delete outputs_;
   }
@@ -3056,6 +3065,22 @@ const char* Resource::_InternalParse(const char* begin, const char* end, void* o
             {parser_till_end, object}, ptr - size, ptr));
         break;
       }
+      // string message = 3;
+      case 3: {
+        if (static_cast<::google::protobuf::uint8>(tag) != 26) goto handle_unusual;
+        ptr = ::google::protobuf::io::ReadSize(ptr, &size);
+        GOOGLE_PROTOBUF_PARSER_ASSERT(ptr);
+        ctx->extra_parse_data().SetFieldName("flyteidl.admin.Resource.message");
+        object = msg->mutable_message();
+        if (size > end - ptr + ::google::protobuf::internal::ParseContext::kSlopBytes) {
+          parser_till_end = ::google::protobuf::internal::GreedyStringParserUTF8;
+          goto string_till_end;
+        }
+        GOOGLE_PROTOBUF_PARSER_ASSERT(::google::protobuf::internal::StringCheckUTF8(ptr, size, ctx));
+        ::google::protobuf::internal::InlineGreedyStringParser(object, ptr, size, ctx);
+        ptr += size;
+        break;
+      }
       default: {
       handle_unusual:
         if ((tag & 7) == 4 || tag == 0) {
@@ -3071,6 +3096,10 @@ const char* Resource::_InternalParse(const char* begin, const char* end, void* o
     }  // switch
   }  // while
   return ptr;
+string_till_end:
+  static_cast<::std::string*>(object)->clear();
+  static_cast<::std::string*>(object)->reserve(size);
+  goto len_delim_till_end;
 len_delim_till_end:
   return ctx->StoreAndTailCall(ptr, end, {_InternalParse, msg},
                                {parser_till_end, object}, size);
@@ -3111,6 +3140,21 @@ bool Resource::MergePartialFromCodedStream(
         break;
       }
 
+      // string message = 3;
+      case 3: {
+        if (static_cast< ::google::protobuf::uint8>(tag) == (26 & 0xFF)) {
+          DO_(::google::protobuf::internal::WireFormatLite::ReadString(
+                input, this->mutable_message()));
+          DO_(::google::protobuf::internal::WireFormatLite::VerifyUtf8String(
+            this->message().data(), static_cast<int>(this->message().length()),
+            ::google::protobuf::internal::WireFormatLite::PARSE,
+            "flyteidl.admin.Resource.message"));
+        } else {
+          goto handle_unusual;
+        }
+        break;
+      }
+
       default: {
       handle_unusual:
         if (tag == 0) {
@@ -3150,6 +3194,16 @@ void Resource::SerializeWithCachedSizes(
       2, HasBitSetters::outputs(this), output);
   }
 
+  // string message = 3;
+  if (this->message().size() > 0) {
+    ::google::protobuf::internal::WireFormatLite::VerifyUtf8String(
+      this->message().data(), static_cast<int>(this->message().length()),
+      ::google::protobuf::internal::WireFormatLite::SERIALIZE,
+      "flyteidl.admin.Resource.message");
+    ::google::protobuf::internal::WireFormatLite::WriteStringMaybeAliased(
+      3, this->message(), output);
+  }
+
   if (_internal_metadata_.have_unknown_fields()) {
     ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
         _internal_metadata_.unknown_fields(), output);
@@ -3176,6 +3230,17 @@ ::google::protobuf::uint8* Resource::InternalSerializeWithCachedSizesToArray(
         2, HasBitSetters::outputs(this), target);
   }
 
+  // string message = 3;
+  if (this->message().size() > 0) {
+    ::google::protobuf::internal::WireFormatLite::VerifyUtf8String(
+      this->message().data(), static_cast<int>(this->message().length()),
+      ::google::protobuf::internal::WireFormatLite::SERIALIZE,
+      "flyteidl.admin.Resource.message");
+    target =
+      ::google::protobuf::internal::WireFormatLite::WriteStringToArray(
+        3, this->message(), target);
+  }
+
   if (_internal_metadata_.have_unknown_fields()) {
     target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
         _internal_metadata_.unknown_fields(), target);
@@ -3197,6 +3262,13 @@ size_t Resource::ByteSizeLong() const {
   // Prevent compiler warnings about cached_has_bits being unused
   (void) cached_has_bits;
 
+  // string message = 3;
+  if (this->message().size() > 0) {
+    total_size += 1 +
+      ::google::protobuf::internal::WireFormatLite::StringSize(
+        this->message());
+  }
+
   // .flyteidl.core.LiteralMap outputs = 2;
   if (this->has_outputs()) {
     total_size += 1 +
@@ -3237,6 +3309,10 @@ void Resource::MergeFrom(const Resource& from) {
   ::google::protobuf::uint32 cached_has_bits = 0;
   (void) cached_has_bits;
 
+  if (from.message().size() > 0) {
+
+    message_.AssignWithDefault(&::google::protobuf::internal::GetEmptyStringAlreadyInited(), from.message_);
+  }
   if (from.has_outputs()) {
     mutable_outputs()->::flyteidl::core::LiteralMap::MergeFrom(from.outputs());
   }
@@ -3270,6 +3346,8 @@ void Resource::Swap(Resource* other) {
 void Resource::InternalSwap(Resource* other) {
   using std::swap;
   _internal_metadata_.Swap(&other->_internal_metadata_);
+  message_.Swap(&other->message_, &::google::protobuf::internal::GetEmptyStringAlreadyInited(),
+    GetArenaNoVirtual());
   swap(outputs_, other->outputs_);
   swap(state_, other->state_);
 }
diff --git a/gen/pb-cpp/flyteidl/admin/agent.pb.h b/gen/pb-cpp/flyteidl/admin/agent.pb.h
index 5ee08079f..5d43236ad 100644
--- a/gen/pb-cpp/flyteidl/admin/agent.pb.h
+++ b/gen/pb-cpp/flyteidl/admin/agent.pb.h
@@ -1015,6 +1015,20 @@ class Resource final :
 
   // accessors -------------------------------------------------------
 
+  // string message = 3;
+  void clear_message();
+  static const int kMessageFieldNumber = 3;
+  const ::std::string& message() const;
+  void set_message(const ::std::string& value);
+  #if LANG_CXX11
+  void set_message(::std::string&& value);
+  #endif
+  void set_message(const char* value);
+  void set_message(const char* value, size_t size);
+  ::std::string* mutable_message();
+  ::std::string* release_message();
+  void set_allocated_message(::std::string* message);
+
   // .flyteidl.core.LiteralMap outputs = 2;
   bool has_outputs() const;
   void clear_outputs();
@@ -1035,6 +1049,7 @@ class Resource final :
   class HasBitSetters;
 
   ::google::protobuf::internal::InternalMetadataWithArena _internal_metadata_;
+  ::google::protobuf::internal::ArenaStringPtr message_;
   ::flyteidl::core::LiteralMap* outputs_;
   int state_;
   mutable ::google::protobuf::internal::CachedSize _cached_size_;
@@ -1985,6 +2000,59 @@ inline void Resource::set_allocated_outputs(::flyteidl::core::LiteralMap* output
   // @@protoc_insertion_point(field_set_allocated:flyteidl.admin.Resource.outputs)
 }
 
+// string message = 3;
+inline void Resource::clear_message() {
+  message_.ClearToEmptyNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
+}
+inline const ::std::string& Resource::message() const {
+  // @@protoc_insertion_point(field_get:flyteidl.admin.Resource.message)
+  return message_.GetNoArena();
+}
+inline void Resource::set_message(const ::std::string& value) {
+  
+  message_.SetNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited(), value);
+  // @@protoc_insertion_point(field_set:flyteidl.admin.Resource.message)
+}
+#if LANG_CXX11
+inline void Resource::set_message(::std::string&& value) {
+  
+  message_.SetNoArena(
+    &::google::protobuf::internal::GetEmptyStringAlreadyInited(), ::std::move(value));
+  // @@protoc_insertion_point(field_set_rvalue:flyteidl.admin.Resource.message)
+}
+#endif
+inline void Resource::set_message(const char* value) {
+  GOOGLE_DCHECK(value != nullptr);
+  
+  message_.SetNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited(), ::std::string(value));
+  // @@protoc_insertion_point(field_set_char:flyteidl.admin.Resource.message)
+}
+inline void Resource::set_message(const char* value, size_t size) {
+  
+  message_.SetNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited(),
+      ::std::string(reinterpret_cast<const char*>(value), size));
+  // @@protoc_insertion_point(field_set_pointer:flyteidl.admin.Resource.message)
+}
+inline ::std::string* Resource::mutable_message() {
+  
+  // @@protoc_insertion_point(field_mutable:flyteidl.admin.Resource.message)
+  return message_.MutableNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
+}
+inline ::std::string* Resource::release_message() {
+  // @@protoc_insertion_point(field_release:flyteidl.admin.Resource.message)
+  
+  return message_.ReleaseNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited());
+}
+inline void Resource::set_allocated_message(::std::string* message) {
+  if (message != nullptr) {
+    
+  } else {
+    
+  }
+  message_.SetAllocatedNoArena(&::google::protobuf::internal::GetEmptyStringAlreadyInited(), message);
+  // @@protoc_insertion_point(field_set_allocated:flyteidl.admin.Resource.message)
+}
+
 // -------------------------------------------------------------------
 
 // DeleteTaskRequest
diff --git a/gen/pb-go/flyteidl/admin/agent.pb.go b/gen/pb-go/flyteidl/admin/agent.pb.go
index 21152a7c9..a99d01b89 100644
--- a/gen/pb-go/flyteidl/admin/agent.pb.go
+++ b/gen/pb-go/flyteidl/admin/agent.pb.go
@@ -349,10 +349,12 @@ type Resource struct {
 	// The outputs of the execution. It's typically used by sql task. Agent service will create a
 	// Structured dataset pointing to the query result table.
 	// +optional
-	Outputs              *core.LiteralMap `protobuf:"bytes,2,opt,name=outputs,proto3" json:"outputs,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}         `json:"-"`
-	XXX_unrecognized     []byte           `json:"-"`
-	XXX_sizecache        int32            `json:"-"`
+	Outputs *core.LiteralMap `protobuf:"bytes,2,opt,name=outputs,proto3" json:"outputs,omitempty"`
+	// A descriptive message for the current state. e.g. waiting for cluster.
+	Message              string   `protobuf:"bytes,3,opt,name=message,proto3" json:"message,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
 }
 
 func (m *Resource) Reset()         { *m = Resource{} }
@@ -394,6 +396,13 @@ func (m *Resource) GetOutputs() *core.LiteralMap {
 	return nil
 }
 
+func (m *Resource) GetMessage() string {
+	if m != nil {
+		return m.Message
+	}
+	return ""
+}
+
 // A message used to delete a task.
 type DeleteTaskRequest struct {
 	// A predefined yet extensible Task type identifier.
@@ -494,51 +503,52 @@ func init() {
 func init() { proto.RegisterFile("flyteidl/admin/agent.proto", fileDescriptor_c434e52bb0028071) }
 
 var fileDescriptor_c434e52bb0028071 = []byte{
-	// 724 bytes of a gzipped FileDescriptorProto
+	// 738 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x55, 0x5d, 0x6f, 0xe2, 0x46,
 	0x14, 0x2d, 0x10, 0x08, 0x5c, 0xf2, 0x01, 0xd3, 0xa0, 0x3a, 0x24, 0xad, 0x22, 0xaa, 0x56, 0x51,
 	0xab, 0x1a, 0x85, 0xb4, 0x4d, 0xd2, 0x87, 0x56, 0x24, 0xb8, 0x08, 0x89, 0xa0, 0x68, 0x02, 0x55,
-	0x5b, 0xa9, 0x45, 0x83, 0xb9, 0xb0, 0x16, 0x66, 0xec, 0xf5, 0x8c, 0x51, 0x78, 0xde, 0x3f, 0xb1,
-	0x3f, 0x77, 0xe5, 0xf1, 0x47, 0x00, 0xb1, 0xab, 0x44, 0xfb, 0x86, 0xef, 0x39, 0xf7, 0xcc, 0x99,
-	0x73, 0xaf, 0x0d, 0x54, 0x27, 0xf6, 0x52, 0xa2, 0x35, 0xb6, 0xeb, 0x6c, 0x3c, 0xb7, 0x78, 0x9d,
-	0x4d, 0x91, 0x4b, 0xdd, 0xf5, 0x1c, 0xe9, 0x90, 0x83, 0x18, 0xd3, 0x15, 0x56, 0x3d, 0x4d, 0xb8,
-	0xa6, 0xe3, 0x61, 0xdd, 0xb6, 0x24, 0x7a, 0xcc, 0x16, 0x21, 0xbb, 0x7a, 0xbc, 0x8e, 0x4a, 0x26,
-	0x66, 0x31, 0xf4, 0xf5, 0x3a, 0x64, 0x71, 0x89, 0xde, 0x84, 0x99, 0x18, 0xc1, 0xdf, 0x6c, 0xc0,
-	0x63, 0xe4, 0xd2, 0x9a, 0x58, 0xe8, 0x85, 0x78, 0xed, 0x7d, 0x16, 0x2a, 0x7d, 0x26, 0x66, 0xc6,
-	0x13, 0x9a, 0xbe, 0xb4, 0x1c, 0x7e, 0x8f, 0x92, 0x8d, 0x99, 0x64, 0x84, 0x42, 0x39, 0x38, 0x67,
-	0x88, 0x31, 0x32, 0xb4, 0xc6, 0x5a, 0xea, 0x2c, 0x75, 0x5e, 0x6c, 0x7c, 0xaf, 0x27, 0xee, 0x03,
-	0x55, 0x7d, 0x4d, 0xa0, 0x93, 0x1c, 0x41, 0x0f, 0xe5, 0x3a, 0x40, 0x4e, 0xa1, 0xc0, 0xd9, 0x1c,
-	0x85, 0xcb, 0x4c, 0xd4, 0xd2, 0x67, 0xa9, 0xf3, 0x02, 0x7d, 0x2e, 0x90, 0x0e, 0xe4, 0x6c, 0x36,
-	0x42, 0x5b, 0x68, 0x99, 0xb3, 0xcc, 0x79, 0xb1, 0x71, 0xa1, 0xaf, 0x87, 0xa4, 0x6f, 0x35, 0xaa,
-	0x77, 0x55, 0x8f, 0xc1, 0xa5, 0xb7, 0xa4, 0x91, 0x00, 0xf9, 0x1b, 0x8a, 0x8c, 0x73, 0x47, 0xb2,
-	0x80, 0x29, 0xb4, 0x1d, 0xa5, 0xf7, 0xeb, 0xcb, 0xf4, 0x9a, 0xcf, 0x8d, 0xa1, 0xe8, 0xaa, 0x14,
-	0xd1, 0xe1, 0xcb, 0xd9, 0xb5, 0x18, 0x0a, 0xf4, 0x16, 0x96, 0x89, 0x43, 0x66, 0x9a, 0x8e, 0xcf,
-	0xa5, 0x96, 0x55, 0x97, 0x29, 0xcf, 0xae, 0xc5, 0x63, 0x88, 0x34, 0x43, 0x80, 0x48, 0xa8, 0x20,
-	0x5f, 0x58, 0x9e, 0xc3, 0xe7, 0xc8, 0xe5, 0x70, 0xc1, 0x3c, 0x8b, 0x8d, 0x6c, 0x14, 0x5a, 0x4e,
-	0x79, 0xfa, 0xe3, 0x65, 0x9e, 0x8c, 0x67, 0x89, 0xbf, 0x62, 0x85, 0xd0, 0xdc, 0x11, 0x6e, 0x81,
-	0xaa, 0x37, 0x50, 0x5c, 0x89, 0x85, 0x94, 0x20, 0x33, 0xc3, 0xa5, 0x9a, 0x5e, 0x81, 0x06, 0x3f,
-	0xc9, 0x11, 0x64, 0x17, 0xcc, 0xf6, 0xe3, 0x29, 0x84, 0x0f, 0xbf, 0xa5, 0xaf, 0x53, 0xd5, 0xdf,
-	0xa1, 0xb4, 0x99, 0xc0, 0xab, 0xfa, 0xdb, 0x70, 0xfc, 0x51, 0xb7, 0xaf, 0x11, 0xaa, 0xbd, 0x4b,
-	0x43, 0xf9, 0xce, 0x43, 0x26, 0x31, 0xc8, 0x84, 0xe2, 0x5b, 0x1f, 0x85, 0x24, 0x17, 0x90, 0xb3,
-	0xb8, 0xeb, 0x4b, 0x11, 0xed, 0xe2, 0xf1, 0xc6, 0x2e, 0x76, 0xc3, 0x37, 0xe7, 0x9e, 0xb9, 0x34,
-	0x22, 0x92, 0x2b, 0xc8, 0x4b, 0x9c, 0xbb, 0x36, 0x93, 0xe1, 0x29, 0xc5, 0xc6, 0xc9, 0x96, 0x05,
-	0xee, 0x47, 0x14, 0x9a, 0x90, 0xc9, 0xb7, 0xb0, 0xef, 0xf8, 0xd2, 0xf5, 0xe5, 0xd0, 0xf5, 0x70,
-	0x62, 0x3d, 0x69, 0x19, 0xe5, 0x71, 0x2f, 0x2c, 0x3e, 0xa8, 0x1a, 0xf9, 0x0f, 0xbe, 0xda, 0x78,
-	0x4f, 0xe6, 0xd1, 0xd4, 0xb4, 0x1d, 0x75, 0xd8, 0x77, 0x2f, 0x1a, 0x31, 0xad, 0xc8, 0x6d, 0xe5,
-	0xda, 0x0d, 0x90, 0xd5, 0x10, 0x84, 0xeb, 0x70, 0xa1, 0x9c, 0x79, 0x28, 0x1c, 0xdf, 0x33, 0x51,
-	0x1d, 0xa7, 0xc2, 0xd8, 0xa3, 0x7b, 0x71, 0x31, 0x68, 0xaf, 0x51, 0x38, 0x68, 0xa3, 0x5c, 0x0d,
-	0xef, 0x04, 0x0a, 0xca, 0xab, 0x5c, 0xba, 0x18, 0x0d, 0x21, 0x1f, 0x14, 0xfa, 0x4b, 0x77, 0x8b,
-	0x66, 0x7a, 0x8b, 0x66, 0x1b, 0x0e, 0x13, 0xcd, 0xc8, 0xcb, 0xcf, 0x90, 0x8f, 0x29, 0xd1, 0x4c,
-	0xb4, 0xcd, 0x1b, 0xd3, 0x08, 0xa7, 0x09, 0xb3, 0x66, 0x43, 0x3e, 0xae, 0x92, 0x1f, 0x21, 0x2b,
-	0x64, 0x30, 0x9d, 0xa0, 0xfd, 0xa0, 0x51, 0xd9, 0x6c, 0x7f, 0x0c, 0x40, 0x1a, 0x72, 0xc8, 0x25,
-	0xec, 0x86, 0xf9, 0x8b, 0x68, 0x98, 0x9f, 0xd8, 0x80, 0x98, 0x59, 0x1b, 0x40, 0xb9, 0x85, 0x36,
-	0xae, 0xaf, 0xd2, 0xe7, 0xa7, 0x71, 0x04, 0x64, 0x55, 0x36, 0x0c, 0xe4, 0x87, 0xff, 0x21, 0xab,
-	0x1c, 0x93, 0x0a, 0x94, 0xa9, 0xd1, 0xa7, 0xff, 0x34, 0x6f, 0xbb, 0xc6, 0xf0, 0xcf, 0x66, 0xa7,
-	0x3b, 0xa0, 0x46, 0xe9, 0x8b, 0xa0, 0xfc, 0x60, 0xd0, 0xfb, 0x66, 0xcf, 0xe8, 0xf5, 0x93, 0x72,
-	0x8a, 0x14, 0x61, 0xf7, 0xc1, 0xe8, 0xb5, 0x3a, 0xbd, 0x76, 0x29, 0x1d, 0x3c, 0xd0, 0x41, 0xaf,
-	0x17, 0x3c, 0x64, 0xc8, 0x3e, 0x14, 0x1e, 0x07, 0x77, 0x77, 0x86, 0xd1, 0x32, 0x5a, 0xa5, 0x9d,
-	0xdb, 0xab, 0x7f, 0x7f, 0x99, 0x5a, 0xf2, 0x8d, 0x3f, 0xd2, 0x4d, 0x67, 0x5e, 0x57, 0x97, 0x77,
-	0xbc, 0x69, 0x3d, 0xf9, 0xd2, 0x4f, 0x91, 0xd7, 0xdd, 0xd1, 0x4f, 0x53, 0xa7, 0xbe, 0xfe, 0x07,
-	0x34, 0xca, 0xa9, 0x6f, 0xfe, 0xe5, 0x87, 0x00, 0x00, 0x00, 0xff, 0xff, 0xe4, 0x4c, 0xa8, 0x65,
-	0x99, 0x06, 0x00, 0x00,
+	0x5b, 0xa9, 0x45, 0x83, 0xb9, 0x50, 0x0b, 0x33, 0x76, 0x3d, 0x63, 0x14, 0x9e, 0x57, 0xda, 0xdf,
+	0xb0, 0x3f, 0x77, 0xe5, 0xf1, 0x47, 0x00, 0xb1, 0xab, 0x44, 0xfb, 0xe6, 0xb9, 0xe7, 0xde, 0x33,
+	0x67, 0xce, 0x19, 0x6b, 0xa0, 0x3a, 0xb1, 0x97, 0x12, 0xad, 0xb1, 0x5d, 0x67, 0xe3, 0xb9, 0xc5,
+	0xeb, 0x6c, 0x8a, 0x5c, 0xea, 0xae, 0xe7, 0x48, 0x87, 0x1c, 0xc4, 0x98, 0xae, 0xb0, 0xea, 0x69,
+	0xd2, 0x6b, 0x3a, 0x1e, 0xd6, 0x6d, 0x4b, 0xa2, 0xc7, 0x6c, 0x11, 0x76, 0x57, 0x8f, 0xd7, 0x51,
+	0xc9, 0xc4, 0x2c, 0x86, 0xbe, 0x5c, 0x87, 0x2c, 0x2e, 0xd1, 0x9b, 0x30, 0x13, 0x23, 0xf8, 0xab,
+	0x0d, 0x78, 0x8c, 0x5c, 0x5a, 0x13, 0x0b, 0xbd, 0x10, 0xaf, 0xbd, 0xcb, 0x42, 0xa5, 0xcf, 0xc4,
+	0xcc, 0x78, 0x42, 0xd3, 0x97, 0x96, 0xc3, 0xef, 0x51, 0xb2, 0x31, 0x93, 0x8c, 0x50, 0x28, 0x07,
+	0xfb, 0x0c, 0x31, 0x46, 0x86, 0xd6, 0x58, 0x4b, 0x9d, 0xa5, 0xce, 0x8b, 0x8d, 0x6f, 0xf5, 0x44,
+	0x7d, 0xc0, 0xaa, 0xaf, 0x11, 0x74, 0x92, 0x2d, 0xe8, 0xa1, 0x5c, 0x07, 0xc8, 0x29, 0x14, 0x38,
+	0x9b, 0xa3, 0x70, 0x99, 0x89, 0x5a, 0xfa, 0x2c, 0x75, 0x5e, 0xa0, 0xcf, 0x05, 0xd2, 0x81, 0x9c,
+	0xcd, 0x46, 0x68, 0x0b, 0x2d, 0x73, 0x96, 0x39, 0x2f, 0x36, 0x2e, 0xf4, 0x75, 0x93, 0xf4, 0xad,
+	0x42, 0xf5, 0xae, 0x9a, 0x31, 0xb8, 0xf4, 0x96, 0x34, 0x22, 0x20, 0x7f, 0x42, 0x91, 0x71, 0xee,
+	0x48, 0x16, 0x74, 0x0a, 0x6d, 0x47, 0xf1, 0xfd, 0xfc, 0x32, 0xbe, 0xe6, 0xf3, 0x60, 0x48, 0xba,
+	0x4a, 0x45, 0x74, 0xf8, 0x7c, 0x76, 0x2d, 0x86, 0x02, 0xbd, 0x85, 0x65, 0xe2, 0x90, 0x99, 0xa6,
+	0xe3, 0x73, 0xa9, 0x65, 0xd5, 0x61, 0xca, 0xb3, 0x6b, 0xf1, 0x18, 0x22, 0xcd, 0x10, 0x20, 0x12,
+	0x2a, 0xc8, 0x17, 0x96, 0xe7, 0xf0, 0x39, 0x72, 0x39, 0x5c, 0x30, 0xcf, 0x62, 0x23, 0x1b, 0x85,
+	0x96, 0x53, 0x9a, 0x7e, 0x7b, 0x99, 0x26, 0xe3, 0x99, 0xe2, 0x8f, 0x98, 0x21, 0x14, 0x77, 0x84,
+	0x5b, 0xa0, 0xea, 0x0d, 0x14, 0x57, 0x6c, 0x21, 0x25, 0xc8, 0xcc, 0x70, 0xa9, 0xd2, 0x2b, 0xd0,
+	0xe0, 0x93, 0x1c, 0x41, 0x76, 0xc1, 0x6c, 0x3f, 0x4e, 0x21, 0x5c, 0xfc, 0x92, 0xbe, 0x4e, 0x55,
+	0x7f, 0x85, 0xd2, 0xa6, 0x03, 0xaf, 0x9a, 0x6f, 0xc3, 0xf1, 0x07, 0xd5, 0xbe, 0x86, 0xa8, 0xf6,
+	0x26, 0x0d, 0xe5, 0x3b, 0x0f, 0x99, 0xc4, 0xc0, 0x13, 0x8a, 0xff, 0xfb, 0x28, 0x24, 0xb9, 0x80,
+	0x9c, 0xc5, 0x5d, 0x5f, 0x8a, 0xe8, 0x2e, 0x1e, 0x6f, 0xdc, 0xc5, 0x6e, 0xf8, 0xe7, 0xdc, 0x33,
+	0x97, 0x46, 0x8d, 0xe4, 0x0a, 0xf2, 0x12, 0xe7, 0xae, 0xcd, 0x64, 0xb8, 0x4b, 0xb1, 0x71, 0xb2,
+	0xe5, 0x02, 0xf7, 0xa3, 0x16, 0x9a, 0x34, 0x93, 0xaf, 0x61, 0xdf, 0xf1, 0xa5, 0xeb, 0xcb, 0xa1,
+	0xeb, 0xe1, 0xc4, 0x7a, 0xd2, 0x32, 0x4a, 0xe3, 0x5e, 0x58, 0x7c, 0x50, 0x35, 0xf2, 0x0f, 0x7c,
+	0xb1, 0xf1, 0x9f, 0xcc, 0xa3, 0xd4, 0xb4, 0x1d, 0xb5, 0xd9, 0x37, 0x2f, 0x8a, 0x98, 0x56, 0xe4,
+	0xb6, 0x72, 0xed, 0x06, 0xc8, 0xaa, 0x09, 0xc2, 0x75, 0xb8, 0x50, 0xca, 0x3c, 0x14, 0x8e, 0xef,
+	0x99, 0xa8, 0xb6, 0x53, 0x66, 0xec, 0xd1, 0xbd, 0xb8, 0x18, 0x8c, 0xd7, 0x28, 0x1c, 0xb4, 0x51,
+	0xae, 0x9a, 0x77, 0x02, 0x05, 0xa5, 0x55, 0x2e, 0x5d, 0x8c, 0x42, 0xc8, 0x07, 0x85, 0xfe, 0xd2,
+	0xdd, 0xc2, 0x99, 0xde, 0xc2, 0xd9, 0x86, 0xc3, 0x84, 0x33, 0xd2, 0xf2, 0x23, 0xe4, 0xe3, 0x96,
+	0x28, 0x13, 0x6d, 0xf3, 0xc4, 0x34, 0xc2, 0x69, 0xd2, 0x59, 0x7b, 0x9b, 0x82, 0x7c, 0x5c, 0x26,
+	0xdf, 0x43, 0x56, 0xc8, 0x20, 0x9e, 0x60, 0xfe, 0xa0, 0x51, 0xd9, 0x9c, 0x7f, 0x0c, 0x40, 0x1a,
+	0xf6, 0x90, 0x4b, 0xd8, 0x0d, 0x03, 0x10, 0x51, 0x9a, 0x1f, 0xb9, 0x02, 0x71, 0x27, 0xd1, 0x60,
+	0x77, 0x8e, 0x42, 0xb0, 0x29, 0x46, 0x21, 0xc6, 0xcb, 0xda, 0x00, 0xca, 0x2d, 0xb4, 0x71, 0xfd,
+	0x96, 0x7d, 0xba, 0x51, 0x47, 0x40, 0x56, 0x69, 0x43, 0xaf, 0xbe, 0xfb, 0x17, 0xb2, 0xea, 0x2c,
+	0xa4, 0x02, 0x65, 0x6a, 0xf4, 0xe9, 0x5f, 0xcd, 0xdb, 0xae, 0x31, 0xfc, 0xbd, 0xd9, 0xe9, 0x0e,
+	0xa8, 0x51, 0xfa, 0x2c, 0x28, 0x3f, 0x18, 0xf4, 0xbe, 0xd9, 0x33, 0x7a, 0xfd, 0xa4, 0x9c, 0x22,
+	0x45, 0xd8, 0x7d, 0x30, 0x7a, 0xad, 0x4e, 0xaf, 0x5d, 0x4a, 0x07, 0x0b, 0x3a, 0xe8, 0xf5, 0x82,
+	0x45, 0x86, 0xec, 0x43, 0xe1, 0x71, 0x70, 0x77, 0x67, 0x18, 0x2d, 0xa3, 0x55, 0xda, 0xb9, 0xbd,
+	0xfa, 0xfb, 0xa7, 0xa9, 0x25, 0xff, 0xf3, 0x47, 0xba, 0xe9, 0xcc, 0xeb, 0xca, 0x16, 0xc7, 0x9b,
+	0xd6, 0x93, 0x47, 0x60, 0x8a, 0xbc, 0xee, 0x8e, 0x7e, 0x98, 0x3a, 0xf5, 0xf5, 0xb7, 0x69, 0x94,
+	0x53, 0xcf, 0xc1, 0xe5, 0xfb, 0x00, 0x00, 0x00, 0xff, 0xff, 0x07, 0x9f, 0x3f, 0x42, 0xb4, 0x06,
+	0x00, 0x00,
 }
diff --git a/gen/pb-go/flyteidl/admin/agent.pb.validate.go b/gen/pb-go/flyteidl/admin/agent.pb.validate.go
index 0ba579cb0..7efab6b18 100644
--- a/gen/pb-go/flyteidl/admin/agent.pb.validate.go
+++ b/gen/pb-go/flyteidl/admin/agent.pb.validate.go
@@ -454,6 +454,8 @@ func (m *Resource) Validate() error {
 		}
 	}
 
+	// no validation rules for Message
+
 	return nil
 }
 
diff --git a/gen/pb-go/flyteidl/service/agent.swagger.json b/gen/pb-go/flyteidl/service/agent.swagger.json
index 0a788c919..2d1d5ea07 100644
--- a/gen/pb-go/flyteidl/service/agent.swagger.json
+++ b/gen/pb-go/flyteidl/service/agent.swagger.json
@@ -197,6 +197,10 @@
         "outputs": {
           "$ref": "#/definitions/coreLiteralMap",
           "title": "The outputs of the execution. It's typically used by sql task. Agent service will create a\nStructured dataset pointing to the query result table.\n+optional"
+        },
+        "message": {
+          "type": "string",
+          "description": "A descriptive message for the current state. e.g. waiting for cluster."
         }
       }
     },
diff --git a/gen/pb-java/flyteidl/admin/Agent.java b/gen/pb-java/flyteidl/admin/Agent.java
index cefc77700..5172925eb 100644
--- a/gen/pb-java/flyteidl/admin/Agent.java
+++ b/gen/pb-java/flyteidl/admin/Agent.java
@@ -5460,6 +5460,24 @@ public interface ResourceOrBuilder extends
      * <code>.flyteidl.core.LiteralMap outputs = 2;</code>
      */
     flyteidl.core.Literals.LiteralMapOrBuilder getOutputsOrBuilder();
+
+    /**
+     * <pre>
+     * A descriptive message for the current state. e.g. waiting for cluster.
+     * </pre>
+     *
+     * <code>string message = 3;</code>
+     */
+    java.lang.String getMessage();
+    /**
+     * <pre>
+     * A descriptive message for the current state. e.g. waiting for cluster.
+     * </pre>
+     *
+     * <code>string message = 3;</code>
+     */
+    com.google.protobuf.ByteString
+        getMessageBytes();
   }
   /**
    * Protobuf type {@code flyteidl.admin.Resource}
@@ -5475,6 +5493,7 @@ private Resource(com.google.protobuf.GeneratedMessageV3.Builder<?> builder) {
     }
     private Resource() {
       state_ = 0;
+      message_ = "";
     }
 
     @java.lang.Override
@@ -5520,6 +5539,12 @@ private Resource(
 
               break;
             }
+            case 26: {
+              java.lang.String s = input.readStringRequireUtf8();
+
+              message_ = s;
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -5616,6 +5641,48 @@ public flyteidl.core.Literals.LiteralMapOrBuilder getOutputsOrBuilder() {
       return getOutputs();
     }
 
+    public static final int MESSAGE_FIELD_NUMBER = 3;
+    private volatile java.lang.Object message_;
+    /**
+     * <pre>
+     * A descriptive message for the current state. e.g. waiting for cluster.
+     * </pre>
+     *
+     * <code>string message = 3;</code>
+     */
+    public java.lang.String getMessage() {
+      java.lang.Object ref = message_;
+      if (ref instanceof java.lang.String) {
+        return (java.lang.String) ref;
+      } else {
+        com.google.protobuf.ByteString bs = 
+            (com.google.protobuf.ByteString) ref;
+        java.lang.String s = bs.toStringUtf8();
+        message_ = s;
+        return s;
+      }
+    }
+    /**
+     * <pre>
+     * A descriptive message for the current state. e.g. waiting for cluster.
+     * </pre>
+     *
+     * <code>string message = 3;</code>
+     */
+    public com.google.protobuf.ByteString
+        getMessageBytes() {
+      java.lang.Object ref = message_;
+      if (ref instanceof java.lang.String) {
+        com.google.protobuf.ByteString b = 
+            com.google.protobuf.ByteString.copyFromUtf8(
+                (java.lang.String) ref);
+        message_ = b;
+        return b;
+      } else {
+        return (com.google.protobuf.ByteString) ref;
+      }
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -5636,6 +5703,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
       if (outputs_ != null) {
         output.writeMessage(2, getOutputs());
       }
+      if (!getMessageBytes().isEmpty()) {
+        com.google.protobuf.GeneratedMessageV3.writeString(output, 3, message_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -5653,6 +5723,9 @@ public int getSerializedSize() {
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(2, getOutputs());
       }
+      if (!getMessageBytes().isEmpty()) {
+        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(3, message_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -5674,6 +5747,8 @@ public boolean equals(final java.lang.Object obj) {
         if (!getOutputs()
             .equals(other.getOutputs())) return false;
       }
+      if (!getMessage()
+          .equals(other.getMessage())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -5691,6 +5766,8 @@ public int hashCode() {
         hash = (37 * hash) + OUTPUTS_FIELD_NUMBER;
         hash = (53 * hash) + getOutputs().hashCode();
       }
+      hash = (37 * hash) + MESSAGE_FIELD_NUMBER;
+      hash = (53 * hash) + getMessage().hashCode();
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -5832,6 +5909,8 @@ public Builder clear() {
           outputs_ = null;
           outputsBuilder_ = null;
         }
+        message_ = "";
+
         return this;
       }
 
@@ -5864,6 +5943,7 @@ public flyteidl.admin.Agent.Resource buildPartial() {
         } else {
           result.outputs_ = outputsBuilder_.build();
         }
+        result.message_ = message_;
         onBuilt();
         return result;
       }
@@ -5918,6 +5998,10 @@ public Builder mergeFrom(flyteidl.admin.Agent.Resource other) {
         if (other.hasOutputs()) {
           mergeOutputs(other.getOutputs());
         }
+        if (!other.getMessage().isEmpty()) {
+          message_ = other.message_;
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -6182,6 +6266,95 @@ public flyteidl.core.Literals.LiteralMapOrBuilder getOutputsOrBuilder() {
         }
         return outputsBuilder_;
       }
+
+      private java.lang.Object message_ = "";
+      /**
+       * <pre>
+       * A descriptive message for the current state. e.g. waiting for cluster.
+       * </pre>
+       *
+       * <code>string message = 3;</code>
+       */
+      public java.lang.String getMessage() {
+        java.lang.Object ref = message_;
+        if (!(ref instanceof java.lang.String)) {
+          com.google.protobuf.ByteString bs =
+              (com.google.protobuf.ByteString) ref;
+          java.lang.String s = bs.toStringUtf8();
+          message_ = s;
+          return s;
+        } else {
+          return (java.lang.String) ref;
+        }
+      }
+      /**
+       * <pre>
+       * A descriptive message for the current state. e.g. waiting for cluster.
+       * </pre>
+       *
+       * <code>string message = 3;</code>
+       */
+      public com.google.protobuf.ByteString
+          getMessageBytes() {
+        java.lang.Object ref = message_;
+        if (ref instanceof String) {
+          com.google.protobuf.ByteString b = 
+              com.google.protobuf.ByteString.copyFromUtf8(
+                  (java.lang.String) ref);
+          message_ = b;
+          return b;
+        } else {
+          return (com.google.protobuf.ByteString) ref;
+        }
+      }
+      /**
+       * <pre>
+       * A descriptive message for the current state. e.g. waiting for cluster.
+       * </pre>
+       *
+       * <code>string message = 3;</code>
+       */
+      public Builder setMessage(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  
+        message_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * A descriptive message for the current state. e.g. waiting for cluster.
+       * </pre>
+       *
+       * <code>string message = 3;</code>
+       */
+      public Builder clearMessage() {
+        
+        message_ = getDefaultInstance().getMessage();
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * A descriptive message for the current state. e.g. waiting for cluster.
+       * </pre>
+       *
+       * <code>string message = 3;</code>
+       */
+      public Builder setMessageBytes(
+          com.google.protobuf.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  checkByteStringIsUtf8(value);
+        
+        message_ = value;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -7424,15 +7597,15 @@ public flyteidl.admin.Agent.DeleteTaskResponse getDefaultInstanceForType() {
       "\016GetTaskRequest\022\021\n\ttask_type\030\001 \001(\t\022\025\n\rre" +
       "source_meta\030\002 \001(\014\"=\n\017GetTaskResponse\022*\n\010" +
       "resource\030\001 \001(\0132\030.flyteidl.admin.Resource" +
-      "\"\\\n\010Resource\022$\n\005state\030\001 \001(\0162\025.flyteidl.a" +
+      "\"m\n\010Resource\022$\n\005state\030\001 \001(\0162\025.flyteidl.a" +
       "dmin.State\022*\n\007outputs\030\002 \001(\0132\031.flyteidl.c" +
-      "ore.LiteralMap\"=\n\021DeleteTaskRequest\022\021\n\tt" +
-      "ask_type\030\001 \001(\t\022\025\n\rresource_meta\030\002 \001(\014\"\024\n" +
-      "\022DeleteTaskResponse*^\n\005State\022\025\n\021RETRYABL" +
-      "E_FAILURE\020\000\022\025\n\021PERMANENT_FAILURE\020\001\022\013\n\007PE" +
-      "NDING\020\002\022\013\n\007RUNNING\020\003\022\r\n\tSUCCEEDED\020\004B7Z5g" +
-      "ithub.com/flyteorg/flyteidl/gen/pb-go/fl" +
-      "yteidl/adminb\006proto3"
+      "ore.LiteralMap\022\017\n\007message\030\003 \001(\t\"=\n\021Delet" +
+      "eTaskRequest\022\021\n\ttask_type\030\001 \001(\t\022\025\n\rresou" +
+      "rce_meta\030\002 \001(\014\"\024\n\022DeleteTaskResponse*^\n\005" +
+      "State\022\025\n\021RETRYABLE_FAILURE\020\000\022\025\n\021PERMANEN" +
+      "T_FAILURE\020\001\022\013\n\007PENDING\020\002\022\013\n\007RUNNING\020\003\022\r\n" +
+      "\tSUCCEEDED\020\004B7Z5github.com/flyteorg/flyt" +
+      "eidl/gen/pb-go/flyteidl/adminb\006proto3"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
         new com.google.protobuf.Descriptors.FileDescriptor.    InternalDescriptorAssigner() {
@@ -7503,7 +7676,7 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors(
     internal_static_flyteidl_admin_Resource_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_flyteidl_admin_Resource_descriptor,
-        new java.lang.String[] { "State", "Outputs", });
+        new java.lang.String[] { "State", "Outputs", "Message", });
     internal_static_flyteidl_admin_DeleteTaskRequest_descriptor =
       getDescriptor().getMessageTypes().get(6);
     internal_static_flyteidl_admin_DeleteTaskRequest_fieldAccessorTable = new
diff --git a/gen/pb-js/flyteidl.d.ts b/gen/pb-js/flyteidl.d.ts
index 7744acd5d..26845bc5d 100644
--- a/gen/pb-js/flyteidl.d.ts
+++ b/gen/pb-js/flyteidl.d.ts
@@ -7913,6 +7913,9 @@ export namespace flyteidl {
 
             /** Resource outputs */
             outputs?: (flyteidl.core.ILiteralMap|null);
+
+            /** Resource message */
+            message?: (string|null);
         }
 
         /** Represents a Resource. */
@@ -7930,6 +7933,9 @@ export namespace flyteidl {
             /** Resource outputs. */
             public outputs?: (flyteidl.core.ILiteralMap|null);
 
+            /** Resource message. */
+            public message: string;
+
             /**
              * Creates a new Resource instance using the specified properties.
              * @param [properties] Properties to set
diff --git a/gen/pb-js/flyteidl.js b/gen/pb-js/flyteidl.js
index 6a2ae8c72..4ffdefc8e 100644
--- a/gen/pb-js/flyteidl.js
+++ b/gen/pb-js/flyteidl.js
@@ -19292,6 +19292,7 @@
                  * @interface IResource
                  * @property {flyteidl.admin.State|null} [state] Resource state
                  * @property {flyteidl.core.ILiteralMap|null} [outputs] Resource outputs
+                 * @property {string|null} [message] Resource message
                  */
     
                 /**
@@ -19325,6 +19326,14 @@
                  */
                 Resource.prototype.outputs = null;
     
+                /**
+                 * Resource message.
+                 * @member {string} message
+                 * @memberof flyteidl.admin.Resource
+                 * @instance
+                 */
+                Resource.prototype.message = "";
+    
                 /**
                  * Creates a new Resource instance using the specified properties.
                  * @function create
@@ -19353,6 +19362,8 @@
                         writer.uint32(/* id 1, wireType 0 =*/8).int32(message.state);
                     if (message.outputs != null && message.hasOwnProperty("outputs"))
                         $root.flyteidl.core.LiteralMap.encode(message.outputs, writer.uint32(/* id 2, wireType 2 =*/18).fork()).ldelim();
+                    if (message.message != null && message.hasOwnProperty("message"))
+                        writer.uint32(/* id 3, wireType 2 =*/26).string(message.message);
                     return writer;
                 };
     
@@ -19380,6 +19391,9 @@
                         case 2:
                             message.outputs = $root.flyteidl.core.LiteralMap.decode(reader, reader.uint32());
                             break;
+                        case 3:
+                            message.message = reader.string();
+                            break;
                         default:
                             reader.skipType(tag & 7);
                             break;
@@ -19415,6 +19429,9 @@
                         if (error)
                             return "outputs." + error;
                     }
+                    if (message.message != null && message.hasOwnProperty("message"))
+                        if (!$util.isString(message.message))
+                            return "message: string expected";
                     return null;
                 };
     
diff --git a/gen/pb_python/flyteidl/admin/agent_pb2.py b/gen/pb_python/flyteidl/admin/agent_pb2.py
index ea0ef0afe..f74b5d3ba 100644
--- a/gen/pb_python/flyteidl/admin/agent_pb2.py
+++ b/gen/pb_python/flyteidl/admin/agent_pb2.py
@@ -17,7 +17,7 @@
 from flyteidl.core import identifier_pb2 as flyteidl_dot_core_dot_identifier__pb2
 
 
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x66lyteidl/admin/agent.proto\x12\x0e\x66lyteidl.admin\x1a\x1c\x66lyteidl/core/literals.proto\x1a\x19\x66lyteidl/core/tasks.proto\x1a\x1d\x66lyteidl/core/interface.proto\x1a\x1e\x66lyteidl/core/identifier.proto\"\x98\x05\n\x15TaskExecutionMetadata\x12R\n\x11task_execution_id\x18\x01 \x01(\x0b\x32&.flyteidl.core.TaskExecutionIdentifierR\x0ftaskExecutionId\x12\x1c\n\tnamespace\x18\x02 \x01(\tR\tnamespace\x12I\n\x06labels\x18\x03 \x03(\x0b\x32\x31.flyteidl.admin.TaskExecutionMetadata.LabelsEntryR\x06labels\x12X\n\x0b\x61nnotations\x18\x04 \x03(\x0b\x32\x36.flyteidl.admin.TaskExecutionMetadata.AnnotationsEntryR\x0b\x61nnotations\x12.\n\x13k8s_service_account\x18\x05 \x01(\tR\x11k8sServiceAccount\x12t\n\x15\x65nvironment_variables\x18\x06 \x03(\x0b\x32?.flyteidl.admin.TaskExecutionMetadata.EnvironmentVariablesEntryR\x14\x65nvironmentVariables\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1a>\n\x10\x41nnotationsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1aG\n\x19\x45nvironmentVariablesEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\x83\x02\n\x11\x43reateTaskRequest\x12\x31\n\x06inputs\x18\x01 \x01(\x0b\x32\x19.flyteidl.core.LiteralMapR\x06inputs\x12\x37\n\x08template\x18\x02 \x01(\x0b\x32\x1b.flyteidl.core.TaskTemplateR\x08template\x12#\n\routput_prefix\x18\x03 \x01(\tR\x0coutputPrefix\x12]\n\x17task_execution_metadata\x18\x04 \x01(\x0b\x32%.flyteidl.admin.TaskExecutionMetadataR\x15taskExecutionMetadata\"9\n\x12\x43reateTaskResponse\x12#\n\rresource_meta\x18\x01 \x01(\x0cR\x0cresourceMeta\"R\n\x0eGetTaskRequest\x12\x1b\n\ttask_type\x18\x01 \x01(\tR\x08taskType\x12#\n\rresource_meta\x18\x02 \x01(\x0cR\x0cresourceMeta\"G\n\x0fGetTaskResponse\x12\x34\n\x08resource\x18\x01 \x01(\x0b\x32\x18.flyteidl.admin.ResourceR\x08resource\"l\n\x08Resource\x12+\n\x05state\x18\x01 \x01(\x0e\x32\x15.flyteidl.admin.StateR\x05state\x12\x33\n\x07outputs\x18\x02 \x01(\x0b\x32\x19.flyteidl.core.LiteralMapR\x07outputs\"U\n\x11\x44\x65leteTaskRequest\x12\x1b\n\ttask_type\x18\x01 \x01(\tR\x08taskType\x12#\n\rresource_meta\x18\x02 \x01(\x0cR\x0cresourceMeta\"\x14\n\x12\x44\x65leteTaskResponse*^\n\x05State\x12\x15\n\x11RETRYABLE_FAILURE\x10\x00\x12\x15\n\x11PERMANENT_FAILURE\x10\x01\x12\x0b\n\x07PENDING\x10\x02\x12\x0b\n\x07RUNNING\x10\x03\x12\r\n\tSUCCEEDED\x10\x04\x42\xb0\x01\n\x12\x63om.flyteidl.adminB\nAgentProtoP\x01Z5github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin\xa2\x02\x03\x46\x41X\xaa\x02\x0e\x46lyteidl.Admin\xca\x02\x0e\x46lyteidl\\Admin\xe2\x02\x1a\x46lyteidl\\Admin\\GPBMetadata\xea\x02\x0f\x46lyteidl::Adminb\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x66lyteidl/admin/agent.proto\x12\x0e\x66lyteidl.admin\x1a\x1c\x66lyteidl/core/literals.proto\x1a\x19\x66lyteidl/core/tasks.proto\x1a\x1d\x66lyteidl/core/interface.proto\x1a\x1e\x66lyteidl/core/identifier.proto\"\x98\x05\n\x15TaskExecutionMetadata\x12R\n\x11task_execution_id\x18\x01 \x01(\x0b\x32&.flyteidl.core.TaskExecutionIdentifierR\x0ftaskExecutionId\x12\x1c\n\tnamespace\x18\x02 \x01(\tR\tnamespace\x12I\n\x06labels\x18\x03 \x03(\x0b\x32\x31.flyteidl.admin.TaskExecutionMetadata.LabelsEntryR\x06labels\x12X\n\x0b\x61nnotations\x18\x04 \x03(\x0b\x32\x36.flyteidl.admin.TaskExecutionMetadata.AnnotationsEntryR\x0b\x61nnotations\x12.\n\x13k8s_service_account\x18\x05 \x01(\tR\x11k8sServiceAccount\x12t\n\x15\x65nvironment_variables\x18\x06 \x03(\x0b\x32?.flyteidl.admin.TaskExecutionMetadata.EnvironmentVariablesEntryR\x14\x65nvironmentVariables\x1a\x39\n\x0bLabelsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1a>\n\x10\x41nnotationsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x1aG\n\x19\x45nvironmentVariablesEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\x83\x02\n\x11\x43reateTaskRequest\x12\x31\n\x06inputs\x18\x01 \x01(\x0b\x32\x19.flyteidl.core.LiteralMapR\x06inputs\x12\x37\n\x08template\x18\x02 \x01(\x0b\x32\x1b.flyteidl.core.TaskTemplateR\x08template\x12#\n\routput_prefix\x18\x03 \x01(\tR\x0coutputPrefix\x12]\n\x17task_execution_metadata\x18\x04 \x01(\x0b\x32%.flyteidl.admin.TaskExecutionMetadataR\x15taskExecutionMetadata\"9\n\x12\x43reateTaskResponse\x12#\n\rresource_meta\x18\x01 \x01(\x0cR\x0cresourceMeta\"R\n\x0eGetTaskRequest\x12\x1b\n\ttask_type\x18\x01 \x01(\tR\x08taskType\x12#\n\rresource_meta\x18\x02 \x01(\x0cR\x0cresourceMeta\"G\n\x0fGetTaskResponse\x12\x34\n\x08resource\x18\x01 \x01(\x0b\x32\x18.flyteidl.admin.ResourceR\x08resource\"\x86\x01\n\x08Resource\x12+\n\x05state\x18\x01 \x01(\x0e\x32\x15.flyteidl.admin.StateR\x05state\x12\x33\n\x07outputs\x18\x02 \x01(\x0b\x32\x19.flyteidl.core.LiteralMapR\x07outputs\x12\x18\n\x07message\x18\x03 \x01(\tR\x07message\"U\n\x11\x44\x65leteTaskRequest\x12\x1b\n\ttask_type\x18\x01 \x01(\tR\x08taskType\x12#\n\rresource_meta\x18\x02 \x01(\x0cR\x0cresourceMeta\"\x14\n\x12\x44\x65leteTaskResponse*^\n\x05State\x12\x15\n\x11RETRYABLE_FAILURE\x10\x00\x12\x15\n\x11PERMANENT_FAILURE\x10\x01\x12\x0b\n\x07PENDING\x10\x02\x12\x0b\n\x07RUNNING\x10\x03\x12\r\n\tSUCCEEDED\x10\x04\x42\xb0\x01\n\x12\x63om.flyteidl.adminB\nAgentProtoP\x01Z5github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin\xa2\x02\x03\x46\x41X\xaa\x02\x0e\x46lyteidl.Admin\xca\x02\x0e\x46lyteidl\\Admin\xe2\x02\x1a\x46lyteidl\\Admin\\GPBMetadata\xea\x02\x0f\x46lyteidl::Adminb\x06proto3')
 
 _globals = globals()
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
@@ -32,8 +32,8 @@
   _TASKEXECUTIONMETADATA_ANNOTATIONSENTRY._serialized_options = b'8\001'
   _TASKEXECUTIONMETADATA_ENVIRONMENTVARIABLESENTRY._options = None
   _TASKEXECUTIONMETADATA_ENVIRONMENTVARIABLESENTRY._serialized_options = b'8\001'
-  _globals['_STATE']._serialized_start=1530
-  _globals['_STATE']._serialized_end=1624
+  _globals['_STATE']._serialized_start=1557
+  _globals['_STATE']._serialized_end=1651
   _globals['_TASKEXECUTIONMETADATA']._serialized_start=167
   _globals['_TASKEXECUTIONMETADATA']._serialized_end=831
   _globals['_TASKEXECUTIONMETADATA_LABELSENTRY']._serialized_start=637
@@ -50,10 +50,10 @@
   _globals['_GETTASKREQUEST']._serialized_end=1236
   _globals['_GETTASKRESPONSE']._serialized_start=1238
   _globals['_GETTASKRESPONSE']._serialized_end=1309
-  _globals['_RESOURCE']._serialized_start=1311
-  _globals['_RESOURCE']._serialized_end=1419
-  _globals['_DELETETASKREQUEST']._serialized_start=1421
-  _globals['_DELETETASKREQUEST']._serialized_end=1506
-  _globals['_DELETETASKRESPONSE']._serialized_start=1508
-  _globals['_DELETETASKRESPONSE']._serialized_end=1528
+  _globals['_RESOURCE']._serialized_start=1312
+  _globals['_RESOURCE']._serialized_end=1446
+  _globals['_DELETETASKREQUEST']._serialized_start=1448
+  _globals['_DELETETASKREQUEST']._serialized_end=1533
+  _globals['_DELETETASKRESPONSE']._serialized_start=1535
+  _globals['_DELETETASKRESPONSE']._serialized_end=1555
 # @@protoc_insertion_point(module_scope)
diff --git a/gen/pb_python/flyteidl/admin/agent_pb2.pyi b/gen/pb_python/flyteidl/admin/agent_pb2.pyi
index 830158d0b..6ca7ba23a 100644
--- a/gen/pb_python/flyteidl/admin/agent_pb2.pyi
+++ b/gen/pb_python/flyteidl/admin/agent_pb2.pyi
@@ -93,12 +93,14 @@ class GetTaskResponse(_message.Message):
     def __init__(self, resource: _Optional[_Union[Resource, _Mapping]] = ...) -> None: ...
 
 class Resource(_message.Message):
-    __slots__ = ["state", "outputs"]
+    __slots__ = ["state", "outputs", "message"]
     STATE_FIELD_NUMBER: _ClassVar[int]
     OUTPUTS_FIELD_NUMBER: _ClassVar[int]
+    MESSAGE_FIELD_NUMBER: _ClassVar[int]
     state: State
     outputs: _literals_pb2.LiteralMap
-    def __init__(self, state: _Optional[_Union[State, str]] = ..., outputs: _Optional[_Union[_literals_pb2.LiteralMap, _Mapping]] = ...) -> None: ...
+    message: str
+    def __init__(self, state: _Optional[_Union[State, str]] = ..., outputs: _Optional[_Union[_literals_pb2.LiteralMap, _Mapping]] = ..., message: _Optional[str] = ...) -> None: ...
 
 class DeleteTaskRequest(_message.Message):
     __slots__ = ["task_type", "resource_meta"]
diff --git a/gen/pb_rust/flyteidl.admin.rs b/gen/pb_rust/flyteidl.admin.rs
index a6acaa927..947fd4abb 100644
--- a/gen/pb_rust/flyteidl.admin.rs
+++ b/gen/pb_rust/flyteidl.admin.rs
@@ -78,6 +78,9 @@ pub struct Resource {
     /// +optional
     #[prost(message, optional, tag="2")]
     pub outputs: ::core::option::Option<super::core::LiteralMap>,
+    /// A descriptive message for the current state. e.g. waiting for cluster.
+    #[prost(string, tag="3")]
+    pub message: ::prost::alloc::string::String,
 }
 /// A message used to delete a task.
 #[allow(clippy::derive_partial_eq_without_eq)]
diff --git a/protos/flyteidl/admin/agent.proto b/protos/flyteidl/admin/agent.proto
index e9a484917..db94ddc02 100644
--- a/protos/flyteidl/admin/agent.proto
+++ b/protos/flyteidl/admin/agent.proto
@@ -73,6 +73,8 @@ message Resource {
   // Structured dataset pointing to the query result table.
   // +optional
   core.LiteralMap outputs = 2;
+  // A descriptive message for the current state. e.g. waiting for cluster.
+  string message = 3;
 }
 
 // A message used to delete a task.