Index

heartbeat

Heartbeat + routines package (Diapason-style ambient automation).

Functions:

run_routine

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

Execute one routine. Returns structured result.

Source code in src/diapason/heartbeat/kinds.py

def run_routine(
    routine: Routine,
    *,
    system: Any = None,
    force: bool = False,
    workspace: str | None = None,
    config: Any = None,
) -> dict[str, Any]:
    """Execute one routine. Returns structured result."""
    if not routine.enabled and not force:
        record_run(
            routine.id,
            success=True,
            result="disabled",
            skipped=True,
            workspace=workspace,
        )
        return {"ok": True, "skipped": True, "reason": "disabled", "content": ""}

    ok_idle, idle_reason = _precheck_idle(routine, force=force)
    if not ok_idle:
        record_run(
            routine.id,
            success=True,
            result=idle_reason,
            skipped=True,
            workspace=workspace,
        )
        return {"ok": True, "skipped": True, "reason": idle_reason, "content": ""}

    quiet = _quiet_from_config(config) and not force
    kind = (routine.kind or "prompt").strip().lower()

    try:
        if kind in ("morning-digest", "morning-brief"):
            content = _run_morning_digest(system=system, speak=False)
        elif kind == "reminder":
            content = str(
                (routine.payload or {}).get("message")
                or (routine.payload or {}).get("prompt")
                or routine.name
                or "Reminder"
            )
        elif kind in ("prompt", "calendar-ping", "calendar_ping"):
            content = _run_prompt(routine, system=system)
        elif kind == "shell":
            content = "shell kind disabled by default (set allow_shell in config)."
            record_run(
                routine.id,
                success=False,
                result=content,
                skipped=True,
                workspace=workspace,
            )
            return {
                "ok": False,
                "skipped": True,
                "reason": "shell_denied",
                "content": content,
            }
        else:
            # Un kind inconnu était enregistré comme un SUCCÈS, sans un mot
            # dans les journaux. Neuf des douze routines de l'utilisateur
            # étaient dans ce cas : elles « réussissaient » chaque jour sans
            # rien faire, et rien ne le lui apprenait. Un échec silencieux qui
            # se déclare réussi est pire qu'un échec bruyant.
            content = (
                f"Type de routine inconnu : « {kind} ». "
                f"Types exécutables : {', '.join(KINDS_EXECUTABLES)}."
            )
            logger.warning("routine %s : %s", routine.id, content)
            record_run(
                routine.id,
                success=False,
                result=content,
                skipped=False,
                workspace=workspace,
            )
            return {
                "ok": False,
                "skipped": False,
                "reason": "unknown_kind",
                "content": content,
            }

        # SILENT short-circuit (calendar ping with nothing due)
        if (content or "").strip().upper() == "SILENT":
            record_run(
                routine.id,
                success=True,
                result="SILENT",
                skipped=True,
                workspace=workspace,
            )
            return {"ok": True, "skipped": True, "reason": "silent", "content": ""}

        deliver = _should_deliver(routine, force=force, config=config)
        if quiet:
            # Still ran; suppress ambient delivery
            record_run(
                routine.id,
                success=True,
                result=f"[quiet] {content}",
                skipped=False,
                workspace=workspace,
            )
            return {
                "ok": True,
                "skipped": False,
                "quiet": True,
                "delivered": False,
                "content": content,
            }

        record_run(
            routine.id,
            success=True,
            result=content,
            skipped=False,
            workspace=workspace,
        )
        return {
            "ok": True,
            "skipped": False,
            "quiet": False,
            "delivered": deliver,
            "content": content,
        }
    except Exception as exc:
        logger.exception("routine %s failed", routine.id)
        record_run(
            routine.id,
            success=False,
            result=str(exc),
            skipped=False,
            workspace=workspace,
        )
        return {"ok": False, "skipped": False, "content": str(exc), "error": str(exc)}

clear_done

clear_done(workspace: Path | str | None = None) -> int

Remove checked items under ## Now. Returns count removed.

Source code in src/diapason/heartbeat/markdown.py

def clear_done(workspace: Path | str | None = None) -> int:
    """Remove checked items under ## Now. Returns count removed."""
    path = ensure_heartbeat_file(workspace)
    lines = path.read_text(encoding="utf-8").splitlines()
    start, end = _section_bounds(lines, "## Now")
    if start < 0:
        return 0
    kept: list[str] = []
    removed = 0
    for i, line in enumerate(lines):
        if start <= i < end:
            m = _TASK_RE.match(line)
            if m and m.group(2).lower() == "x":
                removed += 1
                continue
        kept.append(line)
    path.write_text("\n".join(kept) + "\n", encoding="utf-8")
    return removed

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

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

sync_heartbeat_and_routines

sync_heartbeat_and_routines(
    scheduler: Any,
    config: Any = None,
    *,
    workspace: str | None = None,
) -> dict[str, Any]

Upsert heartbeat:tick + routine:{id} tasks. Returns summary.

Source code in src/diapason/heartbeat/sync.py

def sync_heartbeat_and_routines(
    scheduler: Any,
    config: Any = None,
    *,
    workspace: str | None = None,
) -> dict[str, Any]:
    """Upsert heartbeat:tick + routine:{id} tasks. Returns summary."""
    try:
        from diapason.core.config import load_config

        cfg = config or load_config()
    except Exception:
        cfg = config

    hb = getattr(cfg, "heartbeat", None) if cfg is not None else None
    rt = getattr(cfg, "routines", None) if cfg is not None else None
    ws = workspace or (
        str(getattr(hb, "workspace_dir", "") or "").strip() or None if hb else None
    )

    ensure_heartbeat_file(ws)
    ensure_routines_file(ws)

    summary: dict[str, Any] = {"heartbeat": None, "routines": []}

    hb_enabled = bool(getattr(hb, "enabled", False)) if hb is not None else False
    interval = int(getattr(hb, "interval_seconds", 1800) if hb is not None else 1800)
    summary["heartbeat"] = _upsert_task(
        scheduler,
        task_id=HEARTBEAT_TASK_ID,
        prompt=HEARTBEAT_PROMPT,
        schedule_type="interval",
        schedule_value=str(max(60, interval)),
        metadata={"diapason_kind": "heartbeat"},
        active=hb_enabled,
    )

    routines_enabled = bool(getattr(rt, "enabled", True)) if rt is not None else True
    for routine in load_routines(ws):
        task_id = f"routine:{routine.id}"
        active = routines_enabled and routine.enabled
        _upsert_task(
            scheduler,
            task_id=task_id,
            prompt=f"[ROUTINE:{routine.id}]",
            schedule_type="cron",
            schedule_value=routine.cron,
            metadata={
                "diapason_kind": "routine",
                "routine_id": routine.id,
                "kind": routine.kind,
            },
            active=active,
        )
        summary["routines"].append(
            {"id": routine.id, "active": active, "task_id": task_id}
        )

    logger.info(
        "Synced heartbeat=%s routines=%d",
        summary["heartbeat"],
        len(summary["routines"]),
    )
    return summary