feat: wire --resume/--continue CLI flags and add cron scheduler daemon (#16)
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
This commit is contained in:
Chao Qin
@@ -26,6 +26,7 @@ dependencies = [
|
||||
"pyperclip>=1.9.0",
|
||||
"pyyaml>=6.0",
|
||||
"watchfiles>=0.20.0",
|
||||
"croniter>=2.0.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()),
|
||||
}
|
||||
@@ -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(),
|
||||
|
||||
@@ -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})"
|
||||
)
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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}")
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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() == []
|
||||
@@ -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"
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user