Skip to content

kinds

kinds

Routine kind handlers (morning-digest, reminder, prompt, …).

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