1 files changed, 160 insertions, 0 deletions
diff --git a/scripts/heartbeat.py b/scripts/heartbeat.py
new file mode 100644
index 0000000..ea74648
--- /dev/null
+++ b/scripts/heartbeat.py
@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+"""Variable heartbeat for Sénia.
+
+Runs from cron. Decides whether Sénia should wake up; if so, invokes claude-code
+non-interactively and pipes the output to ntfy. The ntfy delivery is the wrapper's
+job — invisible to Sénia. She just answers; the script delivers.
+"""
+
+import os
+import sqlite3
+import subprocess
+import sys
+import urllib.request
+from email.header import Header
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+# ---------- config ----------
+
+SENIA_DB = Path("/etc/senia/data/senia.db")
+STATE_DIR = Path("/etc/senia/data/heartbeat")
+LAST_WAKE_FILE = STATE_DIR / "last_wake_at"
+WORKSPACE = Path("/workspace")
+
+QUIET_MAX_HOURS = 4
+
+NTFY_URL = os.environ["NTFY_URL"]       # e.g. https://ntfy.example.tld/senia
+NTFY_TOKEN = os.environ["NTFY_TOKEN"]
+
+
+# ---------- helpers ----------
+
+def ongoing_tasks() -> list[dict]:
+    if not SENIA_DB.exists():
+        return []
+    conn = sqlite3.connect(SENIA_DB)
+    conn.row_factory = sqlite3.Row
+    try:
+        return [dict(r) for r in conn.execute(
+            "SELECT * FROM tasks ORDER BY due_at IS NULL, due_at"
+        ).fetchall()]
+    finally:
+        conn.close()
+
+
+def last_wake_at() -> datetime | None:
+    if not LAST_WAKE_FILE.exists():
+        return None
+    try:
+        return datetime.fromisoformat(LAST_WAKE_FILE.read_text().strip())
+    except (ValueError, OSError):
+        return None
+
+
+def record_wake(ts: datetime) -> None:
+    STATE_DIR.mkdir(parents=True, exist_ok=True)
+    LAST_WAKE_FILE.write_text(ts.isoformat())
+
+
+def should_wake(tasks, last_wake, now) -> tuple[bool, str]:
+    if tasks:
+        return True, f"There are {len(tasks)} pending task(s) to consider."
+    if last_wake is None:
+        return True, "First wake-up since the script was set up."
+    elapsed = now - last_wake
+    if elapsed >= timedelta(hours=QUIET_MAX_HOURS):
+        hrs = elapsed.total_seconds() / 3600
+        return True, f"It's been {hrs:.1f} hours since the last check-in."
+    return False, ""
+
+
+# ---------- the wake-up prompt ----------
+
+WAKE_PROMPT_TEMPLATE = """\
+This is an autonomous heartbeat wake-up. There is no user message; you're being given a moment to check in on things.
+
+Current time: {now}
+Time since last wake-up: {since_last}
+Reason this wake-up fired: {reason}
+
+Decide what (if anything) is worth doing in this moment. Briefly explain your decision and any actions you may take.
+"""
+
+
+def build_prompt(now, last_wake, reason) -> str:
+    if last_wake is None:
+        since_last = "never (first wake-up)"
+    else:
+        hrs = (now - last_wake).total_seconds() / 3600
+        since_last = f"{hrs:.1f} hours"
+    return WAKE_PROMPT_TEMPLATE.format(
+        now=now.isoformat(timespec="seconds"),
+        since_last=since_last,
+        reason=reason,
+    )
+
+
+# ---------- claude invocation + ntfy delivery ----------
+
+def wake_senia(prompt: str) -> str | None:
+    """Run claude-code non-interactively. Returns stdout text, or None on error."""
+    try:
+        result = subprocess.run(
+            ["claude", "-p", prompt, "--output-format", "text"],
+            cwd=WORKSPACE,
+            capture_output=True,
+            text=True,
+            timeout=300,
+        )
+    except subprocess.TimeoutExpired:
+        print("claude timed out", file=sys.stderr)
+        return None
+    if result.returncode != 0:
+        print(f"claude exited {result.returncode}\n{result.stderr}", file=sys.stderr)
+        return None
+    return result.stdout.strip()
+
+
+def send_to_ntfy(message: str) -> None:
+    req = urllib.request.Request(
+        NTFY_URL,
+        data=message.encode("utf-8"),
+        method="POST",
+        headers={
+            "Authorization": f"Bearer {NTFY_TOKEN}",
+            "Content-Type": "text/plain; charset=utf-8",
+            "Title": Header("Sénia", "utf-8").encode(),
+        },
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        if resp.status >= 300:
+            print(f"ntfy returned {resp.status}", file=sys.stderr)
+
+
+# ---------- main ----------
+
+def main() -> int:
+    now = datetime.now(timezone.utc).astimezone()
+    tasks = ongoing_tasks()
+    last_wake = last_wake_at()
+
+    wake, reason = should_wake(tasks, last_wake, now)
+    if not wake:
+        return 0
+
+    response = wake_senia(build_prompt(now, last_wake, reason))
+    if response is None:
+        return 1   # don't record wake on failure; we'll retry next tick
+
+    record_wake(now)
+
+    # "(quiet)" is the agreed-upon "nothing worth saying" sentinel.
+    # Anything else gets delivered.
+    if response and response.lower() != "(quiet)":
+        send_to_ntfy(response)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())