[Feature] Add agent hub and task scheduler#9
Conversation
…l-tier execution Complete scheduler module (src/leapflow/scheduler/) for armed skills: Core types and triggers: - ArmedTask, TaskState, ExecutionTier, Trigger/ComputeBackend/SkillExecutor Protocols - IntervalTrigger: parse '30m', '2h', 'every 5m' duration expressions - CronTrigger: croniter-backed (with basic HH:MM fallback) - EventTrigger: fnmatch-based pattern matching on SystemEvent types - ConditionTrigger: safe expression parser (no eval) for declarative conditions Local execution (Tier 1): - TaskStore: DuckDB persistence with armed_tasks table, atomic operations - LocalScheduler: 60s async tick loop, at-most-once (advance before execute), fast-forward on startup for overdue tasks, max_runs enforcement Cloud execution (Tier 2): - ComputeBackend Protocol for pluggable cloud providers - ModelScopeStudioBackend: Docker-based private Studio deployment - WorkerPackager: generates Dockerfile + worker.py + requirements.txt - CloudDispatcher: orchestrates package → create → inject secrets → deploy Unified orchestration: - TaskCoordinator: tier routing (auto/local/cloud), trigger expression parser - CLI: 'leap arm' and 'leap tasks' commands with status/logs/cancel - Interactive REPL integration: arm/tasks commands registered - Config: scheduler_enabled/tick_seconds/grace_seconds/default_tier
…nd improve cloud dispatch Critical fixes: - Align parse_trigger_expression() output keys with Trigger.deserialize(): cron_expression → expression, event_name → event_pattern (fixes KeyError) - Wire LocalScheduler + CloudDispatcher into CLI _get_coordinator() so 'leap arm' actually works instead of always raising RuntimeError Medium: - CloudDispatcher._build_task_config: derive check_interval from trigger type (interval→capped at 1h, cron→300s, condition/event→60s) Minor: - CloudDispatcher.deploy: cleanup temp package dir with shutil.rmtree in finally block to prevent disk space leaks
…ct detection, security enhancement SkillManifest extensions (backward compatible): - content_hash: SHA256[:16] of semantic content for integrity tracking - updated_by / updated_at: authorship tracking per modification - tags: classification labels for skill discovery Push version conflict detection: - Pre-check remote version before push (raises VersionConflictError) - CLI catches conflict, offers force-push with confirmation - --force flag bypasses check ContentSanitizer enhancement: - SSH private key detection (-----BEGIN...PRIVATE KEY-----) - AWS access key (AKIA...) and secret key patterns - Hardcoded IP address detection (non-loopback) - has_blocking_findings(): HIGH severity blocks push by default Content-hash divergence in sync: - Same version but different content_hash → detected as 'conflict' - SyncEngine skips conflicts with warning (manual resolution required) Author auto-fill on push: - Fetches username via backend.authenticate() - Sets author (first push), updated_by, updated_at automatically Design: relies on ModelScope ORG RBAC for access control (read/write/admin), LeapFlow only adds application-layer semantics the platform cannot provide.
… sync conflicts - serializer.py: add content_hash/updated_by/updated_at/tags to _manifest_to_dict, _dict_to_manifest, and import_skill (round-trip verified) - sync.py: _resolve_conflicts manual strategy now preserves direction='conflict' actions so content-hash divergence warnings reach execute_plan
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a long-horizon async task scheduler supporting both local and cloud execution (via ModelScope Studio), complete with DuckDB persistence, multiple trigger types (interval, cron, event, condition), and interactive CLI commands (arm and tasks). It also adds team collaboration fields and version conflict checks to the Hub client and sync logic. The review feedback highlights three key issues: a race condition in the local scheduler that can overwrite a task's SUSPENDED state back to ARMED or DONE after execution, a potential AttributeError in the CLI trigger formatter if trigger_config is passed as a string, and fragile SELECT * queries in the database store that map rows to dataclasses by index.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| 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) |
There was a problem hiding this comment.
There is a race condition when updating the task state after execution. If a task is cancelled (state set to SUSPENDED) while it is executing, this unconditional update will overwrite the SUSPENDED state back to ARMED (or DONE). This causes the cancellation to be ignored, and the task will run again on the next tick.
To fix this, check if the loaded task state is SUSPENDED before resetting it to ARMED.
| 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 |
There was a problem hiding this comment.
If task.trigger_config is stored or passed as a string (which is explicitly handled as a possibility in other parts of the scheduler codebase), calling .get() directly on it will raise an AttributeError and crash the CLI command.
Safely parse the trigger_config if it is a string before accessing its keys.
| result = self._con.execute( | ||
| "SELECT * FROM armed_tasks WHERE task_id = ?", [task_id] | ||
| ).fetchone() |
There was a problem hiding this comment.
Using SELECT * is fragile when mapping query results to a dataclass by index (as done in _row_to_task). If the database schema is migrated, columns are reordered, or new columns are added in the future, the indices will map to the wrong fields, leading to silent data corruption or runtime crashes.
Explicitly list the columns in the SELECT statement to guarantee the returned column order. The same should be applied to load_all and get_due_tasks.
1 participant
No description provided.