A Daily Digest of What Actually Got Merged, Built from MQTT and a Local LLM

The problem

The team received a daily email, generated by a small script that queried the CI system for the day’s merged pull requests. The email listed each merge as a title, a link, and the number of green checks. It was accurate. It was also not useful, because the information it carried — that a pull request merged and that its checks passed — was the information everyone already had, from the CI system itself, in real time, the moment it happened.

What the email did not say was the part a person actually wants in the morning: what changed, in plain language, in each of the repositories they care about. Not the check count. Not the title, which is often a conventional-commit prefix and a noun. The change.

The constraint that shaped the solution was that the summary had to be generated by a model the team could run locally, on the same hardware that ran the rest of the pipeline, without sending the diff to a cloud API. The diff contains internal identifiers and, occasionally, things that should not leave the network. A local model was not a preference; it was the requirement.

Working through it

The event feed, and why MQTT was already there

The CI system published merge events to an MQTT broker as part of its notification fan-out, the same feed that drove the dashboard’s live activity panel. The events carried the repository name, the pull request number, the title, the author, the merge commit, and a diff stat. The broker was already running, already authenticated, already subscribed to by one consumer. Adding a second consumer was cheaper than building a new integration against the CI system’s API, which would have meant handling pagination, rate limits, and a webhook endpoint, for the same data.

The subscriber is a single long-running process that connects to the broker, subscribes to the merge topic, and writes each event to a SQLite database. It does not summarise, does not send email, does not do anything except store the event with the timestamp it arrived. The separation matters, because the subscriber is the only component that has to be up when the event happens, and it is the one component whose failure loses data.

# subscriber.py — stores merge events, nothing else.
import json
import sqlite3
import paho.mqtt.client as mqtt
 
DB = "merges.sqlite3"
TOPIC = "ci/merge/+"
 
def on_message(client, _userdata, msg):
    event = json.loads(msg.payload)
    with sqlite3.connect(DB) as db:
        db.execute(
            "INSERT OR REPLACE INTO merges "
            "(repo, pr, title, author, sha, diffstat, merged_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (event["repo"], event["pr"], event["title"],
             event["author"], event["sha"], event["diffstat"],
             event["merged_at"]),
        )
 
client = mqtt.Client()
client.on_message = on_message
client.connect("mqtt.internal")
client.subscribe(TOPIC)
client.loop_forever()

INSERT OR REPLACE keyed on the repository and pull request number means a redelivered event does not produce a duplicate row, which is the failure mode that matters for a feed that is at-least-once.

The nightly job: group, then summarise

The nightly job runs at 06:00, reads the day’s rows from SQLite, groups them by repository, and for each repository asks the local model to write a short summary of the day’s merges. The prompt is the group of diff stats and titles, and the instruction is two sentences per merge, plain language, no check counts, no conventional-commit prefixes.

# digest.py — runs nightly, one LLM call per repository.
import sqlite3
from local_llm import complete   # the team's local model client
 
PROMPT = (
    "For each merged pull request below, write two sentences in plain "
    "language describing what changed. No check counts, no prefixes.\n\n"
    "{entries}\n\nSummaries:"
)
 
def build_digest(date: str) -> str:
    with sqlite3.connect(DB) as db:
        rows = db.execute(
            "SELECT repo, title, diffstat FROM merges "
            "WHERE date(merged_at) = ? ORDER BY repo, merged_at",
            (date,),
        ).fetchall()
 
    by_repo: dict[str, list] = {}
    for repo, title, diffstat in rows:
        by_repo.setdefault(repo, []).append(f"- {title}\n  {diffstat}")
 
    sections = []
    for repo, entries in by_repo.items():
        summary = complete(PROMPT.format(entries="\n".join(entries)))
        sections.append(f"## {repo}\n{summary}")
    return "\n\n".join(sections)

The job makes one model call per repository, not one per pull request, because the model is local and slow, and because a per-repository summary is the unit a reader actually wants: what happened in this repository today, not a list of isolated two-sentence fragments.

Why the summary is from the diff stat, not the diff

The event carries a diff stat, not the full diff. This is a deliberate limitation, and it is the part of the design that is worth naming. A full diff is the more faithful input to a summariser, and it is also the input that contains the internal identifiers and the occasional secret that should not be prompted, even to a local model, because the model’s output is emailed to a list of people, and a summariser can and does echo input back verbatim.

The diff stat — files touched, lines added and removed — is enough for a two-sentence summary of what changed, and it is the input that cannot leak what it does not contain. The trade is that the summary says “changed the authentication module, 140 lines across 6 files” rather than “removed the session-token refresh path”. For a morning digest, the first is the right granularity, and it is the granularity that keeps the input safe by construction rather than by a review of every diff that passes through.

Keeping the pipeline honest with a check on the stored events

The component whose failure loses data is the subscriber, and the property worth pinning is that every merge event that arrived was stored, not that the summary is good, which is not testable in a meaningful way. The check counts the events the broker delivered against the rows in the database, over a window, and fails if the two diverge:

# tests/test_subscriber_stores_every_event.py
def test_no_event_is_lost(tmp_path, monkeypatch):
    db_path = tmp_path / "merges.sqlite3"
    monkeypatch.setattr(subscriber, "DB", str(db_path))
    subscriber.init_db()
 
    delivered = [make_event(repo=f"repo{i % 3}", pr=i) for i in range(50)]
    for event in delivered:
        subscriber.on_message(None, None, FakeMessage(event))
 
    with sqlite3.connect(db_path) as db:
        stored = db.execute("SELECT COUNT(*) FROM merges").fetchone()[0]
    assert stored == len(delivered)

The test is not a test of the summariser. It is a test of the one component whose silent failure is unrecoverable, and it is the check that would catch a schema change, a dropped column, or a WHERE clause that quietly filters out a class of event.

The solution

The state of the fix:

1. A single MQTT subscriber stores every merge event to SQLite, keyed so a redelivery does not duplicate the row. It is the only component that has to be up when the event happens, and the only one whose failure loses data.
2. A nightly job groups the day’s rows by repository and makes one local-LLM call per repository, producing a two-sentence plain-language summary per merge.
3. The model’s input is the diff stat, not the diff, so the summary cannot echo internal identifiers or secrets it was never given.
4. A test asserts that every delivered event is stored, so a silent loss in the subscriber fails the build rather than the next morning’s digest.

The general shape of the problem is a notification that carries the status of a process rather than the output of it, and a constraint that the enrichment has to happen locally. The answer is the same in most cases: store the raw event at the point it happens, do the expensive transformation on a schedule where it is cheap, and keep the transformation’s input as small as the output allows.

Conclusion

A daily email that lists merges as titles and check counts says what happened in CI, which is the information everyone already has in real time. The part a person wants in the morning is what changed, in plain language, per repository, and that part requires a model.

The pipeline is three components: a subscriber that stores the event, a nightly job that groups and summarises, and a local model that writes the summary from the diff stat rather than the diff, so the input cannot leak what it does not contain. One cron entry, no new service, no cloud API. The email now says what changed, and it says it from a model that never left the network.