orchestrata_ultimate.tools.schedule_cron_tool

Reference › orchestrata_ultimate › tools › schedule_cron_tool

ScheduleCronTool (SPEC-068) lets a caller create, update, pause, resume and delete cron-style scheduled jobs, compute upcoming fire times in any IANA timezone, trigger runs on demand and read execution history. Job and execution state lives in Redis, in keys namespaced so a caller or test can scope a run to one namespace and wipe it with ScheduleCronStore.purge_namespace. It is registered by agents.tool_registry and exported from the package __init__. tests/tools/test_schedule_cron_tool.py covers it.

The module records several forks from the spec. Persistence is Redis rather than SQLite. There is no croniter, apscheduler or pytz: CronExpression is a hand-written five-field parser supporting *, lists, ranges, steps and month and weekday names, and rejects the six-field form and the L, W and # characters with a clear error. Timezones use the standard-library zone database with IANA names only. There is no background scheduler daemon and no missed-job recovery scan; the tool stores next_run_at, the MissedJobPolicy, and history.

execute takes an action and a params dictionary and dispatches to schedule_job, unschedule_job, update_job, list_jobs, get_job_status, run_job_now, cancel_job, get_next_runs, validate_cron_expression, pause_job, resume_job, get_execution_history or clear_history. It never raises: a ScheduleCronError, a wrong-parameter error or any other exception becomes a failed ToolResult. Data types include CronJob, JobExecution, the enums ConcurrencyPolicy and JobStatus, and the result dataclasses.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

CronParseError

Bases: ValueError · Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-004']) · Line 126

Raised when a cron expression cannot be parsed (ERR-001).

ScheduleCronErrorCode

Decorators: traces(spec='068', story='US-068-002', reqs=['REQ-068-020', 'REQ-068-021', 'REQ-068-022']) · Line 135

Error codes for the ScheduleCronTool pipeline (spec.md Error Conditions).

ScheduleCronError

Bases: Exception · Decorators: traces(spec='068', story='US-068-002', reqs=['REQ-068-020']) · Line 149

Raised for a spec.md ERR-00N domain condition; rendered as a ToolResult.

MissedJobPolicy (enum)

Bases: StrEnum · Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-012']) · Line 170

Policy for handling missed jobs (bookkeeping only — see module fork note).

ConcurrencyPolicy (enum)

Bases: StrEnum · Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-013']) · Line 180

Policy for handling concurrent job execution.

JobStatus (enum)

Bases: StrEnum · Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-015']) · Line 190

Job execution status.

CronJob (dataclass)

Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-009', 'REQ-068-019', 'REQ-068-020', 'REQ-068-021', 'REQ-068-022']), dataclass · Line 210

A scheduled cron job definition.

JobExecution (dataclass)

Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-010', 'REQ-068-017']), dataclass · Line 235

A single recorded job execution.

JobScheduleResult (dataclass)

Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-001', 'REQ-068-005']), dataclass · Line 252

Result of scheduling (or updating) a job.

JobExecutionResult (dataclass)

Decorators: traces(spec='068', story='US-068-002', reqs=['REQ-068-016', 'REQ-068-017']), dataclass · Line 263

Result of an on-demand job execution.

CronExpression (dataclass)

Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-001', 'REQ-068-002', 'REQ-068-003', 'REQ-068-007']), dataclass(frozen=True) · Line 344

A parsed 5-field cron expression (minute hour day month weekday).

ScheduleCronStore

Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-009', 'REQ-068-010', 'REQ-068-013', 'REQ-068-014']) · Line 511

Redis persistence for cron jobs + execution history + concurrency locks.

ScheduleCronTool

Bases: BaseTool · Decorators: traces(spec='068', story='US-068-001', reqs=['REQ-068-001', 'REQ-068-005', 'REQ-068-009', 'REQ-068-019']) · Line 799

Schedule and manage recurring jobs using cron expressions.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/schedule_cron_tool.py

"""ScheduleCronTool — cron-style job scheduling with timezone support.

Spec: SPEC-068 (ScheduleCronTool). A SPEC-062 ``BaseTool`` that lets a caller
create/list/update/pause/resume/delete cron-style scheduled jobs, compute
their upcoming fire times in any IANA timezone, trigger on-demand execution,
and inspect execution history — all persisted in Redis (this project's hot
store), namespaced so concurrent callers (and tests) never collide.

**Forks from spec.md (reported to the orchestrator, not silently guessed):**

* **Persistence is Redis, not SQLite.** spec.md's Technical Design assumes a
  SQLite database + ``aiosqlite``. Per the build brief, this project's real
  hot store is Redis (see ``core.redis_factory``), so job/execution state
  lives there instead — one JSON blob per job/execution plus small index
  structures (a set of job ids, per-job and global sorted sets of execution
  ids ordered by start time). Every key is namespaced
  (``...:schedule_cron:{namespace}:...``) so a caller (a test, in practice)
  can scope an entire run to one throwaway namespace and wipe it with
  :meth:`ScheduleCronStore.purge_namespace`.
* **No ``croniter``/``apscheduler``/``pytz`` dependency.** Those are spec.md's
  named runtime dependencies, but adding them requires a ``pyproject.toml``
  edit the build brief explicitly forbids. Cron parsing is a small hand-rolled
  5-field parser (:class:`CronExpression`) supporting ``*``, lists (``,``),
  ranges (``-``), steps (``/``), and named months/weekdays (``JAN``-``DEC``,
  ``SUN``-``SAT``); timezones use the stdlib ``zoneinfo`` (IANA names only).
  **Not implemented** (and not stamped with ``@traces``): the extended
  6-field (seconds) format and the ``L``/``W``/``#`` special characters —
  parsing rejects them with a clear error rather than silently mis-scheduling.
  UTC offset notation (``+05:30``) timezones are likewise not accepted; only
  IANA zone names are (``ZoneInfo`` validates them).
* **No background scheduler daemon / missed-job recovery scan.** spec.md's
  "Background Scheduler" and "Recovery Service" are a long-running process
  that continuously triggers due jobs and, on its own startup, detects and
  replays missed runs. This module is a stateless-per-call tool with no
  process of its own to run that loop in — it implements the bookkeeping a
  daemon would need (``next_run_at`` computation, ``miss_policy`` stored on
  the job, execution history) and exposes on-demand triggering
  (:meth:`ScheduleCronTool.run_job_now`), but actually driving jobs at their
  scheduled time, and replaying missed runs after downtime, needs a real
  scheduler process wired to this tool — reported as a wiring fork, not
  invented here.
* **Concurrency protection is a simple Redis lock, not a full policy engine.**
  ``FORBID``/``SKIP`` acquire a ``SET NX EX`` lock keyed by job id (TTL =
  ``timeout_seconds``, so a stuck lock self-expires); ``ALLOW`` and ``QUEUE``
  both run unlocked (a real FIFO queue for ``QUEUE`` is out of MVP scope).
* **Cancellation is in-process only.** :meth:`ScheduleCronTool.run_job_now`
  registers its own ``asyncio.Task`` (and live subprocess) in an in-memory
  dict keyed by job id; :meth:`ScheduleCronTool.cancel_job` cancels that task
  and kills the subprocess. There is no cross-process job registry, so a job
  triggered by a different tool instance/process cannot be cancelled from
  here — another consequence of there being no daemon process.
* **Command execution model.** ``command`` is run via
  ``asyncio.create_subprocess_shell`` with the job's ``args`` dict passed to
  the child as ``SCHEDULE_CRON_JOB_ARGS`` (JSON) plus
  ``SCHEDULE_CRON_JOB_ID``/``SCHEDULE_CRON_EXECUTION_ID`` env vars — spec.md
  does not specify an args-passing convention, so this is this module's
  design choice. Command allowlisting (SR-004) is not implemented.
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import uuid
from collections.abc import Awaitable, Callable
from dataclasses import asdict, dataclass, field, replace
from datetime import UTC, datetime, timedelta
from enum import StrEnum
from typing import TYPE_CHECKING, Any, cast
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.redis_factory import create_async_redis
from orchestrata_ultimate.tools.base import (
    BaseTool,
    ToolCategory,
    ToolMetadata,
    ToolPriority,
    ToolResult,
)
from orchestrata_ultimate.tools.registry import ToolRegistry

if TYPE_CHECKING:
    import redis.asyncio as redis_asyncio

_NAME_PATTERN = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")

# ── cron field vocabularies ──────────────────────────────────────────────

_MONTH_NAMES = {
    name: i + 1
    for i, name in enumerate(
        [
            "JAN",
            "FEB",
            "MAR",
            "APR",
            "MAY",
            "JUN",
            "JUL",
            "AUG",
            "SEP",
            "OCT",
            "NOV",
            "DEC",
        ]
    )
}
_WEEKDAY_NAMES = {
    name: i for i, name in enumerate(["SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT"])
}

#: Unsupported special characters, scoped per-field so they don't false-positive
#: on legitimate name tokens (e.g. "JUL" contains 'L', "WED" contains 'W').
#: Real cron dialects use 'L'/'W' only on the day-of-month field and '#' only
#: on the day-of-week field.
_UNSUPPORTED_SPECIAL_CHARS = {"day": "LW", "weekday": "#"}


# ── errors ────────────────────────────────────────────────────────────────


@traces(spec="068", story="US-068-001", reqs=["REQ-068-004"])
class CronParseError(ValueError):
    """Raised when a cron expression cannot be parsed (ERR-001)."""


@traces(
    spec="068",
    story="US-068-002",
    reqs=["REQ-068-020", "REQ-068-021", "REQ-068-022"],
)
class ScheduleCronErrorCode:
    """Error codes for the ScheduleCronTool pipeline (spec.md Error Conditions)."""

    INVALID_CRON_EXPRESSION = "invalid_cron_expression"
    INVALID_TIMEZONE = "invalid_timezone"
    JOB_NOT_FOUND = "job_not_found"
    JOB_ALREADY_EXISTS = "job_already_exists"
    CONCURRENT_EXECUTION = "concurrent_execution"
    JOB_TIMEOUT = "job_timeout"
    INVALID_CONFIGURATION = "invalid_configuration"
    MAX_RETRIES_EXCEEDED = "max_retries_exceeded"


@traces(spec="068", story="US-068-002", reqs=["REQ-068-020"])
class ScheduleCronError(Exception):
    """Raised for a spec.md ERR-00N domain condition; rendered as a ToolResult."""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message

    def to_result(self, tool_name: str) -> ToolResult:
        return ToolResult(
            tool_name=tool_name,
            success=False,
            output="",
            error=f"{self.code}: {self.message}",
        )


# ── enums (spec.md Tool Interface) ───────────────────────────────────────


@traces(spec="068", story="US-068-001", reqs=["REQ-068-012"])
class MissedJobPolicy(StrEnum):
    """Policy for handling missed jobs (bookkeeping only — see module fork note)."""

    SKIP = "skip"
    RUN_ONCE = "run_once"
    RUN_ALL = "run_all"
    NOTIFY_ONLY = "notify_only"


@traces(spec="068", story="US-068-001", reqs=["REQ-068-013"])
class ConcurrencyPolicy(StrEnum):
    """Policy for handling concurrent job execution."""

    ALLOW = "allow"
    SKIP = "skip"
    QUEUE = "queue"
    FORBID = "forbid"


@traces(spec="068", story="US-068-001", reqs=["REQ-068-015"])
class JobStatus(StrEnum):
    """Job execution status."""

    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    RECOVERED = "recovered"


# ── data models (spec.md Tool Interface) ─────────────────────────────────


@traces(
    spec="068",
    story="US-068-001",
    reqs=["REQ-068-009", "REQ-068-019", "REQ-068-020", "REQ-068-021", "REQ-068-022"],
)
@dataclass
class CronJob:
    """A scheduled cron job definition."""

    id: str
    name: str
    cron_expression: str
    timezone: str
    command: str
    args: dict[str, Any] = field(default_factory=dict)
    enabled: bool = True
    miss_policy: MissedJobPolicy = MissedJobPolicy.RUN_ONCE
    concurrency_policy: ConcurrencyPolicy = ConcurrencyPolicy.ALLOW
    max_retries: int = 3
    retry_delay_seconds: int = 60
    timeout_seconds: int = 300
    description: str = ""
    tags: list[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    last_run_at: datetime | None = None
    next_run_at: datetime | None = None


@traces(spec="068", story="US-068-001", reqs=["REQ-068-010", "REQ-068-017"])
@dataclass
class JobExecution:
    """A single recorded job execution."""

    id: str
    job_id: str
    status: JobStatus
    started_at: datetime
    completed_at: datetime | None = None
    error_message: str | None = None
    output: str | None = None
    retry_count: int = 0
    is_recovery: bool = False
    execution_node: str | None = None


@traces(spec="068", story="US-068-001", reqs=["REQ-068-001", "REQ-068-005"])
@dataclass
class JobScheduleResult:
    """Result of scheduling (or updating) a job."""

    success: bool
    job_id: str | None = None
    next_run_at: datetime | None = None
    error_message: str | None = None


@traces(spec="068", story="US-068-002", reqs=["REQ-068-016", "REQ-068-017"])
@dataclass
class JobExecutionResult:
    """Result of an on-demand job execution."""

    success: bool
    execution_id: str | None = None
    output: str | None = None
    error_message: str | None = None
    duration_seconds: float | None = None


# ── cron expression parsing ───────────────────────────────────────────────


@traces(spec="068", story="FR-001", reqs=["REQ-068-002"])
def _resolve_token(token: str, names: dict[str, int] | None) -> int:
    token = token.strip()
    if names is not None and token.upper() in names:
        return names[token.upper()]
    if not token.isdigit():
        raise CronParseError(f"invalid value {token!r}")
    return int(token)


@traces(spec="068", story="FR-001", reqs=["REQ-068-001", "REQ-068-002"])
def _parse_field(
    spec: str,
    min_v: int,
    max_v: int,
    *,
    field_name: str,
    names: dict[str, int] | None = None,
) -> frozenset[int]:
    """Parse one cron field (``*``, lists, ranges, steps, names) into a value set."""
    spec = spec.strip()
    if not spec:
        raise CronParseError(f"{field_name}: field must not be blank")
    special = _UNSUPPORTED_SPECIAL_CHARS.get(field_name, "")
    if special and any(ch in spec for ch in special):
        raise CronParseError(
            f"{field_name}: {'/'.join(special)!r} special characters are not supported"
        )
    values: set[int] = set()
    for part in spec.split(","):
        part = part.strip()
        if not part:
            raise CronParseError(f"{field_name}: empty item in list {spec!r}")
        step = 1
        base = part
        if "/" in part:
            base, step_s = part.split("/", 1)
            if not step_s.isdigit() or int(step_s) <= 0:
                raise CronParseError(f"{field_name}: invalid step {step_s!r}")
            step = int(step_s)
        if base == "*":
            lo, hi = min_v, max_v
        elif "-" in base:
            lo_s, hi_s = base.split("-", 1)
            lo = _resolve_token(lo_s, names)
            hi = _resolve_token(hi_s, names)
            if lo > hi:
                raise CronParseError(f"{field_name}: range start {lo} > end {hi}")
        else:
            lo = hi = _resolve_token(base, names)
            if step != 1:
                hi = max_v
        if lo < min_v or hi > max_v:
            raise CronParseError(
                f"{field_name}: value out of range [{min_v},{max_v}]: {part!r}"
            )
        values.update(range(lo, hi + 1, step))
    if field_name == "weekday":
        values = {0 if v == 7 else v for v in values}
    return frozenset(values)


@traces(
    spec="068",
    story="US-068-001",
    reqs=["REQ-068-001", "REQ-068-002", "REQ-068-003", "REQ-068-007"],
)
@dataclass(frozen=True)
class CronExpression:
    """A parsed 5-field cron expression (minute hour day month weekday)."""

    raw: str
    minutes: frozenset[int]
    hours: frozenset[int]
    days: frozenset[int]
    months: frozenset[int]
    weekdays: frozenset[int]
    day_star: bool
    weekday_star: bool

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-001", "REQ-068-002"])
    @classmethod
    def parse(cls, expression: str) -> CronExpression:
        """Parse a standard 5-field cron expression.

        Raises:
            CronParseError: malformed field count, out-of-range value, bad
                step/range, or an unsupported ``L``/``W``/``#`` character
                (ERR-001).
        """
        if not isinstance(expression, str) or not expression.strip():
            raise CronParseError("cron expression must not be blank")
        fields = expression.strip().split()
        if len(fields) != 5:
            raise CronParseError(
                f"expression must have 5 fields, got {len(fields)}: {expression!r}"
            )
        minute_s, hour_s, day_s, month_s, weekday_s = fields
        return cls(
            raw=expression.strip(),
            minutes=_parse_field(minute_s, 0, 59, field_name="minute"),
            hours=_parse_field(hour_s, 0, 23, field_name="hour"),
            days=_parse_field(day_s, 1, 31, field_name="day"),
            months=_parse_field(month_s, 1, 12, field_name="month", names=_MONTH_NAMES),
            weekdays=_parse_field(
                weekday_s, 0, 7, field_name="weekday", names=_WEEKDAY_NAMES
            ),
            day_star=(day_s.strip() == "*"),
            weekday_star=(weekday_s.strip() == "*"),
        )

    @traces(spec="068", story="FR-001", reqs=["REQ-068-001"])
    def _date_matches(self, dt: datetime) -> bool:
        if dt.month not in self.months:
            return False
        dom_match = dt.day in self.days
        cron_weekday = (dt.weekday() + 1) % 7  # Mon=0..Sun=6 -> Sun=0..Sat=6
        dow_match = cron_weekday in self.weekdays
        if self.day_star and self.weekday_star:
            return True
        if self.day_star:
            return dow_match
        if self.weekday_star:
            return dom_match
        return dom_match or dow_match

    @traces(spec="068", story="US-068-002", reqs=["REQ-068-007", "REQ-068-025"])
    def next_fire_time(
        self, after: datetime, tz: ZoneInfo, *, max_lookahead_days: int = 366 * 4
    ) -> datetime | None:
        """Compute the next fire time strictly after ``after``, in UTC.

        ``after`` may be in any timezone; the search itself is carried out in
        wall-clock terms in ``tz`` (so DST transitions in that zone are
        respected), and the result is converted back to UTC (spec.md FR-002:
        "Store all timestamps in UTC internally").
        """
        local = after.astimezone(tz)
        current = local.replace(second=0, microsecond=0) + timedelta(minutes=1)
        end = local + timedelta(days=max_lookahead_days)
        while current <= end:
            if not self._date_matches(current):
                current = (current + timedelta(days=1)).replace(hour=0, minute=0)
                continue
            if current.hour in self.hours and current.minute in self.minutes:
                return current.astimezone(UTC)
            current += timedelta(minutes=1)
        return None


# ── Redis persistence ─────────────────────────────────────────────────────


@traces(spec="068", story="FR-003", reqs=["REQ-068-009"])
def _job_to_json(job: CronJob) -> str:
    data = asdict(job)
    data["miss_policy"] = job.miss_policy.value
    data["concurrency_policy"] = job.concurrency_policy.value
    data["created_at"] = job.created_at.isoformat()
    data["updated_at"] = job.updated_at.isoformat()
    data["last_run_at"] = job.last_run_at.isoformat() if job.last_run_at else None
    data["next_run_at"] = job.next_run_at.isoformat() if job.next_run_at else None
    return json.dumps(data)


@traces(spec="068", story="FR-003", reqs=["REQ-068-009"])
def _job_from_json(raw: str) -> CronJob:
    data = json.loads(raw)
    return CronJob(
        id=data["id"],
        name=data["name"],
        cron_expression=data["cron_expression"],
        timezone=data["timezone"],
        command=data["command"],
        args=dict(data.get("args") or {}),
        enabled=bool(data.get("enabled", True)),
        miss_policy=MissedJobPolicy(data.get("miss_policy", "run_once")),
        concurrency_policy=ConcurrencyPolicy(data.get("concurrency_policy", "allow")),
        max_retries=int(data.get("max_retries", 3)),
        retry_delay_seconds=int(data.get("retry_delay_seconds", 60)),
        timeout_seconds=int(data.get("timeout_seconds", 300)),
        description=str(data.get("description", "")),
        tags=list(data.get("tags") or []),
        created_at=datetime.fromisoformat(data["created_at"]),
        updated_at=datetime.fromisoformat(data["updated_at"]),
        last_run_at=(
            datetime.fromisoformat(data["last_run_at"])
            if data.get("last_run_at")
            else None
        ),
        next_run_at=(
            datetime.fromisoformat(data["next_run_at"])
            if data.get("next_run_at")
            else None
        ),
    )


@traces(spec="068", story="FR-003", reqs=["REQ-068-010"])
def _execution_to_json(execution: JobExecution) -> str:
    data = asdict(execution)
    data["status"] = execution.status.value
    data["started_at"] = execution.started_at.isoformat()
    data["completed_at"] = (
        execution.completed_at.isoformat() if execution.completed_at else None
    )
    return json.dumps(data)


@traces(spec="068", story="FR-003", reqs=["REQ-068-010"])
def _execution_from_json(raw: str) -> JobExecution:
    data = json.loads(raw)
    return JobExecution(
        id=data["id"],
        job_id=data["job_id"],
        status=JobStatus(data["status"]),
        started_at=datetime.fromisoformat(data["started_at"]),
        completed_at=(
            datetime.fromisoformat(data["completed_at"])
            if data.get("completed_at")
            else None
        ),
        error_message=data.get("error_message"),
        output=data.get("output"),
        retry_count=int(data.get("retry_count", 0)),
        is_recovery=bool(data.get("is_recovery", False)),
        execution_node=data.get("execution_node"),
    )


@traces(
    spec="068",
    story="US-068-001",
    reqs=["REQ-068-009", "REQ-068-010", "REQ-068-013", "REQ-068-014"],
)
class ScheduleCronStore:
    """Redis persistence for cron jobs + execution history + concurrency locks.

    Every key is prefixed ``orchestrata_ultimate:schedule_cron:{namespace}:``
    so callers (tests, in particular) can scope an entire run to one
    namespace and wipe it in one shot via :meth:`purge_namespace`.
    """

    def __init__(
        self, redis_client: redis_asyncio.Redis, namespace: str = "default"
    ) -> None:
        self._redis = redis_client
        self._ns = namespace

    @traces(spec="068", story="FR-003", reqs=["REQ-068-009"])
    def _job_key(self, job_id: str) -> str:
        return f"orchestrata_ultimate:schedule_cron:{self._ns}:job:{job_id}"

    @traces(spec="068", story="FR-003", reqs=["REQ-068-009"])
    def _jobs_set_key(self) -> str:
        return f"orchestrata_ultimate:schedule_cron:{self._ns}:jobs"

    @traces(spec="068", story="FR-003", reqs=["REQ-068-010"])
    def _exec_key(self, exec_id: str) -> str:
        return f"orchestrata_ultimate:schedule_cron:{self._ns}:exec:{exec_id}"

    @traces(spec="068", story="FR-003", reqs=["REQ-068-010"])
    def _exec_by_job_key(self, job_id: str) -> str:
        return f"orchestrata_ultimate:schedule_cron:{self._ns}:exec_by_job:{job_id}"

    @traces(spec="068", story="FR-003", reqs=["REQ-068-010"])
    def _exec_all_key(self) -> str:
        return f"orchestrata_ultimate:schedule_cron:{self._ns}:exec_all"

    @traces(spec="068", story="FR-005", reqs=["REQ-068-013", "REQ-068-014"])
    def _lock_key(self, job_id: str) -> str:
        return f"orchestrata_ultimate:schedule_cron:{self._ns}:lock:{job_id}"

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-009"])
    async def save_job(self, job: CronJob) -> None:
        pipe = self._redis.pipeline()
        pipe.set(self._job_key(job.id), _job_to_json(job))
        pipe.sadd(self._jobs_set_key(), job.id)
        await pipe.execute()

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-009"])
    async def get_job(self, job_id: str) -> CronJob | None:
        raw = await self._redis.get(self._job_key(job_id))
        return _job_from_json(raw) if raw else None

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-022"])
    async def delete_job(self, job_id: str) -> bool:
        pipe = self._redis.pipeline()
        pipe.delete(self._job_key(job_id))
        pipe.srem(self._jobs_set_key(), job_id)
        results = await pipe.execute()
        return bool(results[0])

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-019"])
    async def list_jobs(self) -> list[CronJob]:
        # redis-py types smembers() as the shared sync/async command-mixin
        # union `Awaitable[Set] | Set`; cast narrows it to the async branch
        # so pyright accepts the `await` (this client is always asyncio).
        job_ids = await cast(
            "Awaitable[set[str]]", self._redis.smembers(self._jobs_set_key())
        )
        if not job_ids:
            return []
        raws = await self._redis.mget([self._job_key(jid) for jid in job_ids])
        return [_job_from_json(r) for r in raws if r]

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-019"])
    async def find_by_name(self, name: str) -> CronJob | None:
        for job in await self.list_jobs():
            if job.name == name:
                return job
        return None

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-010"])
    async def save_execution(self, execution: JobExecution) -> None:
        ts = execution.started_at.timestamp()
        pipe = self._redis.pipeline()
        pipe.set(self._exec_key(execution.id), _execution_to_json(execution))
        pipe.zadd(self._exec_by_job_key(execution.job_id), {execution.id: ts})
        pipe.zadd(self._exec_all_key(), {execution.id: ts})
        await pipe.execute()

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-010"])
    async def get_execution(self, exec_id: str) -> JobExecution | None:
        raw = await self._redis.get(self._exec_key(exec_id))
        return _execution_from_json(raw) if raw else None

    @traces(spec="068", story="FR-008", reqs=["REQ-068-026"])
    async def _hydrate(self, exec_ids: list[str]) -> list[JobExecution]:
        if not exec_ids:
            return []
        raws = await self._redis.mget([self._exec_key(e) for e in exec_ids])
        return [_execution_from_json(r) for r in raws if r]

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-026"])
    async def list_executions_for_job(
        self, job_id: str, limit: int
    ) -> list[JobExecution]:
        if limit <= 0:
            return []
        exec_ids = await self._redis.zrevrange(
            self._exec_by_job_key(job_id), 0, limit - 1
        )
        return await self._hydrate(exec_ids)

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-026"])
    async def list_all_executions(self, limit: int) -> list[JobExecution]:
        if limit <= 0:
            return []
        exec_ids = await self._redis.zrevrange(self._exec_all_key(), 0, limit - 1)
        return await self._hydrate(exec_ids)

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-022"])
    async def delete_all_executions_for_job(self, job_id: str) -> int:
        exec_ids = await self._redis.zrange(self._exec_by_job_key(job_id), 0, -1)
        count = 0
        for exec_id in exec_ids:
            pipe = self._redis.pipeline()
            pipe.delete(self._exec_key(exec_id))
            pipe.zrem(self._exec_all_key(), exec_id)
            await pipe.execute()
            count += 1
        await self._redis.delete(self._exec_by_job_key(job_id))
        return count

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-027"])
    async def delete_executions_older_than(
        self, job_id: str | None, cutoff: datetime
    ) -> int:
        cutoff_ts = cutoff.timestamp()
        zkey = (
            self._exec_by_job_key(job_id)
            if job_id is not None
            else self._exec_all_key()
        )
        exec_ids = await self._redis.zrangebyscore(zkey, "-inf", cutoff_ts)
        count = 0
        for exec_id in exec_ids:
            execution = await self.get_execution(exec_id)
            pipe = self._redis.pipeline()
            pipe.delete(self._exec_key(exec_id))
            pipe.zrem(self._exec_all_key(), exec_id)
            if execution is not None:
                pipe.zrem(self._exec_by_job_key(execution.job_id), exec_id)
            await pipe.execute()
            count += 1
        return count

    @traces(spec="068", story="US-068-002", reqs=["REQ-068-013"])
    async def acquire_lock(self, job_id: str, ttl_seconds: int) -> bool:
        acquired = await self._redis.set(
            self._lock_key(job_id), "1", nx=True, ex=max(ttl_seconds, 1)
        )
        return bool(acquired)

    @traces(spec="068", story="US-068-002", reqs=["REQ-068-013"])
    async def release_lock(self, job_id: str) -> None:
        await self._redis.delete(self._lock_key(job_id))

    @traces(spec="068", story="US-068-002", reqs=["REQ-068-014"])
    async def is_locked(self, job_id: str) -> bool:
        return bool(await self._redis.exists(self._lock_key(job_id)))

    async def purge_namespace(self) -> int:
        """Delete every key under this namespace (test/cleanup helper)."""
        pattern = f"orchestrata_ultimate:schedule_cron:{self._ns}:*"
        count = 0
        async for key in self._redis.scan_iter(match=pattern, count=200):
            await self._redis.delete(key)
            count += 1
        return count


# ── public dict rendering (API-contract shape) ───────────────────────────


def _job_public_dict(job: CronJob) -> dict[str, Any]:
    return {
        "id": job.id,
        "name": job.name,
        "cron_expression": job.cron_expression,
        "timezone": job.timezone,
        "command": job.command,
        "args": dict(job.args),
        "enabled": job.enabled,
        "status": "active" if job.enabled else "paused",
        "miss_policy": job.miss_policy.value,
        "concurrency_policy": job.concurrency_policy.value,
        "max_retries": job.max_retries,
        "retry_delay_seconds": job.retry_delay_seconds,
        "timeout_seconds": job.timeout_seconds,
        "description": job.description,
        "tags": list(job.tags),
        "created_at": job.created_at.isoformat(),
        "updated_at": job.updated_at.isoformat(),
        "last_run_at": job.last_run_at.isoformat() if job.last_run_at else None,
        "next_run_at": job.next_run_at.isoformat() if job.next_run_at else None,
    }


@traces(spec="068", story="FR-008", reqs=["REQ-068-026"])
def _execution_public_dict(execution: JobExecution) -> dict[str, Any]:
    return {
        "id": execution.id,
        "job_id": execution.job_id,
        "status": execution.status.value,
        "started_at": execution.started_at.isoformat(),
        "completed_at": (
            execution.completed_at.isoformat() if execution.completed_at else None
        ),
        "error_message": execution.error_message,
        "output": execution.output,
        "retry_count": execution.retry_count,
        "is_recovery": execution.is_recovery,
        "execution_node": execution.execution_node,
    }


def _json_default(obj: Any) -> Any:
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, StrEnum):
        return obj.value
    raise TypeError(f"not JSON serializable: {obj!r}")


_UPDATABLE_JOB_FIELDS = frozenset(
    {
        "name",
        "cron_expression",
        "timezone",
        "command",
        "args",
        "enabled",
        "miss_policy",
        "concurrency_policy",
        "max_retries",
        "retry_delay_seconds",
        "timeout_seconds",
        "description",
        "tags",
    }
)

_SCHEDULE_CRON_PARAMETERS: dict[str, Any] = {
    "type": "object",
    "properties": {
        "action": {
            "type": "string",
            "enum": [
                "schedule_job",
                "unschedule_job",
                "update_job",
                "list_jobs",
                "get_job_status",
                "run_job_now",
                "cancel_job",
                "get_next_runs",
                "validate_cron_expression",
                "pause_job",
                "resume_job",
                "get_execution_history",
                "clear_history",
            ],
            "description": "Which ScheduleCronTool operation to perform.",
        },
        "params": {
            "type": "object",
            "description": "Action-specific keyword arguments (see the method of the same name).",
        },
    },
    "required": ["action"],
}


# ── the tool ──────────────────────────────────────────────────────────────


@traces(
    spec="068",
    story="US-068-001",
    reqs=["REQ-068-001", "REQ-068-005", "REQ-068-009", "REQ-068-019"],
)
class ScheduleCronTool(BaseTool):
    """Schedule and manage recurring jobs using cron expressions."""

    def __init__(self, store: ScheduleCronStore, metadata: ToolMetadata) -> None:
        self._store = store
        self._metadata = metadata
        self._running_tasks: dict[str, asyncio.Task[Any]] = {}
        self._running_procs: dict[str, asyncio.subprocess.Process] = {}
        self._actions: dict[str, Callable[..., Awaitable[Any]]] = {
            "schedule_job": self.schedule_job,
            "unschedule_job": self.unschedule_job,
            "update_job": self.update_job,
            "list_jobs": self.list_jobs,
            "get_job_status": self.get_job_status,
            "run_job_now": self.run_job_now,
            "cancel_job": self.cancel_job,
            "get_next_runs": self.get_next_runs,
            "validate_cron_expression": self.validate_cron_expression,
            "pause_job": self.pause_job,
            "resume_job": self.resume_job,
            "get_execution_history": self.get_execution_history,
            "clear_history": self.clear_history,
        }

    @property
    def metadata(self) -> ToolMetadata:
        return self._metadata

    # ── BaseTool contract ────────────────────────────────────────────

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-001"])
    async def execute(self, arguments: dict[str, Any]) -> ToolResult:
        """Dispatch a tool-call ``{"action": ..., "params": {...}}`` request.

        Never raises: any :class:`ScheduleCronError` (or unforeseen
        exception) becomes a failed :class:`ToolResult` instead.
        """
        try:
            action = arguments.get("action")
            if action not in self._actions:
                raise ScheduleCronError(
                    ScheduleCronErrorCode.INVALID_CONFIGURATION,
                    f"unknown action: {action!r}",
                )
            params = dict(arguments.get("params") or {})
            method = self._actions[action]
            result = await method(**params)
            if isinstance(result, ToolResult):
                return result
            payload = (
                asdict(result) if hasattr(result, "__dataclass_fields__") else result
            )
            return ToolResult(
                tool_name=self.metadata.name,
                success=True,
                output=json.dumps(payload, default=_json_default),
            )
        except ScheduleCronError as exc:
            return exc.to_result(self.metadata.name)
        except TypeError as exc:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"invalid params for action {arguments.get('action')!r}: {exc}",
            )
        except Exception as exc:  # noqa: BLE001 - never propagate; report as ToolResult
            return ToolResult(
                tool_name=self.metadata.name, success=False, output="", error=str(exc)
            )

    # ── job CRUD (spec.md Tool Interface) ────────────────────────────

    @traces(
        spec="068",
        story="US-068-001",
        reqs=["REQ-068-001", "REQ-068-005", "REQ-068-009", "REQ-068-020"],
    )
    async def schedule_job(
        self,
        name: str,
        cron_expression: str,
        command: str,
        timezone: str = "UTC",
        args: dict[str, Any] | None = None,
        miss_policy: str = "run_once",
        concurrency_policy: str = "allow",
        max_retries: int = 3,
        retry_delay_seconds: int = 60,
        timeout_seconds: int = 300,
        description: str = "",
        tags: list[str] | None = None,
        enabled: bool = True,
        force: bool = False,
    ) -> JobScheduleResult:
        """Schedule a new cron job (FR-001/FR-002/FR-003).

        ``force`` is a brief-driven extension beyond spec.md's literal
        signature: by default a job whose ``name`` matches an existing job
        in this namespace is rejected (ERR-004 JOB_ALREADY_EXISTS); pass
        ``force=True`` to add a duplicate-named job anyway.
        """
        if not name or not _NAME_PATTERN.match(name):
            return JobScheduleResult(
                success=False, error_message=f"invalid job name: {name!r}"
            )
        try:
            cron = CronExpression.parse(cron_expression)
        except CronParseError as exc:
            return JobScheduleResult(
                success=False, error_message=f"INVALID_CRON_EXPRESSION: {exc}"
            )
        try:
            tz = ZoneInfo(timezone)
        except ZoneInfoNotFoundError, ValueError:
            return JobScheduleResult(
                success=False, error_message=f"INVALID_TIMEZONE: {timezone!r}"
            )
        try:
            miss = MissedJobPolicy(miss_policy)
            conc = ConcurrencyPolicy(concurrency_policy)
        except ValueError as exc:
            return JobScheduleResult(
                success=False, error_message=f"INVALID_CONFIGURATION: {exc}"
            )

        if not force and await self._store.find_by_name(name) is not None:
            return JobScheduleResult(
                success=False,
                error_message=(
                    f"JOB_ALREADY_EXISTS: a job named {name!r} already exists "
                    "(pass force=True to add a duplicate)"
                ),
            )

        now = datetime.now(UTC)
        next_run = cron.next_fire_time(now.astimezone(tz), tz) if enabled else None
        job = CronJob(
            id=f"job_{uuid.uuid4().hex}",
            name=name,
            cron_expression=cron.raw,
            timezone=timezone,
            command=command,
            args=dict(args or {}),
            enabled=enabled,
            miss_policy=miss,
            concurrency_policy=conc,
            max_retries=max_retries,
            retry_delay_seconds=retry_delay_seconds,
            timeout_seconds=timeout_seconds,
            description=description,
            tags=list(tags or []),
            created_at=now,
            updated_at=now,
            next_run_at=next_run,
        )
        await self._store.save_job(job)
        return JobScheduleResult(success=True, job_id=job.id, next_run_at=next_run)

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-022"])
    async def unschedule_job(self, job_id: str) -> ToolResult:
        """Remove a scheduled job and its execution history (FR-008/AC-006)."""
        job = await self._store.get_job(job_id)
        if job is None:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"JOB_NOT_FOUND: {job_id}",
            )
        await self._store.delete_job(job_id)
        await self._store.delete_all_executions_for_job(job_id)
        return ToolResult(
            tool_name=self.metadata.name,
            success=True,
            output=json.dumps({"job_id": job_id, "deleted": True}),
        )

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-020"])
    async def update_job(self, job_id: str, **updates: Any) -> ToolResult:
        """Update an existing job's configuration (FR-008)."""
        job = await self._store.get_job(job_id)
        if job is None:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"JOB_NOT_FOUND: {job_id}",
            )
        unknown = set(updates) - _UPDATABLE_JOB_FIELDS
        if unknown:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"INVALID_CONFIGURATION: unknown fields {sorted(unknown)}",
            )

        merged = _job_public_dict(job) | updates
        try:
            cron = CronExpression.parse(str(merged["cron_expression"]))
            tz = ZoneInfo(str(merged["timezone"]))
            miss = MissedJobPolicy(merged["miss_policy"])
            conc = ConcurrencyPolicy(merged["concurrency_policy"])
        except (CronParseError, ZoneInfoNotFoundError, ValueError) as exc:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"INVALID_CONFIGURATION: {exc}",
            )

        now = datetime.now(UTC)
        enabled = bool(merged["enabled"])
        next_run = cron.next_fire_time(now.astimezone(tz), tz) if enabled else None
        updated = replace(
            job,
            name=str(merged["name"]),
            cron_expression=cron.raw,
            timezone=str(merged["timezone"]),
            command=str(merged["command"]),
            args=dict(merged["args"] or {}),
            enabled=enabled,
            miss_policy=miss,
            concurrency_policy=conc,
            max_retries=int(merged["max_retries"]),
            retry_delay_seconds=int(merged["retry_delay_seconds"]),
            timeout_seconds=int(merged["timeout_seconds"]),
            description=str(merged["description"]),
            tags=list(merged["tags"] or []),
            updated_at=now,
            next_run_at=next_run,
        )
        await self._store.save_job(updated)
        return ToolResult(
            tool_name=self.metadata.name,
            success=True,
            output=json.dumps(_job_public_dict(updated)),
        )

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-019"])
    async def list_jobs(
        self,
        enabled_only: bool = False,
        tags: list[str] | None = None,
        limit: int = 100,
    ) -> list[dict[str, Any]]:
        """List scheduled jobs, optionally filtered (FR-008)."""
        jobs = await self._store.list_jobs()
        if enabled_only:
            jobs = [j for j in jobs if j.enabled]
        if tags:
            wanted = set(tags)
            jobs = [j for j in jobs if wanted & set(j.tags)]
        jobs.sort(key=lambda j: j.created_at)
        return [_job_public_dict(j) for j in jobs[:limit]]

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-009", "REQ-068-014"])
    async def get_job_status(self, job_id: str) -> dict[str, Any]:
        """Get a job's current status plus recent execution history.

        Raises:
            ScheduleCronError: ``JOB_NOT_FOUND`` (ERR-003).
        """
        job = await self._store.get_job(job_id)
        if job is None:
            raise ScheduleCronError(
                ScheduleCronErrorCode.JOB_NOT_FOUND, f"no such job: {job_id}"
            )
        executions = await self._store.list_executions_for_job(job_id, limit=10)
        payload = _job_public_dict(job)
        payload["is_locked"] = await self._store.is_locked(job_id)
        payload["recent_executions"] = [_execution_public_dict(e) for e in executions]
        return payload

    # ── pause/resume ─────────────────────────────────────────────────

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-021"])
    async def pause_job(self, job_id: str) -> ToolResult:
        """Pause a job (disable without removing) (AC-006)."""
        return await self._set_enabled(job_id, False)

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-021"])
    async def resume_job(self, job_id: str) -> ToolResult:
        """Resume a paused job (AC-006)."""
        return await self._set_enabled(job_id, True)

    @traces(spec="068", story="FR-008", reqs=["REQ-068-021"])
    async def _set_enabled(self, job_id: str, enabled: bool) -> ToolResult:
        job = await self._store.get_job(job_id)
        if job is None:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"JOB_NOT_FOUND: {job_id}",
            )
        now = datetime.now(UTC)
        next_run = job.next_run_at
        if enabled:
            cron = CronExpression.parse(job.cron_expression)
            tz = ZoneInfo(job.timezone)
            next_run = cron.next_fire_time(now.astimezone(tz), tz)
        else:
            next_run = None
        updated = replace(job, enabled=enabled, updated_at=now, next_run_at=next_run)
        await self._store.save_job(updated)
        return ToolResult(
            tool_name=self.metadata.name,
            success=True,
            output=json.dumps(_job_public_dict(updated)),
        )

    # ── cron computation ─────────────────────────────────────────────

    @traces(spec="068", story="US-068-002", reqs=["REQ-068-025"])
    async def get_next_runs(self, job_id: str, count: int = 5) -> list[datetime]:
        """Get upcoming scheduled run times, in UTC (FR-008).

        Raises:
            ScheduleCronError: ``JOB_NOT_FOUND`` (ERR-003).
        """
        job = await self._store.get_job(job_id)
        if job is None:
            raise ScheduleCronError(
                ScheduleCronErrorCode.JOB_NOT_FOUND, f"no such job: {job_id}"
            )
        cron = CronExpression.parse(job.cron_expression)
        tz = ZoneInfo(job.timezone)
        results: list[datetime] = []
        cursor = datetime.now(UTC).astimezone(tz)
        for _ in range(max(count, 0)):
            nxt = cron.next_fire_time(cursor, tz)
            if nxt is None:
                break
            results.append(nxt)
            cursor = nxt.astimezone(tz)
        return results

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-003"])
    async def validate_cron_expression(
        self, expression: str, timezone: str = "UTC"
    ) -> dict[str, Any]:
        """Validate a cron expression + timezone before scheduling (FR-001)."""
        errors: list[str] = []
        try:
            cron = CronExpression.parse(expression)
        except CronParseError as exc:
            return {"valid": False, "errors": [str(exc)], "next_run_at": None}
        try:
            tz = ZoneInfo(timezone)
        except ZoneInfoNotFoundError, ValueError:
            errors.append(f"invalid timezone: {timezone!r}")
            return {"valid": False, "errors": errors, "next_run_at": None}
        next_run = cron.next_fire_time(datetime.now(UTC).astimezone(tz), tz)
        return {
            "valid": True,
            "errors": [],
            "next_run_at": next_run.isoformat() if next_run else None,
        }

    # ── execution ─────────────────────────────────────────────────────

    @traces(
        spec="068", story="FR-006", reqs=["REQ-068-015", "REQ-068-016", "REQ-068-017"]
    )
    async def _run_command_once(
        self, job: CronJob, exec_id: str
    ) -> tuple[bool, str, str | None]:
        """Run ``job.command`` once, honoring ``job.timeout_seconds`` (FR-006)."""
        env = {
            **os.environ,
            "SCHEDULE_CRON_JOB_ID": job.id,
            "SCHEDULE_CRON_EXECUTION_ID": exec_id,
            "SCHEDULE_CRON_JOB_ARGS": json.dumps(job.args),
        }
        try:
            proc = await asyncio.create_subprocess_shell(
                job.command,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env=env,
            )
        except OSError as exc:
            return False, "", f"COMMAND_NOT_FOUND: {exc}"

        self._running_procs[job.id] = proc
        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=job.timeout_seconds
            )
        except TimeoutError:
            proc.kill()
            await proc.wait()
            return False, "", "JOB_TIMEOUT: exceeded timeout_seconds"
        finally:
            self._running_procs.pop(job.id, None)

        output = stdout.decode("utf-8", errors="replace")
        if proc.returncode != 0:
            err = (
                stderr.decode("utf-8", errors="replace")
                or f"exit code {proc.returncode}"
            )
            return False, output, err
        return True, output, None

    @traces(
        spec="068",
        story="US-068-002",
        reqs=[
            "REQ-068-013",
            "REQ-068-014",
            "REQ-068-015",
            "REQ-068-016",
            "REQ-068-017",
            "REQ-068-023",
        ],
    )
    async def run_job_now(self, job_id: str, force: bool = False) -> JobExecutionResult:
        """Trigger immediate job execution, with retry + concurrency handling (FR-006/FR-007).

        Raises:
            ScheduleCronError: ``JOB_NOT_FOUND`` (ERR-003).
        """
        job = await self._store.get_job(job_id)
        if job is None:
            raise ScheduleCronError(
                ScheduleCronErrorCode.JOB_NOT_FOUND, f"no such job: {job_id}"
            )
        if not job.enabled and not force:
            return JobExecutionResult(
                success=False,
                error_message="job is disabled; pass force=True to override",
            )

        locking = not force and job.concurrency_policy in (
            ConcurrencyPolicy.FORBID,
            ConcurrencyPolicy.SKIP,
        )
        lock_acquired = True
        if locking:
            lock_acquired = await self._store.acquire_lock(job_id, job.timeout_seconds)
            if not lock_acquired:
                return JobExecutionResult(
                    success=False,
                    error_message="CONCURRENT_EXECUTION: another instance is already running",
                )

        exec_id = f"exec_{uuid.uuid4().hex}"
        started_at = datetime.now(UTC)
        execution = JobExecution(
            id=exec_id, job_id=job_id, status=JobStatus.RUNNING, started_at=started_at
        )
        await self._store.save_execution(execution)

        task = asyncio.current_task()
        if task is not None:
            self._running_tasks[job_id] = task

        attempt = 0
        success = False
        output = ""
        error_message: str | None = None
        try:
            while True:
                success, output, error_message = await self._run_command_once(
                    job, exec_id
                )
                if success or attempt >= job.max_retries:
                    break
                attempt += 1
                execution = replace(execution, retry_count=attempt)
                await self._store.save_execution(execution)
                await asyncio.sleep(job.retry_delay_seconds)
        except asyncio.CancelledError:
            proc = self._running_procs.pop(job_id, None)
            if proc is not None:
                proc.kill()
            execution = replace(
                execution,
                status=JobStatus.CANCELLED,
                completed_at=datetime.now(UTC),
                error_message="JOB_CANCELLED: cancelled during execution",
                retry_count=attempt,
            )
            await self._store.save_execution(execution)
            raise
        finally:
            self._running_tasks.pop(job_id, None)
            if locking and lock_acquired:
                await self._store.release_lock(job_id)

        completed_at = datetime.now(UTC)
        final_status = JobStatus.COMPLETED if success else JobStatus.FAILED
        execution = replace(
            execution,
            status=final_status,
            completed_at=completed_at,
            output=output,
            error_message=error_message,
            retry_count=attempt,
        )
        await self._store.save_execution(execution)
        await self._store.save_job(
            replace(job, last_run_at=completed_at, updated_at=completed_at)
        )

        duration = (completed_at - started_at).total_seconds()
        if success:
            return JobExecutionResult(
                success=True,
                execution_id=exec_id,
                output=output,
                duration_seconds=duration,
            )
        return JobExecutionResult(
            success=False,
            execution_id=exec_id,
            error_message=error_message
            or "MAX_RETRIES_EXCEEDED: job failed after all retries",
            duration_seconds=duration,
        )

    @traces(spec="068", story="US-068-002", reqs=["REQ-068-024"])
    async def cancel_job(self, job_id: str) -> ToolResult:
        """Cancel a job's in-flight execution, if one is running in this process.

        In-process only (see module fork note): a job triggered from a
        different :class:`ScheduleCronTool` instance/process cannot be
        cancelled from here.
        """
        job = await self._store.get_job(job_id)
        if job is None:
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error=f"JOB_NOT_FOUND: {job_id}",
            )
        task = self._running_tasks.get(job_id)
        if task is None or task.done():
            return ToolResult(
                tool_name=self.metadata.name,
                success=False,
                output="",
                error="no running execution to cancel for this job",
            )
        task.cancel()
        return ToolResult(
            tool_name=self.metadata.name,
            success=True,
            output=json.dumps({"job_id": job_id, "cancelled": True}),
        )

    # ── execution history ─────────────────────────────────────────────

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-026"])
    async def get_execution_history(
        self,
        job_id: str | None = None,
        status: str | None = None,
        limit: int = 50,
    ) -> list[dict[str, Any]]:
        """Get job execution history, optionally filtered (FR-008)."""
        fetch_limit = max(limit * 4, limit) if status else limit
        executions = (
            await self._store.list_executions_for_job(job_id, fetch_limit)
            if job_id is not None
            else await self._store.list_all_executions(fetch_limit)
        )
        if status:
            wanted = JobStatus(status)
            executions = [e for e in executions if e.status == wanted]
        return [_execution_public_dict(e) for e in executions[:limit]]

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-027"])
    async def clear_history(
        self, job_id: str | None = None, older_than_days: int = 30
    ) -> ToolResult:
        """Clear execution history records older than a cutoff (FR-008)."""
        cutoff = datetime.now(UTC) - timedelta(days=older_than_days)
        count = await self._store.delete_executions_older_than(job_id, cutoff)
        return ToolResult(
            tool_name=self.metadata.name,
            success=True,
            output=json.dumps({"deleted": count}),
        )

    # ── construction / registration ───────────────────────────────────

    @traces(spec="068", story="US-068-001", reqs=["REQ-068-009"])
    @classmethod
    def create(
        cls,
        redis_client: redis_asyncio.Redis | None = None,
        namespace: str = "default",
        host: str | None = None,
        port: int | None = None,
        db: int | None = None,
    ) -> ScheduleCronTool:
        """Build a fully-wired ``ScheduleCronTool`` over Redis.

        ``redis_client`` lets a caller (a test, in practice) supply an
        already-connected client pointed at a specific DB; otherwise one is
        built via :func:`~orchestrata_ultimate.core.redis_factory.create_async_redis`
        (which itself reads ``REDIS_HOST``/``REDIS_PORT``/``REDIS_DB`` from
        the environment when ``host``/``port``/``db`` are omitted).
        """
        client = (
            redis_client
            if redis_client is not None
            else create_async_redis(host=host, port=port, db=db)
        )
        store = ScheduleCronStore(client, namespace=namespace)
        metadata = ToolMetadata(
            name="schedule_cron",
            description=(
                "Schedule recurring jobs using cron expressions with timezone "
                "support: create/list/update/pause/resume/delete jobs, compute "
                "upcoming fire times, trigger on-demand execution, and inspect "
                "execution history."
            ),
            category=ToolCategory.PLANNING,
            parameters=_SCHEDULE_CRON_PARAMETERS,
            priority=ToolPriority.P2,
            tags=("scheduling", "cron", "planning", "timezone"),
        )
        return cls(store=store, metadata=metadata)


@traces(spec="068", story="US-068-001", reqs=["REQ-068-009"])
def register_schedule_cron_tool(
    registry: ToolRegistry,
    *,
    redis_client: redis_asyncio.Redis | None = None,
    namespace: str = "default",
    host: str | None = None,
    port: int | None = None,
    db: int | None = None,
) -> ScheduleCronTool:
    """Build a :class:`ScheduleCronTool` and register it into ``registry``."""
    tool = ScheduleCronTool.create(
        redis_client=redis_client, namespace=namespace, host=host, port=port, db=db
    )
    registry.register(tool, replace=True)
    return tool