From e418d7c588bc2215db0c13b9be6cdc6a3a7b8ade Mon Sep 17 00:00:00 2001
From: machichima <nary12321@gmail.com>
Date: Tue, 23 Jun 2026 16:24:46 +0800
Subject: [PATCH 1/4] feat: add sleep plugins and examples

Signed-off-by: machichima <nary12321@gmail.com>
---
 .../flytekit-sleep/examples/sleep_example.py  | 40 +++++++++++
 .../flytekit-sleep/examples/sleep_fanout.py   | 49 +++++++++++++
 .../flytekitplugins/__init__.py               |  0
 .../flytekitplugins/sleep/__init__.py         | 11 +++
 .../flytekitplugins/sleep/task.py             | 46 +++++++++++++
 plugins/flytekit-sleep/setup.py               | 36 ++++++++++
 plugins/flytekit-sleep/tests/__init__.py      |  0
 plugins/flytekit-sleep/tests/test_sleep.py    | 69 +++++++++++++++++++
 8 files changed, 251 insertions(+)
 create mode 100644 plugins/flytekit-sleep/examples/sleep_example.py
 create mode 100644 plugins/flytekit-sleep/examples/sleep_fanout.py
 create mode 100644 plugins/flytekit-sleep/flytekitplugins/__init__.py
 create mode 100644 plugins/flytekit-sleep/flytekitplugins/sleep/__init__.py
 create mode 100644 plugins/flytekit-sleep/flytekitplugins/sleep/task.py
 create mode 100644 plugins/flytekit-sleep/setup.py
 create mode 100644 plugins/flytekit-sleep/tests/__init__.py
 create mode 100644 plugins/flytekit-sleep/tests/test_sleep.py

diff --git a/plugins/flytekit-sleep/examples/sleep_example.py b/plugins/flytekit-sleep/examples/sleep_example.py
new file mode 100644
index 0000000000..e327841c41
--- /dev/null
+++ b/plugins/flytekit-sleep/examples/sleep_example.py
@@ -0,0 +1,40 @@
+"""
+Sleep Plugin Example
+====================
+
+The ``core-sleep`` plugin executes entirely in the backend — no task pod is created.
+The sleep duration is a normal task input, so it can be a dynamic workflow value.
+"""
+
+import os
+from datetime import timedelta
+
+from flytekitplugins.sleep import Sleep
+
+from flytekit import task, workflow
+
+
+@task(
+    cache_version="2",
+    task_config=Sleep()
+)
+def sleep_for(duration: timedelta) -> None:
+    # This body only runs during local execution.
+    # On the cluster, the backend sleeps for `duration` without running this.
+    print(f"[local] sleeping for {duration}")
+
+
+@workflow
+def wf(duration: timedelta) -> None:
+    sleep_for(duration=duration)
+
+
+if __name__ == "__main__":
+    from click.testing import CliRunner
+
+    from flytekit.clis.sdk_in_container import pyflyte
+
+    runner = CliRunner()
+    path = os.path.realpath(__file__)
+    result = runner.invoke(pyflyte.main, ["--config", os.path.expanduser("~/.flyte/config-sandbox.yaml"), "run", "--remote", path, "wf", "--duration", "10s"])
+    print("Remote Execution: ", result.output)
diff --git a/plugins/flytekit-sleep/examples/sleep_fanout.py b/plugins/flytekit-sleep/examples/sleep_fanout.py
new file mode 100644
index 0000000000..78f55f49aa
--- /dev/null
+++ b/plugins/flytekit-sleep/examples/sleep_fanout.py
@@ -0,0 +1,49 @@
+"""
+Sleep Fanout Example
+====================
+
+Fan out N core-sleep leaves in parallel using map_task.
+No task pods are created for the leaves — the backend handles sleep directly.
+
+Usage (remote):
+    python sleep_fanout.py
+"""
+
+import os
+from datetime import timedelta
+
+from flytekitplugins.sleep import Sleep
+
+from flytekit import task, workflow
+
+N_CHILDREN = 20
+
+
+@task(task_config=Sleep())
+def sleep_leaf(duration: timedelta) -> None:
+    pass
+
+
+@workflow
+def wf(sleep_duration: timedelta = timedelta(seconds=10)) -> None:
+    for _ in range(N_CHILDREN):
+        sleep_leaf(duration=sleep_duration)
+
+
+if __name__ == "__main__":
+    from click.testing import CliRunner
+
+    from flytekit.clis.sdk_in_container import pyflyte
+
+    runner = CliRunner()
+    path = os.path.realpath(__file__)
+
+    result = runner.invoke(pyflyte.main, [
+        "--config", os.path.expanduser("~/.flyte/config-sandbox.yaml"),
+        "run", "--remote",
+        path, "wf",
+        "--sleep_duration", "10s",
+    ])
+    print(result.output)
+    if result.exception:
+        raise result.exception
diff --git a/plugins/flytekit-sleep/flytekitplugins/__init__.py b/plugins/flytekit-sleep/flytekitplugins/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/plugins/flytekit-sleep/flytekitplugins/sleep/__init__.py b/plugins/flytekit-sleep/flytekitplugins/sleep/__init__.py
new file mode 100644
index 0000000000..7be9202a64
--- /dev/null
+++ b/plugins/flytekit-sleep/flytekitplugins/sleep/__init__.py
@@ -0,0 +1,11 @@
+"""
+.. currentmodule:: flytekitplugins.sleep
+
+.. autosummary::
+   :template: custom.rst
+   :toctree: generated/
+
+   Sleep
+"""
+
+from .task import Sleep
diff --git a/plugins/flytekit-sleep/flytekitplugins/sleep/task.py b/plugins/flytekit-sleep/flytekitplugins/sleep/task.py
new file mode 100644
index 0000000000..254a2b88c7
--- /dev/null
+++ b/plugins/flytekit-sleep/flytekitplugins/sleep/task.py
@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, Optional
+
+from flytekit.configuration import SerializationSettings
+from flytekit.core.python_function_task import PythonFunctionTask
+from flytekit.core.task import TaskPlugins
+
+
+@dataclass
+class Sleep:
+    """
+    Route a task to the backend ``core-sleep`` plugin.
+
+    The sleep duration is provided as a normal task input, not plugin config.
+    No container is launched; the backend handles the sleep directly.
+
+    Usage::
+
+        from flytekitplugins.sleep import Sleep
+        from flytekit import task
+        from datetime import timedelta
+
+        @task(task_config=Sleep())
+        def sleep_for(duration: timedelta) -> None:
+            pass  # only runs locally; backend executes the sleep
+    """
+
+
+class SleepFunctionTask(PythonFunctionTask[Sleep]):
+    _TASK_TYPE = "core-sleep"
+
+    def __init__(self, task_config: Optional[Sleep], task_function: Callable, **kwargs):
+        super().__init__(
+            task_config=task_config or Sleep(),
+            task_function=task_function,
+            task_type=self._TASK_TYPE,
+            **kwargs,
+        )
+
+    def get_custom(self, settings: SerializationSettings) -> Dict[str, Any]:
+        return {}
+
+
+TaskPlugins.register_pythontask_plugin(Sleep, SleepFunctionTask)
diff --git a/plugins/flytekit-sleep/setup.py b/plugins/flytekit-sleep/setup.py
new file mode 100644
index 0000000000..c4810fa68c
--- /dev/null
+++ b/plugins/flytekit-sleep/setup.py
@@ -0,0 +1,36 @@
+from setuptools import setup
+
+PLUGIN_NAME = "sleep"
+
+microlib_name = f"flytekitplugins-{PLUGIN_NAME}"
+
+__version__ = "0.0.0+develop"
+
+setup(
+    title="Sleep",
+    title_expanded="Flytekit Sleep Plugin",
+    name=microlib_name,
+    version=__version__,
+    author="flyteorg",
+    author_email="admin@flyte.org",
+    description="This package holds the core-sleep plugin for flytekit",
+    namespace_packages=["flytekitplugins"],
+    packages=[f"flytekitplugins.{PLUGIN_NAME}"],
+    install_requires=["flytekit>=1.15.1"],
+    license="apache2",
+    python_requires=">=3.9",
+    classifiers=[
+        "Intended Audience :: Science/Research",
+        "Intended Audience :: Developers",
+        "License :: OSI Approved :: Apache Software License",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Topic :: Scientific/Engineering",
+        "Topic :: Scientific/Engineering :: Artificial Intelligence",
+        "Topic :: Software Development",
+        "Topic :: Software Development :: Libraries",
+        "Topic :: Software Development :: Libraries :: Python Modules",
+    ],
+    entry_points={"flytekit.plugins": [f"{PLUGIN_NAME}=flytekitplugins.{PLUGIN_NAME}"]},
+)
diff --git a/plugins/flytekit-sleep/tests/__init__.py b/plugins/flytekit-sleep/tests/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/plugins/flytekit-sleep/tests/test_sleep.py b/plugins/flytekit-sleep/tests/test_sleep.py
new file mode 100644
index 0000000000..ea320b1f45
--- /dev/null
+++ b/plugins/flytekit-sleep/tests/test_sleep.py
@@ -0,0 +1,69 @@
+from collections import OrderedDict
+from datetime import timedelta
+
+from flytekitplugins.sleep import Sleep
+from flytekitplugins.sleep.task import SleepFunctionTask
+
+from flytekit import task, workflow
+from flytekit.configuration import Image, ImageConfig, SerializationSettings
+from flytekit.extend import get_serializable
+
+default_img = Image(name="default", fqn="test", tag="tag")
+serialization_settings = SerializationSettings(
+    project="proj",
+    domain="dom",
+    version="123",
+    image_config=ImageConfig(default_image=default_img, images=[default_img]),
+    env={},
+)
+
+
+def test_task_type():
+    @task(task_config=Sleep())
+    def sleep_for(duration: timedelta) -> None:
+        pass
+
+    assert isinstance(sleep_for, SleepFunctionTask)
+    assert sleep_for.task_type == "core-sleep"
+
+
+def test_serialization():
+    @task(task_config=Sleep())
+    def sleep_for(duration: timedelta) -> None:
+        pass
+
+    task_spec = get_serializable(OrderedDict(), serialization_settings, sleep_for)
+
+    assert task_spec.template.type == "core-sleep"
+    assert task_spec.template.custom == {}
+
+    inputs = task_spec.template.interface.inputs
+    assert "duration" in inputs
+    assert len(inputs) == 1
+
+    outputs = task_spec.template.interface.outputs
+    assert len(outputs) == 0
+
+
+def test_local_execution_calls_function():
+    called = []
+
+    @task(task_config=Sleep())
+    def sleep_for(duration: timedelta) -> None:
+        called.append(duration)
+
+    sleep_for(duration=timedelta(seconds=1))
+    assert called == [timedelta(seconds=1)]
+
+
+def test_workflow_integration():
+    @task(task_config=Sleep())
+    def sleep_for(duration: timedelta) -> None:
+        pass
+
+    @workflow
+    def wf(duration: timedelta) -> None:
+        sleep_for(duration=duration)
+
+    spec = get_serializable(OrderedDict(), serialization_settings, wf)
+    assert len(spec.template.nodes) == 1

From bc64e9473fc6499da38ea3134e6922a23c0846a7 Mon Sep 17 00:00:00 2001
From: machichima <nary12321@gmail.com>
Date: Tue, 23 Jun 2026 17:43:14 +0800
Subject: [PATCH 2/4] example: update example

Signed-off-by: machichima <nary12321@gmail.com>
---
 .../flytekit-sleep/examples/sleep_fanout.py   | 26 ++++++++++++++-----
 1 file changed, 20 insertions(+), 6 deletions(-)

diff --git a/plugins/flytekit-sleep/examples/sleep_fanout.py b/plugins/flytekit-sleep/examples/sleep_fanout.py
index 78f55f49aa..095b690090 100644
--- a/plugins/flytekit-sleep/examples/sleep_fanout.py
+++ b/plugins/flytekit-sleep/examples/sleep_fanout.py
@@ -11,23 +11,36 @@
 
 import os
 from datetime import timedelta
+from typing import List
 
+import flytekit as fl
 from flytekitplugins.sleep import Sleep
 
-from flytekit import task, workflow
+from flytekit import map_task, task, workflow
 
-N_CHILDREN = 20
+sleep_image = fl.ImageSpec(
+    registry="ghcr.io/machichima",
+    name="sleep-fanout",
+    apt_packages=["git"],
+    packages=["git+https://github.com/machichima/flytekit.git@add-sleep-plugin#subdirectory=plugins/flytekit-sleep"],
+    env={"REBUILD": "1"},
+)
 
 
-@task(task_config=Sleep())
+@task(container_image=sleep_image)
+def make_durations(duration: timedelta, n: int) -> List[timedelta]:
+    return [duration] * n
+
+
+@task(task_config=Sleep(), container_image=sleep_image)
 def sleep_leaf(duration: timedelta) -> None:
     pass
 
 
 @workflow
-def wf(sleep_duration: timedelta = timedelta(seconds=10)) -> None:
-    for _ in range(N_CHILDREN):
-        sleep_leaf(duration=sleep_duration)
+def wf(sleep_duration: timedelta = timedelta(seconds=10), n_children: int = 400) -> None:
+    durations = make_durations(duration=sleep_duration, n=n_children)
+    map_task(sleep_leaf)(duration=durations)
 
 
 if __name__ == "__main__":
@@ -43,6 +56,7 @@ def wf(sleep_duration: timedelta = timedelta(seconds=10)) -> None:
         "run", "--remote",
         path, "wf",
         "--sleep_duration", "10s",
+        "--n_children", "400",
     ])
     print(result.output)
     if result.exception:

From aa7d666c84c8840c02103ed5a894b187f29dcdb2 Mon Sep 17 00:00:00 2001
From: machichima <nary12321@gmail.com>
Date: Tue, 23 Jun 2026 17:51:55 +0800
Subject: [PATCH 3/4] example: add dynamic task example

Signed-off-by: machichima <nary12321@gmail.com>
---
 .../examples/sleep_fanout_dynamic.py          | 68 +++++++++++++++++++
 1 file changed, 68 insertions(+)
 create mode 100644 plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py

diff --git a/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py b/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py
new file mode 100644
index 0000000000..b67de7e26e
--- /dev/null
+++ b/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py
@@ -0,0 +1,68 @@
+"""
+Sleep Fanout (Dynamic) Example
+===============================
+
+Fan out N core-sleep leaves in parallel using @dynamic.
+n_children is a runtime input — the @dynamic task runs in a container to
+expand the subworkflow, then each leaf runs as core-sleep (no pod).
+
+Requires flytekitplugins-sleep in the container image. The ImageSpec below
+installs it from the local wheel via copy + pip install.
+
+Usage (remote):
+    python sleep_fanout_dynamic.py
+"""
+
+import os
+from datetime import timedelta
+
+import flytekit as fl
+from flytekitplugins.sleep import Sleep
+
+from flytekit import dynamic, task, workflow
+
+fanout_image = fl.ImageSpec(
+    registry="ghcr.io/machichima",
+    name="sleep-fanout",
+    apt_packages=["git"],
+    packages=["git+https://github.com/machichima/flytekit.git@add-sleep-plugin#subdirectory=plugins/flytekit-sleep"],
+)
+
+
+@task(task_config=Sleep())
+def sleep_leaf(duration: timedelta) -> None:
+    pass
+
+
+@dynamic(container_image=fanout_image)
+def sleep_fanout(n_children: int, sleep_duration: timedelta) -> None:
+    for _ in range(n_children):
+        sleep_leaf(duration=sleep_duration)
+
+
+@workflow
+def wf(
+    n_children: int = 10,
+    sleep_duration: timedelta = timedelta(seconds=10),
+) -> None:
+    sleep_fanout(n_children=n_children, sleep_duration=sleep_duration)
+
+
+if __name__ == "__main__":
+    from click.testing import CliRunner
+
+    from flytekit.clis.sdk_in_container import pyflyte
+
+    runner = CliRunner()
+    path = os.path.realpath(__file__)
+
+    result = runner.invoke(pyflyte.main, [
+        "--config", os.path.expanduser("~/.flyte/config-sandbox.yaml"),
+        "run", "--remote",
+        path, "wf",
+        "--n_children", "400",
+        "--sleep_duration", "10s",
+    ])
+    print(result.output)
+    if result.exception:
+        raise result.exception

From 05d9ebbed60566d5f8a0674d5f8587b7d7761ee2 Mon Sep 17 00:00:00 2001
From: machichima <nary12321@gmail.com>
Date: Tue, 21 Jul 2026 16:35:48 +0800
Subject: [PATCH 4/4] refactor: lint error

Signed-off-by: machichima <nary12321@gmail.com>
---
 .../flytekit-sleep/examples/sleep_example.py  | 19 +++++++++++----
 .../flytekit-sleep/examples/sleep_fanout.py   | 24 ++++++++++++-------
 .../examples/sleep_fanout_dynamic.py          | 24 ++++++++++++-------
 3 files changed, 46 insertions(+), 21 deletions(-)

diff --git a/plugins/flytekit-sleep/examples/sleep_example.py b/plugins/flytekit-sleep/examples/sleep_example.py
index e327841c41..c50fc19818 100644
--- a/plugins/flytekit-sleep/examples/sleep_example.py
+++ b/plugins/flytekit-sleep/examples/sleep_example.py
@@ -14,10 +14,7 @@
 from flytekit import task, workflow
 
 
-@task(
-    cache_version="2",
-    task_config=Sleep()
-)
+@task(cache_version="2", task_config=Sleep())
 def sleep_for(duration: timedelta) -> None:
     # This body only runs during local execution.
     # On the cluster, the backend sleeps for `duration` without running this.
@@ -36,5 +33,17 @@ def wf(duration: timedelta) -> None:
 
     runner = CliRunner()
     path = os.path.realpath(__file__)
-    result = runner.invoke(pyflyte.main, ["--config", os.path.expanduser("~/.flyte/config-sandbox.yaml"), "run", "--remote", path, "wf", "--duration", "10s"])
+    result = runner.invoke(
+        pyflyte.main,
+        [
+            "--config",
+            os.path.expanduser("~/.flyte/config-sandbox.yaml"),
+            "run",
+            "--remote",
+            path,
+            "wf",
+            "--duration",
+            "10s",
+        ],
+    )
     print("Remote Execution: ", result.output)
diff --git a/plugins/flytekit-sleep/examples/sleep_fanout.py b/plugins/flytekit-sleep/examples/sleep_fanout.py
index 095b690090..bfc6b737cd 100644
--- a/plugins/flytekit-sleep/examples/sleep_fanout.py
+++ b/plugins/flytekit-sleep/examples/sleep_fanout.py
@@ -13,9 +13,9 @@
 from datetime import timedelta
 from typing import List
 
-import flytekit as fl
 from flytekitplugins.sleep import Sleep
 
+import flytekit as fl
 from flytekit import map_task, task, workflow
 
 sleep_image = fl.ImageSpec(
@@ -51,13 +51,21 @@ def wf(sleep_duration: timedelta = timedelta(seconds=10), n_children: int = 400)
     runner = CliRunner()
     path = os.path.realpath(__file__)
 
-    result = runner.invoke(pyflyte.main, [
-        "--config", os.path.expanduser("~/.flyte/config-sandbox.yaml"),
-        "run", "--remote",
-        path, "wf",
-        "--sleep_duration", "10s",
-        "--n_children", "400",
-    ])
+    result = runner.invoke(
+        pyflyte.main,
+        [
+            "--config",
+            os.path.expanduser("~/.flyte/config-sandbox.yaml"),
+            "run",
+            "--remote",
+            path,
+            "wf",
+            "--sleep_duration",
+            "10s",
+            "--n_children",
+            "400",
+        ],
+    )
     print(result.output)
     if result.exception:
         raise result.exception
diff --git a/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py b/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py
index b67de7e26e..65d2e1c2f3 100644
--- a/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py
+++ b/plugins/flytekit-sleep/examples/sleep_fanout_dynamic.py
@@ -16,9 +16,9 @@
 import os
 from datetime import timedelta
 
-import flytekit as fl
 from flytekitplugins.sleep import Sleep
 
+import flytekit as fl
 from flytekit import dynamic, task, workflow
 
 fanout_image = fl.ImageSpec(
@@ -56,13 +56,21 @@ def wf(
     runner = CliRunner()
     path = os.path.realpath(__file__)
 
-    result = runner.invoke(pyflyte.main, [
-        "--config", os.path.expanduser("~/.flyte/config-sandbox.yaml"),
-        "run", "--remote",
-        path, "wf",
-        "--n_children", "400",
-        "--sleep_duration", "10s",
-    ])
+    result = runner.invoke(
+        pyflyte.main,
+        [
+            "--config",
+            os.path.expanduser("~/.flyte/config-sandbox.yaml"),
+            "run",
+            "--remote",
+            path,
+            "wf",
+            "--n_children",
+            "400",
+            "--sleep_duration",
+            "10s",
+        ],
+    )
     print(result.output)
     if result.exception:
         raise result.exception