From 1929ad805106dc2f45b79fc6b3cc36e111004259 Mon Sep 17 00:00:00 2001 From: Jiabin Tang <56426168+tjb-tech@users.noreply.github.com> Date: Sun, 10 May 2026 16:55:10 +0800 Subject: [PATCH] feat(ohmo): add agent-turn cron delivery (#247) --- ohmo/gateway/notify.py | 78 +++++++++++++ ohmo/runtime.py | 6 +- src/openharness/cli.py | 16 ++- src/openharness/services/cron.py | 28 ++++- src/openharness/services/cron_scheduler.py | 116 ++++++++++++++++++- src/openharness/tools/cron_create_tool.py | 64 ++++++++-- src/openharness/tools/cron_list_tool.py | 18 ++- src/openharness/tools/remote_trigger_tool.py | 4 +- tests/test_services/test_cron.py | 25 ++++ tests/test_tools/test_core_tools.py | 31 ++++- 10 files changed, 363 insertions(+), 23 deletions(-) create mode 100644 ohmo/gateway/notify.py diff --git a/ohmo/gateway/notify.py b/ohmo/gateway/notify.py new file mode 100644 index 0000000..5aadb62 --- /dev/null +++ b/ohmo/gateway/notify.py @@ -0,0 +1,78 @@ +"""Proactive notification helpers for ohmo gateway channels.""" + +from __future__ import annotations + +import asyncio +import json +import logging +from pathlib import Path +from typing import Any + +from ohmo.gateway.config import load_gateway_config + +logger = logging.getLogger(__name__) + + +class OhmoNotificationError(RuntimeError): + """Raised when a proactive notification cannot be delivered.""" + + +def _chunk_text(text: str, *, max_chars: int = 1800) -> list[str]: + """Split text into message-sized chunks without losing content.""" + stripped = text.strip() + if not stripped: + return [] + chunks: list[str] = [] + remaining = stripped + while len(remaining) > max_chars: + split_at = remaining.rfind("\n", 0, max_chars) + if split_at < max_chars // 2: + split_at = max_chars + chunks.append(remaining[:split_at].strip()) + remaining = remaining[split_at:].strip() + if remaining: + chunks.append(remaining) + return chunks + + +def _send_feishu_text_sync(*, user_open_id: str, content: str, workspace: str | Path | None = None) -> None: + """Send a Feishu direct message using ohmo gateway Feishu credentials.""" + try: + import lark_oapi as lark + from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody + except ImportError as exc: # pragma: no cover - depends on optional extra + raise OhmoNotificationError("Feishu SDK is not installed. Run: pip install lark-oapi") from exc + + config = load_gateway_config(workspace) + feishu_config: dict[str, Any] = config.channel_configs.get("feishu", {}) + app_id = str(feishu_config.get("app_id") or "").strip() + app_secret = str(feishu_config.get("app_secret") or "").strip() + if not app_id or not app_secret: + raise OhmoNotificationError("Feishu app_id/app_secret are not configured in ohmo gateway config.") + + client = lark.Client.builder().app_id(app_id).app_secret(app_secret).log_level(lark.LogLevel.INFO).build() + for chunk in _chunk_text(content): + request = ( + CreateMessageRequest.builder() + .receive_id_type("open_id") + .request_body( + CreateMessageRequestBody.builder() + .receive_id(user_open_id) + .msg_type("text") + .content(json.dumps({"text": chunk}, ensure_ascii=False)) + .build() + ) + .build() + ) + response = client.im.v1.message.create(request) + if not response.success(): + log_id = response.get_log_id() if hasattr(response, "get_log_id") else "" + raise OhmoNotificationError( + f"send Feishu DM failed: code={response.code}, msg={response.msg}, log_id={log_id}" + ) + + +async def send_feishu_dm(*, user_open_id: str, content: str, workspace: str | Path | None = None) -> None: + """Send a proactive Feishu direct message to a user open_id.""" + await asyncio.to_thread(_send_feishu_text_sync, user_open_id=user_open_id, content=content, workspace=workspace) + logger.info("Sent proactive Feishu DM to open_id=%s", user_open_id) diff --git a/ohmo/runtime.py b/ohmo/runtime.py index fa0ed73..21fc5ea 100644 --- a/ohmo/runtime.py +++ b/ohmo/runtime.py @@ -183,7 +183,10 @@ async def run_ohmo_print_mode( async def _print_system(message: str) -> None: print(message, file=sys.stderr) + saw_error = False + async def _render_event(event) -> None: + nonlocal saw_error if isinstance(event, AssistantTextDelta): sys.stdout.write(event.text) sys.stdout.flush() @@ -191,6 +194,7 @@ async def run_ohmo_print_mode( sys.stdout.write("\n") sys.stdout.flush() elif isinstance(event, ErrorEvent): + saw_error = True print(event.message, file=sys.stderr) elif isinstance(event, CompactProgressEvent): if event.message: @@ -209,6 +213,6 @@ async def run_ohmo_print_mode( clear_output=_clear_output, ) await close_runtime(bundle) - return 0 + return 1 if saw_error else 0 finally: os.chdir(previous_cwd) diff --git a/src/openharness/cli.py b/src/openharness/cli.py index c94546f..f0373ed 100644 --- a/src/openharness/cli.py +++ b/src/openharness/cli.py @@ -930,8 +930,20 @@ def cron_list_cmd() -> None: 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']}") + timezone = f" ({job['timezone']})" if job.get("timezone") else "" + print(f" [{enabled}] {job['name']} {job.get('schedule', '?')}{timezone}") + print(f" cmd: {job.get('command') or '(agent_turn)'}") + payload = job.get("payload") + if isinstance(payload, dict): + print( + f" payload: {payload.get('kind', 'agent_turn')} -> " + f"{payload.get('channel', '?')}:{payload.get('to', '?')}" + ) + notify = job.get("notify") + if isinstance(notify, dict): + notify_type = notify.get("type", "?") + target = notify.get("user_open_id") or notify.get("open_id") or notify.get("chat_id") or "?" + print(f" notify: {notify_type} -> {target}") print(f" last: {last}{status_indicator} next: {job.get('next_run', 'n/a')[:19]}") diff --git a/src/openharness/services/cron.py b/src/openharness/services/cron.py index 4cf61a4..33288d9 100644 --- a/src/openharness/services/cron.py +++ b/src/openharness/services/cron.py @@ -4,6 +4,7 @@ from __future__ import annotations import json from datetime import datetime, timezone +from zoneinfo import ZoneInfo from pathlib import Path from typing import Any @@ -44,9 +45,28 @@ def validate_cron_expression(expression: str) -> bool: 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.""" +def validate_timezone(tz: str | None) -> bool: + """Return True if *tz* is a valid IANA timezone or empty.""" + if not tz: + return True + try: + ZoneInfo(tz) + except Exception: + return False + return True + + +def next_run_time(expression: str, base: datetime | None = None, tz: str | None = None) -> datetime: + """Return the next run time for a cron expression. + + The returned datetime is always UTC. If *tz* is provided, the cron expression + is interpreted in that IANA timezone. + """ base = base or datetime.now(timezone.utc) + if tz: + local_base = base.astimezone(ZoneInfo(tz)) + local_next = croniter(expression, local_base).get_next(datetime) + return local_next.astimezone(timezone.utc) return croniter(expression, base).get_next(datetime) @@ -61,7 +81,7 @@ def upsert_cron_job(job: dict[str, Any]) -> None: schedule = job.get("schedule", "") if validate_cron_expression(schedule): - job["next_run"] = next_run_time(schedule).isoformat() + job["next_run"] = next_run_time(schedule, tz=job.get("timezone") or job.get("tz")).isoformat() with exclusive_file_lock(_cron_lock_path()): jobs = [existing for existing in load_cron_jobs() if existing.get("name") != job.get("name")] @@ -112,6 +132,6 @@ def mark_job_run(name: str, *, success: bool) -> None: 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() + job["next_run"] = next_run_time(schedule, now, tz=job.get("timezone") or job.get("tz")).isoformat() save_cron_jobs(jobs) return diff --git a/src/openharness/services/cron_scheduler.py b/src/openharness/services/cron_scheduler.py index 1d697ef..3c8d361 100644 --- a/src/openharness/services/cron_scheduler.py +++ b/src/openharness/services/cron_scheduler.py @@ -12,6 +12,7 @@ import asyncio import json import logging import os +import shlex import signal import sys import time @@ -28,6 +29,14 @@ from openharness.services.cron import ( from openharness.sandbox import SandboxUnavailableError from openharness.utils.shell import create_shell_subprocess +try: + from ohmo.gateway.config import load_gateway_config +except Exception: # pragma: no cover - ohmo is optional for non-ohmo cron users + load_gateway_config = None # type: ignore[assignment] + + +NOTIFICATION_OUTPUT_LIMIT = 3500 + logger = logging.getLogger(__name__) TICK_INTERVAL_SECONDS = 30 @@ -147,12 +156,113 @@ def stop_scheduler() -> bool: # Job execution # --------------------------------------------------------------------------- + +def _format_notification(job: dict[str, Any], entry: dict[str, Any]) -> str: + """Build a concise notification body for a completed cron job.""" + status = entry.get("status", "?") + rc = entry.get("returncode", "?") + lines = [ + f"⏰ Cron job finished: {job.get('name', '?')}", + f"Status: {status} (rc={rc})", + f"Started: {entry.get('started_at', '?')}", + f"Ended: {entry.get('ended_at', '?')}", + ] + stdout = str(entry.get("stdout") or "").strip() + stderr = str(entry.get("stderr") or "").strip() + if stdout: + lines.extend(["", "Output:", stdout[-NOTIFICATION_OUTPUT_LIMIT:]]) + if stderr: + lines.extend(["", "Stderr:", stderr[-NOTIFICATION_OUTPUT_LIMIT:]]) + if not stdout and not stderr: + lines.extend(["", "(no output)"]) + return "\n".join(lines) + + +async def _notify_job_result(job: dict[str, Any], entry: dict[str, Any]) -> None: + """Deliver an optional post-run notification for a cron job.""" + notify = job.get("notify") + payload = job.get("payload") + if not isinstance(notify, dict) and isinstance(payload, dict) and payload.get("deliver"): + notify = {"type": payload.get("channel"), "to": payload.get("to")} + if not isinstance(notify, dict): + return + notify_type = str(notify.get("type") or "").strip().lower() + try: + if notify_type in {"feishu_dm", "feishu"}: + from ohmo.gateway.notify import send_feishu_dm + + user_open_id = str( + notify.get("user_open_id") or notify.get("open_id") or notify.get("to") or "" + ).strip() + if not user_open_id: + raise ValueError("missing notify.user_open_id") + workspace = notify.get("workspace") + await send_feishu_dm( + user_open_id=user_open_id, + content=_format_notification(job, entry), + workspace=str(workspace) if workspace else None, + ) + elif notify_type: + raise ValueError(f"unsupported notify.type: {notify_type}") + except Exception as exc: + logger.error("Failed to notify cron job %r result: %s", job.get("name"), exc) + entry["notification_status"] = "failed" + entry["notification_error"] = str(exc) + else: + entry["notification_status"] = "sent" + + +def _command_for_job(job: dict[str, Any]) -> str: + """Return the shell command used to execute a job.""" + command = job.get("command") + if command: + return str(command) + payload = job.get("payload") + if not isinstance(payload, dict) or payload.get("kind", "agent_turn") != "agent_turn": + raise ValueError("cron job has no command or agent_turn payload") + message = str(payload.get("message") or "").strip() + if not message: + raise ValueError("agent_turn cron job is missing payload.message") + cwd = str(job.get("cwd") or ".") + parts = ["ohmo"] + profile = payload.get("profile") or job.get("provider_profile") + if profile is None and load_gateway_config is not None: + profile = load_gateway_config().provider_profile + if profile: + parts.extend(["--profile", str(profile)]) + parts.extend( + [ + "--cwd", + cwd, + "--print", + message, + ] + ) + return " ".join(shlex.quote(part) for part in parts) + + 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) + try: + command = _command_for_job(job) + except Exception as exc: + entry = { + "name": name, + "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) + await _notify_job_result(job, entry) + append_history(entry) + return entry logger.info("Executing cron job %r: %s", name, command) try: @@ -183,6 +293,7 @@ async def execute_job(job: dict[str, Any]) -> dict[str, Any]: "stderr": "Job timed out after 300s", } mark_job_run(name, success=False) + await _notify_job_result(job, entry) append_history(entry) return entry except SandboxUnavailableError as exc: @@ -197,6 +308,7 @@ async def execute_job(job: dict[str, Any]) -> dict[str, Any]: "stderr": str(exc), } mark_job_run(name, success=False) + await _notify_job_result(job, entry) append_history(entry) return entry except Exception as exc: @@ -211,6 +323,7 @@ async def execute_job(job: dict[str, Any]) -> dict[str, Any]: "stderr": str(exc), } mark_job_run(name, success=False) + await _notify_job_result(job, entry) append_history(entry) return entry @@ -226,6 +339,7 @@ async def execute_job(job: dict[str, Any]) -> dict[str, Any]: "stderr": (stderr.decode("utf-8", errors="replace")[-2000:] if stderr else ""), } mark_job_run(name, success=success) + await _notify_job_result(job, entry) append_history(entry) logger.info("Job %r finished: %s (rc=%s)", name, entry["status"], process.returncode) return entry diff --git a/src/openharness/tools/cron_create_tool.py b/src/openharness/tools/cron_create_tool.py index 6b271e3..9997a1e 100644 --- a/src/openharness/tools/cron_create_tool.py +++ b/src/openharness/tools/cron_create_tool.py @@ -2,9 +2,11 @@ from __future__ import annotations +from typing import Any + from pydantic import BaseModel, Field -from openharness.services.cron import upsert_cron_job, validate_cron_expression +from openharness.services.cron import upsert_cron_job, validate_cron_expression, validate_timezone from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult @@ -18,9 +20,25 @@ class CronCreateToolInput(BaseModel): "'0 9 * * 1-5' for weekdays at 9am)" ), ) - command: str = Field(description="Shell command to run when triggered") + command: str | None = Field(default=None, description="Shell command to run when triggered") + message: str | None = Field(default=None, description="Instruction for an agent_turn cron job") + timezone: str | None = Field(default=None, description="IANA timezone for interpreting cron schedule") cwd: str | None = Field(default=None, description="Optional working directory override") enabled: bool = Field(default=True, description="Whether the job is active") + payload: dict[str, Any] | None = Field( + default=None, + description=( + "Optional nanobot-style payload. Example: " + "{'kind': 'agent_turn', 'message': 'check GitHub', 'deliver': True, 'channel': 'feishu', 'to': 'ou_xxx'}." + ), + ) + notify: dict[str, Any] | None = Field( + default=None, + description=( + "Optional notification target. Example: " + "{'type': 'feishu_dm', 'user_open_id': 'ou_xxx'} to send job output to a Feishu private chat." + ), + ) class CronCreateTool(BaseTool): @@ -47,16 +65,40 @@ class CronCreateTool(BaseTool): ), is_error=True, ) + if not validate_timezone(arguments.timezone): + return ToolResult(output=f"Invalid timezone: {arguments.timezone!r}", 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, - } - ) + payload = dict(arguments.payload or {}) + if arguments.message: + payload.setdefault("kind", "agent_turn") + payload.setdefault("message", arguments.message) + if arguments.notify is not None: + payload.setdefault("deliver", True) + if str(arguments.notify.get("type") or "").strip().lower() == "feishu_dm": + payload.setdefault("channel", "feishu") + payload.setdefault("to", arguments.notify.get("user_open_id") or arguments.notify.get("open_id")) + + if payload and not payload.get("message") and not arguments.command: + return ToolResult(output="Cron job requires payload.message, message, or command.", is_error=True) + if not payload and not arguments.command: + return ToolResult(output="Cron job requires command or message.", is_error=True) + + job = { + "name": arguments.name, + "schedule": arguments.schedule, + "cwd": arguments.cwd or str(context.cwd), + "enabled": arguments.enabled, + } + if arguments.timezone: + job["timezone"] = arguments.timezone + if arguments.command is not None: + job["command"] = arguments.command + if payload: + payload.setdefault("kind", "agent_turn") + job["payload"] = payload + if arguments.notify is not None: + job["notify"] = arguments.notify + upsert_cron_job(job) 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 65fc7fa..3056c4e 100644 --- a/src/openharness/tools/cron_list_tool.py +++ b/src/openharness/tools/cron_list_tool.py @@ -47,9 +47,23 @@ class CronListTool(BaseTool): next_run = next_run[:19] last_status = job.get("last_status", "") status_str = f" ({last_status})" if last_status else "" + notify = job.get("notify") + notify_line = "" + if isinstance(notify, dict): + notify_type = notify.get("type", "?") + target = notify.get("user_open_id") or notify.get("open_id") or notify.get("chat_id") or "?" + notify_line = f"\n notify: {notify_type} -> {target}" + timezone = f" ({job['timezone']})" if job.get("timezone") else "" + payload = job.get("payload") + payload_line = "" + if isinstance(payload, dict): + payload_line = f"\n payload: {payload.get('kind', 'agent_turn')} -> {payload.get('channel', '?')}:{payload.get('to', '?')}" + command = job.get("command") or "(agent_turn)" lines.append( - f"[{enabled}] {job['name']} {job.get('schedule', '?')}\n" - f" cmd: {job['command']}\n" + f"[{enabled}] {job['name']} {job.get('schedule', '?')}{timezone}\n" + f" cmd: {command}" + f"{payload_line}" + f"{notify_line}\n" f" last: {last_run}{status_str} next: {next_run}" ) return ToolResult(output="\n".join(lines)) diff --git a/src/openharness/tools/remote_trigger_tool.py b/src/openharness/tools/remote_trigger_tool.py index 15fd9d3..22f74a3 100644 --- a/src/openharness/tools/remote_trigger_tool.py +++ b/src/openharness/tools/remote_trigger_tool.py @@ -8,6 +8,7 @@ from pathlib import Path from pydantic import BaseModel, Field from openharness.services.cron import get_cron_job +from openharness.services.cron_scheduler import _command_for_job from openharness.sandbox import SandboxUnavailableError from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult from openharness.utils.shell import create_shell_subprocess @@ -38,8 +39,9 @@ class RemoteTriggerTool(BaseTool): cwd = Path(job.get("cwd") or context.cwd).expanduser() try: + command = _command_for_job(job) process = await create_shell_subprocess( - str(job["command"]), + command, cwd=cwd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, diff --git a/tests/test_services/test_cron.py b/tests/test_services/test_cron.py index cea7db1..8bb4c46 100644 --- a/tests/test_services/test_cron.py +++ b/tests/test_services/test_cron.py @@ -17,6 +17,7 @@ from openharness.services.cron import ( set_job_enabled, upsert_cron_job, validate_cron_expression, + validate_timezone, ) @@ -49,6 +50,15 @@ class TestValidation: nxt = next_run_time("0 * * * *", base) assert nxt == datetime(2026, 1, 1, 1, 0, 0, tzinfo=timezone.utc) + def test_next_run_time_with_timezone_returns_utc(self) -> None: + base = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + nxt = next_run_time("0 18 * * *", base, tz="Asia/Hong_Kong") + assert nxt == datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc) + + def test_validate_timezone(self) -> None: + assert validate_timezone("Asia/Hong_Kong") + assert not validate_timezone("Asia/HongKong") + class TestCRUD: def test_empty_load(self) -> None: @@ -63,6 +73,21 @@ class TestCRUD: assert "next_run" in jobs[0] assert "created_at" in jobs[0] + def test_upsert_preserves_notify_target(self) -> None: + notify = {"type": "feishu_dm", "user_open_id": "ou_test"} + upsert_cron_job({"name": "test-job", "schedule": "*/5 * * * *", "command": "echo hi", "notify": notify}) + job = get_cron_job("test-job") + assert job is not None + assert job["notify"] == notify + + def test_upsert_preserves_agent_turn_payload(self) -> None: + payload = {"kind": "agent_turn", "message": "check GitHub", "deliver": True, "channel": "feishu", "to": "ou_test"} + upsert_cron_job({"name": "test-job", "schedule": "0 18 * * *", "timezone": "Asia/Hong_Kong", "payload": payload}) + job = get_cron_job("test-job") + assert job is not None + assert job["payload"] == payload + assert job["timezone"] == "Asia/Hong_Kong" + 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"}) diff --git a/tests/test_tools/test_core_tools.py b/tests/test_tools/test_core_tools.py index 4986cb4..7c7e060 100644 --- a/tests/test_tools/test_core_tools.py +++ b/tests/test_tools/test_core_tools.py @@ -304,13 +304,19 @@ 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="0 0 * * *", command="printf 'CRON_OK'"), + CronCreateToolInput( + name="nightly", + schedule="0 0 * * *", + command="printf 'CRON_OK'", + notify={"type": "feishu_dm", "user_open_id": "ou_test"}, + ), context, ) assert create_result.is_error is False list_result = await CronListTool().execute(CronListToolInput(), context) assert "nightly" in list_result.output + assert "feishu_dm" in list_result.output trigger_result = await RemoteTriggerTool().execute( RemoteTriggerToolInput(name="nightly"), @@ -324,3 +330,26 @@ async def test_cron_and_remote_trigger_tools(tmp_path: Path, monkeypatch): context, ) assert delete_result.is_error is False + + +@pytest.mark.asyncio +async def test_cron_create_agent_turn_payload(tmp_path: Path, monkeypatch): + monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) + context = ToolExecutionContext(cwd=tmp_path) + + create_result = await CronCreateTool().execute( + CronCreateToolInput( + name="daily-summary", + schedule="0 18 * * *", + timezone="Asia/Hong_Kong", + message="check GitHub", + payload={"deliver": True, "channel": "feishu", "to": "ou_test"}, + ), + context, + ) + assert create_result.is_error is False + + list_result = await CronListTool().execute(CronListToolInput(), context) + assert "daily-summary" in list_result.output + assert "Asia/Hong_Kong" in list_result.output + assert "payload: agent_turn -> feishu:ou_test" in list_result.output