diff --git a/applications/periodic_job/example.py b/applications/periodic_job/example.py
new file mode 100644
index 0000000..d601ff3
--- /dev/null
+++ b/applications/periodic_job/example.py
@@ -0,0 +1,52 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""主备定时任务 —— 怎么调用。
+
+redis 由调用方注入（生产用真 Redis；测试可注入 FakeRedis）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import socket
+from typing import Any
+
+from openjiuwen_runtime.foundation.periodic import create_single_leader_job
+
+
+def _instance_id() -> str:
+    """本机工号：集群里每台机器必须不一样，否则抽签分不清谁是谁。"""
+    return socket.gethostname()
+
+
+async def example_create_single_leader_job(redis: Any) -> None:
+    async def on_tick() -> None:
+        # 到点干活；需要时间就自己 time.time()
+        print("[demo] tick")
+
+    job = create_single_leader_job(
+        redis,  # 【必填】Redis（报名 / 抽签 / 执行锁）
+        name="demo",  # 【必填】任务名；不传 lock_key 时 → 锁名 lock:demo
+        on_tick=on_tick,  # 【必填】到点回调（无参）
+        instance_id=_instance_id(),  # 【必填】本机实例 ID
+        # interval_sec=1,            # 【选填】默认 1；执行锁 TTL = 此值
+        # gather_window_sec=0.08,    # 【选填】默认 0.08；开火前提前醒来报名
+        # lock_key="",               # 【选填】空则 lock:{name}；一般不用改
+        # run_on_start=False,        # 【选填】True=启动立刻跑一轮（多用于测试）
+    )
+    await job.start()
+    try:
+        await asyncio.sleep(5)  # 演示跑几秒；生产里挂在服务生命周期上
+    finally:
+        await job.stop()
+
+
+if __name__ == "__main__":
+    # 注入 redis 后跑：
+    #   import redis.asyncio as redis
+    #   r = redis.from_url("redis://127.0.0.1:6379/0")
+    #   asyncio.run(example_create_single_leader_job(r))
+    raise SystemExit(
+        "请注入 redis 后调用：asyncio.run(example_create_single_leader_job(your_redis))"
+    )
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/__init__.py b/foundation/openjiuwen_runtime/foundation/periodic/__init__.py
new file mode 100644
index 0000000..f6dd588
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/__init__.py
@@ -0,0 +1,26 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""进程内周期任务 SDK（Schedule + Coordinator + JobRunner）。
+
+主备模式：``SingleLeaderCoordinator``（等待窗口抽签 + 锁续期）。
+时间：本机墙钟 ``time.time``。
+
+对外入口：``create_single_leader_job``（唯一配置面）。
+"""
+
+from .coordinator import Coordinator, SingleLeaderCoordinator
+from .factory import create_single_leader_job
+from .lock import TickLock
+from .runner import JobRunner
+from .schedule import IntervalSchedule, Schedule
+
+__all__ = (
+    "Coordinator",
+    "IntervalSchedule",
+    "JobRunner",
+    "Schedule",
+    "SingleLeaderCoordinator",
+    "TickLock",
+    "create_single_leader_job",
+)
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/coordinator/__init__.py b/foundation/openjiuwen_runtime/foundation/periodic/coordinator/__init__.py
new file mode 100644
index 0000000..0c11ecc
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/coordinator/__init__.py
@@ -0,0 +1,10 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+from .base import Coordinator
+from .single_leader import SingleLeaderCoordinator
+
+__all__ = (
+    "Coordinator",
+    "SingleLeaderCoordinator",
+)
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/coordinator/base.py b/foundation/openjiuwen_runtime/foundation/periodic/coordinator/base.py
new file mode 100644
index 0000000..b4ad9ba
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/coordinator/base.py
@@ -0,0 +1,28 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""协调器协议。"""
+
+from __future__ import annotations
+
+from typing import Optional, Protocol
+
+
+class Coordinator(Protocol):
+    async def try_claim(
+        self,
+        *,
+        now: float,
+        instance_id: str,
+        planned_fire: float | None = None,
+    ) -> Optional[str]:
+        """试着领取本轮执行权；成功返回锁 token，失败返回 None。
+
+        ``planned_fire``：本拍语义上的开火整点（如 10.000）。
+        提前醒来时 ``now`` 可能是 T-窗口，epoch / 等到点应以 ``planned_fire`` 为准。
+        """
+        ...
+
+    async def release(self, token: str) -> None:
+        """交回执行权（按 token 校验后放锁）。"""
+        ...
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/coordinator/single_leader.py b/foundation/openjiuwen_runtime/foundation/periodic/coordinator/single_leader.py
new file mode 100644
index 0000000..46ad9db
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/coordinator/single_leader.py
@@ -0,0 +1,151 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""主备协调：提前报名 → 等到开火点 → 抽签选主 → 持锁续期执行。
+
+流程（每拍，配合 JobRunner 提前 ``gather_window`` 醒来）：
+1. ``planned_fire`` 为本拍整点 T；``now`` 多为 T-窗口
+2. ``SADD candidates:{epoch}`` 报名（epoch 取自 T）
+3. 睡到 T（剩余窗口），让网络慢的实例也能进来
+4. Lua 原子抽签：``SRANDMEMBER`` + ``SET NX winner:{epoch}``
+5. 只有 winner 去 ``SET NX`` 执行锁，并启动续期；别人空转
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Optional
+
+from openjiuwen_runtime.foundation.log import get_logger
+from openjiuwen_runtime.foundation.periodic.lock import TickLock
+
+logger = get_logger(__name__)
+
+_ELECT_LUA = """
+local existing = redis.call('GET', KEYS[1])
+if existing then
+    return existing
+end
+local pick = redis.call('SRANDMEMBER', KEYS[2])
+if not pick then
+    return false
+end
+local ok = redis.call('SET', KEYS[1], pick, 'NX', 'EX', tonumber(ARGV[1]))
+if ok then
+    return pick
+end
+return redis.call('GET', KEYS[1])
+"""
+
+
+class SingleLeaderCoordinator:
+    """主备：开火前窗口内集齐候选人，到点后随机选唯一执行者。"""
+
+    def __init__(
+        self,
+        redis: Any,
+        *,
+        lock_key: str,
+        lock_ttl_sec: int = 1,
+        token_prefix: str = "job",
+        instance_id: str = "",
+        gather_window_sec: float = 0.08,
+        meta_ttl_sec: int = 3,
+    ) -> None:
+        self._redis = redis
+        self._instance_id = instance_id
+        self._lock_key = lock_key
+        self._gather_window_sec = max(float(gather_window_sec), 0.0)
+        self._meta_ttl_sec = max(int(meta_ttl_sec), 1)
+        self._lock = TickLock(
+            redis,
+            lock_key=lock_key,
+            lock_ttl_sec=lock_ttl_sec,
+            token_prefix=token_prefix,
+            instance_id=instance_id,
+        )
+
+    def _candidates_key(self, epoch: int) -> str:
+        return f"{self._lock_key}:candidates:{epoch}"
+
+    def _winner_key(self, epoch: int) -> str:
+        return f"{self._lock_key}:winner:{epoch}"
+
+    async def try_claim(
+        self,
+        *,
+        now: float,
+        instance_id: str,
+        planned_fire: float | None = None,
+    ) -> Optional[str]:
+        iid = instance_id or self._instance_id
+        fire_at = float(planned_fire) if planned_fire is not None else float(now)
+        epoch = int(fire_at)
+        cand_key = self._candidates_key(epoch)
+        winner_key = self._winner_key(epoch)
+
+        await self._redis.sadd(cand_key, iid)
+        try:
+            await self._redis.expire(cand_key, self._meta_ttl_sec)
+        except Exception:
+            logger.debug("candidates expire failed: key=%s", cand_key)
+
+        if planned_fire is not None:
+            delay = fire_at - now
+        else:
+            delay = self._gather_window_sec
+        if delay > 0:
+            await asyncio.sleep(delay)
+
+        winner = await self._elect(winner_key, cand_key)
+        if winner is None:
+            logger.debug("no candidates for epoch=%s instance=%s", epoch, iid)
+            return None
+
+        winner_s = winner.decode() if isinstance(winner, (bytes, bytearray)) else str(winner)
+        if winner_s != iid:
+            logger.debug(
+                "not elected: epoch=%s instance=%s winner=%s",
+                epoch,
+                iid,
+                winner_s,
+            )
+            return None
+
+        token = await self._lock.try_acquire()
+        if token is None:
+            logger.warning(
+                "elected but lock busy: epoch=%s instance=%s key=%s",
+                epoch,
+                iid,
+                self._lock_key,
+            )
+            return None
+
+        self._lock.start_renew(token)
+        logger.info(
+            "single_leader claimed: epoch=%s instance=%s key=%s",
+            epoch,
+            iid,
+            self._lock_key,
+        )
+        return token
+
+    async def _elect(self, winner_key: str, cand_key: str) -> Any:
+        return await self._redis.eval(
+            _ELECT_LUA,
+            2,
+            winner_key,
+            cand_key,
+            str(self._meta_ttl_sec),
+        )
+
+    async def release(self, token: str) -> None:
+        try:
+            await self._lock.release_if_owner(token)
+        except Exception:
+            logger.exception(
+                "single_leader release failed: key=%s token=%s",
+                self._lock.lock_key,
+                token,
+            )
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/factory.py b/foundation/openjiuwen_runtime/foundation/periodic/factory.py
new file mode 100644
index 0000000..641c530
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/factory.py
@@ -0,0 +1,60 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""工厂：一条调用组装好主备定时任务。
+
+对外配置只认本函数参数；内部零件（Runner / Schedule / Coordinator）不另搞 Config 袋。
+"""
+
+from __future__ import annotations
+
+from typing import Any, Awaitable, Callable
+
+from openjiuwen_runtime.foundation.periodic.coordinator.single_leader import (
+    SingleLeaderCoordinator,
+)
+from openjiuwen_runtime.foundation.periodic.runner import JobRunner
+from openjiuwen_runtime.foundation.periodic.schedule.interval import IntervalSchedule
+
+# 内部常量（不对外暴露）
+_META_TTL_SEC = 3
+
+
+def create_single_leader_job(
+    redis: Any,  # Redis 客户端（要能 async：set/eval/sadd…）
+    *,
+    name: str,  # 任务名；默认锁 key 为 lock:{name}
+    on_tick: Callable[[], Awaitable[None]],  # 到点回调：async def on_tick() -> None
+    instance_id: str,  # 本机实例 ID，报名/抽签用，集群内需唯一
+    interval_sec: int = 1,  # 每隔多少秒响一次；锁 TTL 与此相同
+    gather_window_sec: float = 0.08,  # 开火前集合窗口：提前醒来报名，到整秒抽签
+    lock_key: str = "",  # 执行锁 Redis key；空则用 lock:{name}
+    run_on_start: bool = False,  # True 启动后立刻跑一轮（一般仅测试）
+) -> JobRunner:
+    """创建主备周期任务，返回可 start/stop 的 JobRunner。
+
+    调用方通常只需：
+        job = create_single_leader_job(redis, name="x", on_tick=..., instance_id="n1")
+        await job.start()
+        ...
+        await job.stop()
+    """
+    interval = max(int(interval_sec), 1)
+    key = (lock_key or f"lock:{name}").rstrip(":")
+    return JobRunner(
+        name=name,
+        schedule=IntervalSchedule(interval),
+        coordinator=SingleLeaderCoordinator(
+            redis,
+            lock_key=key,
+            lock_ttl_sec=interval,
+            token_prefix=name,
+            instance_id=instance_id,
+            gather_window_sec=gather_window_sec,
+            meta_ttl_sec=_META_TTL_SEC,
+        ),
+        on_tick=on_tick,
+        instance_id=instance_id,
+        gather_window_sec=gather_window_sec,
+        run_on_start=run_on_start,
+    )
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/lock.py b/foundation/openjiuwen_runtime/foundation/periodic/lock.py
new file mode 100644
index 0000000..16f13e9
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/lock.py
@@ -0,0 +1,140 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""Tick 锁：SET NX EX；持有期间可续期；释放时校验 token。"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Optional
+from uuid import uuid4
+
+from openjiuwen_runtime.foundation.log import get_logger
+
+logger = get_logger(__name__)
+
+_RELEASE_IF_OWNER_LUA = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('DEL', KEYS[1])
+end
+return 0
+"""
+
+_RENEW_IF_OWNER_LUA = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+    return redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
+end
+return 0
+"""
+
+
+class TickLock:
+    """执行权短锁；支持后台自动续期（lease）。"""
+
+    def __init__(
+        self,
+        redis: Any,
+        *,
+        lock_key: str,
+        lock_ttl_sec: int,
+        token_prefix: str,
+        instance_id: str,
+        renew_interval_sec: Optional[float] = None,
+    ) -> None:
+        self._redis = redis
+        self._lock_key = lock_key
+        self._lock_ttl_sec = max(int(lock_ttl_sec), 1)
+        self._token_prefix = token_prefix
+        self._instance_id = instance_id
+        # 默认约 TTL/3 续一次，保证业务跑久一点也不会丢锁
+        self._renew_interval_sec = (
+            float(renew_interval_sec)
+            if renew_interval_sec is not None
+            else max(0.2, self._lock_ttl_sec / 3.0)
+        )
+        self._renew_task: Optional[asyncio.Task[Any]] = None
+        self._renew_token: Optional[str] = None
+        self._lost = False
+
+    @property
+    def lock_key(self) -> str:
+        return self._lock_key
+
+    @property
+    def lost(self) -> bool:
+        return self._lost
+
+    def new_token(self) -> str:
+        return f"{self._token_prefix}:{self._instance_id}:{uuid4()}"
+
+    async def try_acquire(self, token: Optional[str] = None) -> Optional[str]:
+        """抢锁成功返回 token，失败返回 None。"""
+        tok = token or self.new_token()
+        ok = await self._redis.set(self._lock_key, tok, nx=True, ex=self._lock_ttl_sec)
+        if ok:
+            self._lost = False
+            logger.info("tick lock acquired: key=%s token=%s", self._lock_key, tok)
+            return tok
+        logger.debug("tick lock miss: key=%s", self._lock_key)
+        return None
+
+    async def renew_once(self, token: str) -> bool:
+        """仍是 owner 则续 TTL，返回 True；失锁返回 False。"""
+        result = await self._redis.eval(
+            _RENEW_IF_OWNER_LUA,
+            1,
+            self._lock_key,
+            token,
+            str(self._lock_ttl_sec),
+        )
+        return int(result or 0) == 1
+
+    def start_renew(self, token: str) -> None:
+        """启动后台续期；同一把锁只跑一个续期任务。"""
+        self.stop_renew()
+        self._renew_token = token
+        self._lost = False
+        self._renew_task = asyncio.create_task(
+            self._renew_loop(token),
+            name=f"tick-lock-renew-{self._lock_key}",
+        )
+
+    def stop_renew(self) -> None:
+        """停止后台续期（不放锁）。"""
+        task = self._renew_task
+        self._renew_task = None
+        self._renew_token = None
+        if task is not None and not task.done():
+            task.cancel()
+
+    async def _renew_loop(self, token: str) -> None:
+        try:
+            while True:
+                await asyncio.sleep(self._renew_interval_sec)
+                ok = await self.renew_once(token)
+                if not ok:
+                    self._lost = True
+                    logger.warning(
+                        "tick lock lost on renew: key=%s token=%s",
+                        self._lock_key,
+                        token,
+                    )
+                    return
+                logger.debug("tick lock renewed: key=%s", self._lock_key)
+        except asyncio.CancelledError:
+            return
+
+    async def release_if_owner(self, token: str) -> bool:
+        """先停续期，再仅当仍是自己的 token 时删除锁。"""
+        self.stop_renew()
+        result = await self._redis.eval(_RELEASE_IF_OWNER_LUA, 1, self._lock_key, token)
+        released = int(result or 0) == 1
+        if released:
+            logger.debug("tick lock released: key=%s token=%s", self._lock_key, token)
+        else:
+            logger.debug(
+                "tick lock not released (not owner or gone): key=%s token=%s",
+                self._lock_key,
+                token,
+            )
+        return released
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/runner.py b/foundation/openjiuwen_runtime/foundation/periodic/runner.py
new file mode 100644
index 0000000..e275479
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/runner.py
@@ -0,0 +1,202 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""JobRunner：提前窗口醒来 → 协调 → 回调 → 放锁。
+
+使用本机墙钟 ``time.time``。睡觉一次睡到目标点；
+stop 通过 cancel 打断，或醒来后检查退出。
+上一拍未结束则跳过本拍（写死）。
+
+对外配置请看工厂 ``create_single_leader_job``；本类只收组装后的零件。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import Any, Awaitable, Callable, Optional
+
+from openjiuwen_runtime.foundation.log import get_logger
+from openjiuwen_runtime.foundation.periodic.coordinator.base import Coordinator
+from openjiuwen_runtime.foundation.periodic.schedule.base import Schedule
+
+logger = get_logger(__name__)
+
+_STOP_TIMEOUT_SEC = 3.0
+
+
+class JobRunner:
+    """进程内周期任务生命周期管理。"""
+
+    def __init__(
+        self,
+        *,
+        name: str,
+        schedule: Schedule,
+        coordinator: Coordinator,
+        on_tick: Callable[[], Awaitable[None]],
+        instance_id: str,
+        gather_window_sec: float = 0.0,
+        run_on_start: bool = False,
+        stop_timeout_sec: float = _STOP_TIMEOUT_SEC,
+    ) -> None:
+        self._name = name
+        self._schedule = schedule
+        self._coordinator = coordinator
+        self._on_tick = on_tick
+        self._instance_id = instance_id
+        self._gather_window_sec = max(float(gather_window_sec), 0.0)
+        self._run_on_start = bool(run_on_start)
+        self._stop_timeout_sec = float(stop_timeout_sec)
+        self._stopped = asyncio.Event()
+        self._task: Optional[asyncio.Task[Any]] = None
+        self._busy = False
+        self._last_now: Optional[float] = None
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def _now(self) -> float:
+        return float(time.time())
+
+    async def start(self) -> None:
+        if self._task and not self._task.done():
+            return
+        self._stopped.clear()
+        self._task = asyncio.create_task(
+            self._run_forever(),
+            name=f"periodic-{self._name}-{self._instance_id}",
+        )
+        logger.info(
+            "JobRunner started: job=%s instance=%s",
+            self._name,
+            self._instance_id,
+        )
+
+    async def stop(self) -> None:
+        self._stopped.set()
+        task = self._task
+        self._task = None
+        if task is None:
+            return
+        task.cancel()
+        try:
+            await asyncio.wait_for(task, timeout=self._stop_timeout_sec)
+        except (asyncio.CancelledError, asyncio.TimeoutError):
+            pass
+        except Exception:
+            logger.exception(
+                "JobRunner stop wait failed: job=%s instance=%s",
+                self._name,
+                self._instance_id,
+            )
+        logger.info(
+            "JobRunner stopped: job=%s instance=%s",
+            self._name,
+            self._instance_id,
+        )
+
+    async def _sleep_until(self, target: float) -> None:
+        """一次睡到目标时间；stop 靠 cancel 打断，或醒来后由循环检查 _stopped。"""
+        if self._stopped.is_set():
+            return
+        delay = target - self._now()
+        if delay <= 0:
+            return
+        await asyncio.sleep(delay)
+
+    async def _run_forever(self) -> None:
+        if self._run_on_start and not self._stopped.is_set():
+            await self._safe_tick(planned_fire=self._now())
+
+        while not self._stopped.is_set():
+            try:
+                now = self._now()
+                if self._last_now is not None and now < self._last_now:
+                    logger.warning(
+                        "clock went backwards: job=%s last=%s now=%s",
+                        self._name,
+                        self._last_now,
+                        now,
+                    )
+                    now = self._now()
+                self._last_now = now
+
+                next_ts = self._schedule.next_fire_time(now)
+                if next_ts <= now:
+                    next_ts = self._schedule.next_fire_time(self._now())
+
+                gather = min(self._gather_window_sec, max(next_ts - now, 0.0))
+                wake_at = next_ts - gather
+                if wake_at > now:
+                    await self._sleep_until(wake_at)
+                if self._stopped.is_set():
+                    break
+
+                now2 = self._now()
+                await self._safe_tick(planned_fire=next_ts, now=now2)
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception(
+                    "JobRunner loop error: job=%s instance=%s",
+                    self._name,
+                    self._instance_id,
+                )
+
+    async def _safe_tick(self, *, planned_fire: float, now: Optional[float] = None) -> None:
+        now_v = now if now is not None else self._now()
+        claim = await self._coordinator.try_claim(
+            now=now_v,
+            instance_id=self._instance_id,
+            planned_fire=planned_fire,
+        )
+        if claim is None:
+            logger.debug(
+                "job lock miss: job=%s instance=%s",
+                self._name,
+                self._instance_id,
+            )
+            return
+
+        if self._busy:
+            logger.info(
+                "tick skipped overlap: job=%s instance=%s",
+                self._name,
+                self._instance_id,
+            )
+            await self._coordinator.release(claim)
+            return
+
+        self._busy = True
+        ok = False
+        t0 = time.monotonic()
+        delay_ms = max(0.0, (now_v - planned_fire) * 1000)
+        try:
+            await self._on_tick()
+            ok = True
+        except Exception:
+            logger.exception(
+                "on_tick failed: job=%s instance=%s",
+                self._name,
+                self._instance_id,
+            )
+        finally:
+            duration_ms = (time.monotonic() - t0) * 1000
+            self._busy = False
+            try:
+                await self._coordinator.release(claim)
+            except Exception:
+                logger.exception(
+                    "release after tick failed: job=%s",
+                    self._name,
+                )
+            logger.info(
+                "tick done: job=%s instance=%s ok=%s delay_ms=%.1f duration_ms=%.1f",
+                self._name,
+                self._instance_id,
+                ok,
+                delay_ms,
+                duration_ms,
+            )
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/schedule/__init__.py b/foundation/openjiuwen_runtime/foundation/periodic/schedule/__init__.py
new file mode 100644
index 0000000..bec2bdd
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/schedule/__init__.py
@@ -0,0 +1,7 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+from .base import Schedule
+from .interval import IntervalSchedule
+
+__all__ = ("IntervalSchedule", "Schedule")
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/schedule/base.py b/foundation/openjiuwen_runtime/foundation/periodic/schedule/base.py
new file mode 100644
index 0000000..256edaf
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/schedule/base.py
@@ -0,0 +1,14 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""调度器协议。"""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+
+class Schedule(Protocol):
+    def next_fire_time(self, now: float) -> float:
+        """返回严格大于 now 的下次触发 unix 秒。"""
+        ...
diff --git a/foundation/openjiuwen_runtime/foundation/periodic/schedule/interval.py b/foundation/openjiuwen_runtime/foundation/periodic/schedule/interval.py
new file mode 100644
index 0000000..485a258
--- /dev/null
+++ b/foundation/openjiuwen_runtime/foundation/periodic/schedule/interval.py
@@ -0,0 +1,23 @@
+# coding: utf-8
+# Copyright (c) Huawei Technologies Co., Ltd. 2026-2026. All rights reserved
+
+"""固定间隔、挂钟边界对齐的调度。"""
+
+from __future__ import annotations
+
+import math
+
+
+class IntervalSchedule:
+    """下一个 interval 边界触发（整秒/整 N 秒对齐）。"""
+
+    def __init__(self, interval_sec: int = 1) -> None:
+        self._interval_sec = max(int(interval_sec), 1)
+
+    @property
+    def interval_sec(self) -> int:
+        return self._interval_sec
+
+    def next_fire_time(self, now: float) -> float:
+        interval = self._interval_sec
+        return (math.floor(now / interval) + 1) * interval