Skip to content

runner

runner

Heartbeat tick — drain one pending HEARTBEAT.md task.

Functions:

run_heartbeat_tick 

run_heartbeat_tick(
    *,
    system: Any = None,
    force: bool = False,
    workspace: str | None = None,
    config: Any = None,
) -> dict[str, Any]

Process the first pending ## Now item. No-op if empty.

Source code in src/diapason/heartbeat/runner.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def run_heartbeat_tick(
    *,
    system: Any = None,
    force: bool = False,
    workspace: str | None = None,
    config: Any = None,
) -> dict[str, Any]:
    """Process the first pending ## Now item. No-op if empty."""
    try:
        from diapason.core.config import load_config

        cfg = config or load_config()
        hb = getattr(cfg, "heartbeat", None)
        enabled = True if hb is None else bool(getattr(hb, "enabled", True))
        if not enabled and not force:
            return {"ok": True, "skipped": True, "reason": "disabled", "content": ""}
        quiet = in_quiet_hours(
            enabled=bool(getattr(hb, "quiet_hours_enabled", True)) if hb else True,
            start=str(getattr(hb, "quiet_hours_start", "22:00") if hb else "22:00"),
            end=str(getattr(hb, "quiet_hours_end", "07:00") if hb else "07:00"),
        )
        allowlist = []
        if hb is not None:
            raw = str(getattr(hb, "tool_allowlist", "") or "")
            allowlist = [t.strip() for t in raw.split(",") if t.strip()]
    except Exception:
        quiet = False
        allowlist = []

    ws = workspace or _workspace_from_config(config)
    ensure_heartbeat_file(ws)
    pending = pending_now(ws)
    if not pending:
        return {"ok": True, "skipped": True, "reason": "empty", "content": ""}

    task = pending[0]
    content = ""
    try:
        if system is not None:
            tools = allowlist or None
            content = str(
                system.ask(
                    (
                        "You are handling a heartbeat queue item. "
                        "Do the following briefly and safely. "
                        "Do not send email/SMS. Do not invent facts.\n\n"
                        f"TASK: {task.text}"
                    ),
                    agent="simple",
                    tools=tools,
                )
            )
        else:
            content = f"Acknowledged heartbeat task: {task.text}"
    except Exception as exc:
        logger.exception("heartbeat task failed")
        content = f"Heartbeat task failed: {exc}"
        append_changelog(f"FAILED {task.text} — {exc}", ws)
        return {
            "ok": False,
            "skipped": False,
            "task": task.text,
            "content": content,
            "error": str(exc),
        }

    mark_done(task, ws)
    note = content.strip().splitlines()[0][:160] if content.strip() else "done"
    append_changelog(f"DONE {task.text} — {note}", ws)

    return {
        "ok": True,
        "skipped": False,
        "task": task.text,
        "content": content,
        "quiet": quiet and not force,
        "delivered": not (quiet and not force),
    }