diff --git a/.env.example b/.env.example index a4eeda4..d8d210d 100644 --- a/.env.example +++ b/.env.example @@ -291,3 +291,11 @@ LEAPFLOW_RECORDING_MODE=video # HuggingFace Hub credentials (future) # HF_TOKEN=hf_xxxxx # HF_ENDPOINT=https://huggingface.co + +# ═════════════════════════════════════════════════════════════════════ +# Scheduler — Long-horizon async task execution +# ═════════════════════════════════════════════════════════════════════ +# LEAPFLOW_SCHEDULER_ENABLED=true +# LEAPFLOW_SCHEDULER_TICK_SECONDS=60 +# LEAPFLOW_SCHEDULER_GRACE_SECONDS=120.0 +# LEAPFLOW_SCHEDULER_DEFAULT_TIER=auto diff --git a/.gitignore b/.gitignore index 5054237..ccaffcb 100644 --- a/.gitignore +++ b/.gitignore @@ -236,3 +236,4 @@ leapflow.sock temp .DS_Store +AGENTS.md diff --git a/src/leapflow/_env_template.py b/src/leapflow/_env_template.py index 4b26ed8..c03290a 100644 --- a/src/leapflow/_env_template.py +++ b/src/leapflow/_env_template.py @@ -278,4 +278,12 @@ # HuggingFace Hub credentials (future) # HF_TOKEN=hf_xxxxx # HF_ENDPOINT=https://huggingface.co + +# ═════════════════════════════════════════════════════════════════════ +# Scheduler — Long-horizon async task execution +# ═════════════════════════════════════════════════════════════════════ +# LEAPFLOW_SCHEDULER_ENABLED=true +# LEAPFLOW_SCHEDULER_TICK_SECONDS=60 +# LEAPFLOW_SCHEDULER_GRACE_SECONDS=120.0 +# LEAPFLOW_SCHEDULER_DEFAULT_TIER=auto """ diff --git a/src/leapflow/cli/commands/hub.py b/src/leapflow/cli/commands/hub.py index c5a6fcf..a6b9940 100644 --- a/src/leapflow/cli/commands/hub.py +++ b/src/leapflow/cli/commands/hub.py @@ -162,6 +162,7 @@ async def _hub_whoami(ctx: "Context", args: List[str]) -> int: async def _hub_push(ctx: "Context", args: List[str]) -> int: """Push a skill to the Hub.""" from leapflow.hub import ContentSanitizer, SkillSerializer, Visibility + from leapflow.hub.protocol import VersionConflictError if not args: print(" Usage: hub push [--visibility private|public] [--version v1.0.0]") @@ -215,6 +216,14 @@ async def _hub_push(ctx: "Context", args: List[str]) -> int: warnings = sanitizer.scan(bundle) _print_warnings(warnings, "Sanitization") + # Step 3b: Check for blocking findings + if sanitizer.has_blocking_findings(warnings) and "--force" not in args: + print(" HIGH severity findings detected. Use --force to push anyway.") + confirmed = await _confirm("Push despite high-severity warnings?", dangerous=True, skill_name=skill_name) + if not confirmed: + print(" Push aborted.") + return 0 + # Step 4: Show summary visibility = Visibility(visibility_str) client = _build_hub_client(ctx) @@ -241,8 +250,38 @@ async def _hub_push(ctx: "Context", args: List[str]) -> int: print(" Cancelled.") return 0 - # Step 6: Push - result = await client.push(bundle, skill_name=bundle.manifest.name, visibility=visibility) + # Step 6: Auto-fill author info + import dataclasses + from datetime import datetime, timezone + + try: + user_info = await client.backend.authenticate() + username = user_info.username + except Exception: + username = "" + + if username: + manifest = dataclasses.replace( + bundle.manifest, + author=bundle.manifest.author or username, + updated_by=username, + updated_at=datetime.now(timezone.utc).isoformat(), + ) + bundle = dataclasses.replace(bundle, manifest=manifest) + + # Step 7: Push + force = "--force" in args + try: + result = await client.push(bundle, skill_name=bundle.manifest.name, visibility=visibility, force=force) + except VersionConflictError as e: + print(f" Version conflict: {e}") + confirmed = await _confirm("Force push anyway?") + if confirmed: + result = await client.push(bundle, skill_name=bundle.manifest.name, visibility=visibility, force=True) + else: + print(" Push aborted.") + return 0 + print(f"\n Pushed successfully!") print(f" Repo: {result.repo_id}") print(f" Version: {result.version}") diff --git a/src/leapflow/cli/commands/interactive.py b/src/leapflow/cli/commands/interactive.py index c65cb8b..0776a87 100644 --- a/src/leapflow/cli/commands/interactive.py +++ b/src/leapflow/cli/commands/interactive.py @@ -174,6 +174,8 @@ def _bridge_online() -> bool: print(" skills disable — Disable a learned skill") print(" skills delete — Delete a learned skill") print(" hub — Hub operations (push/pull/sync/search)") + print(" arm --trigger — Arm a skill for scheduled execution") + print(" tasks — List/manage scheduled tasks") print(" shortcut list — List quick-reply shortcuts") print(" shortcut add

= — Add shortcut (pattern = reply)") print(" shortcut remove

— Remove a shortcut") @@ -405,6 +407,19 @@ def _bridge_online() -> bool: await cmd_hub(ctx, hub_args) continue + # ── Scheduler commands ── + if line.startswith("arm"): + from leapflow.cli.commands.scheduler import cmd_arm + arm_args = line.split()[1:] if len(line.split()) > 1 else [] + await cmd_arm(ctx, arm_args) + continue + + if line.startswith("tasks"): + from leapflow.cli.commands.scheduler import cmd_tasks + tasks_args = line.split()[1:] if len(line.split()) > 1 else [] + await cmd_tasks(ctx, tasks_args) + continue + # ── Run command ── if line.startswith("run "): trigger_or_name = line[4:].strip() diff --git a/src/leapflow/cli/commands/scheduler.py b/src/leapflow/cli/commands/scheduler.py new file mode 100644 index 0000000..8c59271 --- /dev/null +++ b/src/leapflow/cli/commands/scheduler.py @@ -0,0 +1,289 @@ +"""Scheduler CLI commands — arm tasks and manage scheduled execution. + +Provides ``leap arm`` and ``leap tasks`` subcommands for the interactive REPL. +""" + +from __future__ import annotations + +import time +from typing import TYPE_CHECKING, List + +if TYPE_CHECKING: + from leapflow.cli.context import Context + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _format_duration(seconds: float) -> str: + """Format a duration in seconds into a human-readable string.""" + if seconds <= 0: + return "now" + if seconds < 60: + return f"{int(seconds)}s" + if seconds < 3600: + return f"in {int(seconds / 60)}m" + if seconds < 86400: + h = int(seconds / 3600) + m = int((seconds % 3600) / 60) + return f"in {h}h{m}m" if m else f"in {h}h" + d = int(seconds / 86400) + return f"in {d}d" + + +def _format_trigger(task) -> str: + """Format trigger info for display.""" + if task.trigger_type == "interval": + sec = task.trigger_config.get("interval_seconds", 0) + if sec < 60: + return f"every {int(sec)}s" + if sec < 3600: + return f"every {int(sec / 60)}m" + if sec < 86400: + return f"every {int(sec / 3600)}h" + return f"every {int(sec / 86400)}d" + if task.trigger_type == "cron": + return task.trigger_config.get("expression", "cron") + if task.trigger_type == "event": + return f"event:{task.trigger_config.get('event_pattern', '?')}" + if task.trigger_type == "condition": + expr = task.trigger_config.get("expression", "?") + return f"cond:{expr[:20]}" + return task.trigger_type + + +def _get_coordinator(ctx: "Context"): + """Get or create a TaskCoordinator from context.""" + from leapflow.scheduler.coordinator import TaskCoordinator + from leapflow.scheduler.local_scheduler import LocalScheduler + from leapflow.scheduler.store import TaskStore + + # Use existing coordinator if available + if hasattr(ctx, "coordinator") and ctx.coordinator is not None: + return ctx.coordinator + + # Build one from settings + db_path = ctx.settings.data_dir / "scheduler.duckdb" + store = TaskStore(db_path) + + # Local scheduler with a simple skill executor + class _SimpleExecutor: + """Minimal skill executor for scheduled tasks.""" + + async def execute(self, skill_name: str, parameters: dict) -> dict: + # Try to execute via session if available + if ctx.session: + try: + result = await ctx.session.execute_skill(skill_name, params=parameters) + return {"ok": True, "output": str(result)[:200]} + except Exception as e: + return {"ok": False, "error": str(e)} + return {"ok": False, "error": "No session available"} + + local_scheduler = LocalScheduler( + store=store, + executor=_SimpleExecutor(), + tick_seconds=ctx.settings.scheduler_tick_seconds, + grace_seconds=ctx.settings.scheduler_grace_seconds, + ) + + # Cloud dispatcher (optional, only if compute backend available) + cloud_dispatcher = None + try: + from leapflow.scheduler.cloud_dispatcher import CloudDispatcher + from leapflow.scheduler.compute.modelscope_studio import ModelScopeStudioBackend + from leapflow.scheduler.worker_packager import WorkerPackager + + backend = ModelScopeStudioBackend() + packager = WorkerPackager() + cloud_dispatcher = CloudDispatcher(backend, packager) + except (ImportError, Exception): + pass # Cloud not available — local-only mode + + coordinator = TaskCoordinator( + store=store, + local_scheduler=local_scheduler, + cloud_dispatcher=cloud_dispatcher, + default_tier=ctx.settings.scheduler_default_tier, + ) + ctx.coordinator = coordinator + return coordinator + + +# --------------------------------------------------------------------------- +# Commands +# --------------------------------------------------------------------------- + + +async def cmd_arm(ctx: "Context", args: List[str]) -> int: + """leap arm --trigger "" [--local|--cloud] [--max-runs N] + + Arm a skill for scheduled execution. + """ + if not args or "--help" in args or "-h" in args: + print("Usage: arm --trigger \"\" [--local|--cloud] [--max-runs N]") + print() + print("Trigger formats:") + print(" 30m / 2h / 1d — interval") + print(" every 5m — interval") + print(" 0 9 * * * — cron (5 fields)") + print(" event:ci.passed — event") + print(" condition:expr > val — condition") + print() + print("Options:") + print(" --local Force local execution") + print(" --cloud Force cloud execution") + print(" --max-runs N Stop after N executions (-1 = unlimited)") + return 0 + + # Parse arguments + skill_name = args[0] + trigger_expr = "" + execution_tier = "auto" + max_runs = -1 + + i = 1 + while i < len(args): + if args[i] == "--trigger" and i + 1 < len(args): + trigger_expr = args[i + 1] + i += 2 + elif args[i] == "--local": + execution_tier = "local" + i += 1 + elif args[i] == "--cloud": + execution_tier = "cloud" + i += 1 + elif args[i] == "--max-runs" and i + 1 < len(args): + try: + max_runs = int(args[i + 1]) + except ValueError: + print(f" Error: --max-runs must be an integer, got '{args[i + 1]}'") + return 1 + i += 2 + else: + # If no --trigger flag, treat remaining as trigger expression + if not trigger_expr: + trigger_expr = " ".join(args[i:]) + break + i += 1 + + if not trigger_expr: + print(" Error: trigger expression required.") + print(" Usage: arm --trigger \"30m\"") + return 1 + + try: + coordinator = _get_coordinator(ctx) + task = await coordinator.arm( + skill_name=skill_name, + trigger_expr=trigger_expr, + execution_tier=execution_tier, + max_runs=max_runs, + ) + now = time.time() + next_in = _format_duration(task.next_due_at - now) + print(f" Armed: {task.task_id[:8]} skill={skill_name} tier={task.execution_tier} next={next_in}") + except ValueError as e: + print(f" Error: {e}") + return 1 + except RuntimeError as e: + print(f" Error: {e}") + return 1 + + return 0 + + +async def cmd_tasks(ctx: "Context", args: List[str]) -> int: + """leap tasks [status | logs | cancel ] + + Manage scheduled tasks. + """ + coordinator = _get_coordinator(ctx) + + if not args: + # List all tasks + tasks = await coordinator.list_tasks() + if not tasks: + print(" No scheduled tasks.") + return 0 + + now = time.time() + print(f" {'ID':<10} {'Skill':<20} {'Trigger':<16} {'Tier':<7} {'State':<10} Next Due") + print(f" {'-' * 80}") + for t in tasks: + tid = t.task_id[:8] + skill = t.skill_name[:18] + trigger = _format_trigger(t)[:14] + tier = t.execution_tier[:5] + state = t.state[:8] + if t.next_due_at > 0: + next_due = _format_duration(t.next_due_at - now) + else: + next_due = "-" + print(f" {tid:<10} {skill:<20} {trigger:<16} {tier:<7} {state:<10} {next_due}") + print(f"\n {len(tasks)} task(s).") + return 0 + + subcmd = args[0].lower() + + if subcmd == "status" and len(args) > 1: + task_id = args[1] + try: + status = await coordinator.status(task_id) + t = status.task + print(f" Task: {t.task_id}") + print(f" Skill: {t.skill_name}") + print(f" Trigger: {_format_trigger(t)}") + print(f" Tier: {t.execution_tier}") + print(f" State: {t.state}") + print(f" Runs: {t.run_count}" + (f" / {t.max_runs}" if t.max_runs > 0 else "")) + print(f" Running: {status.is_running}") + if t.next_due_at > 0: + now = time.time() + print(f" Next due: {_format_duration(t.next_due_at - now)}") + if t.cloud_worker_id: + print(f" Worker: {t.cloud_worker_id}") + except ValueError as e: + print(f" Error: {e}") + return 1 + return 0 + + if subcmd == "logs" and len(args) > 1: + task_id = args[1] + tail = 50 + if len(args) > 2: + try: + tail = int(args[2]) + except ValueError: + pass + try: + logs = await coordinator.logs(task_id, tail=tail) + if not logs: + print(" No logs available.") + else: + for line in logs: + print(f" {line}") + except ValueError as e: + print(f" Error: {e}") + return 1 + return 0 + + if subcmd == "cancel" and len(args) > 1: + task_id = args[1] + try: + await coordinator.cancel(task_id) + print(f" Cancelled: {task_id[:8]}") + except ValueError as e: + print(f" Error: {e}") + return 1 + return 0 + + # Unknown subcommand + print("Usage: tasks [status | logs | cancel ]") + print(" (no args) — List all scheduled tasks") + print(" status — Show detailed status") + print(" logs — Show recent logs") + print(" cancel — Cancel a task") + return 0 diff --git a/src/leapflow/config.py b/src/leapflow/config.py index ff3095f..649fd71 100644 --- a/src/leapflow/config.py +++ b/src/leapflow/config.py @@ -335,6 +335,12 @@ class Settings: hub_repo_prefix: str = "leapflow-" hub_search_sources: str = "modelscope" # comma-separated backend names for multi-source search + # ── Scheduler ── + scheduler_enabled: bool = True + scheduler_tick_seconds: int = 60 + scheduler_grace_seconds: float = 120.0 + scheduler_default_tier: str = "auto" # auto | local | cloud + @property def has_llm_credentials(self) -> bool: return bool(self.llm_api_key.strip()) @@ -653,6 +659,12 @@ def load_config(*, env_file: str | Path | None = None) -> Settings: hub_repo_prefix = os.getenv("LEAPFLOW_HUB_REPO_PREFIX", "leapflow-") hub_search_sources = os.getenv("LEAPFLOW_HUB_SEARCH_SOURCES", "modelscope") + # Scheduler + scheduler_enabled = _bool("LEAPFLOW_SCHEDULER_ENABLED", "true") + scheduler_tick_seconds = int(os.getenv("LEAPFLOW_SCHEDULER_TICK_SECONDS", "60")) + scheduler_grace_seconds = float(os.getenv("LEAPFLOW_SCHEDULER_GRACE_SECONDS", "120.0")) + scheduler_default_tier = os.getenv("LEAPFLOW_SCHEDULER_DEFAULT_TIER", "auto") + settings = Settings( llm_api_key=api_key, llm_base_url=base_url.rstrip("/"), @@ -840,6 +852,11 @@ def load_config(*, env_file: str | Path | None = None) -> Settings: hub_sync_copilot=hub_sync_copilot, hub_repo_prefix=hub_repo_prefix, hub_search_sources=hub_search_sources, + # Scheduler + scheduler_enabled=scheduler_enabled, + scheduler_tick_seconds=scheduler_tick_seconds, + scheduler_grace_seconds=scheduler_grace_seconds, + scheduler_default_tier=scheduler_default_tier, ) if not settings.llm_api_key: diff --git a/src/leapflow/hub/client.py b/src/leapflow/hub/client.py index 596d6e0..40fb132 100644 --- a/src/leapflow/hub/client.py +++ b/src/leapflow/hub/client.py @@ -17,6 +17,7 @@ SkillManifest, SkillSummary, UserInfo, + VersionConflictError, Visibility, ) @@ -189,6 +190,8 @@ async def push( skill_name: str | None = None, repo_id: str | None = None, visibility: Visibility | None = None, + *, + force: bool = False, ) -> PushResult: """Push a skill bundle to the remote hub. @@ -197,9 +200,13 @@ async def push( skill_name: Skill name (used to construct repo_id if repo_id not given). repo_id: Explicit repo_id (overrides auto-construction). visibility: Repository visibility (defaults to client default). + force: Skip version conflict check if True. Returns: PushResult with published version and URL. + + Raises: + VersionConflictError: If remote has same or newer version (unless force=True). """ if repo_id is None: name = skill_name or bundle.manifest.name @@ -207,8 +214,42 @@ async def push( vis = visibility or self._default_visibility + if not force: + conflict = await self._check_push_conflict(repo_id, bundle.manifest.version) + if conflict: + raise VersionConflictError(conflict) + return await self.backend.push_skill(bundle, repo_id, vis) + async def _check_push_conflict(self, repo_id: str, local_version: str) -> str | None: + """Check if remote already has same or newer version.""" + try: + versions = await self.backend.get_skill_versions(repo_id) + if not versions: + return None + latest = versions[0] + if self._version_gte(latest.version, local_version): + return ( + f"Remote already has v{latest.version} " + f"(pushing v{local_version}). " + f"Increment version or use force=True." + ) + except Exception: + return None # Repo doesn't exist yet or network error + return None + + @staticmethod + def _version_gte(remote: str, local: str) -> bool: + """Check if remote version >= local version (semver comparison).""" + def _parse(v: str): + v = v.lstrip("vV") + parts = v.split(".") + return tuple(int(p) for p in parts if p.isdigit()) + try: + return _parse(remote) >= _parse(local) + except (ValueError, IndexError): + return remote >= local # Fallback to string comparison + async def pull( self, repo_id: str, diff --git a/src/leapflow/hub/protocol.py b/src/leapflow/hub/protocol.py index b40ddae..ce15140 100644 --- a/src/leapflow/hub/protocol.py +++ b/src/leapflow/hub/protocol.py @@ -95,6 +95,11 @@ class SkillManifest: author: str = "" hub_type: str = "" repo_id: str = "" + # ── Team collaboration fields ── + content_hash: str = "" # SHA256[:16] of (source_code + parameters + triggers) + updated_by: str = "" # Last modifier username + updated_at: str = "" # Last modification ISO timestamp + tags: List[str] = field(default_factory=list) # Classification tags @dataclass(frozen=True) @@ -209,3 +214,8 @@ async def delete_skill(self, repo_id: str) -> None: PermissionError: If the user lacks delete permissions. """ ... + + +class VersionConflictError(Exception): + """Raised when push detects a version conflict with remote.""" + pass diff --git a/src/leapflow/hub/security.py b/src/leapflow/hub/security.py index 127134c..b59fec0 100644 --- a/src/leapflow/hub/security.py +++ b/src/leapflow/hub/security.py @@ -76,6 +76,38 @@ class ContentSanitizer: "pii", "Possible phone number: {match}", ), + # SSH private keys + ( + re.compile(r"-----BEGIN\s+(?:RSA|DSA|EC|OPENSSH)\s+PRIVATE\s+KEY-----"), + "high", + "token", + "SSH private key detected", + ), + # AWS access key + ( + re.compile(r"AKIA[0-9A-Z]{16}"), + "high", + "token", + "AWS access key: {match}", + ), + # AWS secret key pattern + ( + re.compile(r"(?i)aws[_-]?secret[_-]?access[_-]?key\s*[=:]\s*[^\s]{20,}"), + "high", + "token", + "AWS secret key pattern: {match}", + ), + # Hardcoded IP addresses (non-loopback, non-placeholder) + ( + re.compile( + r"\b(?!127\.0\.0\.1|0\.0\.0\.0|localhost)" + r"(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}" + r"(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\b" + ), + "low", + "pii", + "IP address: {match}", + ), ] def scan(self, bundle: SkillBundle) -> List[SanitizationWarning]: @@ -128,6 +160,10 @@ def _scan_content( ) ) + def has_blocking_findings(self, warnings: List[SanitizationWarning]) -> bool: + """Check if any findings are severe enough to block push without --force.""" + return any(w.severity == "high" for w in warnings) + # ─── Security Auditor ──────────────────────────────────────────────────────── diff --git a/src/leapflow/hub/serializer.py b/src/leapflow/hub/serializer.py index 21cbe14..c4bee6a 100644 --- a/src/leapflow/hub/serializer.py +++ b/src/leapflow/hub/serializer.py @@ -6,6 +6,8 @@ from __future__ import annotations +import dataclasses +import hashlib import json import logging from typing import TYPE_CHECKING, Dict @@ -27,6 +29,15 @@ logger.debug("PyYAML not available; using JSON fallback for manifest serialization") +def compute_content_hash(bundle: "SkillBundle") -> str: + """Compute SHA256[:16] over skill's semantic content for integrity tracking.""" + hasher = hashlib.sha256() + hasher.update((bundle.source_code or "").encode()) + hasher.update(json.dumps(bundle.manifest.parameters, sort_keys=True).encode()) + hasher.update(json.dumps(sorted(bundle.manifest.triggers)).encode()) + return hasher.hexdigest()[:16] + + # ─── YAML Helpers ──────────────────────────────────────────────────────────── @@ -45,6 +56,10 @@ def _manifest_to_dict(manifest: SkillManifest) -> dict: "author": manifest.author, "hub_type": manifest.hub_type, "repo_id": manifest.repo_id, + "content_hash": manifest.content_hash, + "updated_by": manifest.updated_by, + "updated_at": manifest.updated_at, + "tags": manifest.tags, } @@ -63,6 +78,10 @@ def _dict_to_manifest(data: dict) -> SkillManifest: author=data.get("author", ""), hub_type=data.get("hub_type", ""), repo_id=data.get("repo_id", ""), + content_hash=data.get("content_hash", ""), + updated_by=data.get("updated_by", ""), + updated_at=data.get("updated_at", ""), + tags=data.get("tags", []) or [], ) @@ -121,7 +140,7 @@ def export_skill(self, stored_skill: dict) -> SkillBundle: author=stored_skill.get("author", ""), ) - return SkillBundle( + bundle = SkillBundle( manifest=manifest, source_code=stored_skill.get("source_code", ""), trajectory_skeleton=stored_skill.get("trajectory_skeleton", ""), @@ -129,6 +148,13 @@ def export_skill(self, stored_skill: dict) -> SkillBundle: readme=stored_skill.get("readme", ""), ) + # Auto-fill content hash if not already set + if not bundle.manifest.content_hash: + updated_manifest = dataclasses.replace(bundle.manifest, content_hash=compute_content_hash(bundle)) + bundle = dataclasses.replace(bundle, manifest=updated_manifest) + + return bundle + def import_skill(self, bundle: SkillBundle) -> dict: """Convert bundle back to fields suitable for SkillLibraryStore.save(). @@ -153,6 +179,10 @@ def import_skill(self, bundle: SkillBundle) -> dict: "readme": bundle.readme, "hub_type": m.hub_type, "repo_id": m.repo_id, + "content_hash": m.content_hash, + "updated_by": m.updated_by, + "updated_at": m.updated_at, + "tags": m.tags, } def bundle_to_files(self, bundle: SkillBundle) -> Dict[str, str]: diff --git a/src/leapflow/hub/sync.py b/src/leapflow/hub/sync.py index 6ad8937..fdcae84 100644 --- a/src/leapflow/hub/sync.py +++ b/src/leapflow/hub/sync.py @@ -25,7 +25,7 @@ class SyncAction: """A single sync operation to perform.""" - direction: str # "push" | "pull" + direction: str # "push" | "pull" | "conflict" skill_name: str local_version: str # "" if not local remote_version: str # "" if not remote @@ -227,9 +227,19 @@ async def compute_plan( ) ) else: - # Same version — resolve based on strategy for edge cases - # (versions equal means no action needed) - pass + # Same version — check content hash for silent divergence + local_hash = getattr(local, "content_hash", "") + remote_hash = getattr(remote, "content_hash", "") if hasattr(remote, "content_hash") else "" + if local_hash and remote_hash and local_hash != remote_hash: + actions.append(SyncAction( + direction="conflict", + skill_name=name, + local_version=local.version, + remote_version=remote.version, + reason="content_diverged", + repo_id=getattr(remote, "repo_id", ""), + )) + # else: truly identical, skip # Process skills only on remote if not push_only: @@ -265,12 +275,13 @@ def _resolve_conflicts(self, actions: List[SyncAction]) -> List[SyncAction]: (user must resolve manually). """ if self._strategy == "manual": - # In manual mode, skip any action where both versions exist - # and a clear winner isn't determined by version comparison + # In manual mode, keep clear-cut actions and explicit conflicts + # (conflicts are surfaced to user for manual resolution) return [ a for a in actions - if a.reason in ("local_only", "remote_only") + if a.direction == "conflict" + or a.reason in ("local_only", "remote_only") or (a.reason == "local_newer" and a.direction == "push") or (a.reason == "remote_newer" and a.direction == "pull") ] @@ -298,7 +309,11 @@ async def execute_plan( for action in plan.actions: try: - if action.direction == "push": + if action.direction == "conflict": + logger.warning("Conflict: skill '%s' v%s has diverged content. Resolve manually.", action.skill_name, action.local_version) + completed.append(f"CONFLICT: {action.skill_name} (content diverged, resolve manually)") + continue + elif action.direction == "push": desc = await self._execute_push(action) else: desc = await self._execute_pull(action) diff --git a/src/leapflow/scheduler/__init__.py b/src/leapflow/scheduler/__init__.py new file mode 100644 index 0000000..fe7bcbc --- /dev/null +++ b/src/leapflow/scheduler/__init__.py @@ -0,0 +1,42 @@ +"""Long-horizon async task scheduler — local and cloud execution.""" + +from leapflow.scheduler.types import ( + ArmedTask, + TaskState, + ExecutionTier, + TaskStatus, + Trigger, + ComputeBackend as ComputeBackendProtocol, + SkillExecutor, +) +from leapflow.scheduler.store import TaskStore +from leapflow.scheduler.local_scheduler import LocalScheduler +from leapflow.scheduler.cloud_dispatcher import CloudDispatcher +from leapflow.scheduler.worker_packager import WorkerPackager +from leapflow.scheduler.coordinator import TaskCoordinator +from leapflow.scheduler.triggers import create_trigger +from leapflow.scheduler.compute import ComputeBackend +from leapflow.scheduler.compute.modelscope_studio import ModelScopeStudioBackend + +__all__ = [ + # Types & enums + "ArmedTask", + "TaskState", + "ExecutionTier", + "TaskStatus", + "Trigger", + "ComputeBackend", + "ComputeBackendProtocol", + "SkillExecutor", + # Store + "TaskStore", + # Schedulers & dispatchers + "LocalScheduler", + "CloudDispatcher", + "WorkerPackager", + "TaskCoordinator", + # Trigger factory + "create_trigger", + # Compute backends + "ModelScopeStudioBackend", +] diff --git a/src/leapflow/scheduler/cloud_dispatcher.py b/src/leapflow/scheduler/cloud_dispatcher.py new file mode 100644 index 0000000..ca673a9 --- /dev/null +++ b/src/leapflow/scheduler/cloud_dispatcher.py @@ -0,0 +1,194 @@ +"""Cloud dispatcher — orchestrates cloud task deployment lifecycle. + +Workflow: package → create worker → inject secrets → deploy → monitor. +Provides a high-level interface over ComputeBackend and WorkerPackager. +""" + +from __future__ import annotations + +import json +import logging +from typing import Dict, List, Optional + +from leapflow.scheduler.compute.protocol import ComputeBackend +from leapflow.scheduler.types import ArmedTask +from leapflow.scheduler.worker_packager import WorkerPackager + +logger = logging.getLogger(__name__) + + +class CloudDispatcher: + """Orchestrates cloud task deployment lifecycle. + + Coordinates the WorkerPackager and a ComputeBackend to deploy tasks + as self-contained cloud workers with full secret injection. + + Usage: + backend = ModelScopeStudioBackend() + packager = WorkerPackager() + dispatcher = CloudDispatcher(backend, packager) + worker_id = await dispatcher.deploy(task) + """ + + def __init__( + self, + compute_backend: ComputeBackend, + packager: Optional[WorkerPackager] = None, + ) -> None: + """Initialize the dispatcher. + + Args: + compute_backend: Backend responsible for creating/managing workers. + packager: Worker packager (defaults to a new WorkerPackager instance). + """ + self._backend = compute_backend + self._packager = packager or WorkerPackager() + + @property + def backend_type(self) -> str: + """Return the underlying compute backend type.""" + return self._backend.backend_type + + async def deploy( + self, + task: ArmedTask, + skill_source: str = "", + context: Optional[dict] = None, + ) -> str: + """Deploy a task as a cloud worker. + + Performs the full lifecycle: package → create → inject secrets → deploy. + + Args: + task: The armed task to deploy. + skill_source: Optional skill source code to bundle. + context: Optional context snapshot for the worker. + + Returns: + The worker_id of the deployed instance. + + Raises: + RuntimeError: If any step in the deployment pipeline fails. + """ + worker_id = f"leapflow-task-{task.task_id[:8]}" + + # 1. Package the worker + logger.info("Packaging worker for task %s...", task.task_id[:8]) + package_path = self._packager.package(task, skill_source, context) + + try: + # 2. Create remote worker + logger.info("Creating remote worker: %s", worker_id) + await self._backend.create_worker( + worker_id, package_path, visibility="private" + ) + + # 3. Inject secrets (full task config as env var) + logger.info("Injecting secrets into worker: %s", worker_id) + task_config = self._build_task_config(task) + secrets: Dict[str, str] = { + "LEAPFLOW_TASK_CONFIG": json.dumps(task_config, ensure_ascii=False), + } + await self._backend.inject_secrets(worker_id, secrets) + + # 4. Deploy + logger.info("Deploying worker: %s", worker_id) + await self._backend.deploy(worker_id) + + logger.info( + "Successfully deployed task %s as worker %s", + task.task_id[:8], + worker_id, + ) + return worker_id + finally: + # Cleanup temp package directory + try: + import shutil + shutil.rmtree(package_path, ignore_errors=True) + except Exception: + logger.debug("Failed to cleanup package at %s", package_path) + + async def status(self, worker_id: str) -> str: + """Get the current status of a deployed worker. + + Args: + worker_id: The worker identifier. + + Returns: + Status string: 'building' | 'running' | 'stopped' | 'failed' | 'unknown'. + """ + return await self._backend.get_status(worker_id) + + async def logs(self, worker_id: str, tail: int = 50) -> List[str]: + """Retrieve recent log lines from a deployed worker. + + Args: + worker_id: The worker identifier. + tail: Number of recent log lines to retrieve. + + Returns: + List of log line strings. + """ + return await self._backend.get_logs(worker_id, tail=tail) + + async def stop(self, worker_id: str) -> None: + """Stop a running worker without destroying it. + + Args: + worker_id: The worker identifier. + """ + logger.info("Stopping worker: %s", worker_id) + await self._backend.stop(worker_id) + + async def destroy(self, worker_id: str) -> None: + """Stop and permanently delete a worker. + + Args: + worker_id: The worker identifier. + """ + logger.info("Destroying worker: %s", worker_id) + await self._backend.destroy(worker_id) + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + @staticmethod + def _build_task_config(task: ArmedTask) -> dict: + """Build the LEAPFLOW_TASK_CONFIG payload from an ArmedTask. + + Ensures all fields are JSON-serializable plain dicts/primitives. + """ + trigger_config = task.trigger_config + if isinstance(trigger_config, str): + try: + trigger_config = json.loads(trigger_config) + except (json.JSONDecodeError, TypeError): + trigger_config = {"raw": trigger_config} + + parameters = task.parameters + if isinstance(parameters, str): + try: + parameters = json.loads(parameters) + except (json.JSONDecodeError, TypeError): + parameters = {"raw": parameters} + + # Derive check_interval from trigger configuration + check_interval = 60 # default + if task.trigger_type == "interval": + interval_s = trigger_config.get("interval_seconds", 60) + check_interval = max(30, min(int(interval_s), 3600)) # 30s ~ 1h range + elif task.trigger_type == "cron": + check_interval = 300 # 5min check for cron (platform handles exact timing) + elif task.trigger_type in ("condition", "event"): + check_interval = 60 # condition/event need frequent polling + + return { + "task_id": task.task_id, + "skill_name": task.skill_name, + "trigger_config": trigger_config, + "parameters": parameters, + "max_runs": task.max_runs, + "check_interval": check_interval, + } diff --git a/src/leapflow/scheduler/compute/__init__.py b/src/leapflow/scheduler/compute/__init__.py new file mode 100644 index 0000000..aaed145 --- /dev/null +++ b/src/leapflow/scheduler/compute/__init__.py @@ -0,0 +1,5 @@ +"""Compute backends for cloud task execution.""" + +from leapflow.scheduler.compute.protocol import ComputeBackend + +__all__ = ["ComputeBackend"] diff --git a/src/leapflow/scheduler/compute/modelscope_studio.py b/src/leapflow/scheduler/compute/modelscope_studio.py new file mode 100644 index 0000000..bda9ea4 --- /dev/null +++ b/src/leapflow/scheduler/compute/modelscope_studio.py @@ -0,0 +1,232 @@ +"""ModelScope Studio compute backend implementation. + +Each task is deployed as a private Docker-based Studio that runs a LeapFlow +worker loop. Secrets are injected as environment variables into the container. + +Uses asyncio.to_thread to wrap synchronous ModelScope Hub SDK calls. +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path +from typing import Any, Dict, List + +logger = logging.getLogger(__name__) + +_SDK_INSTALL_HINT = ( + "ModelScope Hub SDK not found. Install it with:\n" + " pip install modelscope-hub\n" + "or:\n" + " uv pip install modelscope-hub" +) + +# Status mapping from ModelScope Studio states to our canonical states. +_STATUS_MAP: Dict[str, str] = { + "building": "building", + "running": "running", + "stopped": "stopped", + "failed": "failed", + "pending": "building", + "deploying": "building", + "error": "failed", +} + + +class ModelScopeStudioBackend: + """ComputeBackend implementation using ModelScope Hub Studio (Docker mode). + + Each task is deployed as a private Docker-based Studio that runs + a LeapFlow worker loop. Secrets are injected as environment variables. + + Conforms to the ComputeBackend protocol defined in + ``leapflow.scheduler.compute.protocol``. + """ + + backend_type = "modelscope_studio" + + def __init__(self) -> None: + """Initialize the backend with lazy SDK import. + + Raises: + ImportError: If the modelscope_hub SDK is not installed. + """ + self._api: Any = None + self._ensure_sdk() + + def _ensure_sdk(self) -> None: + """Verify SDK availability and create API instance.""" + try: + from modelscope_hub import HubApi # type: ignore[import-untyped] + + self._api = HubApi() + except ImportError: + raise ImportError(_SDK_INSTALL_HINT) from None + + # ------------------------------------------------------------------ + # ComputeBackend interface + # ------------------------------------------------------------------ + + async def create_worker( + self, + worker_id: str, + package_path: Path, + *, + visibility: str = "private", + ) -> str: + """Create a Studio repo with Docker SDK type and upload the package. + + Args: + worker_id: Repo identifier for the Studio. + package_path: Path to the worker package directory. + visibility: 'private' or 'public'. + + Returns: + The worker_id as confirmation. + """ + # Create the Studio repository + await asyncio.to_thread( + self._api.create_repo, + repo_id=worker_id, + repo_type="studio", + sdk_type="docker", + visibility=visibility, + exist_ok=True, + ) + logger.info("Created Studio repo: %s (visibility=%s)", worker_id, visibility) + + # Upload the deployable package + await asyncio.to_thread( + self._api.upload_folder, + repo_id=worker_id, + repo_type="studio", + folder_path=str(package_path), + commit_message=f"Deploy worker package for {worker_id}", + ) + logger.info("Uploaded worker package to Studio: %s", worker_id) + + return worker_id + + async def inject_secrets( + self, + worker_id: str, + secrets: Dict[str, str], + ) -> None: + """Add secrets to the Studio (available as env vars in the container). + + Args: + worker_id: Target Studio repo identifier. + secrets: Key-value pairs to inject. + """ + for key, value in secrets.items(): + await asyncio.to_thread( + self._api.add_secret, + repo_id=worker_id, + repo_type="studio", + key=key, + value=value, + ) + logger.info( + "Injected %d secret(s) into Studio: %s", len(secrets), worker_id + ) + + async def deploy(self, worker_id: str) -> None: + """Trigger Studio deployment. + + Args: + worker_id: Target Studio repo identifier. + """ + await asyncio.to_thread( + self._api.deploy_repo, + repo_id=worker_id, + repo_type="studio", + ) + logger.info("Deployed Studio: %s", worker_id) + + async def stop(self, worker_id: str) -> None: + """Stop a running Studio. + + Args: + worker_id: Target Studio repo identifier. + """ + await asyncio.to_thread( + self._api.stop_repo, + repo_id=worker_id, + repo_type="studio", + ) + logger.info("Stopped Studio: %s", worker_id) + + async def get_status(self, worker_id: str) -> str: + """Get Studio status. + + Args: + worker_id: Target Studio repo identifier. + + Returns: + Canonical status: 'building' | 'running' | 'stopped' | 'failed' | 'unknown'. + """ + try: + info = await asyncio.to_thread( + self._api.get_repo, + repo_id=worker_id, + repo_type="studio", + ) + raw_status = str(info.get("status", "unknown")).lower() + return _STATUS_MAP.get(raw_status, "unknown") + except Exception as e: + logger.warning("Failed to get status for '%s': %s", worker_id, e) + return "unknown" + + async def get_logs(self, worker_id: str, *, tail: int = 50) -> List[str]: + """Get Studio run logs. + + Args: + worker_id: Target Studio repo identifier. + tail: Number of recent log lines to retrieve. + + Returns: + List of log line strings. + """ + try: + logs = await asyncio.to_thread( + self._api.get_repo_logs, + repo_id=worker_id, + repo_type="studio", + log_type="run", + page_size=tail, + ) + # SDK may return a list of dicts or list of strings + if isinstance(logs, list): + return [ + str(entry.get("message", entry)) if isinstance(entry, dict) else str(entry) + for entry in logs[-tail:] + ] + return [] + except Exception as e: + logger.warning("Failed to get logs for '%s': %s", worker_id, e) + return [] + + async def destroy(self, worker_id: str) -> None: + """Stop and permanently delete the Studio. + + Args: + worker_id: Target Studio repo identifier. + """ + # Best-effort stop before deletion + try: + await self.stop(worker_id) + except Exception: + pass # Already stopped or non-existent — proceed to delete + + try: + await asyncio.to_thread( + self._api.delete_repo, + repo_id=worker_id, + repo_type="studio", + ) + logger.info("Destroyed Studio: %s", worker_id) + except Exception as e: + raise RuntimeError( + f"Failed to destroy Studio '{worker_id}': {e}" + ) from e diff --git a/src/leapflow/scheduler/compute/protocol.py b/src/leapflow/scheduler/compute/protocol.py new file mode 100644 index 0000000..f084086 --- /dev/null +++ b/src/leapflow/scheduler/compute/protocol.py @@ -0,0 +1,103 @@ +"""Compute backend protocol for cloud task execution. + +Defines the abstract interface that all compute backends must implement. +Backends manage the full lifecycle of remote workers: create → deploy → monitor → destroy. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Dict, List, Protocol, runtime_checkable + + +@runtime_checkable +class ComputeBackend(Protocol): + """Abstract compute backend — ModelScope Studio, HF Spaces, local Docker. + + Implementations wrap platform-specific SDKs to provide a uniform + interface for deploying and managing cloud workers. + """ + + @property + def backend_type(self) -> str: + """Identifier string for this backend (e.g. 'modelscope_studio').""" + ... + + async def create_worker( + self, + worker_id: str, + package_path: Path, + *, + visibility: str = "private", + ) -> str: + """Create a remote worker instance. + + Args: + worker_id: Unique identifier for the worker. + package_path: Path to the deployable package directory. + visibility: 'private' or 'public'. + + Returns: + Instance identifier or URL. + """ + ... + + async def inject_secrets( + self, + worker_id: str, + secrets: Dict[str, str], + ) -> None: + """Inject environment variables/secrets into the worker. + + Args: + worker_id: Target worker identifier. + secrets: Key-value pairs to inject as environment variables. + """ + ... + + async def deploy(self, worker_id: str) -> None: + """Trigger deployment/start of the worker. + + Args: + worker_id: Target worker identifier. + """ + ... + + async def stop(self, worker_id: str) -> None: + """Stop a running worker without destroying it. + + Args: + worker_id: Target worker identifier. + """ + ... + + async def get_status(self, worker_id: str) -> str: + """Get worker status. + + Args: + worker_id: Target worker identifier. + + Returns: + One of: 'building' | 'running' | 'stopped' | 'failed' | 'unknown'. + """ + ... + + async def get_logs(self, worker_id: str, *, tail: int = 50) -> List[str]: + """Retrieve recent log lines from the worker. + + Args: + worker_id: Target worker identifier. + tail: Number of recent log lines to retrieve. + + Returns: + List of log line strings. + """ + ... + + async def destroy(self, worker_id: str) -> None: + """Stop and permanently delete the worker and its resources. + + Args: + worker_id: Target worker identifier. + """ + ... diff --git a/src/leapflow/scheduler/coordinator.py b/src/leapflow/scheduler/coordinator.py new file mode 100644 index 0000000..69d817d --- /dev/null +++ b/src/leapflow/scheduler/coordinator.py @@ -0,0 +1,295 @@ +"""Unified task orchestrator — routes armed tasks to local or cloud execution. + +Tier decision heuristic: +- AUTO: interval < 1h → local; interval >= 1h or condition → cloud +- LOCAL/CLOUD: explicit override + +Both tiers share the same TaskStore as single source of truth. +""" + +from __future__ import annotations + +import logging +import re +import time +from typing import List, Optional + +from leapflow.scheduler.store import TaskStore +from leapflow.scheduler.triggers import create_trigger +from leapflow.scheduler.types import ArmedTask, ExecutionTier, TaskState, TaskStatus + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Trigger expression parser +# --------------------------------------------------------------------------- + +_INTERVAL_PATTERN = re.compile( + r"^(?:every\s+)?(\d+(?:\.\d+)?)\s*([smhd]|sec|min|hour|hours|day|days|minutes?)$", + re.IGNORECASE, +) +_CRON_PATTERN = re.compile(r"^(\S+\s+){4}\S+$") # 5 space-separated fields +_EVENT_PREFIX = "event:" +_CONDITION_PREFIX = "condition:" + +_UNIT_TO_SECONDS = { + "s": 1.0, "sec": 1.0, + "m": 60.0, "min": 60.0, "minute": 60.0, "minutes": 60.0, + "h": 3600.0, "hour": 3600.0, "hours": 3600.0, + "d": 86400.0, "day": 86400.0, "days": 86400.0, +} + + +def parse_trigger_expression(expr: str) -> tuple[str, dict]: + """Parse a human-friendly trigger expression into (trigger_type, config). + + Supported formats: + - "30m" / "2h" / "1d" → IntervalTrigger + - "every 5m" → IntervalTrigger + - "0 9 * * *" → CronTrigger (5-field cron) + - "event:ci.passed" → EventTrigger + - "condition:file_count > 50" → ConditionTrigger + + Returns: + Tuple of (trigger_type, trigger_config dict). + + Raises: + ValueError: If expression cannot be parsed. + """ + expr = expr.strip() + if not expr: + raise ValueError("Trigger expression cannot be empty.") + + # Event trigger + if expr.lower().startswith(_EVENT_PREFIX): + event_name = expr[len(_EVENT_PREFIX):].strip() + if not event_name: + raise ValueError("Event trigger must specify an event name, e.g. 'event:ci.passed'") + return "event", {"event_pattern": event_name} + + # Condition trigger + if expr.lower().startswith(_CONDITION_PREFIX): + condition_expr = expr[len(_CONDITION_PREFIX):].strip() + if not condition_expr: + raise ValueError( + "Condition trigger must specify an expression, e.g. 'condition:file_count > 50'" + ) + return "condition", {"expression": condition_expr} + + # Cron trigger (5 fields separated by spaces) + if _CRON_PATTERN.match(expr): + return "cron", {"expression": expr} + + # Interval trigger + match = _INTERVAL_PATTERN.match(expr) + if match: + value = float(match.group(1)) + unit = match.group(2).lower() + multiplier = _UNIT_TO_SECONDS.get(unit) + if multiplier is None: + raise ValueError(f"Unknown interval unit: {unit!r}") + interval_seconds = value * multiplier + return "interval", {"interval_seconds": interval_seconds} + + raise ValueError( + f"Cannot parse trigger expression: {expr!r}. " + f"Supported formats: '30m', 'every 5m', '0 9 * * *', " + f"'event:', 'condition:'" + ) + + +# --------------------------------------------------------------------------- +# TaskCoordinator +# --------------------------------------------------------------------------- + + +class TaskCoordinator: + """Unified task orchestrator — routes armed tasks to local or cloud execution. + + Tier decision heuristic: + - AUTO: interval < 1h → local; interval >= 1h or condition/event → cloud + - LOCAL/CLOUD: explicit override + + Both tiers share the same TaskStore as single source of truth. + """ + + def __init__( + self, + store: TaskStore, + local_scheduler: Optional["LocalScheduler"] = None, + cloud_dispatcher: Optional["CloudDispatcher"] = None, + default_tier: str = "auto", + ) -> None: + self._store = store + self._local = local_scheduler + self._cloud = cloud_dispatcher + self._default_tier = default_tier + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def arm( + self, + skill_name: str, + trigger_expr: str, + *, + execution_tier: str = "auto", + max_runs: int = -1, + parameters: Optional[dict] = None, + context_snapshot: Optional[dict] = None, + ) -> ArmedTask: + """Create and register an armed task. + + 1. Parse trigger expression → Trigger object + 2. Decide execution tier + 3. Create ArmedTask with unique ID + 4. Persist to store + 5. Route to local_scheduler.register() or cloud_dispatcher.deploy() + """ + # 1. Parse trigger + trigger_type, trigger_config = parse_trigger_expression(trigger_expr) + + # 2. Decide tier + tier = execution_tier if execution_tier != "auto" else self._default_tier + if tier == "auto": + tier = self._decide_tier(trigger_type, trigger_config) + + # 3. Validate tier availability + if tier == ExecutionTier.LOCAL.value and self._local is None: + raise RuntimeError( + "Local scheduler not available. Use --cloud or configure a local scheduler." + ) + if tier == ExecutionTier.CLOUD.value and self._cloud is None: + raise RuntimeError( + "Cloud dispatcher not available. Use --local or configure a cloud backend." + ) + + # 4. Create trigger instance to compute next_due_at + now = time.time() + trigger = create_trigger(trigger_type, trigger_config) + trigger.advance(now) + + # 5. Create ArmedTask + task = ArmedTask( + skill_name=skill_name, + trigger_type=trigger_type, + trigger_config=trigger_config, + state=TaskState.ARMED.value, + execution_tier=tier, + context_snapshot=context_snapshot or {}, + parameters=parameters or {}, + max_runs=max_runs, + next_due_at=trigger.next_due_at, + ) + + # 6. Persist + self._store.save(task) + + # 7. Route to execution backend + if tier == ExecutionTier.LOCAL.value: + assert self._local is not None + await self._local.register(task) + logger.info("Armed task %s → local (skill=%s)", task.task_id[:8], skill_name) + elif tier == ExecutionTier.CLOUD.value: + assert self._cloud is not None + worker_id = await self._cloud.deploy(task) + task.cloud_worker_id = worker_id + self._store.save(task) + logger.info("Armed task %s → cloud (skill=%s)", task.task_id[:8], skill_name) + + return task + + async def cancel(self, task_id: str) -> None: + """Cancel a task (local: suspend, cloud: stop worker).""" + task = self._store.load(task_id) + if task is None: + raise ValueError(f"Task not found: {task_id}") + + if task.execution_tier == ExecutionTier.LOCAL.value and self._local: + await self._local.cancel(task_id) + elif task.execution_tier == ExecutionTier.CLOUD.value and self._cloud: + if task.cloud_worker_id: + await self._cloud.stop(task.cloud_worker_id) + self._store.update_state(task_id, TaskState.SUSPENDED.value) + else: + self._store.update_state(task_id, TaskState.SUSPENDED.value) + + logger.info("Cancelled task %s", task_id[:8]) + + async def status(self, task_id: str) -> TaskStatus: + """Unified status query.""" + task = self._store.load(task_id) + if task is None: + raise ValueError(f"Task not found: {task_id}") + + is_running = task.state == TaskState.EXECUTING.value + logs_tail: List[str] = [] + + if task.execution_tier == ExecutionTier.CLOUD.value and self._cloud and task.cloud_worker_id: + try: + cloud_status = await self._cloud.status(task.cloud_worker_id) + is_running = cloud_status in ("running", "building") + except Exception: + pass + + return TaskStatus(task=task, is_running=is_running, logs_tail=logs_tail) + + async def list_tasks(self) -> List[ArmedTask]: + """List all tasks from store.""" + return self._store.load_all() + + async def logs(self, task_id: str, tail: int = 50) -> List[str]: + """Get logs (local: from last execution, cloud: from Studio logs).""" + task = self._store.load(task_id) + if task is None: + raise ValueError(f"Task not found: {task_id}") + + if task.execution_tier == ExecutionTier.CLOUD.value and self._cloud and task.cloud_worker_id: + return await self._cloud.logs(task.cloud_worker_id, tail=tail) + + # Local tasks: no log store yet, return placeholder + return [f"[local] Task {task_id[:8]}: state={task.state}, runs={task.run_count}"] + + # ------------------------------------------------------------------ + # Tier decision heuristic + # ------------------------------------------------------------------ + + def _decide_tier(self, trigger_type: str, trigger_config: dict) -> str: + """Heuristic tier decision. + + - condition/event → cloud (needs persistent monitoring) + - interval >= 3600s → cloud + - interval < 3600s → local + - cron with daily+ frequency → cloud + """ + if trigger_type in ("condition", "event"): + # Needs persistent monitoring — prefer cloud + return ExecutionTier.CLOUD.value if self._cloud else ExecutionTier.LOCAL.value + + if trigger_type == "interval": + interval_s = trigger_config.get("interval_seconds", 0) + if interval_s >= 3600: + return ExecutionTier.CLOUD.value if self._cloud else ExecutionTier.LOCAL.value + return ExecutionTier.LOCAL.value if self._local else ExecutionTier.CLOUD.value + + if trigger_type == "cron": + # Heuristic: check if frequency is daily or less often + cron_expr = trigger_config.get("expression", "") + parts = cron_expr.split() + if len(parts) >= 5: + # If minute and hour are specific (not */N), it's likely daily+ + minute_field, hour_field = parts[0], parts[1] + if "*" not in minute_field and "*" not in hour_field: + return ExecutionTier.CLOUD.value if self._cloud else ExecutionTier.LOCAL.value + return ExecutionTier.LOCAL.value if self._local else ExecutionTier.CLOUD.value + + # Default: local if available + return ExecutionTier.LOCAL.value if self._local else ExecutionTier.CLOUD.value + + +# Deferred imports for type hints +from leapflow.scheduler.cloud_dispatcher import CloudDispatcher # noqa: E402 +from leapflow.scheduler.local_scheduler import LocalScheduler # noqa: E402 + +__all__ = ["TaskCoordinator", "parse_trigger_expression"] diff --git a/src/leapflow/scheduler/local_scheduler.py b/src/leapflow/scheduler/local_scheduler.py new file mode 100644 index 0000000..4d89aa0 --- /dev/null +++ b/src/leapflow/scheduler/local_scheduler.py @@ -0,0 +1,185 @@ +"""Local async scheduler — runs as background task in event loop. + +Design principles: +- At-most-once: advance next_due BEFORE execute (crash-safe) +- Fast-forward: on startup, skip overdue tasks beyond grace period +- Non-blocking: tick runs in background, never blocks REPL +- Confidence gating: low-confidence tasks emit notification instead of executing +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import time +from typing import Optional + +from leapflow.scheduler.store import TaskStore +from leapflow.scheduler.triggers import create_trigger +from leapflow.scheduler.types import ArmedTask, SkillExecutor, TaskState + +logger = logging.getLogger(__name__) + + +class LocalScheduler: + """Local async scheduler — runs as background task in event loop. + + Tick-based design: every ``tick_seconds`` (default 60s), the scheduler + queries the TaskStore for due tasks and dispatches them through the + SkillExecutor. + """ + + def __init__( + self, + store: TaskStore, + executor: SkillExecutor, + *, + tick_seconds: int = 60, + grace_seconds: float = 120.0, + ) -> None: + self._store = store + self._executor = executor + self._tick_seconds = tick_seconds + self._grace_seconds = grace_seconds + self._task: Optional[asyncio.Task] = None # type: ignore[type-arg] + self._running = False + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def start(self) -> None: + """Start background tick loop.""" + self._running = True + self._fast_forward() + self._task = asyncio.create_task(self._tick_loop()) + logger.info("LocalScheduler started (tick=%ds)", self._tick_seconds) + + async def stop(self) -> None: + """Gracefully stop the tick loop.""" + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + logger.info("LocalScheduler stopped") + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + async def register(self, task: ArmedTask) -> None: + """Register a new armed task.""" + self._store.save(task) + logger.info( + "Registered task %s (skill=%s, trigger=%s)", + task.task_id[:8], + task.skill_name, + task.trigger_type, + ) + + async def cancel(self, task_id: str) -> None: + """Cancel (suspend) a task.""" + self._store.update_state(task_id, TaskState.SUSPENDED.value) + logger.info("Cancelled task %s", task_id[:8]) + + # ------------------------------------------------------------------ + # Tick loop + # ------------------------------------------------------------------ + + async def _tick_loop(self) -> None: + """Background loop: check and execute due tasks every tick.""" + while self._running: + try: + await self._tick() + except Exception as e: + logger.error("Scheduler tick error: %s", e, exc_info=True) + await asyncio.sleep(self._tick_seconds) + + async def _tick(self) -> None: + """Single tick: find due tasks, advance, execute.""" + now = time.time() + due_tasks = self._store.get_due_tasks(now) + + for task in due_tasks: + await self._execute_task(task, now) + + async def _execute_task(self, task: ArmedTask, now: float) -> None: + """Execute a single due task with at-most-once semantics.""" + # At-most-once: advance BEFORE execute + trigger = create_trigger( + task.trigger_type, + task.trigger_config if isinstance(task.trigger_config, dict) else json.loads(task.trigger_config), + ) + trigger.advance(now) + new_due = trigger.next_due_at + self._store.advance_next_due(task.task_id, new_due) + + # Execute + try: + self._store.update_state(task.task_id, TaskState.EXECUTING.value) + + parameters = ( + task.parameters + if isinstance(task.parameters, dict) + else json.loads(task.parameters) + ) + result = await self._executor.execute(task.skill_name, parameters) + self._store.increment_run_count(task.task_id) + + # Check max_runs exhaustion + updated = self._store.load(task.task_id) + if updated and updated.max_runs > 0 and updated.run_count >= updated.max_runs: + self._store.update_state(task.task_id, TaskState.DONE.value) + logger.info( + "Task %s completed (max_runs reached)", task.task_id[:8] + ) + else: + self._store.update_state(task.task_id, TaskState.ARMED.value) + + logger.info( + "Task %s executed: ok=%s", + task.task_id[:8], + result.get("ok", False), + ) + except Exception as e: + self._store.update_state(task.task_id, TaskState.FAILED.value) + logger.error("Task %s failed: %s", task.task_id[:8], e) + + # ------------------------------------------------------------------ + # Fast-forward + # ------------------------------------------------------------------ + + def _fast_forward(self) -> None: + """On startup: advance overdue tasks past their grace period.""" + now = time.time() + all_tasks = self._store.load_all() + forwarded = 0 + + for task in all_tasks: + if task.state != TaskState.ARMED.value: + continue + if task.next_due_at <= 0: + continue + if now - task.next_due_at <= self._grace_seconds: + continue + + # Overdue beyond grace — fast forward + trigger = create_trigger( + task.trigger_type, + task.trigger_config if isinstance(task.trigger_config, dict) else json.loads(task.trigger_config), + ) + trigger.advance(now) + self._store.advance_next_due(task.task_id, trigger.next_due_at) + forwarded += 1 + logger.info( + "Fast-forwarded task %s to %.0f", + task.task_id[:8], + trigger.next_due_at, + ) + + if forwarded: + logger.info("Fast-forwarded %d overdue tasks", forwarded) diff --git a/src/leapflow/scheduler/store.py b/src/leapflow/scheduler/store.py new file mode 100644 index 0000000..1578929 --- /dev/null +++ b/src/leapflow/scheduler/store.py @@ -0,0 +1,208 @@ +"""DuckDB-backed persistence for armed tasks. + +Provides atomic CRUD and query operations for the scheduler. +At-most-once guarantee: advance_next_due() atomically updates +before execution to prevent double-fire on crash recovery. +""" + +from __future__ import annotations + +import json +import logging +import time +from pathlib import Path +from typing import List, Optional + +import duckdb + +from leapflow.scheduler.types import ArmedTask + +logger = logging.getLogger(__name__) + + +class TaskStore: + """DuckDB-backed persistence for armed tasks. + + Provides atomic CRUD and query operations. + At-most-once guarantee: advance_next_due() atomically updates + before execution to prevent double-fire on crash recovery. + """ + + def __init__(self, db_path: str | Path) -> None: + """Initialize store, creating table if not exists.""" + self._path = Path(db_path) + self._path.parent.mkdir(parents=True, exist_ok=True) + self._con = duckdb.connect(str(self._path)) + self._ensure_table() + + def close(self) -> None: + """Close the DuckDB connection.""" + self._con.close() + + # ------------------------------------------------------------------ + # Schema + # ------------------------------------------------------------------ + + def _ensure_table(self) -> None: + """CREATE TABLE IF NOT EXISTS armed_tasks (...)""" + self._con.execute(""" + CREATE TABLE IF NOT EXISTS armed_tasks ( + task_id TEXT PRIMARY KEY, + skill_name TEXT NOT NULL, + trigger_type TEXT NOT NULL, + trigger_config TEXT NOT NULL, + state TEXT DEFAULT 'armed', + execution_tier TEXT DEFAULT 'auto', + context_snapshot TEXT DEFAULT '{}', + confidence DOUBLE DEFAULT 0.0, + created_at DOUBLE NOT NULL, + next_due_at DOUBLE DEFAULT 0.0, + last_run_at DOUBLE DEFAULT 0.0, + run_count INTEGER DEFAULT 0, + max_runs INTEGER DEFAULT -1, + grace_seconds DOUBLE DEFAULT 120.0, + parameters TEXT DEFAULT '{}', + cloud_worker_id TEXT DEFAULT '', + metadata TEXT DEFAULT '{}' + ) + """) + + # ------------------------------------------------------------------ + # CRUD + # ------------------------------------------------------------------ + + def save(self, task: ArmedTask) -> None: + """Insert or update a task.""" + self._con.execute( + """ + INSERT OR REPLACE INTO armed_tasks ( + task_id, skill_name, trigger_type, trigger_config, + state, execution_tier, context_snapshot, confidence, + created_at, next_due_at, last_run_at, run_count, + max_runs, grace_seconds, parameters, cloud_worker_id, metadata + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + task.task_id, + task.skill_name, + task.trigger_type, + json.dumps(task.trigger_config) if isinstance(task.trigger_config, dict) else task.trigger_config, + task.state, + task.execution_tier, + json.dumps(task.context_snapshot) if isinstance(task.context_snapshot, dict) else task.context_snapshot, + task.confidence, + task.created_at, + task.next_due_at, + task.last_run_at, + task.run_count, + task.max_runs, + task.grace_seconds, + json.dumps(task.parameters) if isinstance(task.parameters, dict) else task.parameters, + task.cloud_worker_id, + json.dumps(task.metadata) if isinstance(task.metadata, dict) else task.metadata, + ], + ) + + def load(self, task_id: str) -> Optional[ArmedTask]: + """Load a single task by ID.""" + result = self._con.execute( + "SELECT * FROM armed_tasks WHERE task_id = ?", [task_id] + ).fetchone() + if result is None: + return None + return self._row_to_task(result) + + def load_all(self) -> List[ArmedTask]: + """Load all tasks.""" + rows = self._con.execute("SELECT * FROM armed_tasks").fetchall() + return [self._row_to_task(row) for row in rows] + + def delete(self, task_id: str) -> None: + """Remove a task.""" + self._con.execute( + "DELETE FROM armed_tasks WHERE task_id = ?", [task_id] + ) + + # ------------------------------------------------------------------ + # Query + # ------------------------------------------------------------------ + + def get_due_tasks(self, now: float) -> List[ArmedTask]: + """Get tasks where next_due_at <= now AND state = 'armed'.""" + rows = self._con.execute( + """ + SELECT * FROM armed_tasks + WHERE next_due_at <= ? AND next_due_at > 0 AND state = 'armed' + ORDER BY next_due_at ASC + """, + [now], + ).fetchall() + return [self._row_to_task(row) for row in rows] + + # ------------------------------------------------------------------ + # Atomic updates + # ------------------------------------------------------------------ + + def advance_next_due(self, task_id: str, new_due_at: float) -> None: + """Atomically advance next_due_at (at-most-once guarantee).""" + self._con.execute( + "UPDATE armed_tasks SET next_due_at = ? WHERE task_id = ?", + [new_due_at, task_id], + ) + + def update_state(self, task_id: str, new_state: str) -> None: + """Update task state.""" + self._con.execute( + "UPDATE armed_tasks SET state = ? WHERE task_id = ?", + [new_state, task_id], + ) + + def increment_run_count(self, task_id: str) -> None: + """Increment run_count and update last_run_at.""" + now = time.time() + self._con.execute( + """ + UPDATE armed_tasks + SET run_count = run_count + 1, last_run_at = ? + WHERE task_id = ? + """, + [now, task_id], + ) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _row_to_task(self, row: tuple) -> ArmedTask: + """Convert a DuckDB row tuple to an ArmedTask dataclass.""" + return ArmedTask( + task_id=row[0], + skill_name=row[1], + trigger_type=row[2], + trigger_config=self._safe_json_loads(row[3]), + state=row[4], + execution_tier=row[5], + context_snapshot=self._safe_json_loads(row[6]), + confidence=row[7], + created_at=row[8], + next_due_at=row[9], + last_run_at=row[10], + run_count=row[11], + max_runs=row[12], + grace_seconds=row[13], + parameters=self._safe_json_loads(row[14]), + cloud_worker_id=row[15], + metadata=self._safe_json_loads(row[16]), + ) + + @staticmethod + def _safe_json_loads(value: str | dict | None) -> dict: + """Safely parse a JSON string, returning empty dict on failure.""" + if value is None: + return {} + if isinstance(value, dict): + return value + try: + return json.loads(value) + except (json.JSONDecodeError, TypeError): + return {} diff --git a/src/leapflow/scheduler/triggers/__init__.py b/src/leapflow/scheduler/triggers/__init__.py new file mode 100644 index 0000000..4e909c6 --- /dev/null +++ b/src/leapflow/scheduler/triggers/__init__.py @@ -0,0 +1,66 @@ +"""Trigger factory: create Trigger instances from type string and config dict.""" + +from __future__ import annotations + +from typing import Dict, Type + +from leapflow.scheduler.types import Trigger +from leapflow.scheduler.triggers.condition import ConditionTrigger +from leapflow.scheduler.triggers.cron import CronTrigger +from leapflow.scheduler.triggers.event import EventTrigger +from leapflow.scheduler.triggers.interval import IntervalTrigger + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +_TRIGGER_REGISTRY: Dict[str, Type] = { + "interval": IntervalTrigger, + "cron": CronTrigger, + "event": EventTrigger, + "condition": ConditionTrigger, +} + + +def register_trigger(trigger_type: str, cls: Type) -> None: + """Register a custom trigger class for a given type string. + + Args: + trigger_type: The type identifier (used in serialized configs). + cls: The trigger class that implements the Trigger protocol. + """ + _TRIGGER_REGISTRY[trigger_type] = cls + + +def create_trigger(trigger_type: str, config: dict) -> Trigger: + """Factory: create a Trigger instance from type string and config dict. + + Args: + trigger_type: One of "interval", "cron", "event", "condition", + or any registered custom type. + config: Configuration dict to pass to the trigger's deserialize method. + + Returns: + A Trigger instance ready for use. + + Raises: + ValueError: If trigger_type is not registered. + """ + cls = _TRIGGER_REGISTRY.get(trigger_type) + if cls is None: + available = ", ".join(sorted(_TRIGGER_REGISTRY.keys())) + raise ValueError( + f"Unknown trigger type: {trigger_type!r}. " + f"Available: {available}" + ) + return cls.deserialize(config) # type: ignore[return-value] + + +__all__ = [ + "create_trigger", + "register_trigger", + "IntervalTrigger", + "CronTrigger", + "EventTrigger", + "ConditionTrigger", +] diff --git a/src/leapflow/scheduler/triggers/condition.py b/src/leapflow/scheduler/triggers/condition.py new file mode 100644 index 0000000..c897875 --- /dev/null +++ b/src/leapflow/scheduler/triggers/condition.py @@ -0,0 +1,208 @@ +"""Condition-based trigger: fires when a declarative condition is met. + +Supports simple comparison expressions like: + "file_count > 50" + "disk_usage > 90" + "queue_length >= 100" + +Does NOT use eval() — implements a safe expression parser. +""" + +from __future__ import annotations + +import operator +import re +import time +from typing import Any, Callable, ClassVar, Dict, Optional + + +# Supported comparison operators +_OPERATORS: Dict[str, Callable[[Any, Any], bool]] = { + ">": operator.gt, + ">=": operator.ge, + "<": operator.lt, + "<=": operator.le, + "==": operator.eq, + "!=": operator.ne, +} + +# Pattern: metric_name value (with optional % suffix on value) +_CONDITION_RE = re.compile( + r"^\s*([a-zA-Z_][a-zA-Z0-9_.]*)\s*" # metric name + r"([><=!]+)\s*" # operator + r"(\d+(?:\.\d+)?)\s*%?\s*$" # numeric value (optional % ignored) +) + + +class ConditionTrigger: + """Fires when a declarative condition expression evaluates to True. + + The condition is evaluated against a context dict that maps metric + names to their current numeric values. The context is updated + externally via :meth:`update_context`. + + Example expressions: + "file_count > 50" + "disk_usage > 90" + "queue_length >= 100" + + Security: Does NOT use eval(). Implements a simple comparison parser. + """ + + trigger_type: ClassVar[str] = "condition" + + def __init__( + self, + expression: str, + *, + check_interval: float = 60.0, + next_due_at: float = 0.0, + ) -> None: + if not expression: + raise ValueError("expression must not be empty") + self._expression = expression.strip() + self._check_interval = check_interval + self._next_due_at = next_due_at if next_due_at > 0 else time.time() + self._context: Dict[str, float] = {} + + # Parse and validate expression at construction time + self._parsed = self._parse_expression(self._expression) + if self._parsed is None: + raise ValueError( + f"Cannot parse condition expression: {self._expression!r}. " + f"Expected format: 'metric_name value' where op is one of " + f"{list(_OPERATORS.keys())}" + ) + + # ------------------------------------------------------------------ + # Trigger Protocol + # ------------------------------------------------------------------ + + @property + def trigger_type(self) -> str: # type: ignore[override] + return "condition" + + def is_due(self, now: float) -> bool: + """Evaluate the condition against current context. + + Only checks when now >= next_due_at (respects check_interval). + Returns True if the condition is satisfied. + """ + if now < self._next_due_at: + return False + return self._evaluate() + + def advance(self, now: float) -> None: + """Schedule the next check at now + check_interval.""" + self._next_due_at = now + self._check_interval + + @property + def next_due_at(self) -> float: + return self._next_due_at + + def serialize(self) -> dict: + """Serialize to JSON-compatible dict.""" + return { + "trigger_type": "condition", + "expression": self._expression, + "check_interval": self._check_interval, + "next_due_at": self._next_due_at, + } + + # ------------------------------------------------------------------ + # Context management + # ------------------------------------------------------------------ + + def update_context(self, context: Dict[str, float]) -> None: + """Update the metric context used for condition evaluation. + + Args: + context: Dict mapping metric names to numeric values. + """ + self._context.update(context) + + def set_metric(self, name: str, value: float) -> None: + """Set a single metric value in the context. + + Args: + name: Metric name (must match expression's left-hand side). + value: Current numeric value. + """ + self._context[name] = value + + # ------------------------------------------------------------------ + # Expression parsing (safe, no eval) + # ------------------------------------------------------------------ + + @staticmethod + def _parse_expression( + expression: str, + ) -> Optional[tuple[str, Callable[[Any, Any], bool], float]]: + """Parse a condition expression into (metric, operator_fn, threshold). + + Returns None if the expression cannot be parsed. + """ + match = _CONDITION_RE.match(expression) + if not match: + return None + + metric_name = match.group(1) + op_str = match.group(2) + threshold = float(match.group(3)) + + op_fn = _OPERATORS.get(op_str) + if op_fn is None: + return None + + return (metric_name, op_fn, threshold) + + def _evaluate(self) -> bool: + """Evaluate the parsed condition against current context.""" + if self._parsed is None: + return False + + metric_name, op_fn, threshold = self._parsed + current_value = self._context.get(metric_name) + if current_value is None: + # Metric not in context — condition cannot be satisfied + return False + + return op_fn(float(current_value), threshold) + + # ------------------------------------------------------------------ + # Factory + # ------------------------------------------------------------------ + + @classmethod + def deserialize(cls, data: dict) -> "ConditionTrigger": + """Reconstruct from serialized dict.""" + return cls( + expression=data["expression"], + check_interval=data.get("check_interval", 60.0), + next_due_at=data.get("next_due_at", 0.0), + ) + + # ------------------------------------------------------------------ + # Properties + # ------------------------------------------------------------------ + + @property + def expression(self) -> str: + """The condition expression string.""" + return self._expression + + @property + def check_interval(self) -> float: + """Seconds between condition evaluations.""" + return self._check_interval + + @property + def context(self) -> Dict[str, float]: + """Current metric context (read-only copy).""" + return dict(self._context) + + def __repr__(self) -> str: + return ( + f"ConditionTrigger(expression={self._expression!r}, " + f"check_interval={self._check_interval})" + ) diff --git a/src/leapflow/scheduler/triggers/cron.py b/src/leapflow/scheduler/triggers/cron.py new file mode 100644 index 0000000..e83129f --- /dev/null +++ b/src/leapflow/scheduler/triggers/cron.py @@ -0,0 +1,160 @@ +"""Cron-expression trigger with graceful fallback. + +Uses ``croniter`` when available for full cron expression support. +Falls back to a minimal daily HH:MM parser otherwise. +""" + +from __future__ import annotations + +import time +from datetime import datetime, timezone +from typing import ClassVar, Optional + +try: + from croniter import croniter as _croniter # type: ignore[import-untyped] + + _HAS_CRONITER = True +except ImportError: + _croniter = None + _HAS_CRONITER = False + + +class CronTrigger: + """Cron-based trigger: fires according to a cron expression. + + Full cron support requires the ``croniter`` package. Without it, + only simple "HH:MM" daily schedules are supported. + """ + + trigger_type: ClassVar[str] = "cron" + + def __init__(self, expression: str, *, next_due_at: float = 0.0) -> None: + self._expression = expression.strip() + self._next_due_at = next_due_at + + # Validate and compute first due time if not provided + if self._next_due_at <= 0.0: + self._next_due_at = self._compute_next(time.time()) + + # ------------------------------------------------------------------ + # Trigger Protocol + # ------------------------------------------------------------------ + + @property + def trigger_type(self) -> str: # type: ignore[override] + return "cron" + + def is_due(self, now: float) -> bool: + """Return True if now >= next_due_at.""" + return now >= self._next_due_at + + def advance(self, now: float) -> None: + """Compute and set the next trigger time after *now*.""" + self._next_due_at = self._compute_next(now) + + @property + def next_due_at(self) -> float: + return self._next_due_at + + def serialize(self) -> dict: + """Serialize to JSON-compatible dict.""" + return { + "trigger_type": "cron", + "expression": self._expression, + "next_due_at": self._next_due_at, + } + + # ------------------------------------------------------------------ + # Factory + # ------------------------------------------------------------------ + + @classmethod + def deserialize(cls, data: dict) -> "CronTrigger": + """Reconstruct from serialized dict.""" + return cls( + expression=data["expression"], + next_due_at=data.get("next_due_at", 0.0), + ) + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _compute_next(self, after: float) -> float: + """Compute the next trigger time after the given timestamp.""" + if _HAS_CRONITER: + return self._compute_next_croniter(after) + return self._compute_next_fallback(after) + + def _compute_next_croniter(self, after: float) -> float: + """Use croniter for full cron expression support.""" + dt = datetime.fromtimestamp(after, tz=timezone.utc) + cron = _croniter(self._expression, dt) + next_dt = cron.get_next(datetime) + return next_dt.timestamp() + + def _compute_next_fallback(self, after: float) -> float: + """Fallback: support only 'HH:MM' daily schedules or '* * * * *' style. + + Parses simple HH:MM format and schedules for the next occurrence. + For full cron expressions without croniter, raises a clear error. + """ + parsed = self._try_parse_hhmm(self._expression) + if parsed is None: + raise ValueError( + f"Cannot parse cron expression {self._expression!r} without " + f"the 'croniter' package. Install it or use HH:MM format." + ) + hour, minute = parsed + dt_after = datetime.fromtimestamp(after, tz=timezone.utc) + # Build candidate for today + candidate = dt_after.replace( + hour=hour, minute=minute, second=0, microsecond=0 + ) + if candidate.timestamp() <= after: + # Already passed today, schedule for tomorrow + from datetime import timedelta + + candidate = candidate + timedelta(days=1) + return candidate.timestamp() + + @staticmethod + def _try_parse_hhmm(expression: str) -> Optional[tuple[int, int]]: + """Try to parse 'HH:MM' or cron '0 9 * * *' into (hour, minute). + + Returns None if parsing fails. + """ + # Try HH:MM format + if ":" in expression and len(expression) <= 5: + parts = expression.split(":") + if len(parts) == 2: + try: + h, m = int(parts[0]), int(parts[1]) + if 0 <= h <= 23 and 0 <= m <= 59: + return (h, m) + except ValueError: + pass + + # Try standard 5-field cron where day/month/dow are all * + fields = expression.split() + if len(fields) == 5 and fields[2] == "*" and fields[3] == "*" and fields[4] == "*": + try: + minute = int(fields[0]) + hour = int(fields[1]) + if 0 <= hour <= 23 and 0 <= minute <= 59: + return (hour, minute) + except ValueError: + pass + + return None + + @property + def expression(self) -> str: + """The cron expression string.""" + return self._expression + + def __repr__(self) -> str: + return ( + f"CronTrigger(expression={self._expression!r}, " + f"next_due_at={self._next_due_at})" + ) diff --git a/src/leapflow/scheduler/triggers/event.py b/src/leapflow/scheduler/triggers/event.py new file mode 100644 index 0000000..2664623 --- /dev/null +++ b/src/leapflow/scheduler/triggers/event.py @@ -0,0 +1,136 @@ +"""Event-based trigger: fires when an external event matches a pattern.""" + +from __future__ import annotations + +import fnmatch +import time +from typing import ClassVar + + +class EventTrigger: + """Fires when an external event matching the configured pattern occurs. + + The trigger is activated by calling :meth:`set_triggered` from an + EventBus subscriber or similar mechanism. Pattern matching uses + fnmatch-style glob patterns (e.g. "ci.passed", "fs.change:*.pdf"). + """ + + trigger_type: ClassVar[str] = "event" + + def __init__( + self, + event_pattern: str, + *, + next_due_at: float = 0.0, + ) -> None: + if not event_pattern: + raise ValueError("event_pattern must not be empty") + self._event_pattern = event_pattern + self._next_due_at = next_due_at + self._triggered = False + self._last_event: str = "" + + # ------------------------------------------------------------------ + # Trigger Protocol + # ------------------------------------------------------------------ + + @property + def trigger_type(self) -> str: # type: ignore[override] + return "event" + + def is_due(self, now: float) -> bool: + """Return True if the trigger has been activated by a matching event.""" + return self._triggered + + def advance(self, now: float) -> None: + """Reset the triggered flag after the task has been executed.""" + self._triggered = False + self._next_due_at = 0.0 # No predictable next time for events + + @property + def next_due_at(self) -> float: + return self._next_due_at + + def serialize(self) -> dict: + """Serialize to JSON-compatible dict.""" + return { + "trigger_type": "event", + "event_pattern": self._event_pattern, + "next_due_at": self._next_due_at, + } + + # ------------------------------------------------------------------ + # Event interface + # ------------------------------------------------------------------ + + def set_triggered(self, event_name: str = "") -> None: + """Mark this trigger as fired. + + Called by external EventBus subscribers when a matching event arrives. + Optionally records the event name for diagnostics. + + Args: + event_name: The specific event that triggered this (for logging). + """ + self._triggered = True + self._last_event = event_name + self._next_due_at = time.time() + + def matches(self, event_name: str) -> bool: + """Check if an event name matches this trigger's pattern. + + Uses fnmatch-style glob matching. The pattern may contain ':' as + a namespace separator (e.g. "fs.change:*.pdf" matches + "fs.change:report.pdf"). + + Args: + event_name: The event to check against the pattern. + """ + return fnmatch.fnmatch(event_name, self._event_pattern) + + def notify(self, event_name: str) -> bool: + """Convenience: check match and set_triggered if matched. + + Returns True if the event matched and the trigger was activated. + """ + if self.matches(event_name): + self.set_triggered(event_name) + return True + return False + + # ------------------------------------------------------------------ + # Factory + # ------------------------------------------------------------------ + + @classmethod + def deserialize(cls, data: dict) -> "EventTrigger": + """Reconstruct from serialized dict.""" + return cls( + event_pattern=data["event_pattern"], + next_due_at=data.get("next_due_at", 0.0), + ) + + # ------------------------------------------------------------------ + # Properties + # ------------------------------------------------------------------ + + @property + def event_pattern(self) -> str: + """The event pattern this trigger listens for.""" + return self._event_pattern + + @property + def is_triggered(self) -> bool: + """Whether the trigger is currently in fired state.""" + return self._triggered + + @property + def last_event(self) -> str: + """The last event name that activated this trigger.""" + return self._last_event + + def __repr__(self) -> str: + return ( + f"EventTrigger(event_pattern={self._event_pattern!r}, " + f"triggered={self._triggered})" + ) diff --git a/src/leapflow/scheduler/triggers/interval.py b/src/leapflow/scheduler/triggers/interval.py new file mode 100644 index 0000000..3f3e34d --- /dev/null +++ b/src/leapflow/scheduler/triggers/interval.py @@ -0,0 +1,128 @@ +"""Interval-based trigger: fires every N seconds/minutes/hours/days.""" + +from __future__ import annotations + +import re +from typing import ClassVar, Dict + + +_UNIT_MAP: Dict[str, float] = { + "s": 1.0, + "sec": 1.0, + "second": 1.0, + "seconds": 1.0, + "m": 60.0, + "min": 60.0, + "minute": 60.0, + "minutes": 60.0, + "h": 3600.0, + "hr": 3600.0, + "hour": 3600.0, + "hours": 3600.0, + "d": 86400.0, + "day": 86400.0, + "days": 86400.0, +} + +_INTERVAL_RE = re.compile( + r"(?:every\s+)?(\d+(?:\.\d+)?)\s*([a-zA-Z]+)", re.IGNORECASE +) + + +def _parse_interval(spec: str) -> float: + """Parse an interval specification string into seconds. + + Supported formats: + "30m", "2h", "1d", "every 5m", "90s", "1.5h" + """ + match = _INTERVAL_RE.match(spec.strip()) + if not match: + # Try pure numeric (seconds) + try: + return float(spec.strip()) + except ValueError: + raise ValueError(f"Cannot parse interval: {spec!r}") + + value = float(match.group(1)) + unit = match.group(2).lower() + multiplier = _UNIT_MAP.get(unit) + if multiplier is None: + raise ValueError(f"Unknown interval unit: {unit!r} in {spec!r}") + return value * multiplier + + +class IntervalTrigger: + """Fires at fixed time intervals. + + Accepts human-readable specs like "30m", "2h", "1d", "every 5m". + """ + + trigger_type: ClassVar[str] = "interval" + + def __init__(self, interval_seconds: float, *, next_due_at: float = 0.0) -> None: + if interval_seconds <= 0: + raise ValueError("interval_seconds must be positive") + self._interval_seconds = interval_seconds + self._next_due_at = next_due_at + + # ------------------------------------------------------------------ + # Trigger Protocol + # ------------------------------------------------------------------ + + @property + def trigger_type(self) -> str: # type: ignore[override] + return "interval" + + def is_due(self, now: float) -> bool: + """Return True if now >= next_due_at.""" + return now >= self._next_due_at + + def advance(self, now: float) -> None: + """Set next trigger point to now + interval.""" + self._next_due_at = now + self._interval_seconds + + @property + def next_due_at(self) -> float: + return self._next_due_at + + def serialize(self) -> dict: + """Serialize to JSON-compatible dict.""" + return { + "trigger_type": "interval", + "interval_seconds": self._interval_seconds, + "next_due_at": self._next_due_at, + } + + # ------------------------------------------------------------------ + # Factory + # ------------------------------------------------------------------ + + @classmethod + def deserialize(cls, data: dict) -> "IntervalTrigger": + """Reconstruct from serialized dict.""" + return cls( + interval_seconds=data["interval_seconds"], + next_due_at=data.get("next_due_at", 0.0), + ) + + @classmethod + def from_spec(cls, spec: str, *, now: float = 0.0) -> "IntervalTrigger": + """Create from a human-readable spec string. + + Args: + spec: e.g. "30m", "2h", "every 5m" + now: current timestamp; first trigger at now + interval + """ + seconds = _parse_interval(spec) + return cls(interval_seconds=seconds, next_due_at=now + seconds) + + @property + def interval_seconds(self) -> float: + """The configured interval in seconds.""" + return self._interval_seconds + + def __repr__(self) -> str: + return ( + f"IntervalTrigger(interval_seconds={self._interval_seconds}, " + f"next_due_at={self._next_due_at})" + ) diff --git a/src/leapflow/scheduler/types.py b/src/leapflow/scheduler/types.py new file mode 100644 index 0000000..0ba96bf --- /dev/null +++ b/src/leapflow/scheduler/types.py @@ -0,0 +1,162 @@ +"""Core type definitions for the long-horizon async task scheduler.""" + +from __future__ import annotations + +import uuid +import time +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Dict, List, Protocol, runtime_checkable + + +# --------------------------------------------------------------------------- +# Enums +# --------------------------------------------------------------------------- + + +class TaskState(str, Enum): + """Lifecycle states for a scheduled task.""" + + ARMED = "armed" + WATCHING = "watching" + DUE = "due" + CONFIRMING = "confirming" + EXECUTING = "executing" + DONE = "done" + FAILED = "failed" + SUSPENDED = "suspended" + + +class ExecutionTier(str, Enum): + """Where the task should be executed.""" + + LOCAL = "local" + CLOUD = "cloud" + AUTO = "auto" + + +# --------------------------------------------------------------------------- +# Dataclasses +# --------------------------------------------------------------------------- + + +@dataclass +class ArmedTask: + """Mutable representation of a scheduled task with full lifecycle state.""" + + skill_name: str + trigger_type: str + trigger_config: dict + task_id: str = field(default_factory=lambda: uuid.uuid4().hex) + state: str = "armed" + execution_tier: str = "auto" + context_snapshot: dict = field(default_factory=dict) + confidence: float = 0.0 + created_at: float = field(default_factory=time.time) + next_due_at: float = 0.0 + last_run_at: float = 0.0 + run_count: int = 0 + max_runs: int = -1 # -1 = unlimited + grace_seconds: float = 120.0 + parameters: dict = field(default_factory=dict) + cloud_worker_id: str = "" + metadata: dict = field(default_factory=dict) + + +@dataclass +class TaskStatus: + """Runtime status snapshot for a task.""" + + task: ArmedTask + is_running: bool = False + logs_tail: List[str] = field(default_factory=list) + + +# --------------------------------------------------------------------------- +# Protocols +# --------------------------------------------------------------------------- + + +@runtime_checkable +class Trigger(Protocol): + """Protocol that all trigger implementations must satisfy. + + Triggers determine *when* a task becomes due for execution. + All triggers must be serializable to/from plain dicts for persistence. + """ + + @property + def trigger_type(self) -> str: + """Unique string identifier for the trigger kind.""" + ... + + def is_due(self, now: float) -> bool: + """Return True if the trigger condition is currently met.""" + ... + + def advance(self, now: float) -> None: + """Advance internal state to the next trigger point after firing.""" + ... + + @property + def next_due_at(self) -> float: + """Timestamp (epoch seconds) of the next expected trigger.""" + ... + + def serialize(self) -> dict: + """Serialize trigger state to a JSON-compatible dict.""" + ... + + +@runtime_checkable +class ComputeBackend(Protocol): + """Protocol for cloud/remote compute backends. + + Backends manage the lifecycle of remote workers that execute skills. + """ + + @property + def backend_type(self) -> str: + """Identifier string for this backend (e.g. 'cloudflare', 'fly').""" + ... + + async def create_worker( + self, worker_id: str, package_path: Path, visibility: str + ) -> str: + """Create a new remote worker. Returns the worker URL or identifier.""" + ... + + async def inject_secrets( + self, worker_id: str, secrets: Dict[str, str] + ) -> None: + """Inject environment secrets into a worker.""" + ... + + async def deploy(self, worker_id: str) -> None: + """Deploy (or redeploy) the worker.""" + ... + + async def stop(self, worker_id: str) -> None: + """Stop a running worker without destroying it.""" + ... + + async def get_status(self, worker_id: str) -> str: + """Get current status string of the worker.""" + ... + + async def get_logs(self, worker_id: str, tail: int) -> List[str]: + """Retrieve recent log lines from the worker.""" + ... + + async def destroy(self, worker_id: str) -> None: + """Permanently destroy the worker and its resources.""" + ... + + +class SkillExecutor(Protocol): + """Protocol for executing a skill by name with parameters.""" + + async def execute(self, skill_name: str, parameters: dict) -> dict: + """Execute a skill. Returns {"ok": bool, "output": ...}.""" + ... diff --git a/src/leapflow/scheduler/worker_packager.py b/src/leapflow/scheduler/worker_packager.py new file mode 100644 index 0000000..866e67c --- /dev/null +++ b/src/leapflow/scheduler/worker_packager.py @@ -0,0 +1,197 @@ +"""Worker packager — generates self-contained Docker packages for cloud deployment. + +Produces a temporary directory containing all files needed to run a LeapFlow +worker in a Docker container on any supported compute backend. +""" + +from __future__ import annotations + +import json +import logging +import tempfile +from pathlib import Path +from typing import Optional + +from leapflow.scheduler.types import ArmedTask + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Templates +# --------------------------------------------------------------------------- + +_DOCKERFILE_TEMPLATE = """\ +FROM python:3.11-slim +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY . . +CMD ["python", "worker.py"] +""" + +_REQUIREMENTS_TEMPLATE = """\ +# Minimal runtime dependencies for LeapFlow cloud worker +requests>=2.28.0 +""" + +_WORKER_PY_TEMPLATE = '''\ +"""LeapFlow Cloud Worker — long-horizon task executor.""" +import json +import logging +import os +import sys +import time + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", +) +logger = logging.getLogger("leapflow-worker") + + +def main(): + """Main worker loop: parse config, evaluate triggers, execute skill.""" + config = json.loads(os.environ.get("LEAPFLOW_TASK_CONFIG", "{}")) + if not config: + logger.error("No LEAPFLOW_TASK_CONFIG found in environment") + sys.exit(1) + + task_id = config.get("task_id", "unknown") + skill_name = config.get("skill_name", "") + trigger_config = config.get("trigger_config", {}) + parameters = config.get("parameters", {}) + max_runs = config.get("max_runs", -1) + check_interval = config.get("check_interval", 60) + + logger.info("Worker started: task=%s skill=%s", task_id[:8], skill_name) + + run_count = 0 + while True: + # Heartbeat + logger.info("Heartbeat: task=%s runs=%d", task_id[:8], run_count) + + # Check trigger condition + should_run = _check_trigger(trigger_config) + + if should_run: + logger.info("Trigger fired: executing skill \'%s\'", skill_name) + try: + result = _execute_skill(skill_name, parameters) + run_count += 1 + logger.info( + "Execution complete: ok=%s runs=%d", + result.get("ok"), + run_count, + ) + except Exception as e: + logger.error("Execution failed: %s", e) + + if max_runs > 0 and run_count >= max_runs: + logger.info("Max runs reached (%d). Shutting down.", max_runs) + break + + time.sleep(check_interval) + + +def _check_trigger(config): + """Simplified trigger evaluation for cloud worker. + + In cloud mode, the worker itself evaluates timing. For interval/cron + triggers the scheduler has already decided to deploy, so we fire on + each check cycle. + """ + # Future: implement cron expression evaluation here + _ = config.get("trigger_type", "interval") + return True + + +def _execute_skill(skill_name, parameters): + """Execute the target skill. + + In the full implementation this would import and invoke the skill + source code bundled into the package. For now, logs execution. + """ + logger.info("Skill \'%s\' executed with params: %s", skill_name, parameters) + return {"ok": True, "output": f"Executed {skill_name}"} + + +if __name__ == "__main__": + main() +''' + + +# --------------------------------------------------------------------------- +# WorkerPackager +# --------------------------------------------------------------------------- + + +class WorkerPackager: + """Generate self-contained Docker worker packages for cloud deployment. + + Produces a temporary directory containing: + - Dockerfile (Python base + minimal deps) + - worker.py (standard task execution loop) + - requirements.txt (leapflow minimal runtime) + - task_meta.json (non-sensitive task metadata only) + + Sensitive data (full task config) is injected as secrets at deploy time, + not baked into the package. + """ + + def package( + self, + task: ArmedTask, + skill_source: str = "", + context_snapshot: Optional[dict] = None, + ) -> Path: + """Build a deployable worker package. + + Args: + task: The armed task to deploy. + skill_source: Optional skill source code to include. + context_snapshot: Optional context data for the worker. + + Returns: + Path to the temporary directory containing the package files. + """ + tmp_dir = tempfile.mkdtemp(prefix=f"leapflow_worker_{task.task_id[:8]}_") + pkg_path = Path(tmp_dir) + + # 1. Dockerfile + (pkg_path / "Dockerfile").write_text(_DOCKERFILE_TEMPLATE, encoding="utf-8") + + # 2. worker.py + (pkg_path / "worker.py").write_text(_WORKER_PY_TEMPLATE, encoding="utf-8") + + # 3. requirements.txt + (pkg_path / "requirements.txt").write_text( + _REQUIREMENTS_TEMPLATE, encoding="utf-8" + ) + + # 4. Task metadata (non-sensitive — no secrets here) + meta = { + "task_id": task.task_id, + "skill_name": task.skill_name, + "trigger_type": task.trigger_type, + "execution_tier": task.execution_tier, + "created_at": task.created_at, + } + if context_snapshot: + meta["context_snapshot"] = context_snapshot + + (pkg_path / "task_meta.json").write_text( + json.dumps(meta, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + # 5. Optional skill source + if skill_source: + (pkg_path / "skill_source.py").write_text( + skill_source, encoding="utf-8" + ) + + logger.info( + "Packaged worker for task %s at %s", task.task_id[:8], pkg_path + ) + return pkg_path