[BUG] Cache doesn't work when running entities by FlyteRemote

[peterkun23]

Describe the bug

Cache hit never happens when running consecutively the following toy example:

from flytekit import task, HashMethod
from typing_extensions import Annotated
from flytekit.remote import FlyteRemote
from flytekit.configuration import Config, PlatformConfig

class VideoRecord:
    def __init__(self, video_path: str):
        self.video_path = video_path

def hash_video_record(record: VideoRecord) -> str:
    return record.video_path

@task(cache=True, cache_version="1.0")
def bar_1(video_record: Annotated[VideoRecord, HashMethod(hash_video_record)]) -> str:
    print("Running bar_1")
    return video_record.video_path

if __name__ == "__main__":
    video_record = VideoRecord("path/to/video")

    remote = FlyteRemote(
        config=Config(
            platform=PlatformConfig(
                endpoint=endpoint,
                insecure=True,
                insecure_skip_verify=True,
            )
        ),
        default_project=default_project,
        default_domain=default_domain,
    )
    entity = remote.fetch_task(name="toy_example.bar_1", version="1.1")
    remote.execute(
        entity=entity,
        inputs={"video_record": video_record},
        wait=True,
        tags=[],
        overwrite_cache=False,
    )

Expected behavior

I would expect a way to make caching work for this use case (either to implement something in the TypeTransforms -FlytePickleTransformerin this case or something else). I believe the problem is that after calling guess_python_type the HashMethod information is already gone.

Additional context to reproduce

No response

Screenshots

No response

Are you sure this issue hasn't been raised already?

• [X] Yes

Have you read the Code of Conduct?

• [X] Yes

[welcome[bot]]

Thank you for opening your first issue here! 🛠