feat(ohmo): add agent-turn cron delivery (#247)

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)

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)

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]}")
 
 

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

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

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})"

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))

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,

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"})

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