The_miro
/
SmartestHome
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
SmartestHome/digest-engine/run.py

507 lines
22 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""digest-engine entrypoint — one digest run, then exit.
Invoked as `docker compose run --rm digest-engine` by the smart-home-digest.timer
systemd timer, 4x/day. It is deliberately a oneshot, not a daemon: scheduling
lives in systemd, exactly like the restic backup job in
hosts/container-host/scripts/setup-container-host.sh.
Every ingestion source is independently toggleable and independently fallible. A
source that is disabled, misconfigured, or simply broken (a stale Telegram
session, a signal-cli container that is down) logs a warning and contributes an
empty list — it must never take down the rest of the run, because a digest with
most of its sections is worth far more than no digest at all.
Everything here is read-only. See docs/project-plan.md Phase 12 step 8.
WHICH DIGESTS GET GENERATED
---------------------------
Not necessarily all four. Each person in `identity` has a per-person set of the
sections they want (network, household, personal/social, political/news), and
this run generates the union of what the household asked for — a section nobody
wants costs no LLM call and never reaches output/. See preferences.py, including
why every failure in that lookup means "generate everything" rather than
"generate nothing".
Before anything is written to output/, every generated document passes through
synth/counter_run.py — a second LLM call that checks the document against the
same context it was generated from and drops anything that doesn't trace back
to it. That's the final filter against hallucinated quotes, figures, or
theoretical connections; see that module's docstring for how it fails safe.
WHICH RUN IS THIS?
------------------
One feature — the evening recipe suggestion in synth/prompts/household.md — only
applies to one of the four daily runs. The slot is derived here from the local
wall clock rather than passed in by the caller: the systemd unit installed by
hosts/container-host/scripts/setup-container-host.sh runs a bare
`docker compose run --rm digest-engine` with no arguments, and a manual run uses
exactly the same command, so anything argument- or unit-based would have to be
threaded through both and would silently do the wrong thing on a hand-run digest.
The container already has the host's timezone (`TZ` plus a bind-mounted
/etc/localtime), which is the same clock systemd's `OnCalendar` fires against.
The run is attributed to the most recent DIGEST_SCHEDULE slot at or before now,
not to an exact hour match, because the timer is `Persistent=true`: a host that
was asleep at 18:00 fires the run late, and an exact match would drop the evening
feature precisely on the days the digest is read late.
"""
import json
import logging
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from ingest import (
caldav as caldav_ingest,
discord_ingest,
email_imap,
financial,
flight_traffic,
grocy,
naval_traffic,
news_rss,
opnsense_ids,
rci_social,
signal_ingest,
telegram_ingest,
whatsapp_ingest,
)
from synth import counter_run, llm_client
import agenda
import archive
import notify
import preferences
import viewed_tracker
LOG = logging.getLogger("digest")
# (env toggle, context key, module) — order is the order they run in.
SOURCES = (
("ENABLE_EMAIL_INGEST", "email", email_imap),
("ENABLE_SIGNAL_INGEST", "signal", signal_ingest),
("ENABLE_TELEGRAM_INGEST", "telegram", telegram_ingest),
("ENABLE_DISCORD_INGEST", "discord", discord_ingest),
("ENABLE_WHATSAPP_INGEST", "whatsapp", whatsapp_ingest),
("ENABLE_NEWS_INGEST", "news", news_rss),
("ENABLE_RCI_SOCIAL_INGEST", "rci_social", rci_social),
("ENABLE_FINANCIAL_INGEST", "financial", financial),
("ENABLE_FLIGHT_TRAFFIC_INGEST", "flight_traffic", flight_traffic),
("ENABLE_NAVAL_TRAFFIC_INGEST", "naval_traffic", naval_traffic),
("ENABLE_OPNSENSE_IDS_INGEST", "opnsense_ids", opnsense_ids),
("ENABLE_CALDAV_INGEST", "calendar", caldav_ingest),
("ENABLE_GROCY_INGEST", "grocy", grocy),
)
# Which ingested sources feed which section. Two jobs: it is what the personal
# context's `messages` map is built from, and it is how collect() knows that a source
# whose only consumers are switched off for this run needs no fetching at all — a
# digest section nobody asked for should cost neither an LLM call nor the WAN round
# trips its sources would have made. `email` appears twice on purpose: mail is both
# personal correspondence and, occasionally, politically relevant evidence.
SECTION_SOURCES = {
"personal": ("email", "signal", "telegram", "discord", "whatsapp"),
# `calendar` is here for the agendas, not for the diary: the political section
# owns the branch agenda's contents and the to-do list that comes out of it (see
# agenda.py), and it cannot match an agenda to a meeting it has never fetched.
"political": ("news", "rci_social", "financial", "email", "flight_traffic",
"naval_traffic", "calendar"),
"household": ("calendar", "grocy"),
"network": ("opnsense_ids",),
}
DEFAULT_SCHEDULE = "00,06,12,18"
DEFAULT_EVENING_HOUR = 18
def env_flag(name, default="false"):
return os.environ.get(name, default).strip().lower() == "true"
def schedule_hours():
hours = sorted(
{int(part.strip()) for part in os.environ.get("DIGEST_SCHEDULE", DEFAULT_SCHEDULE).split(",")
if part.strip().isdigit() and 0 <= int(part.strip()) <= 23}
)
if not hours:
LOG.warning("DIGEST_SCHEDULE is unusable, falling back to %s", DEFAULT_SCHEDULE)
return [int(part) for part in DEFAULT_SCHEDULE.split(",")]
return hours
def slot_hour(now_local, hours):
earlier = [hour for hour in hours if hour <= now_local.hour]
# Before the first slot of the day the run still belongs to yesterday's last one.
return earlier[-1] if earlier else hours[-1]
def resolve_slot():
now_local = datetime.now().astimezone()
hours = schedule_hours()
try:
evening_hour = int(os.environ.get("DIGEST_EVENING_HOUR", DEFAULT_EVENING_HOUR))
except ValueError:
# Unlike an ingestion module, this runs outside collect()'s backstop, and a
# typo in one env var must not take down a whole digest.
LOG.warning("DIGEST_EVENING_HOUR is not a number, falling back to %s", DEFAULT_EVENING_HOUR)
evening_hour = DEFAULT_EVENING_HOUR
if evening_hour not in hours:
LOG.warning(
"DIGEST_EVENING_HOUR=%s is not one of the DIGEST_SCHEDULE slots %s, "
"so no run will ever be the evening one",
evening_hour,
hours,
)
current = slot_hour(now_local, hours)
is_evening = current == evening_hour or env_flag("DIGEST_FORCE_EVENING")
LOG.info("run attributed to the %02d:00 slot (evening run: %s)", current, is_evening)
return current, evening_hour, is_evening
def collect(lookback_hours, sections=None):
"""`sections` is the set of digest sections this run is generating; a source no
enabled section reads is skipped entirely rather than fetched and then dropped.
None means "generate everything", i.e. fetch everything that is switched on.
"""
needed = None
if sections is not None:
needed = {source for section in sections for source in SECTION_SOURCES.get(section, ())}
collected = {}
for toggle, key, module in SOURCES:
if needed is not None and key not in needed:
LOG.info("no enabled digest section reads %s, skipping its ingestion", key)
collected[key] = []
continue
if not env_flag(toggle):
LOG.info("%s is not enabled, skipping %s ingestion", toggle, key)
collected[key] = []
continue
try:
collected[key] = module.fetch(lookback_hours) or []
except Exception:
# Modules already swallow their own failures; this is the backstop for
# anything they miss (an import-time error, an unexpected exception type).
LOG.warning("%s ingestion raised, continuing without it", key, exc_info=True)
collected[key] = []
return collected
def load_previous_run(output_dir):
"""Best-effort read of the run currently pointed at by output/latest — the
candidate for merging if it turns out not to have been viewed. Any failure (first
run ever, corrupt JSON, missing file) is silently "no previous run", which just
means no merge is attempted, never a crash."""
try:
return json.loads((output_dir / "latest" / "digest.json").read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
def should_merge(previous_run, previous_viewed_at):
if previous_run is None:
return False
# None (unknown) is handled identically to "viewed" — see viewed_tracker's
# last_viewed_at() docstring for why treating unknown as unviewed is the riskier
# default, not the safer one.
if previous_viewed_at is None:
return False
return previous_viewed_at < previous_run.get("generated_at", "")
def previous_section_document(previous_run, section):
"""The previous run's own rendered document for one section, at `full` detail —
the richest version, since what gets folded in is content, not layout, and the
merge instruction in synth/llm_client.py asks the model to write fresh prose at
whatever detail_level this pass actually is, not to reuse this verbatim."""
if not previous_run:
return None
for doc in (previous_run.get("sections") or {}).get("full") or []:
if isinstance(doc, dict) and doc.get("section") == section:
return doc
return None
def household_calendar(events):
"""The calendar as the household section sees it: an event that has an agenda says
so, without carrying what the agenda says.
The points and the text go to the political section only (see agenda.upcoming()),
because a branch agenda is party work and reaches the people who asked for that
digest. Enforcing it here rather than only in the prompt means a household-only
run never has the contents in front of it to leak in the first place — a rule the
model cannot break is worth more than one it is told.
"""
trimmed = []
for event in events:
documents = event.get("agenda_documents") if isinstance(event, dict) else None
if not documents:
trimmed.append(event)
continue
copy = dict(event)
copy["agenda_documents"] = [
{
key: document.get(key)
for key in ("filename", "from", "source", "received_at",
"match_confidence", "match_reason", "text_extracted")
}
for document in documents
]
trimmed.append(copy)
return trimmed
def build_section_contexts(collected, lookback_hours, is_evening_run, previous_run=None,
sections=None, section_history=None, unattached_agendas=None,
upcoming_agendas=None):
# The full pantry is only ever needed to work out what the evening recipe still
# requires. On the other three runs it is a few hundred lines of prompt that
# buys nothing, so it is dropped rather than sent and then ignored.
grocy_entries = [
entry
for entry in collected.get("grocy", [])
if is_evening_run or entry.get("category") != "in_stock"
]
contexts = {
"personal": {
"lookback_hours": lookback_hours,
"messages": {key: collected.get(key, []) for key in SECTION_SOURCES["personal"]},
},
"political": {
"lookback_hours": lookback_hours,
"news": collected.get("news", []),
# The organisation's own posts, kept in their own key rather than folded
# into `news`: they are the section's voice, not reporting, and the prompt
# is told to read them differently. See ingest/rci_social.py.
"rci_social": collected.get("rci_social", []),
"financial": collected.get("financial", []),
"mail": collected.get("email", []),
# Extra evidence for the existing political synthesis, not a fourth
# window type — see the traffic-data section of synth/prompts/political.md.
"flight_traffic": collected.get("flight_traffic", []),
"naval_traffic": collected.get("naval_traffic", []),
# What the household is about to sit down and discuss, from the agendas
# that arrived by mail or messenger. Steers curation: a story about the
# subject of Thursday's branch meeting is worth more than one that isn't.
# Topics only — the full agenda text stays in the household section.
"upcoming_agendas": upcoming_agendas or [],
},
# Both keys stay present even when their source is off or broken, so the
# prompt sees the shape it is promised and says "nothing scheduled" instead
# of hallucinating an event.
"household": {
"lookback_hours": lookback_hours,
"calendar": household_calendar(collected.get("calendar", [])),
"grocy": grocy_entries,
# Drives the evening-only recipe/shopping-list section of
# synth/prompts/household.md. Nothing is ever written back to Grocy.
"is_evening_run": is_evening_run,
# Agendas that arrived by mail or message and could not be tied to any
# event — reported rather than dropped, because "an agenda came and I
# can't tell which meeting it's for" is worth a line. Reduced to the same
# fact-level fields as the attached ones: what arrived and from whom, not
# what it says. See household_calendar() above.
"unattached_agendas": [
{
key: document.get(key)
for key in ("filename", "from", "source", "received_at", "text_extracted")
}
for document in (unattached_agendas or [])
],
},
}
# Its own section rather than a couple of lines inside the household one: a
# household member who wants the calendar but not a nightly intrusion-detection
# readout (or the reverse) can only say so if the two are separately generated.
# See synth/prompts/network.md.
contexts["network"] = {
"lookback_hours": lookback_hours,
"network_security": collected.get("opnsense_ids", []),
}
# A section nobody asked for is dropped here, before its context is ever built into
# a prompt — see preferences.py. Done by subtraction from the full set rather than
# by building each context conditionally, so adding a section later can't silently
# forget this filter.
if sections is not None:
contexts = {name: context for name, context in contexts.items() if name in sections}
# Earlier readings from the archive, as their own clearly-labelled block with its
# own timestamps — never merged into this run's entries, so a prompt can't mistake
# last month's figure for something that happened in this window. See archive.py.
for section, block in (section_history or {}).items():
if section in contexts:
contexts[section]["history"] = block
if previous_run is not None:
for section, context in contexts.items():
doc = previous_section_document(previous_run, section)
if doc:
context["previous_unviewed_digest"] = {
"generated_at": doc.get("generated_at"),
"windows": doc.get("windows", []),
"narration": doc.get("narration", ""),
}
return contexts
def write_output(output_dir, run_id, context, documents, sections, people):
run_dir = output_dir / run_id
run_dir.mkdir(parents=True, exist_ok=True)
# Persisted for the Phase 12 step 7 follow-up voice Q&A: a spoken follow-up
# re-queries Ollama against this cached context instead of re-ingesting.
(run_dir / "context.json").write_text(
json.dumps(context, indent=2, ensure_ascii=False, default=str), encoding="utf-8"
)
digest = {
"run_id": run_id,
"generated_at": context["generated_at"],
"sections": documents,
# What this run actually generated, so a surface can tell "nobody asked for
# this section" apart from "it was generated and came back empty".
"sections_generated": sections,
# Who wants which of them, carried alongside the documents so the renderer can
# filter to the person Home Assistant resolved without a second fetch (and
# without digest-web needing to talk to identity at all). Empty when
# preferences could not be read — see preferences.py. THIS IS A DISPLAY
# FILTER, NOT AN ACCESS CONTROL: this whole file is served read-only to the
# LAN, so it hides a section from a screen, not from a person.
"people": people,
}
payload = json.dumps(digest, indent=2, ensure_ascii=False, default=str)
(run_dir / "digest.json").write_text(payload, encoding="utf-8")
# digest-web serves output/ read-only and has no idea which run is newest, so
# the pointer has to live in the served tree itself. latest.json holds the whole
# digest so a template needs exactly one fetch; the `latest` symlink is for
# humans poking around the volume.
(output_dir / "latest.json").write_text(payload, encoding="utf-8")
symlink = output_dir / "latest"
try:
if symlink.is_symlink() or symlink.exists():
symlink.unlink()
symlink.symlink_to(run_id, target_is_directory=True)
except OSError:
LOG.warning("could not update the output/latest symlink", exc_info=True)
return run_dir
def main():
logging.basicConfig(
level=os.environ.get("LOG_LEVEL", "INFO").upper(),
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
)
lookback_hours = float(os.environ.get("DIGEST_LOOKBACK_HOURS", "6"))
output_dir = Path(os.environ.get("DIGEST_OUTPUT_DIR", "/output"))
started_at = datetime.now(timezone.utc)
run_id = started_at.strftime("%Y%m%dT%H%M%SZ")
LOG.info("digest run %s starting (lookback %sh)", run_id, lookback_hours)
current_slot, evening_hour, is_evening_run = resolve_slot()
# Read before ingestion so a section nobody wants costs nothing at all — not an
# LLM call, and not the WAN round trips its sources would have made either.
prefs = preferences.fetch(llm_client.SECTIONS)
sections = preferences.sections_to_generate(prefs, llm_client.SECTIONS)
if not sections:
LOG.warning(
"every household member has opted out of every digest section — "
"generating nothing, which is what was asked for"
)
collected = collect(lookback_hours, sections)
# The long memory: this run goes into the archive, and every item it just
# collected comes back annotated with when this system first saw it. History for
# the sections being generated is read afterwards, so a series includes the
# reading taken a moment ago rather than stopping at the previous run.
archive.record(run_id, collected)
section_history = archive.history(sections)
# A "Tagesordnung" PDF that arrived by mail or WhatsApp belongs to the meeting it
# is for, not to a list of attachments. Done before the contexts are built so the
# calendar entries the household section sees already carry theirs. Only works
# when the message sources are being fetched at all — see SECTION_SOURCES: a run
# generating only the household section deliberately does not go and read the
# household's mail.
unattached_agendas = agenda.attach(collected)
previous_run = load_previous_run(output_dir)
previous_viewed_at = viewed_tracker.last_viewed_at()
merge_previous = should_merge(previous_run, previous_viewed_at)
LOG.info(
"previous run viewed_at=%s, previous run generated_at=%s -> merging: %s",
previous_viewed_at,
previous_run.get("generated_at") if previous_run else None,
merge_previous,
)
context = {
"run_id": run_id,
"generated_at": started_at.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
"lookback_hours": lookback_hours,
"slot_hour": current_slot,
"evening_hour": evening_hour,
"is_evening_run": is_evening_run,
"enabled_sources": [key for toggle, key, _ in SOURCES if env_flag(toggle)],
"item_counts": {key: len(items) for key, items in collected.items()},
"sources": collected,
"merged_unviewed_previous_run": merge_previous,
# Kept in the persisted context (and so in the Phase 12 follow-up voice Q&A's
# view of this run) because "why is there no political section today" is
# answered by this and by nothing else in the bundle.
"sections_generated": sections,
"digest_preferences_known": prefs is not None,
}
section_contexts = build_section_contexts(
collected, lookback_hours, is_evening_run,
previous_run=previous_run if merge_previous else None,
sections=sections,
section_history=section_history,
unattached_agendas=unattached_agendas,
upcoming_agendas=agenda.upcoming(collected, unattached_agendas),
)
documents = llm_client.generate_all(section_contexts, sections=sections)
# The final filter, per docs/project-plan.md: a second, independent pass over
# each document against the same context it was generated from, before anything
# is written to output/. See synth/counter_run.py for what it catches and how it
# fails safe.
documents = counter_run.verify_all(documents, section_contexts)
run_dir = write_output(
output_dir, run_id, context, documents,
sections=sections,
people=prefs["people"] if prefs else [],
)
# After the digest is on disk, never before: a push is a promise that there is
# something to open. Best-effort — see notify.py.
notify.send(context, documents, sections, prefs)
LOG.info(
"digest run %s complete: %s -> %s (sections: %s)",
run_id,
context["item_counts"],
run_dir,
", ".join(sections) or "none",
)
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink