1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
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())
|
