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
|
"""Builds the dynamic per-turn context.
This handles things that change per turn: the current time, today's diary
entries, open tasks. Persona is NOT here; that's a backend configuration
concern (see models/claude_code.py docstring).
"""
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
SENIA_DB = Path("/etc/senia/data/senia.db")
RECENT_FILE = Path("/etc/senia/data/memory/recent.md")
ROLLING_WINDOW_DAYS = 11
def _today_diary_entries() -> list[str]:
if not SENIA_DB.exists():
return []
now = datetime.now(timezone.utc).astimezone()
start = now.replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
conn = sqlite3.connect(SENIA_DB)
try:
return [
f"- {row[0]} - {row[1]}"
for row in conn.execute(
"SELECT ts, content FROM diary_logs WHERE ts >= ? ORDER BY ts",
(start,),
).fetchall()
]
finally:
conn.close()
def _recent_summary() -> str:
if not RECENT_FILE.exists():
return ""
summary = RECENT_FILE.read_text().strip()
return f"## Recent context (last {ROLLING_WINDOW_DAYS} days)\n{summary}"
def build_turn_context(user: str | None = None) -> str:
diary_entries = _today_diary_entries()
diary = (
"\n".join(diary_entries)
if diary_entries
else "(no entries yet today)"
)
recent = _recent_summary()
now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
time_section = f"## Current time\n{now}"
user_section = f"## Speaking with\n{user}" if user else ""
recent_section = _recent_summary()
diary_section = f"## Today's diary so far\n{diary}"
parts = [
time_section,
user_section,
recent_section,
diary_section,
]
return "\n\n".join(p for p in parts if p)
|
