ilia/Hearth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: Hearth consent-only email warmup tool

Browse Source
This commit is contained in:
Ilia 2026-07-09 12:22:15 -04:00
commit 2e2e64c08b
21 changed files with 1513 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
.gitignore vendored Normal file
View File

@ -0,0 +1,9 @@
__pycache__/
*.pyc
.venv/
venv/
.env
hearth.yml
data/
*.db
.DS_Store

10
Dockerfile Normal file
View File

@ -0,0 +1,10 @@
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY hearth ./hearth
ENV PYTHONUNBUFFERED=1
VOLUME ["/app/data"]
HEALTHCHECK --interval=5m --timeout=10s --start-period=30s \
CMD python -m hearth.main --healthcheck || exit 1
CMD ["python", "-m", "hearth.main", "--loop"]

98
README.md Normal file
View File

@ -0,0 +1,98 @@
# Hearth
A small, **consent-only** email warmup service for `levkine.ca`. It exchanges
real-looking (but genuinely sent/replied) mail between mailboxes you own and
a fixed allow-list of people who've explicitly agreed to help, gradually
building the engagement signals (opens, replies, "not spam" marks) that
improve sender reputation with Gmail/Outlook/etc.
## What this is (and isn't)
- **Is:** a private warmup tool, same idea as commercial products like
Mailwarm/Lemwarm — send/reply traffic only between mailboxes you control
or people who opted in, at a low, gradually-ramping volume.
- **Isn't:** a way to email arbitrary/unknown third parties, or to fake
engagement to deceive spam filters at scale. There is no code path in this
repo that can send to, or read mail from, an address that isn't explicitly
listed in `hearth.yml`.
## Guardrails
1. **Hard allow-list** (`hearth.yml`) — the only source of recipients.
`Config.is_known_email()` does an exact (case-insensitive) match; nothing
is ever derived, scraped, or guessed. `mailer.send()` re-validates this
itself, independent of any caller.
2. **Two participant tiers:**
- `full_duplex` — your own mailboxes. Hearth holds IMAP+SMTP credentials
(via a `HEARTH_PASSWORD_<KEY>` env var, sourced from your secrets
manager — never hardcoded), sends new threads, auto-replies to incoming
pool mail, and un-spams anything a provider misfiled.
- `send_only` — a consenting friend/coworker. Hearth only ever sends
**to** their address from one of your `full_duplex` mailboxes. It never
holds credentials for them and never logs into their account — they
reply or mark-not-spam manually, on their own, whenever they want.
3. **Volume caps + ramp curve** — starts at `HEARTH_DAILY_CAP_START`
messages/day total, steps up by `HEARTH_RAMP_STEP_PER_DAY`/day to a hard
`HEARTH_DAILY_CAP_MAX`, only during `HEARTH_BUSINESS_HOURS_START`–`_END`
(optionally weekdays only), spread with jitter rather than bursted.
4. **Kill switch + dry run** — `HEARTH_ENABLED=false` stops everything
immediately; `--dry-run` logs every intended action without sending or
modifying any mailbox.
5. **No bulk/marketing signatures** — short, plain-text, varied content;
no tracking pixels, no unsubscribe footers, no link-heavy bodies.
## Quick start
```bash
cp hearth.yml.example hearth.yml # edit: your mailboxes + consenting participants
cp env.example .env # edit: HEARTH_PASSWORD_<KEY> per full_duplex mailbox
docker compose up -d --build
docker compose logs -f
```
Dry run locally without Docker:
```bash
python3 -m venv .venv && .venv/bin/pip install -r requirements-dev.txt
HEARTH_CONFIG=./hearth.yml HEARTH_DB_PATH=./data/hearth.db \
HEARTH_PASSWORD_ALERTS=... .venv/bin/python -m hearth.main --once --dry-run
```
Run the test suite (includes the allow-list guardrail tests):
```bash
.venv/bin/python -m pytest -q
```
## Adding / removing a participant
- **Your own mailbox (`full_duplex`):** add an entry to `hearth.yml` with
`tier: full_duplex`, IMAP/SMTP host+port, `username`, and a
`password_env` name; set that env var (or vault key, see the `ansible`
repo's `docs/guides/hearth-deploy.md`) to an app password.
- **A consenting friend (`send_only`):** confirm with them first, then add
`tier: send_only` with just their `email` (and optional `display_name`).
No credentials, ever.
- **Removing someone:** delete their block from `hearth.yml` and restart —
Hearth re-reads the allow-list on every start, and the code has no cache
of "recipients I've seen before" outside of what's in that file.
## Architecture
| Module | Responsibility |
|---|---|
| `hearth/config.py` | Loads + validates `hearth.yml`; the only source of recipients |
| `hearth/settings.py` | Env-based runtime settings (caps, ramp, kill switch, dry-run) |
| `hearth/store.py` | SQLite: daily counters, ramp start date, message/thread history, reply queue |
| `hearth/composer.py` | Message content (template bank, optional Ollama/Open WebUI variety) |
| `hearth/mailer.py` | SMTP send, with its own allow-list re-check |
| `hearth/watcher.py` | IMAP poll for `full_duplex` mailboxes: detect pool mail, queue delayed replies, un-spam |
| `hearth/scheduler.py` | Ramp-curve + business-hours + cap logic for starting new threads |
| `hearth/main.py` | CLI entrypoint (`--loop`, `--once`, `--dry-run`, `--healthcheck`) |
## Deployment
See the `ansible` homelab repo's `docs/guides/hearth-deploy.md` for how this
fits into the rest of the `levkine.ca` mail setup (Vault secrets convention,
target host, monitoring notes). `deploy-hearth.sh` in this repo handles the
actual `rsync` + `docker compose up` once a target host is chosen.

36
deploy-hearth.sh Executable file
View File

@ -0,0 +1,36 @@
#!/usr/bin/env bash
# Deploy Hearth to a target host via rsync + docker compose.
# No default host on purpose — pick/provision an LXC first, see
# docs/guides/hearth-deploy.md in the ansible repo, then:
#
# HEARTH_HOST=10.0.10.xx ./deploy-hearth.sh
#
# Assumes: hearth.yml and .env already exist locally (copied from the
# .example files and filled in) — they are rsync'd but never committed.
set -euo pipefail
: "${HEARTH_HOST:?Set HEARTH_HOST=<target LXC IP> (see docs/guides/hearth-deploy.md)}"
REMOTE_DIR="${HEARTH_REMOTE_DIR:-/opt/hearth}"
REPO="$(cd "$(dirname "$0")" && pwd)"
for required in hearth.yml .env; do
if [[ ! -f "${REPO}/${required}" ]]; then
echo "Missing ${REPO}/${required} — copy ${required}.example and fill it in first." >&2
exit 1
fi
done
echo "=== Hearth -> ${HEARTH_HOST}:${REMOTE_DIR} ==="
ssh -o StrictHostKeyChecking=accept-new "root@${HEARTH_HOST}" "mkdir -p ${REMOTE_DIR}/data"
rsync -az --delete \
--exclude='.git' --exclude='.venv' --exclude='__pycache__' \
--exclude='data/' --exclude='.pytest_cache' \
"${REPO}/" "root@${HEARTH_HOST}:${REMOTE_DIR}/"
ssh "root@${HEARTH_HOST}" "cd ${REMOTE_DIR} && docker compose up -d --build"
echo "=== Status ==="
ssh "root@${HEARTH_HOST}" "cd ${REMOTE_DIR} && docker compose ps && docker compose logs --tail=20 hearth"
echo ""
echo "Kill switch: ssh root@${HEARTH_HOST} \"sed -i 's/^HEARTH_ENABLED=.*/HEARTH_ENABLED=false/' ${REMOTE_DIR}/.env && cd ${REMOTE_DIR} && docker compose up -d\""

15
docker-compose.yml Normal file
View File

@ -0,0 +1,15 @@
# Hearth — consent-only email warmup service
name: hearth
services:
hearth:
build: .
container_name: hearth
env_file:
- path: .env
required: false
volumes:
- ./hearth.yml:/app/hearth.yml:ro
- ./data:/app/data
restart: unless-stopped
# No published ports — Hearth only ever talks outbound (SMTP/IMAP) to the
# participants in hearth.yml. It has no HTTP surface to expose.

44
env.example Normal file
View File

@ -0,0 +1,44 @@
# Hearth environment — copy to .env, fill in real values, never commit .env
#
# Global kill switch. Set to false to stop ALL sending/replying immediately
# without touching the config or code.
HEARTH_ENABLED=true
# Path to the participant allow-list (mounted read-only in docker-compose.yml)
HEARTH_CONFIG=/app/hearth.yml
HEARTH_DB_PATH=/app/data/hearth.db
# Scheduling
HEARTH_TIMEZONE=America/Toronto
HEARTH_POLL_INTERVAL_SECONDS=300
HEARTH_BUSINESS_HOURS_START=8
HEARTH_BUSINESS_HOURS_END=20
HEARTH_ACTIVE_WEEKDAYS_ONLY=true
# Warmup ramp curve: daily send cap starts at HEARTH_DAILY_CAP_START on day 1
# and increases by HEARTH_RAMP_STEP_PER_DAY each day up to HEARTH_DAILY_CAP_MAX.
HEARTH_DAILY_CAP_START=2
HEARTH_DAILY_CAP_MAX=20
HEARTH_RAMP_STEP_PER_DAY=1
# How long to wait before auto-replying to an incoming pool message, to look
# like a human reading their email rather than a bot (minutes).
HEARTH_MIN_REPLY_DELAY_MIN=5
HEARTH_MAX_REPLY_DELAY_MIN=180
# Optional: use your existing Ollama / Open WebUI instance to vary message
# text instead of the static template bank. Leave blank to use templates only.
HEARTH_LLM_URL=
HEARTH_LLM_MODEL=llama3.1
# Optional: daily summary via your existing ntfy instance
NTFY_URL=https://push.levkin.ca
NTFY_PUBLISH_TOKEN=
NTFY_TOPIC=hearth-digest
# --- Per-participant mailbox credentials -----------------------------------
# One HEARTH_PASSWORD_<KEY> per full_duplex participant in hearth.yml
# (KEY uppercased, matching the participant's `key:`). send_only participants
# need no password — Hearth never authenticates as them.
HEARTH_PASSWORD_ALERTS=
HEARTH_PASSWORD_PERSONAL_GMAIL=

48
hearth.yml.example Normal file
View File

@ -0,0 +1,48 @@
# Hearth participant allow-list — copy to hearth.yml and fill in real values.
# NEVER commit hearth.yml with real addresses/creds (it's gitignored).
#
# This file is the ONLY place recipients come from. Hearth has no code path
# that discovers, scrapes, or composes an address that isn't listed here.
#
# tier: full_duplex
# Your own mailboxes. Hearth logs in via IMAP+SMTP, sends new threads,
# and auto-replies to incoming pool mail. Needs imap_host/smtp_host/
# username + a HEARTH_PASSWORD_<KEY> env var (see env.example).
#
# tier: send_only
# A friend/coworker who explicitly agreed to help warm up levkine.ca.
# Hearth only ever sends TO this address from one of your full_duplex
# mailboxes — it never has credentials for this inbox, and never logs
# into it. The participant replies/marks-not-spam manually, whenever
# they feel like it, from their own mail client.
participants:
alerts:
tier: full_duplex
email: alerts@levkine.ca
display_name: "Ilia"
imap_host: mail.levkine.ca
imap_port: 993
smtp_host: mail.levkine.ca
smtp_port: 587
username: alerts@levkine.ca
password_env: HEARTH_PASSWORD_ALERTS
personal_gmail:
tier: full_duplex
email: idobkin@gmail.com
display_name: "Ilia"
imap_host: imap.gmail.com
imap_port: 993
smtp_host: smtp.gmail.com
smtp_port: 587
username: idobkin@gmail.com
password_env: HEARTH_PASSWORD_PERSONAL_GMAIL
# Example consenting third party — replace with a real person who has
# actually agreed to this. Delete this entry if you only want to warm up
# between your own mailboxes for now.
friend_example:
tier: send_only
email: friend@example.com
display_name: "Friend"

3
hearth/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""Hearth — consent-only email warmup for levkine.ca."""
__version__ = "0.1.0"

80
hearth/composer.py Normal file
View File

@ -0,0 +1,80 @@
"""Message content generation.
Default is a small hand-written template bank (short, plain-text, varied —
deliberately the opposite of bulk/marketing copy). If HEARTH_LLM_URL is set
(pointing at your existing Ollama / Open WebUI instance), new-thread openers
and replies are generated there instead for more variety, with the template
bank as a fallback on any error/timeout.
"""
from __future__ import annotations
import random
import httpx
from hearth.settings import settings
OPENERS = [
("Quick check-in", "Hey {name}, just checking in — how's your week going?"),
("Question for you", "Hey {name}, quick one for you when you get a sec — how's everything on your end?"),
("Following up", "Hi {name}, following up from last time — anything new on your side?"),
("Random thought", "Hey {name}, random thought but figured I'd send it your way. Hope you're doing well!"),
("Catching up", "Hi {name} — been a while! How have things been?"),
("Quick note", "Hey {name}, quick note to say hi and see how you're doing."),
]
REPLIES = [
"Thanks for the note — all good here, just been busy. How about you?",
"Good timing, I was just thinking about that. Things are going well on my end!",
"Appreciate you checking in! Been keeping busy but can't complain.",
"All good here, thanks! Let's catch up properly soon.",
"Ha, funny you mention that. Things have been steady, how about yours?",
"Doing well, thanks for asking! Hope things are good with you too.",
]
def _first_name(display_name: str, email: str) -> str:
if display_name:
return display_name.split()[0]
return email.split("@")[0].replace(".", " ").replace("_", " ").title()
def _try_llm(prompt: str) -> str | None:
if not settings.llm_url:
return None
try:
resp = httpx.post(
f"{settings.llm_url.rstrip('/')}/api/generate",
json={"model": settings.llm_model, "prompt": prompt, "stream": False},
timeout=15.0,
)
resp.raise_for_status()
text = (resp.json() or {}).get("response", "").strip()
return text or None
except Exception:
return None
def compose_new_thread(recipient_name: str, recipient_email: str) -> tuple[str, str]:
name = _first_name(recipient_name, recipient_email)
subject, template = random.choice(OPENERS)
llm_body = _try_llm(
"Write one short, casual, friendly email (2-3 sentences, plain text, "
f"no subject line) checking in with a friend named {name}. Keep it natural "
"and low-key, like a real person catching up, not a marketing email."
)
body = llm_body or template.format(name=name)
return subject, body
def compose_reply(original_subject: str, recipient_name: str, recipient_email: str) -> tuple[str, str]:
name = _first_name(recipient_name, recipient_email)
subject = original_subject if original_subject.lower().startswith("re:") else f"Re: {original_subject}"
llm_body = _try_llm(
f"Write one short, casual, friendly reply (2-3 sentences, plain text) to {name}, "
"responding to a check-in email. Keep it natural, like a real person, not a bot."
)
body = llm_body or random.choice(REPLIES)
return subject, body

140
hearth/config.py Normal file
View File

@ -0,0 +1,140 @@
"""Participant allow-list loader.
This module is the single source of truth for "who Hearth is allowed to
email". Every other module must go through `Config.participants` /
`Config.get(key)` to resolve an address — there is no function anywhere in
this codebase that accepts a freeform email string and sends to it.
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
import yaml
VALID_TIERS = {"full_duplex", "send_only"}
class ConfigError(ValueError):
pass
@dataclass(frozen=True)
class Participant:
key: str
tier: str
email: str
display_name: str = ""
imap_host: str | None = None
imap_port: int = 993
smtp_host: str | None = None
smtp_port: int = 587
username: str | None = None
password_env: str | None = None
@property
def can_authenticate(self) -> bool:
"""Only full_duplex participants ever get IMAP/SMTP credentials."""
return self.tier == "full_duplex"
@property
def password(self) -> str:
if not self.can_authenticate:
raise ConfigError(
f"participant '{self.key}' is send_only — Hearth never holds "
"credentials for it and must not authenticate as it"
)
if not self.password_env:
raise ConfigError(f"participant '{self.key}' has no password_env set")
value = os.environ.get(self.password_env, "")
if not value:
raise ConfigError(
f"participant '{self.key}' password_env '{self.password_env}' is unset"
)
return value
@dataclass(frozen=True)
class Config:
participants: dict[str, Participant] = field(default_factory=dict)
def get(self, key: str) -> Participant:
try:
return self.participants[key]
except KeyError as exc:
raise ConfigError(f"unknown participant key '{key}' (not in allow-list)") from exc
def is_known_email(self, email: str) -> str | None:
"""Return the participant key for an email address, or None if it's
not on the allow-list. Case-insensitive exact match only — no
pattern/domain matching, so a lookalike address never matches."""
needle = email.strip().lower()
for key, participant in self.participants.items():
if participant.email.strip().lower() == needle:
return key
return None
def full_duplex_keys(self) -> list[str]:
return [k for k, p in self.participants.items() if p.tier == "full_duplex"]
def all_keys(self) -> list[str]:
return list(self.participants.keys())
def load_config(path: str | Path) -> Config:
path = Path(path)
if not path.is_file():
raise ConfigError(
f"config file not found: {path} (copy hearth.yml.example to hearth.yml)"
)
raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
raw_participants = raw.get("participants") or {}
if not isinstance(raw_participants, dict) or not raw_participants:
raise ConfigError("hearth.yml must define at least one participant under 'participants:'")
participants: dict[str, Participant] = {}
seen_emails: dict[str, str] = {}
for key, meta in raw_participants.items():
key = str(key).strip()
if not isinstance(meta, dict):
raise ConfigError(f"participant '{key}' must be a mapping")
tier = str(meta.get("tier", "")).strip()
if tier not in VALID_TIERS:
raise ConfigError(f"participant '{key}' has invalid tier '{tier}' (expected {VALID_TIERS})")
email = str(meta.get("email", "")).strip().lower()
if not email or "@" not in email:
raise ConfigError(f"participant '{key}' has a missing/invalid email")
if email in seen_emails:
raise ConfigError(
f"duplicate email '{email}' used by both '{seen_emails[email]}' and '{key}'"
)
seen_emails[email] = key
if tier == "full_duplex":
for required in ("imap_host", "smtp_host", "username", "password_env"):
if not meta.get(required):
raise ConfigError(f"full_duplex participant '{key}' missing '{required}'")
participants[key] = Participant(
key=key,
tier=tier,
email=email,
display_name=str(meta.get("display_name", "")).strip(),
imap_host=meta.get("imap_host"),
imap_port=int(meta.get("imap_port", 993)),
smtp_host=meta.get("smtp_host"),
smtp_port=int(meta.get("smtp_port", 587)),
username=meta.get("username"),
password_env=meta.get("password_env"),
)
if not any(p.tier == "full_duplex" for p in participants.values()):
raise ConfigError("hearth.yml needs at least one full_duplex participant to send from")
return Config(participants=participants)

97
hearth/mailer.py Normal file
View File

@ -0,0 +1,97 @@
"""SMTP sending — the only place that talks to an SMTP server.
Defense in depth: `send` re-validates the recipient against the allow-list
itself (in addition to callers already having done so), so a bug elsewhere
in the codebase can never result in mail going to an address that isn't in
hearth.yml.
"""
from __future__ import annotations
import logging
import smtplib
import uuid
from dataclasses import dataclass
from email.message import EmailMessage
from email.utils import make_msgid
from hearth.config import Config, ConfigError, Participant
from hearth.settings import settings
log = logging.getLogger("hearth.mailer")
class SendBlocked(RuntimeError):
pass
@dataclass
class SendResult:
message_id: str
dry_run: bool
def _connect(sender: Participant) -> smtplib.SMTP:
if sender.smtp_port == 465:
smtp = smtplib.SMTP_SSL(sender.smtp_host, sender.smtp_port, timeout=20)
else:
smtp = smtplib.SMTP(sender.smtp_host, sender.smtp_port, timeout=20)
smtp.starttls()
smtp.login(sender.username, sender.password)
return smtp
def send(
config: Config,
*,
sender_key: str,
recipient_key: str,
subject: str,
body: str,
in_reply_to: str | None = None,
references: str | None = None,
) -> SendResult:
sender = config.get(sender_key)
recipient = config.get(recipient_key)
if not sender.can_authenticate:
raise SendBlocked(f"sender '{sender_key}' is send_only — Hearth cannot send as it")
# Belt-and-suspenders: confirm the recipient's email is still on the
# allow-list right now, by exact match, not just by key lookup above.
if config.is_known_email(recipient.email) != recipient_key:
raise SendBlocked(f"recipient email for '{recipient_key}' does not match allow-list")
message_id = make_msgid(domain=sender.email.split("@", 1)[-1])
if settings.dry_run or not settings.enabled:
log.info(
"[dry-run] would send %s -> %s subject=%r message_id=%s",
sender.email,
recipient.email,
subject,
message_id,
)
return SendResult(message_id=message_id, dry_run=True)
msg = EmailMessage()
msg["From"] = f"{sender.display_name} <{sender.email}>" if sender.display_name else sender.email
msg["To"] = recipient.email
msg["Subject"] = subject
msg["Message-ID"] = message_id
if in_reply_to:
msg["In-Reply-To"] = in_reply_to
if references:
msg["References"] = references
msg.set_content(body)
smtp = _connect(sender)
try:
smtp.send_message(msg)
finally:
try:
smtp.quit()
except Exception:
pass
log.info("sent %s -> %s subject=%r message_id=%s", sender.email, recipient.email, subject, message_id)
return SendResult(message_id=message_id, dry_run=False)

120
hearth/main.py Normal file
View File

@ -0,0 +1,120 @@
from __future__ import annotations
import argparse
import logging
import sys
import time
from pathlib import Path
from hearth import scheduler, watcher
from hearth.config import ConfigError, load_config
from hearth.settings import settings
from hearth.store import Store
log = logging.getLogger("hearth.main")
HEALTHCHECK_MAX_AGE_SECONDS = 15 * 60
def _heartbeat_path() -> Path:
return Path(settings.db_path).with_name(".heartbeat")
def _ntfy_daily_summary(store: Store) -> None:
if not settings.ntfy_publish_token:
return
summary = store.daily_summary()
try:
import httpx
httpx.post(
f"{settings.ntfy_url.rstrip('/')}/{settings.ntfy_topic}",
headers={
"Authorization": f"Bearer {settings.ntfy_publish_token}",
"Title": "Hearth daily summary",
"Priority": "3",
},
content=(
f"sent={summary['sent']} replied={summary['replied']} "
f"unspammed={summary['unspammed']} bounced={summary['bounced']}"
),
timeout=10,
)
except Exception:
log.exception("ntfy summary failed")
def run_once(store: Store, config) -> None:
watcher.poll_all(config, store)
replied = watcher.send_due_replies(config, store)
started = scheduler.maybe_start_new_thread(config, store)
if replied or started:
log.info("cycle done: replies_sent=%d new_thread=%s", replied, started)
heartbeat = _heartbeat_path()
heartbeat.parent.mkdir(parents=True, exist_ok=True)
heartbeat.write_text(str(time.time()))
def run_loop(store: Store, config) -> None:
last_summary_day = None
while True:
try:
run_once(store, config)
today = __import__("datetime").date.today()
if today != last_summary_day:
_ntfy_daily_summary(store)
last_summary_day = today
except Exception:
log.exception("cycle failed, will retry next interval")
time.sleep(settings.poll_interval_seconds)
def _healthcheck() -> int:
heartbeat = _heartbeat_path()
if not heartbeat.is_file():
return 1
age = time.time() - heartbeat.stat().st_mtime
return 0 if age < HEALTHCHECK_MAX_AGE_SECONDS else 1
def main(argv: list[str] | None = None) -> int:
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
parser = argparse.ArgumentParser(description="Hearth — consent-only email warmup")
parser.add_argument("--loop", action="store_true", help="run forever, polling every HEARTH_POLL_INTERVAL_SECONDS")
parser.add_argument("--once", action="store_true", help="run a single cycle and exit")
parser.add_argument("--dry-run", action="store_true", help="log intended sends without actually sending")
parser.add_argument("--healthcheck", action="store_true", help="exit 0 if the loop is healthy, else 1")
args = parser.parse_args(argv)
if args.healthcheck:
return _healthcheck()
if args.dry_run:
settings.dry_run = True
log.info("dry-run mode enabled via --dry-run")
try:
config = load_config(settings.config_path)
except ConfigError as exc:
log.error("config error: %s", exc)
return 1
store = Store(db_path=settings.db_path)
log.info(
"Hearth starting: %d participant(s) (%d full_duplex), enabled=%s dry_run=%s",
len(config.all_keys()),
len(config.full_duplex_keys()),
settings.enabled,
settings.dry_run,
)
if args.loop:
run_loop(store, config)
else:
run_once(store, config)
return 0
if __name__ == "__main__":
sys.exit(main())

119
hearth/scheduler.py Normal file
View File

@ -0,0 +1,119 @@
"""Decides *whether* to start a new warmup thread this cycle, respecting the
ramp curve, daily cap, and business-hours window. Reply timing/sending is
handled separately in watcher.send_due_replies (already delay-randomized).
"""
from __future__ import annotations
import logging
import random
from datetime import datetime
from zoneinfo import ZoneInfo
from hearth import mailer
from hearth.composer import compose_new_thread
from hearth.config import Config
from hearth.settings import settings
from hearth.store import Store
log = logging.getLogger("hearth.scheduler")
def daily_cap(store: Store) -> int:
day = store.ramp_day()
cap = settings.daily_cap_start + settings.ramp_step_per_day * day
return max(1, int(round(min(cap, settings.daily_cap_max))))
def _now() -> datetime:
try:
return datetime.now(ZoneInfo(settings.timezone))
except Exception:
return datetime.now()
def in_business_hours(now: datetime) -> bool:
if settings.active_weekdays_only and now.weekday() >= 5:
return False
return settings.business_hours_start <= now.hour < settings.business_hours_end
def _minutes_left_in_business_day(now: datetime) -> float:
end = now.replace(hour=settings.business_hours_end, minute=0, second=0, microsecond=0)
if now >= end:
return 0.0
return (end - now).total_seconds() / 60.0
def maybe_start_new_thread(config: Config, store: Store) -> bool:
if not settings.enabled:
log.debug("HEARTH_ENABLED=false — skipping send cycle")
return False
now = _now()
if not in_business_hours(now):
return False
cap = daily_cap(store)
sent_today = store.sent_today_total()
remaining = cap - sent_today
if remaining <= 0:
return False
minutes_left = _minutes_left_in_business_day(now)
poll_minutes = max(settings.poll_interval_seconds / 60.0, 1.0)
if minutes_left <= 0:
return False
# Spread the remaining quota roughly evenly across the remaining polls
# in today's business window, instead of bursting.
probability = min(1.0, (poll_minutes / minutes_left) * remaining)
if random.random() > probability:
return False
eligible_senders = [k for k in config.full_duplex_keys() if store.sent_today(k) < cap]
if not eligible_senders:
return False
sender_key = random.choice(eligible_senders)
candidates = [k for k in config.all_keys() if k != sender_key]
if not candidates:
return False
recipient_key = random.choice(candidates)
recipient = config.get(recipient_key)
subject, body = compose_new_thread(recipient.display_name, recipient.email)
try:
result = mailer.send(
config,
sender_key=sender_key,
recipient_key=recipient_key,
subject=subject,
body=body,
)
except Exception:
log.exception("failed to send new thread %s -> %s", sender_key, recipient_key)
return False
thread_id = store.thread_id_for(None)
store.record_message(
thread_id=thread_id,
from_key=sender_key,
to_key=recipient_key,
direction="sent",
message_id_header=result.message_id,
in_reply_to=None,
subject=subject,
)
if not result.dry_run:
store.increment_sent_today(sender_key)
log.info(
"new thread day=%d cap=%d sent_today=%d -> %s to %s",
store.ramp_day(),
cap,
sent_today + 1,
sender_key,
recipient_key,
)
return True

43
hearth/settings.py Normal file
View File

@ -0,0 +1,43 @@
"""Global runtime settings, sourced from environment variables only.
Nothing in here can name a recipient — that only ever comes from the
participant allow-list loaded in config.py.
"""
from __future__ import annotations
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(extra="ignore")
enabled: bool = Field(default=True, validation_alias="HEARTH_ENABLED")
config_path: str = Field(default="./hearth.yml", validation_alias="HEARTH_CONFIG")
db_path: str = Field(default="./data/hearth.db", validation_alias="HEARTH_DB_PATH")
timezone: str = Field(default="America/Toronto", validation_alias="HEARTH_TIMEZONE")
poll_interval_seconds: int = Field(default=300, validation_alias="HEARTH_POLL_INTERVAL_SECONDS")
business_hours_start: int = Field(default=8, validation_alias="HEARTH_BUSINESS_HOURS_START")
business_hours_end: int = Field(default=20, validation_alias="HEARTH_BUSINESS_HOURS_END")
active_weekdays_only: bool = Field(default=True, validation_alias="HEARTH_ACTIVE_WEEKDAYS_ONLY")
daily_cap_start: int = Field(default=2, validation_alias="HEARTH_DAILY_CAP_START")
daily_cap_max: int = Field(default=20, validation_alias="HEARTH_DAILY_CAP_MAX")
ramp_step_per_day: float = Field(default=1.0, validation_alias="HEARTH_RAMP_STEP_PER_DAY")
min_reply_delay_min: int = Field(default=5, validation_alias="HEARTH_MIN_REPLY_DELAY_MIN")
max_reply_delay_min: int = Field(default=180, validation_alias="HEARTH_MAX_REPLY_DELAY_MIN")
llm_url: str = Field(default="", validation_alias="HEARTH_LLM_URL")
llm_model: str = Field(default="llama3.1", validation_alias="HEARTH_LLM_MODEL")
ntfy_url: str = Field(default="https://push.levkin.ca", validation_alias="NTFY_URL")
ntfy_token: str = Field(default="", validation_alias="NTFY_TOPIC_TOKEN")
ntfy_publish_token: str = Field(default="", validation_alias="NTFY_PUBLISH_TOKEN")
ntfy_topic: str = Field(default="hearth-digest", validation_alias="NTFY_TOPIC")
dry_run: bool = Field(default=False, validation_alias="HEARTH_DRY_RUN")
settings = Settings()

265
hearth/store.py Normal file
View File

@ -0,0 +1,265 @@
"""SQLite-backed state: daily send counters, ramp start date, thread/message
history, and observed bounce/spam-folder events. No participant data is
duplicated here beyond the `key` values already validated by config.py."""
from __future__ import annotations
import sqlite3
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import date, datetime, timezone
from pathlib import Path
SCHEMA = """
CREATE TABLE IF NOT EXISTS state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS daily_counts (
day TEXT NOT NULL,
sender_key TEXT NOT NULL,
sent_count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (day, sender_key)
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
thread_id TEXT NOT NULL,
from_key TEXT NOT NULL,
to_key TEXT NOT NULL,
direction TEXT NOT NULL CHECK (direction IN ('sent', 'received')),
message_id_header TEXT,
in_reply_to TEXT,
subject TEXT,
status TEXT NOT NULL DEFAULT 'sent',
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id);
CREATE INDEX IF NOT EXISTS idx_messages_message_id ON messages(message_id_header);
CREATE TABLE IF NOT EXISTS reply_queue (
message_id_header TEXT PRIMARY KEY,
thread_id TEXT NOT NULL,
mailbox_key TEXT NOT NULL,
sender_key TEXT NOT NULL,
subject TEXT NOT NULL,
references_header TEXT,
scheduled_at TEXT NOT NULL,
created_at TEXT NOT NULL
);
"""
def _now() -> str:
return datetime.now(timezone.utc).isoformat()
@dataclass
class Store:
db_path: str
def __post_init__(self) -> None:
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
with self._conn() as conn:
conn.executescript(SCHEMA)
@contextmanager
def _conn(self):
conn = sqlite3.connect(self.db_path)
try:
yield conn
conn.commit()
finally:
conn.close()
# -- ramp -----------------------------------------------------------
def get_or_set_start_date(self) -> date:
with self._conn() as conn:
row = conn.execute("SELECT value FROM state WHERE key = 'start_date'").fetchone()
if row:
return date.fromisoformat(row[0])
today = date.today()
conn.execute(
"INSERT INTO state (key, value) VALUES ('start_date', ?)", (today.isoformat(),)
)
return today
def ramp_day(self) -> int:
start = self.get_or_set_start_date()
return max(0, (date.today() - start).days)
# -- daily caps -------------------------------------------------------
def sent_today(self, sender_key: str) -> int:
today = date.today().isoformat()
with self._conn() as conn:
row = conn.execute(
"SELECT sent_count FROM daily_counts WHERE day = ? AND sender_key = ?",
(today, sender_key),
).fetchone()
return row[0] if row else 0
def sent_today_total(self) -> int:
today = date.today().isoformat()
with self._conn() as conn:
row = conn.execute(
"SELECT COALESCE(SUM(sent_count), 0) FROM daily_counts WHERE day = ?",
(today,),
).fetchone()
return row[0] if row else 0
def increment_sent_today(self, sender_key: str) -> None:
today = date.today().isoformat()
with self._conn() as conn:
conn.execute(
"""
INSERT INTO daily_counts (day, sender_key, sent_count)
VALUES (?, ?, 1)
ON CONFLICT(day, sender_key) DO UPDATE SET sent_count = sent_count + 1
""",
(today, sender_key),
)
# -- messages ---------------------------------------------------------
def record_message(
self,
*,
thread_id: str,
from_key: str,
to_key: str,
direction: str,
message_id_header: str | None,
in_reply_to: str | None,
subject: str,
status: str = "sent",
) -> None:
with self._conn() as conn:
conn.execute(
"""
INSERT INTO messages
(thread_id, from_key, to_key, direction, message_id_header,
in_reply_to, subject, status, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
thread_id,
from_key,
to_key,
direction,
message_id_header,
in_reply_to,
subject,
status,
_now(),
),
)
def mark_status(self, message_id_header: str, status: str) -> None:
with self._conn() as conn:
conn.execute(
"UPDATE messages SET status = ? WHERE message_id_header = ?",
(status, message_id_header),
)
def already_seen(self, message_id_header: str) -> bool:
with self._conn() as conn:
row = conn.execute(
"SELECT 1 FROM messages WHERE message_id_header = ? LIMIT 1",
(message_id_header,),
).fetchone()
return row is not None
def already_replied(self, in_reply_to: str) -> bool:
with self._conn() as conn:
row = conn.execute(
"SELECT 1 FROM messages WHERE in_reply_to = ? AND direction = 'sent' LIMIT 1",
(in_reply_to,),
).fetchone()
return row is not None
# -- reply queue --------------------------------------------------------
def thread_id_for(self, in_reply_to: str | None) -> str:
"""Resolve the thread a message belongs to, or start a new one."""
if in_reply_to:
with self._conn() as conn:
row = conn.execute(
"SELECT thread_id FROM messages WHERE message_id_header = ? LIMIT 1",
(in_reply_to,),
).fetchone()
if row:
return row[0]
return str(__import__("uuid").uuid4())
def queue_reply(
self,
*,
message_id_header: str,
thread_id: str,
mailbox_key: str,
sender_key: str,
subject: str,
references_header: str | None,
scheduled_at: str,
) -> None:
with self._conn() as conn:
conn.execute(
"""
INSERT OR IGNORE INTO reply_queue
(message_id_header, thread_id, mailbox_key, sender_key,
subject, references_header, scheduled_at, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
message_id_header,
thread_id,
mailbox_key,
sender_key,
subject,
references_header,
scheduled_at,
_now(),
),
)
def due_replies(self) -> list[sqlite3.Row]:
now = _now()
with self._conn() as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT * FROM reply_queue WHERE scheduled_at <= ? ORDER BY scheduled_at",
(now,),
).fetchall()
return rows
def clear_queued_reply(self, message_id_header: str) -> None:
with self._conn() as conn:
conn.execute("DELETE FROM reply_queue WHERE message_id_header = ?", (message_id_header,))
def is_reply_queued(self, message_id_header: str) -> bool:
with self._conn() as conn:
row = conn.execute(
"SELECT 1 FROM reply_queue WHERE message_id_header = ? LIMIT 1",
(message_id_header,),
).fetchone()
return row is not None
def daily_summary(self) -> dict:
today = date.today().isoformat()
with self._conn() as conn:
sent = conn.execute(
"SELECT COUNT(*) FROM messages WHERE direction = 'sent' AND created_at LIKE ?",
(f"{today}%",),
).fetchone()[0]
replied = conn.execute(
"SELECT COUNT(*) FROM messages WHERE direction = 'sent' AND in_reply_to IS NOT NULL AND created_at LIKE ?",
(f"{today}%",),
).fetchone()[0]
unspammed = conn.execute(
"SELECT COUNT(*) FROM messages WHERE status = 'unspammed' AND created_at LIKE ?",
(f"{today}%",),
).fetchone()[0]
bounced = conn.execute(
"SELECT COUNT(*) FROM messages WHERE status = 'bounced' AND created_at LIKE ?",
(f"{today}%",),
).fetchone()[0]
return {"sent": sent, "replied": replied, "unspammed": unspammed, "bounced": bounced}

190
hearth/watcher.py Normal file
View File

@ -0,0 +1,190 @@
"""IMAP polling for full_duplex mailboxes.
Only ever looks at mail from senders already on the allow-list (matched by
exact email address via `Config.is_known_email`). Anything from an address
that isn't a configured participant is left completely untouched — Hearth
does not read, move, or react to a stranger's email.
"""
from __future__ import annotations
import imaplib
import logging
import random
from datetime import datetime, timedelta, timezone
from email import message_from_bytes
from email.utils import parseaddr
from hearth import mailer
from hearth.composer import compose_reply
from hearth.config import Config
from hearth.settings import settings
from hearth.store import Store
log = logging.getLogger("hearth.watcher")
SPAM_FOLDER_CANDIDATES = ["[Gmail]/Spam", "Spam", "Junk", "INBOX.Junk", "INBOX.Spam"]
def _decode(value) -> str:
if value is None:
return ""
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
return str(value)
def _random_reply_delay() -> timedelta:
minutes = random.randint(settings.min_reply_delay_min, settings.max_reply_delay_min)
return timedelta(minutes=minutes)
def _open_imap(participant) -> imaplib.IMAP4:
imap = imaplib.IMAP4_SSL(participant.imap_host, participant.imap_port)
imap.login(participant.username, participant.password)
return imap
def _fetch_unseen(imap: imaplib.IMAP4, folder: str) -> list[bytes]:
typ, _ = imap.select(folder, readonly=False)
if typ != "OK":
return []
typ, data = imap.search(None, "UNSEEN")
if typ != "OK" or not data or not data[0]:
return []
return data[0].split()
def _handle_message(config: Config, store: Store, mailbox_key: str, raw: bytes, *, from_spam: bool) -> None:
msg = message_from_bytes(raw)
_, from_addr = parseaddr(_decode(msg.get("From")))
sender_key = config.is_known_email(from_addr)
if not sender_key:
return # not a pool participant — never touched
if sender_key == mailbox_key:
return # our own mailbox talking to itself, ignore
message_id = _decode(msg.get("Message-ID")).strip()
if not message_id:
return
if store.already_seen(message_id) or store.is_reply_queued(message_id):
return
in_reply_to = _decode(msg.get("In-Reply-To")).strip() or None
subject = _decode(msg.get("Subject")).strip() or "(no subject)"
references = _decode(msg.get("References")).strip() or in_reply_to
thread_id = store.thread_id_for(in_reply_to)
store.record_message(
thread_id=thread_id,
from_key=sender_key,
to_key=mailbox_key,
direction="received",
message_id_header=message_id,
in_reply_to=in_reply_to,
subject=subject,
status="unspammed" if from_spam else "received",
)
if store.already_replied(message_id):
return
scheduled_at = (datetime.now(timezone.utc) + _random_reply_delay()).isoformat()
store.queue_reply(
message_id_header=message_id,
thread_id=thread_id,
mailbox_key=mailbox_key,
sender_key=sender_key,
subject=subject,
references_header=references,
scheduled_at=scheduled_at,
)
log.info(
"queued reply: %s -> %s (subject=%r, in %s)",
mailbox_key,
sender_key,
subject,
"spam->inbox" if from_spam else "inbox",
)
def poll_mailbox(config: Config, store: Store, mailbox_key: str) -> None:
participant = config.get(mailbox_key)
if not participant.can_authenticate:
return
try:
imap = _open_imap(participant)
except Exception:
log.exception("IMAP connect failed for %s", mailbox_key)
return
try:
for uid in _fetch_unseen(imap, "INBOX"):
typ, data = imap.fetch(uid, "(RFC822)")
if typ == "OK" and data and data[0]:
_handle_message(config, store, mailbox_key, data[0][1], from_spam=False)
for folder in SPAM_FOLDER_CANDIDATES:
try:
uids = _fetch_unseen(imap, folder)
except Exception:
continue
for uid in uids:
typ, data = imap.fetch(uid, "(RFC822)")
if typ != "OK" or not data or not data[0]:
continue
raw = data[0][1]
_handle_message(config, store, mailbox_key, raw, from_spam=True)
if not settings.dry_run and settings.enabled:
try:
imap.copy(uid, "INBOX")
imap.store(uid, "+FLAGS", "\\Deleted")
except Exception:
log.exception("failed to move message out of %s for %s", folder, mailbox_key)
if uids:
imap.expunge()
finally:
try:
imap.close()
imap.logout()
except Exception:
pass
def poll_all(config: Config, store: Store) -> None:
for mailbox_key in config.full_duplex_keys():
poll_mailbox(config, store, mailbox_key)
def send_due_replies(config: Config, store: Store) -> int:
sent = 0
for row in store.due_replies():
mailbox = config.get(row["mailbox_key"])
sender = config.get(row["sender_key"])
subject, body = compose_reply(row["subject"], sender.display_name, sender.email)
try:
result = mailer.send(
config,
sender_key=row["mailbox_key"],
recipient_key=row["sender_key"],
subject=subject,
body=body,
in_reply_to=row["message_id_header"],
references=row["references_header"],
)
except Exception:
log.exception("failed to send queued reply %s -> %s", mailbox.email, sender.email)
continue
store.record_message(
thread_id=row["thread_id"],
from_key=row["mailbox_key"],
to_key=row["sender_key"],
direction="sent",
message_id_header=result.message_id,
in_reply_to=row["message_id_header"],
subject=subject,
)
store.clear_queued_reply(row["message_id_header"])
sent += 1
return sent

2
requirements-dev.txt Normal file
View File

@ -0,0 +1,2 @@
-r requirements.txt
pytest==8.3.4

5
requirements.txt Normal file
View File

@ -0,0 +1,5 @@
pydantic==2.10.3
pydantic-settings==2.6.1
PyYAML==6.0.2
httpx==0.28.1
tzdata==2024.2

0
tests/__init__.py Normal file
View File

102
tests/test_config.py Normal file
View File

@ -0,0 +1,102 @@
import textwrap
import pytest
from hearth.config import ConfigError, load_config
VALID_YAML = textwrap.dedent(
"""
participants:
alerts:
tier: full_duplex
email: alerts@levkine.ca
display_name: "Ilia"
imap_host: mail.levkine.ca
smtp_host: mail.levkine.ca
username: alerts@levkine.ca
password_env: TEST_PASSWORD_ALERTS
friend:
tier: send_only
email: friend@example.com
display_name: "Friend"
"""
)
def write_yaml(tmp_path, content):
path = tmp_path / "hearth.yml"
path.write_text(content)
return path
def test_load_valid_config(tmp_path):
path = write_yaml(tmp_path, VALID_YAML)
config = load_config(path)
assert set(config.all_keys()) == {"alerts", "friend"}
assert config.full_duplex_keys() == ["alerts"]
assert config.is_known_email("ALERTS@levkine.ca") == "alerts"
assert config.is_known_email("nobody@nowhere.com") is None
def test_missing_file_raises(tmp_path):
with pytest.raises(ConfigError):
load_config(tmp_path / "does-not-exist.yml")
def test_duplicate_email_rejected(tmp_path):
content = textwrap.dedent(
"""
participants:
alerts:
tier: full_duplex
email: alerts@levkine.ca
imap_host: mail.levkine.ca
smtp_host: mail.levkine.ca
username: alerts@levkine.ca
password_env: TEST_PASSWORD_ALERTS
duplicate:
tier: send_only
email: alerts@levkine.ca
display_name: "Dup"
"""
)
path = write_yaml(tmp_path, content)
with pytest.raises(ConfigError):
load_config(path)
def test_full_duplex_requires_creds_fields(tmp_path):
content = textwrap.dedent(
"""
participants:
broken:
tier: full_duplex
email: broken@levkine.ca
"""
)
path = write_yaml(tmp_path, content)
with pytest.raises(ConfigError):
load_config(path)
def test_requires_at_least_one_full_duplex(tmp_path):
content = textwrap.dedent(
"""
participants:
friend:
tier: send_only
email: friend@example.com
"""
)
path = write_yaml(tmp_path, content)
with pytest.raises(ConfigError):
load_config(path)
def test_send_only_participant_cannot_authenticate(tmp_path):
path = write_yaml(tmp_path, VALID_YAML)
config = load_config(path)
friend = config.get("friend")
assert friend.can_authenticate is False
with pytest.raises(ConfigError):
_ = friend.password

87
tests/test_mailer_guard.py Normal file
View File

@ -0,0 +1,87 @@
import os
import textwrap
import pytest
from hearth import mailer
from hearth.config import ConfigError, load_config
from hearth.settings import settings
YAML = textwrap.dedent(
"""
participants:
alerts:
tier: full_duplex
email: alerts@levkine.ca
imap_host: mail.levkine.ca
smtp_host: mail.levkine.ca
username: alerts@levkine.ca
password_env: TEST_PASSWORD_ALERTS
gmail:
tier: full_duplex
email: idobkin@gmail.com
imap_host: imap.gmail.com
smtp_host: smtp.gmail.com
username: idobkin@gmail.com
password_env: TEST_PASSWORD_GMAIL
friend:
tier: send_only
email: friend@example.com
"""
)
@pytest.fixture(autouse=True)
def _force_dry_run():
original = settings.dry_run
settings.dry_run = True
yield
settings.dry_run = original
@pytest.fixture
def config(tmp_path):
os.environ["TEST_PASSWORD_ALERTS"] = "x"
os.environ["TEST_PASSWORD_GMAIL"] = "x"
path = tmp_path / "hearth.yml"
path.write_text(YAML)
return load_config(path)
def test_send_to_known_participant_succeeds_dry_run(config):
result = mailer.send(
config,
sender_key="alerts",
recipient_key="friend",
subject="hi",
body="hello",
)
assert result.dry_run is True
def test_cannot_send_to_unknown_recipient_key(config):
with pytest.raises(ConfigError):
mailer.send(
config,
sender_key="alerts",
recipient_key="random_stranger",
subject="hi",
body="hello",
)
def test_cannot_send_as_send_only_participant(config):
with pytest.raises(mailer.SendBlocked):
mailer.send(
config,
sender_key="friend",
recipient_key="alerts",
subject="hi",
body="hello",
)
def test_is_known_email_is_exact_match_only(config):
# A lookalike/subdomain address must never resolve to a real participant.
assert config.is_known_email("alerts@levkine.ca.evil.com") is None
assert config.is_known_email("not-friend@example.com") is None