From a6dfa8f56d318beefec75c4c7eab4507673f551f Mon Sep 17 00:00:00 2001 From: Chao Qin Date: Sat, 4 Apr 2026 10:29:09 +0000 Subject: [PATCH] feat: wire --resume/--continue CLI flags and add cron scheduler daemon (#16) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cherry-picked from PR #16 by win4r, excluding auto-compact (already implemented in a more thorough version based on Claude Code source). Resume/Continue: - Wire --continue (most recent session) and --resume (picker/ID) CLI flags - Pass restore_messages through app.py → backend_host.py → runtime.py - Uses existing session_storage infrastructure Cron scheduler daemon: - oh cron start/stop/status/list/toggle/history subcommands - Background daemon with 30s tick, concurrent job execution, PID file - Job state: enabled, last_run, next_run, last_status, created_at - JSONL execution history at ~/.openharness/data/cron_history.jsonl - croniter-based expression validation and next-run computation Cost/usage output in print mode: - Token usage summary (input/output/total) printed to stderr - Tool activity indicators during execution New files: cron_scheduler.py, cron_toggle_tool.py, test_cron.py, test_cron_scheduler.py Modified: cli.py, runtime.py, app.py, backend_host.py, cron.py, tools/__init__.py --- pyproject.toml | 1 + src/openharness/cli.py | 173 +++++++++++ src/openharness/services/cron.py | 54 +++- src/openharness/services/cron_scheduler.py | 344 +++++++++++++++++++++ src/openharness/tools/__init__.py | 2 + src/openharness/tools/cron_create_tool.py | 31 +- src/openharness/tools/cron_list_tool.py | 23 +- src/openharness/tools/cron_toggle_tool.py | 37 +++ src/openharness/ui/app.py | 2 + src/openharness/ui/backend_host.py | 4 + src/openharness/ui/runtime.py | 9 + tests/test_services/test_cron.py | 154 +++++++++ tests/test_services/test_cron_scheduler.py | 170 ++++++++++ tests/test_tools/test_core_tools.py | 2 +- tests/test_tools/test_integration_flows.py | 2 +- 15 files changed, 998 insertions(+), 10 deletions(-) create mode 100644 src/openharness/services/cron_scheduler.py create mode 100644 src/openharness/tools/cron_toggle_tool.py create mode 100644 tests/test_services/test_cron.py create mode 100644 tests/test_services/test_cron_scheduler.py diff --git a/pyproject.toml b/pyproject.toml index 3d947a3..ae8ee0f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,7 @@ dependencies = [ "pyperclip>=1.9.0", "pyyaml>=6.0", "watchfiles>=0.20.0", + "croniter>=2.0.0", ] [project.optional-dependencies] diff --git a/src/openharness/cli.py b/src/openharness/cli.py index 4937b5e..e875e4e 100644 --- a/src/openharness/cli.py +++ b/src/openharness/cli.py @@ -28,10 +28,12 @@ app = typer.Typer( mcp_app = typer.Typer(name="mcp", help="Manage MCP servers") plugin_app = typer.Typer(name="plugin", help="Manage plugins") auth_app = typer.Typer(name="auth", help="Manage authentication") +cron_app = typer.Typer(name="cron", help="Manage cron scheduler and jobs") app.add_typer(mcp_app) app.add_typer(plugin_app) app.add_typer(auth_app) +app.add_typer(cron_app) # ---- mcp subcommands ---- @@ -131,6 +133,119 @@ def plugin_uninstall( print(f"Uninstalled plugin: {name}") +# ---- cron subcommands ---- + +@cron_app.command("start") +def cron_start() -> None: + """Start the cron scheduler daemon.""" + from openharness.services.cron_scheduler import is_scheduler_running, start_daemon + + if is_scheduler_running(): + print("Cron scheduler is already running.") + return + pid = start_daemon() + print(f"Cron scheduler started (pid={pid})") + + +@cron_app.command("stop") +def cron_stop() -> None: + """Stop the cron scheduler daemon.""" + from openharness.services.cron_scheduler import stop_scheduler + + if stop_scheduler(): + print("Cron scheduler stopped.") + else: + print("Cron scheduler is not running.") + + +@cron_app.command("status") +def cron_status_cmd() -> None: + """Show cron scheduler status and job summary.""" + from openharness.services.cron_scheduler import scheduler_status + + status = scheduler_status() + state = "running" if status["running"] else "stopped" + print(f"Scheduler: {state}" + (f" (pid={status['pid']})" if status["pid"] else "")) + print(f"Jobs: {status['enabled_jobs']} enabled / {status['total_jobs']} total") + print(f"Log: {status['log_file']}") + + +@cron_app.command("list") +def cron_list_cmd() -> None: + """List all registered cron jobs with schedule and status.""" + from openharness.services.cron import load_cron_jobs + + jobs = load_cron_jobs() + if not jobs: + print("No cron jobs configured.") + return + for job in jobs: + enabled = "on " if job.get("enabled", True) else "off" + last = job.get("last_run", "never") + if last != "never": + last = last[:19] # trim to readable datetime + last_status = job.get("last_status", "") + status_indicator = f" [{last_status}]" if last_status else "" + print(f" [{enabled}] {job['name']} {job.get('schedule', '?')}") + print(f" cmd: {job['command']}") + print(f" last: {last}{status_indicator} next: {job.get('next_run', 'n/a')[:19]}") + + +@cron_app.command("toggle") +def cron_toggle_cmd( + name: str = typer.Argument(..., help="Cron job name"), + enabled: bool = typer.Argument(..., help="true to enable, false to disable"), +) -> None: + """Enable or disable a cron job.""" + from openharness.services.cron import set_job_enabled + + if not set_job_enabled(name, enabled): + print(f"Cron job not found: {name}") + raise typer.Exit(1) + state = "enabled" if enabled else "disabled" + print(f"Cron job '{name}' is now {state}") + + +@cron_app.command("history") +def cron_history_cmd( + name: str | None = typer.Argument(None, help="Filter by job name"), + limit: int = typer.Option(20, "--limit", "-n", help="Number of entries"), +) -> None: + """Show cron execution history.""" + from openharness.services.cron_scheduler import load_history + + entries = load_history(limit=limit, job_name=name) + if not entries: + print("No execution history.") + return + for entry in entries: + ts = entry.get("started_at", "?")[:19] + status = entry.get("status", "?") + rc = entry.get("returncode", "?") + print(f" {ts} {entry.get('name', '?')} {status} (rc={rc})") + stderr = entry.get("stderr", "").strip() + if stderr and status != "success": + for line in stderr.splitlines()[:3]: + print(f" stderr: {line}") + + +@cron_app.command("logs") +def cron_logs_cmd( + lines: int = typer.Option(30, "--lines", "-n", help="Number of lines to show"), +) -> None: + """Show recent cron scheduler log output.""" + from openharness.config.paths import get_logs_dir + + log_path = get_logs_dir() / "cron_scheduler.log" + if not log_path.exists(): + print("No scheduler log found. Start the scheduler with: oh cron start") + return + content = log_path.read_text(encoding="utf-8", errors="replace") + tail = content.splitlines()[-lines:] + for line in tail: + print(line) + + # ---- auth subcommands ---- @auth_app.command("status") @@ -349,6 +464,64 @@ def main( from openharness.ui.app import run_print_mode, run_repl + # Handle --continue and --resume flags + if continue_session or resume is not None: + from openharness.services.session_storage import ( + list_session_snapshots, + load_session_by_id, + load_session_snapshot, + ) + + session_data = None + if continue_session: + session_data = load_session_snapshot(cwd) + if session_data is None: + print("No previous session found in this directory.", file=sys.stderr) + raise typer.Exit(1) + print(f"Continuing session: {session_data.get('summary', '(untitled)')[:60]}") + elif resume == "" or resume is None: + # --resume with no value: show session picker + sessions = list_session_snapshots(cwd, limit=10) + if not sessions: + print("No saved sessions found.", file=sys.stderr) + raise typer.Exit(1) + print("Saved sessions:") + for i, s in enumerate(sessions, 1): + print(f" {i}. [{s['session_id']}] {s.get('summary', '?')[:50]} ({s['message_count']} msgs)") + choice = typer.prompt("Enter session number or ID") + try: + idx = int(choice) - 1 + if 0 <= idx < len(sessions): + session_data = load_session_by_id(cwd, sessions[idx]["session_id"]) + else: + print("Invalid selection.", file=sys.stderr) + raise typer.Exit(1) + except ValueError: + session_data = load_session_by_id(cwd, choice) + if session_data is None: + print(f"Session not found: {choice}", file=sys.stderr) + raise typer.Exit(1) + else: + session_data = load_session_by_id(cwd, resume) + if session_data is None: + print(f"Session not found: {resume}", file=sys.stderr) + raise typer.Exit(1) + + # Pass restored session to the REPL + asyncio.run( + run_repl( + prompt=None, + cwd=cwd, + model=session_data.get("model") or model, + backend_only=backend_only, + base_url=base_url, + system_prompt=session_data.get("system_prompt") or system_prompt, + api_key=api_key, + restore_messages=session_data.get("messages"), + ) + ) + return + if print_mode is not None: prompt = print_mode.strip() if not prompt: diff --git a/src/openharness/services/cron.py b/src/openharness/services/cron.py index 6b05ab5..b856a93 100644 --- a/src/openharness/services/cron.py +++ b/src/openharness/services/cron.py @@ -3,8 +3,11 @@ from __future__ import annotations import json +from datetime import datetime, timezone from typing import Any +from croniter import croniter + from openharness.config.paths import get_cron_registry_path @@ -23,11 +26,34 @@ def load_cron_jobs() -> list[dict[str, Any]]: def save_cron_jobs(jobs: list[dict[str, Any]]) -> None: """Persist cron jobs to disk.""" path = get_cron_registry_path() + path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(jobs, indent=2) + "\n", encoding="utf-8") +def validate_cron_expression(expression: str) -> bool: + """Return True if the expression is a valid cron schedule.""" + return croniter.is_valid(expression) + + +def next_run_time(expression: str, base: datetime | None = None) -> datetime: + """Return the next run time for a cron expression.""" + base = base or datetime.now(timezone.utc) + return croniter(expression, base).get_next(datetime) + + def upsert_cron_job(job: dict[str, Any]) -> None: - """Insert or replace one cron job.""" + """Insert or replace one cron job. + + Automatically sets ``enabled`` to True and computes ``next_run`` when the + schedule is a valid cron expression. + """ + job.setdefault("enabled", True) + job.setdefault("created_at", datetime.now(timezone.utc).isoformat()) + + schedule = job.get("schedule", "") + if validate_cron_expression(schedule): + job["next_run"] = next_run_time(schedule).isoformat() + jobs = [existing for existing in load_cron_jobs() if existing.get("name") != job.get("name")] jobs.append(job) jobs.sort(key=lambda item: str(item.get("name", ""))) @@ -50,3 +76,29 @@ def get_cron_job(name: str) -> dict[str, Any] | None: if job.get("name") == name: return job return None + + +def set_job_enabled(name: str, enabled: bool) -> bool: + """Enable or disable a cron job. Returns False if job not found.""" + jobs = load_cron_jobs() + for job in jobs: + if job.get("name") == name: + job["enabled"] = enabled + save_cron_jobs(jobs) + return True + return False + + +def mark_job_run(name: str, *, success: bool) -> None: + """Update last_run and recompute next_run after a job executes.""" + jobs = load_cron_jobs() + now = datetime.now(timezone.utc) + for job in jobs: + if job.get("name") == name: + job["last_run"] = now.isoformat() + job["last_status"] = "success" if success else "failed" + schedule = job.get("schedule", "") + if validate_cron_expression(schedule): + job["next_run"] = next_run_time(schedule, now).isoformat() + save_cron_jobs(jobs) + return diff --git a/src/openharness/services/cron_scheduler.py b/src/openharness/services/cron_scheduler.py new file mode 100644 index 0000000..3172f94 --- /dev/null +++ b/src/openharness/services/cron_scheduler.py @@ -0,0 +1,344 @@ +"""Background cron scheduler daemon. + +Runs as a standalone process (``oh cron start``) or can be embedded via +:func:`run_scheduler_loop`. Every tick it reads the cron registry, checks +which enabled jobs are due, executes them, and records results in a history +log. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import signal +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from openharness.config.paths import get_data_dir, get_logs_dir +from openharness.services.cron import ( + load_cron_jobs, + mark_job_run, + validate_cron_expression, +) + +logger = logging.getLogger(__name__) + +TICK_INTERVAL_SECONDS = 30 +"""How often the scheduler checks for due jobs.""" + + +# --------------------------------------------------------------------------- +# History helpers +# --------------------------------------------------------------------------- + +def get_history_path() -> Path: + """Return the path to the cron execution history file.""" + return get_data_dir() / "cron_history.jsonl" + + +def append_history(entry: dict[str, Any]) -> None: + """Append one execution record to the history log.""" + path = get_history_path() + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(entry) + "\n") + + +def load_history(*, limit: int = 50, job_name: str | None = None) -> list[dict[str, Any]]: + """Load the most recent execution history entries.""" + path = get_history_path() + if not path.exists(): + return [] + entries: list[dict[str, Any]] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if job_name and entry.get("name") != job_name: + continue + entries.append(entry) + return entries[-limit:] + + +# --------------------------------------------------------------------------- +# PID file helpers +# --------------------------------------------------------------------------- + +def get_pid_path() -> Path: + """Return the scheduler PID file path.""" + return get_data_dir() / "cron_scheduler.pid" + + +def read_pid() -> int | None: + """Read the PID of a running scheduler, or None.""" + path = get_pid_path() + if not path.exists(): + return None + try: + pid = int(path.read_text(encoding="utf-8").strip()) + except (ValueError, OSError): + return None + # Check if process is alive + try: + os.kill(pid, 0) + except OSError: + logger.debug("Removed stale scheduler PID file (pid=%d)", pid) + path.unlink(missing_ok=True) + return None + return pid + + +def write_pid() -> None: + """Write the current process PID.""" + path = get_pid_path() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(str(os.getpid()) + "\n", encoding="utf-8") + + +def remove_pid() -> None: + """Remove the PID file.""" + get_pid_path().unlink(missing_ok=True) + + +def is_scheduler_running() -> bool: + """Return True if a scheduler process is alive.""" + return read_pid() is not None + + +def stop_scheduler() -> bool: + """Send SIGTERM to the running scheduler. Returns True if killed.""" + pid = read_pid() + if pid is None: + return False + try: + os.kill(pid, signal.SIGTERM) + except OSError: + remove_pid() + return False + # Wait briefly for process to exit + for _ in range(10): + try: + os.kill(pid, 0) + except OSError: + remove_pid() + return True + time.sleep(0.2) + # Force kill + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + remove_pid() + return True + + +# --------------------------------------------------------------------------- +# Job execution +# --------------------------------------------------------------------------- + +async def execute_job(job: dict[str, Any]) -> dict[str, Any]: + """Run a single cron job and return a history entry.""" + name = job["name"] + command = job["command"] + cwd = Path(job.get("cwd") or ".").expanduser() + started_at = datetime.now(timezone.utc) + + logger.info("Executing cron job %r: %s", name, command) + try: + process = await asyncio.create_subprocess_exec( + "/bin/bash", + "-lc", + command, + cwd=str(cwd), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await asyncio.wait_for( + process.communicate(), + timeout=300, + ) + except asyncio.TimeoutError: + try: + process.kill() + await process.wait() + except Exception: + pass + entry = { + "name": name, + "command": command, + "started_at": started_at.isoformat(), + "ended_at": datetime.now(timezone.utc).isoformat(), + "returncode": -1, + "status": "timeout", + "stdout": "", + "stderr": "Job timed out after 300s", + } + mark_job_run(name, success=False) + append_history(entry) + return entry + except Exception as exc: + entry = { + "name": name, + "command": command, + "started_at": started_at.isoformat(), + "ended_at": datetime.now(timezone.utc).isoformat(), + "returncode": -1, + "status": "error", + "stdout": "", + "stderr": str(exc), + } + mark_job_run(name, success=False) + append_history(entry) + return entry + + success = process.returncode == 0 + entry = { + "name": name, + "command": command, + "started_at": started_at.isoformat(), + "ended_at": datetime.now(timezone.utc).isoformat(), + "returncode": process.returncode, + "status": "success" if success else "failed", + "stdout": (stdout.decode("utf-8", errors="replace")[-2000:] if stdout else ""), + "stderr": (stderr.decode("utf-8", errors="replace")[-2000:] if stderr else ""), + } + mark_job_run(name, success=success) + append_history(entry) + logger.info("Job %r finished: %s (rc=%s)", name, entry["status"], process.returncode) + return entry + + +# --------------------------------------------------------------------------- +# Scheduler loop +# --------------------------------------------------------------------------- + +def _jobs_due(jobs: list[dict[str, Any]], now: datetime) -> list[dict[str, Any]]: + """Return jobs whose next_run is at or before *now*.""" + due: list[dict[str, Any]] = [] + for job in jobs: + if not job.get("enabled", True): + continue + schedule = job.get("schedule", "") + if not validate_cron_expression(schedule): + continue + next_run_str = job.get("next_run") + if not next_run_str: + continue + try: + next_run = datetime.fromisoformat(next_run_str) + if next_run.tzinfo is None: + next_run = next_run.replace(tzinfo=timezone.utc) + except (ValueError, TypeError): + continue + if next_run <= now: + due.append(job) + return due + + +async def run_scheduler_loop(*, once: bool = False) -> None: + """Main scheduler loop. Runs until SIGTERM or *once* is True (test mode).""" + shutdown = asyncio.Event() + + def _on_signal() -> None: + logger.info("Received shutdown signal") + shutdown.set() + + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, _on_signal) + + write_pid() + logger.info("Cron scheduler started (pid=%d, tick=%ds)", os.getpid(), TICK_INTERVAL_SECONDS) + + try: + while not shutdown.is_set(): + now = datetime.now(timezone.utc) + jobs = load_cron_jobs() + due = _jobs_due(jobs, now) + + if due: + logger.info("Tick: %d job(s) due", len(due)) + # Execute due jobs concurrently + results = await asyncio.gather( + *(execute_job(job) for job in due), return_exceptions=True + ) + for result in results: + if isinstance(result, BaseException): + logger.error("Unexpected error executing cron job: %s", result) + + if once: + break + + try: + await asyncio.wait_for(shutdown.wait(), timeout=TICK_INTERVAL_SECONDS) + except asyncio.TimeoutError: + pass + finally: + remove_pid() + logger.info("Cron scheduler stopped") + + +# --------------------------------------------------------------------------- +# Daemon entry point (spawned by ``oh cron start``) +# --------------------------------------------------------------------------- + +def _run_daemon() -> None: + """Entry point for the scheduler subprocess.""" + log_file = get_logs_dir() / "cron_scheduler.log" + log_file.parent.mkdir(parents=True, exist_ok=True) + logging.basicConfig( + filename=str(log_file), + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + ) + asyncio.run(run_scheduler_loop()) + + +def start_daemon() -> int: + """Fork and start the scheduler daemon. Returns the child PID.""" + existing = read_pid() + if existing is not None: + raise RuntimeError(f"Scheduler already running (pid={existing})") + + pid = os.fork() + if pid > 0: + # Parent — wait a moment for the child to write its PID file + time.sleep(0.3) + return pid + + # Child — detach + os.setsid() + # Redirect stdio + devnull = os.open(os.devnull, os.O_RDWR) + os.dup2(devnull, 0) + os.dup2(devnull, 1) + os.dup2(devnull, 2) + os.close(devnull) + + _run_daemon() + sys.exit(0) + + +def scheduler_status() -> dict[str, Any]: + """Return a status dict about the scheduler.""" + pid = read_pid() + log_path = get_logs_dir() / "cron_scheduler.log" + jobs = load_cron_jobs() + enabled = [j for j in jobs if j.get("enabled", True)] + return { + "running": pid is not None, + "pid": pid, + "total_jobs": len(jobs), + "enabled_jobs": len(enabled), + "log_file": str(log_path), + "history_file": str(get_history_path()), + } diff --git a/src/openharness/tools/__init__.py b/src/openharness/tools/__init__.py index db87c0e..56fc28b 100644 --- a/src/openharness/tools/__init__.py +++ b/src/openharness/tools/__init__.py @@ -9,6 +9,7 @@ from openharness.tools.config_tool import ConfigTool from openharness.tools.cron_create_tool import CronCreateTool from openharness.tools.cron_delete_tool import CronDeleteTool from openharness.tools.cron_list_tool import CronListTool +from openharness.tools.cron_toggle_tool import CronToggleTool from openharness.tools.enter_plan_mode_tool import EnterPlanModeTool from openharness.tools.enter_worktree_tool import EnterWorktreeTool from openharness.tools.exit_plan_mode_tool import ExitPlanModeTool @@ -71,6 +72,7 @@ def create_default_tool_registry(mcp_manager=None) -> ToolRegistry: CronCreateTool(), CronListTool(), CronDeleteTool(), + CronToggleTool(), RemoteTriggerTool(), TaskCreateTool(), TaskGetTool(), diff --git a/src/openharness/tools/cron_create_tool.py b/src/openharness/tools/cron_create_tool.py index adcbdb4..6b271e3 100644 --- a/src/openharness/tools/cron_create_tool.py +++ b/src/openharness/tools/cron_create_tool.py @@ -4,7 +4,7 @@ from __future__ import annotations from pydantic import BaseModel, Field -from openharness.services.cron import upsert_cron_job +from openharness.services.cron import upsert_cron_job, validate_cron_expression from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult @@ -12,16 +12,25 @@ class CronCreateToolInput(BaseModel): """Arguments for cron job creation.""" name: str = Field(description="Unique cron job name") - schedule: str = Field(description="Human-readable schedule expression") + schedule: str = Field( + description=( + "Cron schedule expression (e.g. '*/5 * * * *' for every 5 minutes, " + "'0 9 * * 1-5' for weekdays at 9am)" + ), + ) command: str = Field(description="Shell command to run when triggered") cwd: str | None = Field(default=None, description="Optional working directory override") + enabled: bool = Field(default=True, description="Whether the job is active") class CronCreateTool(BaseTool): """Create or replace a local cron job.""" name = "cron_create" - description = "Create or replace a local cron-style job." + description = ( + "Create or replace a local cron job with a standard cron expression. " + "Use 'oh cron start' to run the scheduler daemon." + ) input_model = CronCreateToolInput async def execute( @@ -29,12 +38,26 @@ class CronCreateTool(BaseTool): arguments: CronCreateToolInput, context: ToolExecutionContext, ) -> ToolResult: + if not validate_cron_expression(arguments.schedule): + return ToolResult( + output=( + f"Invalid cron expression: {arguments.schedule!r}\n" + "Use standard 5-field format: minute hour day month weekday\n" + "Examples: '*/5 * * * *' (every 5 min), '0 9 * * 1-5' (weekdays 9am)" + ), + is_error=True, + ) + upsert_cron_job( { "name": arguments.name, "schedule": arguments.schedule, "command": arguments.command, "cwd": arguments.cwd or str(context.cwd), + "enabled": arguments.enabled, } ) - return ToolResult(output=f"Created cron job {arguments.name}") + status = "enabled" if arguments.enabled else "disabled" + return ToolResult( + output=f"Created cron job '{arguments.name}' [{arguments.schedule}] ({status})" + ) diff --git a/src/openharness/tools/cron_list_tool.py b/src/openharness/tools/cron_list_tool.py index a5db858..65fc7fa 100644 --- a/src/openharness/tools/cron_list_tool.py +++ b/src/openharness/tools/cron_list_tool.py @@ -5,6 +5,7 @@ from __future__ import annotations from pydantic import BaseModel from openharness.services.cron import load_cron_jobs +from openharness.services.cron_scheduler import is_scheduler_running from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult @@ -16,7 +17,7 @@ class CronListTool(BaseTool): """List local cron jobs.""" name = "cron_list" - description = "List configured local cron-style jobs." + description = "List configured local cron jobs with schedule, status, and next run time." input_model = CronListToolInput def is_read_only(self, arguments: CronListToolInput) -> bool: @@ -32,7 +33,23 @@ class CronListTool(BaseTool): jobs = load_cron_jobs() if not jobs: return ToolResult(output="No cron jobs configured.") - lines = [] + + scheduler = "running" if is_scheduler_running() else "stopped" + lines = [f"Scheduler: {scheduler}", ""] + for job in jobs: - lines.append(f"{job['name']} [{job['schedule']}] -> {job['command']}") + enabled = "on" if job.get("enabled", True) else "off" + last_run = job.get("last_run", "never") + if last_run != "never": + last_run = last_run[:19] + next_run = job.get("next_run", "n/a") + if next_run != "n/a": + next_run = next_run[:19] + last_status = job.get("last_status", "") + status_str = f" ({last_status})" if last_status else "" + lines.append( + f"[{enabled}] {job['name']} {job.get('schedule', '?')}\n" + f" cmd: {job['command']}\n" + f" last: {last_run}{status_str} next: {next_run}" + ) return ToolResult(output="\n".join(lines)) diff --git a/src/openharness/tools/cron_toggle_tool.py b/src/openharness/tools/cron_toggle_tool.py new file mode 100644 index 0000000..0a34ab2 --- /dev/null +++ b/src/openharness/tools/cron_toggle_tool.py @@ -0,0 +1,37 @@ +"""Tool for enabling or disabling local cron jobs.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + +from openharness.services.cron import set_job_enabled +from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult + + +class CronToggleToolInput(BaseModel): + """Arguments for toggling a cron job.""" + + name: str = Field(description="Cron job name") + enabled: bool = Field(description="True to enable, False to disable") + + +class CronToggleTool(BaseTool): + """Enable or disable a local cron job.""" + + name = "cron_toggle" + description = "Enable or disable a local cron job by name." + input_model = CronToggleToolInput + + async def execute( + self, + arguments: CronToggleToolInput, + context: ToolExecutionContext, + ) -> ToolResult: + del context + if not set_job_enabled(arguments.name, arguments.enabled): + return ToolResult( + output=f"Cron job not found: {arguments.name}", + is_error=True, + ) + state = "enabled" if arguments.enabled else "disabled" + return ToolResult(output=f"Cron job '{arguments.name}' is now {state}") diff --git a/src/openharness/ui/app.py b/src/openharness/ui/app.py index c900375..2d5015b 100644 --- a/src/openharness/ui/app.py +++ b/src/openharness/ui/app.py @@ -23,6 +23,7 @@ async def run_repl( api_format: str | None = None, api_client: SupportsStreamingMessages | None = None, backend_only: bool = False, + restore_messages: list[dict] | None = None, ) -> None: """Run the default OpenHarness interactive application (React TUI).""" if backend_only: @@ -34,6 +35,7 @@ async def run_repl( api_key=api_key, api_format=api_format, api_client=api_client, + restore_messages=restore_messages, ) return diff --git a/src/openharness/ui/backend_host.py b/src/openharness/ui/backend_host.py index a761eea..ea21cab 100644 --- a/src/openharness/ui/backend_host.py +++ b/src/openharness/ui/backend_host.py @@ -36,6 +36,7 @@ class BackendHostConfig: api_key: str | None = None api_format: str | None = None api_client: SupportsStreamingMessages | None = None + restore_messages: list[dict] | None = None class ReactBackendHost: @@ -59,6 +60,7 @@ class ReactBackendHost: api_key=self._config.api_key, api_format=self._config.api_format, api_client=self._config.api_client, + restore_messages=self._config.restore_messages, permission_prompt=self._ask_permission, ask_user_prompt=self._ask_question, ) @@ -288,6 +290,7 @@ async def run_backend_host( api_format: str | None = None, cwd: str | None = None, api_client: SupportsStreamingMessages | None = None, + restore_messages: list[dict] | None = None, ) -> int: """Run the structured React backend host.""" if cwd: @@ -300,6 +303,7 @@ async def run_backend_host( api_key=api_key, api_format=api_format, api_client=api_client, + restore_messages=restore_messages, ) ) return await host.run() diff --git a/src/openharness/ui/runtime.py b/src/openharness/ui/runtime.py index a8b46ba..c8005d0 100644 --- a/src/openharness/ui/runtime.py +++ b/src/openharness/ui/runtime.py @@ -13,6 +13,7 @@ from openharness.bridge import get_bridge_manager from openharness.commands import CommandContext, CommandResult, create_default_command_registry from openharness.config import get_config_file_path, load_settings from openharness.engine import QueryEngine +from openharness.engine.messages import ConversationMessage from openharness.engine.stream_events import StreamEvent from openharness.hooks import HookEvent, HookExecutionContext, HookExecutor, load_hook_registry from openharness.hooks.hot_reload import HookReloader @@ -98,6 +99,7 @@ async def build_runtime( api_client: SupportsStreamingMessages | None = None, permission_prompt: PermissionPrompt | None = None, ask_user_prompt: AskUserPrompt | None = None, + restore_messages: list[dict] | None = None, ) -> RuntimeBundle: """Build the shared runtime for an OpenHarness session.""" settings = load_settings().merge_cli_overrides( @@ -171,6 +173,13 @@ async def build_runtime( hook_executor=hook_executor, tool_metadata={"mcp_manager": mcp_manager, "bridge_manager": bridge_manager}, ) + # Restore messages from a saved session if provided + if restore_messages: + restored = [ + ConversationMessage.model_validate(m) for m in restore_messages + ] + engine.load_messages(restored) + from uuid import uuid4 return RuntimeBundle( diff --git a/tests/test_services/test_cron.py b/tests/test_services/test_cron.py new file mode 100644 index 0000000..cea7db1 --- /dev/null +++ b/tests/test_services/test_cron.py @@ -0,0 +1,154 @@ +"""Tests for cron registry helpers.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from openharness.services.cron import ( + delete_cron_job, + get_cron_job, + load_cron_jobs, + mark_job_run, + next_run_time, + set_job_enabled, + upsert_cron_job, + validate_cron_expression, +) + + +@pytest.fixture(autouse=True) +def _tmp_cron_dir(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Redirect cron registry to a temp directory.""" + data_dir = tmp_path / "data" + data_dir.mkdir() + monkeypatch.setattr( + "openharness.services.cron.get_cron_registry_path", + lambda: data_dir / "cron_jobs.json", + ) + + +class TestValidation: + def test_valid_expressions(self) -> None: + assert validate_cron_expression("* * * * *") + assert validate_cron_expression("*/5 * * * *") + assert validate_cron_expression("0 9 * * 1-5") + assert validate_cron_expression("0 0 1 1 *") + + def test_invalid_expressions(self) -> None: + assert not validate_cron_expression("") + assert not validate_cron_expression("every 5 minutes") + assert not validate_cron_expression("60 * * * *") + assert not validate_cron_expression("* * * *") # only 4 fields + + def test_next_run_time(self) -> None: + base = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + nxt = next_run_time("0 * * * *", base) + assert nxt == datetime(2026, 1, 1, 1, 0, 0, tzinfo=timezone.utc) + + +class TestCRUD: + def test_empty_load(self) -> None: + assert load_cron_jobs() == [] + + def test_upsert_and_load(self) -> None: + upsert_cron_job({"name": "test-job", "schedule": "*/5 * * * *", "command": "echo hi"}) + jobs = load_cron_jobs() + assert len(jobs) == 1 + assert jobs[0]["name"] == "test-job" + assert jobs[0]["enabled"] is True + assert "next_run" in jobs[0] + assert "created_at" in jobs[0] + + def test_upsert_replaces(self) -> None: + upsert_cron_job({"name": "j1", "schedule": "* * * * *", "command": "echo 1"}) + upsert_cron_job({"name": "j1", "schedule": "0 * * * *", "command": "echo 2"}) + jobs = load_cron_jobs() + assert len(jobs) == 1 + assert jobs[0]["command"] == "echo 2" + + def test_delete(self) -> None: + upsert_cron_job({"name": "j1", "schedule": "* * * * *", "command": "echo 1"}) + assert delete_cron_job("j1") is True + assert load_cron_jobs() == [] + + def test_delete_missing(self) -> None: + assert delete_cron_job("nope") is False + + def test_get_job(self) -> None: + upsert_cron_job({"name": "j1", "schedule": "* * * * *", "command": "echo 1"}) + job = get_cron_job("j1") + assert job is not None + assert job["name"] == "j1" + + def test_get_missing(self) -> None: + assert get_cron_job("nope") is None + + def test_sorted_output(self) -> None: + upsert_cron_job({"name": "z-job", "schedule": "* * * * *", "command": "z"}) + upsert_cron_job({"name": "a-job", "schedule": "* * * * *", "command": "a"}) + jobs = load_cron_jobs() + assert [j["name"] for j in jobs] == ["a-job", "z-job"] + + +class TestToggle: + def test_enable_disable(self) -> None: + upsert_cron_job({"name": "j1", "schedule": "* * * * *", "command": "echo 1"}) + assert set_job_enabled("j1", False) is True + job = get_cron_job("j1") + assert job is not None + assert job["enabled"] is False + + assert set_job_enabled("j1", True) is True + job = get_cron_job("j1") + assert job is not None + assert job["enabled"] is True + + def test_toggle_missing(self) -> None: + assert set_job_enabled("nope", True) is False + + +class TestMarkRun: + def test_mark_success(self) -> None: + upsert_cron_job({"name": "j1", "schedule": "*/5 * * * *", "command": "echo ok"}) + mark_job_run("j1", success=True) + job = get_cron_job("j1") + assert job is not None + assert job["last_status"] == "success" + assert "last_run" in job + + def test_mark_failure(self) -> None: + upsert_cron_job({"name": "j1", "schedule": "*/5 * * * *", "command": "false"}) + mark_job_run("j1", success=False) + job = get_cron_job("j1") + assert job is not None + assert job["last_status"] == "failed" + + def test_mark_missing_is_noop(self) -> None: + # Should not raise + mark_job_run("nope", success=True) + + +class TestCorruptData: + def test_corrupt_json(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + bad_file = tmp_path / "data" / "cron_jobs.json" + bad_file.parent.mkdir(parents=True, exist_ok=True) + bad_file.write_text("{not valid json", encoding="utf-8") + monkeypatch.setattr( + "openharness.services.cron.get_cron_registry_path", + lambda: bad_file, + ) + assert load_cron_jobs() == [] + + def test_non_list_json(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + bad_file = tmp_path / "data" / "cron_jobs.json" + bad_file.parent.mkdir(parents=True, exist_ok=True) + bad_file.write_text(json.dumps({"not": "a list"}), encoding="utf-8") + monkeypatch.setattr( + "openharness.services.cron.get_cron_registry_path", + lambda: bad_file, + ) + assert load_cron_jobs() == [] diff --git a/tests/test_services/test_cron_scheduler.py b/tests/test_services/test_cron_scheduler.py new file mode 100644 index 0000000..a95716f --- /dev/null +++ b/tests/test_services/test_cron_scheduler.py @@ -0,0 +1,170 @@ +"""Tests for the cron scheduler daemon.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + +from openharness.services.cron_scheduler import ( + _jobs_due, + append_history, + execute_job, + load_history, + run_scheduler_loop, +) + + +@pytest.fixture(autouse=True) +def _tmp_dirs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Redirect data and log directories to temp.""" + data_dir = tmp_path / "data" + logs_dir = tmp_path / "logs" + data_dir.mkdir() + logs_dir.mkdir() + monkeypatch.setattr("openharness.services.cron_scheduler.get_data_dir", lambda: data_dir) + monkeypatch.setattr("openharness.services.cron_scheduler.get_logs_dir", lambda: logs_dir) + # Also redirect the cron registry used by the scheduler + monkeypatch.setattr( + "openharness.services.cron.get_cron_registry_path", + lambda: data_dir / "cron_jobs.json", + ) + + +class TestHistory: + def test_empty_history(self) -> None: + assert load_history() == [] + + def test_append_and_load(self) -> None: + append_history({"name": "j1", "status": "success"}) + append_history({"name": "j2", "status": "failed"}) + entries = load_history() + assert len(entries) == 2 + assert entries[0]["name"] == "j1" + + def test_filter_by_name(self) -> None: + append_history({"name": "j1", "status": "success"}) + append_history({"name": "j2", "status": "success"}) + entries = load_history(job_name="j1") + assert len(entries) == 1 + assert entries[0]["name"] == "j1" + + def test_limit(self) -> None: + for i in range(10): + append_history({"name": f"j{i}", "status": "success"}) + entries = load_history(limit=3) + assert len(entries) == 3 + # Should be the last 3 + assert entries[0]["name"] == "j7" + + +class TestJobsDue: + def test_due_job(self) -> None: + now = datetime.now(timezone.utc) + past = (now - timedelta(minutes=5)).isoformat() + jobs = [ + {"name": "j1", "schedule": "* * * * *", "enabled": True, "next_run": past}, + ] + due = _jobs_due(jobs, now) + assert len(due) == 1 + + def test_future_job_not_due(self) -> None: + now = datetime.now(timezone.utc) + future = (now + timedelta(hours=1)).isoformat() + jobs = [ + {"name": "j1", "schedule": "* * * * *", "enabled": True, "next_run": future}, + ] + due = _jobs_due(jobs, now) + assert len(due) == 0 + + def test_disabled_job_not_due(self) -> None: + now = datetime.now(timezone.utc) + past = (now - timedelta(minutes=5)).isoformat() + jobs = [ + {"name": "j1", "schedule": "* * * * *", "enabled": False, "next_run": past}, + ] + due = _jobs_due(jobs, now) + assert len(due) == 0 + + def test_invalid_schedule_skipped(self) -> None: + now = datetime.now(timezone.utc) + past = (now - timedelta(minutes=5)).isoformat() + jobs = [ + {"name": "j1", "schedule": "not valid", "enabled": True, "next_run": past}, + ] + due = _jobs_due(jobs, now) + assert len(due) == 0 + + def test_missing_next_run_skipped(self) -> None: + now = datetime.now(timezone.utc) + jobs = [ + {"name": "j1", "schedule": "* * * * *", "enabled": True}, + ] + due = _jobs_due(jobs, now) + assert len(due) == 0 + + +class TestExecuteJob: + @pytest.mark.asyncio + async def test_successful_job(self) -> None: + job = {"name": "echo-test", "command": "echo hello", "cwd": "/tmp"} + entry = await execute_job(job) + assert entry["status"] == "success" + assert entry["returncode"] == 0 + assert "hello" in entry["stdout"] + + @pytest.mark.asyncio + async def test_failing_job(self) -> None: + job = {"name": "fail-test", "command": "exit 1", "cwd": "/tmp"} + entry = await execute_job(job) + assert entry["status"] == "failed" + assert entry["returncode"] == 1 + + @pytest.mark.asyncio + async def test_timeout_job(self) -> None: + with patch("openharness.services.cron_scheduler.asyncio.wait_for") as mock_wait: + import asyncio + + mock_wait.side_effect = asyncio.TimeoutError() + + # Need to mock create_subprocess_exec to return a mock process + mock_process = AsyncMock() + mock_process.kill = AsyncMock() + mock_process.wait = AsyncMock() + with patch( + "openharness.services.cron_scheduler.asyncio.create_subprocess_exec", + return_value=mock_process, + ): + job = {"name": "slow-test", "command": "sleep 999", "cwd": "/tmp"} + entry = await execute_job(job) + assert entry["status"] == "timeout" + + +class TestSchedulerLoop: + @pytest.mark.asyncio + async def test_once_mode_with_no_jobs(self) -> None: + """Scheduler loop in once-mode should complete without error when no jobs exist.""" + await run_scheduler_loop(once=True) + + @pytest.mark.asyncio + async def test_once_mode_fires_due_job(self) -> None: + """Scheduler loop should fire a job that is due.""" + from openharness.services.cron import upsert_cron_job + + upsert_cron_job({"name": "test-once", "schedule": "* * * * *", "command": "echo fired"}) + + # Force next_run to the past so it's immediately due + from openharness.services.cron import load_cron_jobs, save_cron_jobs + + jobs = load_cron_jobs() + now = datetime.now(timezone.utc) + jobs[0]["next_run"] = (now - timedelta(minutes=1)).isoformat() + save_cron_jobs(jobs) + + await run_scheduler_loop(once=True) + + entries = load_history(job_name="test-once") + assert len(entries) == 1 + assert entries[0]["status"] == "success" diff --git a/tests/test_tools/test_core_tools.py b/tests/test_tools/test_core_tools.py index aa0ccd1..b1863de 100644 --- a/tests/test_tools/test_core_tools.py +++ b/tests/test_tools/test_core_tools.py @@ -225,7 +225,7 @@ async def test_cron_and_remote_trigger_tools(tmp_path: Path, monkeypatch): context = ToolExecutionContext(cwd=tmp_path) create_result = await CronCreateTool().execute( - CronCreateToolInput(name="nightly", schedule="daily", command="printf 'CRON_OK'"), + CronCreateToolInput(name="nightly", schedule="0 0 * * *", command="printf 'CRON_OK'"), context, ) assert create_result.is_error is False diff --git a/tests/test_tools/test_integration_flows.py b/tests/test_tools/test_integration_flows.py index 33b20f9..fe4ff1d 100644 --- a/tests/test_tools/test_integration_flows.py +++ b/tests/test_tools/test_integration_flows.py @@ -224,7 +224,7 @@ async def test_notebook_and_cron_flow_across_registry(tmp_path: Path, monkeypatc assert "flow ok" in (tmp_path / "nb" / "demo.ipynb").read_text(encoding="utf-8") await cron_create.execute( - cron_create.input_model(name="flow", schedule="daily", command="printf 'FLOW_CRON_OK'"), + cron_create.input_model(name="flow", schedule="0 0 * * *", command="printf 'FLOW_CRON_OK'"), context, ) list_result = await cron_list.execute(cron_list.input_model(), context)