diff --git a/README.md b/README.md
index e486156..4a3e689 100644
--- a/README.md
+++ b/README.md
@@ -653,6 +653,97 @@ tried. `session_util` grew a `blind15_id` covering this in
 [`7e8d126`](https://github.com/session-foundation/libsession-python/commit/7e8d126), which
 no packaged release carries yet.
 
+## Contributor Pull Requests
+
+[`github_prs/digest.py`](github_prs/digest.py) posts one message each weekday morning
+listing the open pull requests in session-foundation's repositories whose author is not
+a maintainer and which have moved in the last three days — the ones nobody on the team
+has a reason to already know about:
+
+```
+**Contributor pull requests** · last 3 days
+🟢 **2** new · ✏️ **1** updated
+**36** open from contributors across the org.
+
+**session-desktop**
+🟢 [#1958](…) @KyWB · 12h · 💬1 · Fix issue #563
+✏️ [#1904](…) @scrense-hash · 3h · 💬2 · feat: add SOCKS5 proxy support
+```
+
+🟢 is a PR the digest has never reported; ✏️ is one it has, which has moved since. A PR
+that has not moved is left out entirely, however wide the window. The backlog line
+counts every open contributor PR regardless, so a quiet day still says how much is
+waiting.
+
+### Weekdays, and the state file
+
+The timer runs `Mon..Fri`, so Monday's run has to cover the weekend — hence a 72-hour
+window rather than a daily one. That window overlaps itself by two days on every run,
+and [`--state`](github_prs/digest.py) is what stops the overlap being noise: it records
+which PRs reached Discord and what each one's `updated_at` was at the time.
+
+`updated_at` moves on *any* change, including one that touches several PRs at once — a
+label sweep, a base branch renamed — so those resurface once even though nobody worked
+on them. The accurate alternative is the head SHA and the comment counts, which are not
+in the search result and cost a request per PR that moved; this is the cheaper half of
+that trade, taken deliberately.
+
+Only what Discord accepted is recorded, so a run that fails on its second message
+re-reports that message's PRs tomorrow rather than losing them. Every way of failing to
+read the state file — missing, unreadable, written by another version — treats every PR
+in the window as new: noisy once, never wrong, which is what makes it a cache rather
+than something to back up.
+
+One search fetches every open PR in the org and the window is applied to the result
+here rather than in the query — that is what buys the backlog count for the cost of a
+single query. Past GitHub's 1000-result search ceiling the digest says the counts are a
+floor instead of failing.
+
+### Who is a maintainer
+
+[`github_prs/maintainers.txt`](github_prs/maintainers.txt), one login per line, matched
+case-insensitively. Bot accounts need no entry — every account GitHub types as a `Bot`
+is dropped, so a renamed Dependabot stays out on its own.
+
+Neither of the two things GitHub could answer this with is a substitute. Org membership
+covers six accounts, two of which are not in the review loop; push access is held by a
+dozen more as outside collaborators, several of them contractors whose PRs are exactly
+what the digest is for. Both would get it wrong in both directions, so the list is
+written by hand — and goes stale silently, since a new maintainer's PRs are reported as
+a stranger's until someone adds them.
+
+### What is left out
+
+Forks, archived repositories and private repositories, by checking the search results
+against the org's repository list rather than by name — so a repository created today
+is covered today and a fork of an upstream project never is. There is no flag to widen
+that: private repositories stay out whatever the token can see.
+
+| flag | |
+| --- | --- |
+| `--window-hours N` | how far back a PR must have moved to be considered (default 72) |
+| `--state PATH` | dedup state; without it every PR in the window is new |
+| `--state-retention-days N` | drop state entries older than this (default 30) |
+| `--dry-run` | print the Discord payload, post nothing |
+| `--org`, `--token`, `--webhook` | override the environment |
+| `--maintainers PATH` | a different list |
+
+| env var | |
+| --- | --- |
+| `GITHUB_PRS_TOKEN` | read-only token; no scope at all is needed, the digest reads public repositories only |
+| `GITHUB_PRS_DISCORD_WEBHOOK_URL` | the channel it posts to (not needed with `--dry-run`) |
+| `GITHUB_PRS_ORG` | optional; defaults to `session-foundation` |
+
+It runs on the same box as the Zendesk digest, under its own user and its own
+environment file — see [deploy/README.md](deploy/README.md). Its HTTP retries,
+Discord posting and dedup state are the same code the Zendesk digest uses, in
+[shared/](shared/).
+
+```sh
+cd github_prs && python -m unittest discover
+cd shared && python -m unittest discover
+```
+
 ## Workflow Failure Notificaiton
 
 If a workflow fails and is in the list of workflows monitored by the failure notificaiton workflow, the failure notificaiton workflow will send a message to a discord webhook.
diff --git a/crowdin/report_multiple_translations.py b/crowdin/report_multiple_translations.py
index 43da1e9..18e1c37 100644
--- a/crowdin/report_multiple_translations.py
+++ b/crowdin/report_multiple_translations.py
@@ -40,16 +40,17 @@
 import collections
 import concurrent.futures
 import datetime as dt
-import email.utils
 import json
 import os
 import subprocess
 import sys
 import threading
-import time
 
 import requests
 
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from shared import discord, retry  # noqa: E402
+
 API = "https://api.crowdin.com/api/v2"
 DEFAULT_PROJECT = "618696"
 KEYRING_ATTRS = ["service", "crowdin", "key", "translation-api-token"]
@@ -81,49 +82,10 @@ def get_token(cli_token):
              "or store it via secret-tool).")
 
 
-def parse_retry_after(value, fallback, cap=30):
-    """Seconds to wait for a Retry-After header, parsed defensively.
-
-    Supports both numeric-seconds and HTTP-date forms (RFC 7231); falls back to
-    `fallback` when the header is missing or unparseable, and clamps the result
-    to `cap` so a bogus/huge value can't stall the run."""
-    wait = fallback
-    if value is not None:
-        try:
-            wait = float(value)
-        except (TypeError, ValueError):
-            try:
-                when = email.utils.parsedate_to_datetime(value)
-                if when.tzinfo is None:
-                    when = when.replace(tzinfo=dt.timezone.utc)
-                wait = (when - dt.datetime.now(dt.timezone.utc)).total_seconds()
-            except (TypeError, ValueError):
-                wait = fallback
-    if wait < 0:
-        wait = fallback
-    return min(wait, cap)
-
-
-def request_with_retry(session, method, url, max_retries=10, **kw):
-    """Request with backoff on 429/5xx AND on network errors (flaky DNS/connection)."""
-    delay = 0.5
-    last_exc = None
-    for _ in range(max_retries):
-        try:
-            r = session.request(method, url, timeout=60, **kw)
-        except requests.exceptions.RequestException as e:
-            last_exc = e
-            time.sleep(delay)
-            delay = min(delay * 2, 30)
-            continue
-        if r.status_code == 429 or r.status_code >= 500:
-            time.sleep(parse_retry_after(r.headers.get("Retry-After"), delay))
-            delay = min(delay * 2, 30)
-            continue
-        r.raise_for_status()
-        return r
-    if last_exc:
-        raise last_exc
+def request_with_retry(session, method, url, **kw):
+    """A Crowdin API call. A non-retryable 4xx raises, so a caller can read the
+    body of what it asked for without checking the status first."""
+    r = retry.request_with_retry(session, method, url, attempts=10, timeout=60, **kw)
     r.raise_for_status()
     return r
 
@@ -366,33 +328,22 @@ def pack_embeds(embeds):
 
 
 def post_to_discord(webhook_url, messages):
-    # Use a fresh, unauthenticated session -- the Crowdin Bearer token must never
-    # be sent to Discord. request_with_retry raises on any non-retryable 4xx, so
-    # we translate that into the concise failure message here.
+    # A fresh, unauthenticated session: the Crowdin Bearer token must never be
+    # sent to Discord.
     with requests.Session() as webhook_session:
-        for payload in messages:
-            try:
-                request_with_retry(webhook_session, "POST", webhook_url, json=payload)
-            except requests.exceptions.RequestException as e:
-                resp = getattr(e, "response", None)
-                if resp is not None:
-                    detail = f"Discord webhook failed ({resp.status_code}): {resp.text[:300]}"
-                else:
-                    detail = f"Discord webhook failed: {e}"
-                # A rich payload can be rejected outright (e.g. an embed exceeded
-                # Discord's size limits). Before crashing, best-effort post a plain
-                # warning so the failure is at least visible in the channel; if even
-                # that fails, fall through to the sys.exit below.
-                try:
-                    request_with_retry(webhook_session, "POST", webhook_url, json={
-                        "content": "⚠️ Crowdin multiple-translations report failed to post "
-                                   "its results (a message was rejected by Discord). "
-                                   "Re-run `report_multiple_translations.py --json` for the "
-                                   "full list.",
-                    })
-                except requests.exceptions.RequestException:
-                    pass
-                sys.exit(detail)
+        posted = discord.post_to_discord(webhook_session, webhook_url, messages)
+        if posted == len(messages):
+            return
+        # A rich payload can be rejected outright, an embed over Discord's size
+        # limits say. A plain warning at least makes the failure visible in the
+        # channel; if that fails too, the exit below still says so.
+        discord.post_to_discord(webhook_session, webhook_url, [{
+            "content": "⚠️ Crowdin multiple-translations report failed to post "
+                       "its results (a message was rejected by Discord). "
+                       "Re-run `report_multiple_translations.py --json` for the "
+                       "full list.",
+        }])
+        sys.exit(f"Discord accepted {posted} of {len(messages)} messages.")
 
 
 # --------------------------------------------------------------------------- #
diff --git a/crowdin/test_report_multiple_translations.py b/crowdin/test_report_multiple_translations.py
new file mode 100644
index 0000000..50a0381
--- /dev/null
+++ b/crowdin/test_report_multiple_translations.py
@@ -0,0 +1,75 @@
+"""
+    cd crowdin && python -m unittest discover
+"""
+import contextlib
+import io
+import os
+import sys
+import unittest
+
+import requests
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import report_multiple_translations as report  # noqa: E402
+from shared.testing import FakeResponse, FakeSession, NoSleep, Patched  # noqa: E402
+
+
+class ApiResponse(FakeResponse):
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise requests.HTTPError(f"{self.status_code}", response=self)
+
+
+class WebhookSession(FakeSession):
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
+class TestRequestWithRetry(unittest.TestCase):
+    def test_a_client_error_raises_rather_than_returning(self):
+        """Callers read the body straight off the response, so a 4xx has to stop them."""
+        with self.assertRaises(requests.HTTPError):
+            report.request_with_retry(FakeSession([ApiResponse({}, status_code=404)]),
+                                      "GET", "https://x")
+
+    def test_crowdins_budget_is_ten_attempts_at_sixty_seconds(self):
+        session = FakeSession([ApiResponse({}, status_code=503)] * 9 + [ApiResponse({"ok": 1})])
+        with NoSleep():
+            self.assertEqual(report.request_with_retry(session, "GET", "https://x").json(), {"ok": 1})
+        self.assertEqual(len(session.calls), 10)
+        self.assertEqual({kw["timeout"] for _, _, kw in session.calls}, {60})
+
+
+class TestPostToDiscord(unittest.TestCase):
+    def post(self, responses, messages):
+        session = WebhookSession(responses)
+        with Patched(report.requests, Session=lambda: session), \
+                contextlib.redirect_stdout(io.StringIO()):
+            report.post_to_discord("https://hook", messages)
+        return session
+
+    def test_every_message_accepted_posts_nothing_else(self):
+        session = self.post([FakeResponse({}, status_code=204)] * 2, [{"embeds": []}] * 2)
+        self.assertEqual(len(session.calls), 2)
+
+    def test_a_rejection_posts_a_plain_warning_then_exits(self):
+        responses = [FakeResponse({}, status_code=204), FakeResponse({}, status_code=400),
+                     FakeResponse({}, status_code=204)]
+        with self.assertRaises(SystemExit) as caught:
+            self.post(responses, [{"embeds": []}] * 3)
+        self.assertIn("1 of 3", str(caught.exception))
+
+    def test_the_warning_is_plain_content_the_webhook_cannot_reject_for_size(self):
+        session = WebhookSession([FakeResponse({}, status_code=400), FakeResponse({}, status_code=204)])
+        with Patched(report.requests, Session=lambda: session), \
+                contextlib.redirect_stdout(io.StringIO()), self.assertRaises(SystemExit):
+            report.post_to_discord("https://hook", [{"embeds": [{"title": "x" * 9000}]}])
+        self.assertEqual(list(session.calls[1][2]["json"]), ["content"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/deploy/README.md b/deploy/README.md
index bde3c0b..4adf7b6 100644
--- a/deploy/README.md
+++ b/deploy/README.md
@@ -1,17 +1,20 @@
 # Self-hosted deployment
 
-The Zendesk triage digest and the `claude:` note webhook, both running on one
-machine.
-
-Two things run here:
+The Zendesk triage digest, the `claude:` note webhook and the contributor pull request
+digest, all on one machine.
 
 | Unit | What it is |
 | --- | --- |
 | `zendesk-relay.service` | Always on. The HTTPS endpoint Zendesk posts note webhooks to. |
 | `zendesk-digest.timer` → `.service` | Weekday mornings. Resolves positive reviews, then posts the digest. |
+| `github-prs-digest.timer` → `.service` | Weekday mornings. Posts the contributor pull request digest. |
+
+`zendesk-alert@.service` and `github-prs-alert@.service` are pulled in by `OnFailure=`
+and report the failed unit to the channel that job posts to.
 
-`zendesk-alert@.service` is pulled in by `OnFailure=` on both, and reports the failed
-unit to the triage channel.
+One clone at `/opt/zendesk` holds all of it — the directory is named after its first
+tenant, not its contents. The venvs are separate, because the two jobs pin `requests`
+differently and a shared one would silently be whichever was installed last.
 
 ## Host requirements
 
@@ -30,7 +33,8 @@ unit to the triage channel.
 - **nginx already installed**, with certbot managing its certificates. This adds one
   server block to it rather than a second web server; two would fight over :443 and
   take the host's other sites down with them.
-- Persistent `/var/lib/zendesk` — it holds the dedup state, the only thing on disk.
+- Persistent `/var/lib/zendesk` and `/var/lib/github-prs` — they hold the two digests'
+  dedup state, the only thing on disk. `StateDirectory=` creates the second one.
 
 Note who else holds root. This box becomes custodian of a Zendesk API token that can
 write a public comment to any ticket, and of a logged-in Claude Code session.
@@ -107,6 +111,39 @@ nginx -t && systemctl reload nginx                # validate what certbot wrote
 > nginx -t && systemctl reload nginx
 > ```
 
+### Adding the pull request digest
+
+Its own user, its own environment file and its own venv, out of the same clone. A
+token that can read the org's repositories has no business in the environment of the
+relay, which is the one process here reachable from the internet.
+
+The account needs no home of its own: nothing in this job shells out to the Claude
+CLI, so the `$HOME` that the Zendesk units bend over backwards to preserve is not
+wanted here at all.
+
+```bash
+useradd --system --no-create-home --home /nonexistent --shell /usr/sbin/nologin ghdigest
+python3 -m venv /opt/github-prs/venv
+/opt/github-prs/venv/bin/pip install -r /opt/zendesk/github_prs/requirements.txt
+chown -R ghdigest:ghdigest /opt/github-prs
+
+install -d -m 750 -o root -g ghdigest /etc/github-prs
+[ -e /etc/github-prs/env ] || install -m 640 -o root -g ghdigest /dev/null /etc/github-prs/env
+"${EDITOR:-nano}" /etc/github-prs/env             # contents under Secrets, below
+
+cp /opt/zendesk/deploy/github-prs-*.service /opt/zendesk/deploy/github-prs-*.timer \
+   /etc/systemd/system/
+systemctl daemon-reload
+systemctl enable --now github-prs-digest.timer
+```
+
+The clone stays owned by `zendesk` and world-readable, which is what lets `ghdigest`
+run out of it. Nothing secret lives there — every secret is under `/etc`.
+
+`/var/lib/github-prs` needs no `install` step: `StateDirectory=github-prs` on the unit
+creates it with the right owner on first start. It holds the dedup state that keeps the
+72-hour window from re-reporting the same PR every weekday morning.
+
 ## Secrets
 
 `/etc/zendesk/env`, mode `640`, `root:zendesk` — readable by the service, not by
@@ -183,6 +220,26 @@ Set `RELAY_DRY_RUN=1` for the first deployment. The whole `claude:` note path ru
 webhook, command parsing, composing, translation — and the Zendesk writes are
 skipped.
 
+### `/etc/github-prs/env`
+
+Mode `640`, `root:ghdigest`, and separate from the Zendesk file rather than merged
+into it — see above.
+
+```sh
+# Read-only, and no scope at all: the digest reads public repositories only. Nothing
+# here ever writes to GitHub.
+GITHUB_PRS_TOKEN=
+
+# The channel the digest posts to. A webhook is bound to the channel it was created
+# in, so this one value decides where the digest goes.
+GITHUB_PRS_DISCORD_WEBHOOK_URL=
+
+# Required, and normally the same webhook: without it alert.py falls back to
+# ZENDESK_DISCORD_WEBHOOK_URL, which is not in this file, and the failure notifier
+# fails instead of reporting.
+ALERT_DISCORD_WEBHOOK_URL=
+```
+
 ## Verifying, in order
 
 **1. Locally, before Zendesk knows the address.** An unsigned request must be refused:
@@ -281,12 +338,39 @@ else: the alert still sends, with the message it always sent.
 ```sh
 cd deploy && python -m unittest discover     # the alert's own tests
 ```
+**6. The pull request digest.** A dry run under the unit's own confinement renders the
+digest and posts nothing. `systemd-run` rather than `runuser` because the token then
+comes from the environment file rather than an argument every process on the box can
+read out of `ps`:
+
+```bash
+systemd-run --pty --uid=ghdigest -p EnvironmentFile=/etc/github-prs/env \
+  -p WorkingDirectory=/opt/zendesk/github_prs \
+  /opt/github-prs/venv/bin/python digest.py --dry-run
+```
+
+Then for real: `systemctl start github-prs-digest.service`,
+`systemctl list-timers github-prs-digest` (expect the next weekday, not tomorrow), and
+`systemctl start github-prs-alert@test.service` for its failure path.
+
+Start it twice. The second run is the one that proves the dedup state: it should report
+everything, then nothing, and say so.
+
+```bash
+journalctl -u github-prs-digest -n 5 --no-pager   # "N new, 0 changed, N unchanged"
+ls -l /var/lib/github-prs/seen.json
+```
+
+A second run that reports everything again means the state was not written — check that
+`StateDirectory=` reached systemd with
+`systemctl show github-prs-digest -p StateDirectory`.
 
 ## Updating
 
 ```bash
 runuser -u zendesk -- git -C /opt/zendesk pull
 /opt/zendesk/venv/bin/pip install -r /opt/zendesk/zendesk_triage/requirements.txt
+/opt/github-prs/venv/bin/pip install -r /opt/zendesk/github_prs/requirements.txt
 systemctl restart zendesk-relay
 ```
 
@@ -347,8 +431,10 @@ What that costs, in order of how much it matters:
 - **The digest is late, not lost.** `Persistent=yes` on the timer means a host that
   was down at 10:00 runs the digest once when it comes back, and the 72-hour window
   covers the gap.
-- **The dedup state may be stale.** Losing `/var/lib/zendesk/seen.json` re-reports the
+- **The dedup state may be stale.** Losing either `seen.json` re-reports that digest's
   window once: noisy, never wrong.
+- **The pull request digest is late, not lost**, on the same `Persistent=yes` as the
+  Zendesk one, and its 72-hour window already covers a weekend's gap.
 
 Failures that are not a whole-host outage report themselves — `OnFailure=` on both
 units posts the failed unit and a `journalctl` line to the triage channel.
\ No newline at end of file
diff --git a/deploy/alert.py b/deploy/alert.py
index 2428ffb..e8254b7 100644
--- a/deploy/alert.py
+++ b/deploy/alert.py
@@ -9,7 +9,8 @@
 
 Posts over ZENDESK_DISCORD_WEBHOOK_URL rather than the bot token, deliberately. This
 is one line of text needing no components, and a failure notifier should depend on as
-little as possible of whatever just broke.
+little as possible of whatever just broke. ALERT_DISCORD_WEBHOOK_URL overrides it, so
+a job that posts to a channel of its own reports its failures there too.
 
 The failed unit's last journal line comes with it, so the channel says what broke
 rather than only that something did. Reading the journal needs the unit to carry
@@ -103,7 +104,8 @@ def main():
     args = [arg.strip() for arg in sys.argv[1:]]
     if not args or len(args) > 2 or not args[0]:
         sys.exit("usage: alert.py <name> [journal-unit]")
-    webhook = triage.get_env("ZENDESK_DISCORD_WEBHOOK_URL")
+    webhook = (os.environ.get("ALERT_DISCORD_WEBHOOK_URL")
+               or triage.get_env("ZENDESK_DISCORD_WEBHOOK_URL"))
     detail = last_job_line(journal_tail(args[-1]))
     message = build_message(args[0], socket.gethostname(), *args[1:], detail=detail)
     # A fresh session, never a Zendesk one — that carries the API-token auth header,
diff --git a/deploy/github-prs-alert@.service b/deploy/github-prs-alert@.service
new file mode 100644
index 0000000..8103550
--- /dev/null
+++ b/deploy/github-prs-alert@.service
@@ -0,0 +1,22 @@
+[Unit]
+Description=Report a failed pull request digest unit to Discord
+Documentation=https://github.com/session-foundation/session-shared-scripts
+
+[Service]
+Type=oneshot
+User=ghdigest
+Group=ghdigest
+EnvironmentFile=/etc/github-prs/env
+
+# Same script as zendesk-alert@, run under this job's environment so the failure
+# lands in the channel the job posts to rather than the triage one.
+ExecStart=/opt/github-prs/venv/bin/python /opt/zendesk/deploy/alert.py %i
+
+# See zendesk-alert@.service: read-only access to the failed unit's journal, which is
+# the line the alert quotes.
+SupplementaryGroups=systemd-journal
+
+NoNewPrivileges=yes
+PrivateTmp=yes
+ProtectSystem=strict
+ProtectHome=yes
diff --git a/deploy/github-prs-digest.service b/deploy/github-prs-digest.service
new file mode 100644
index 0000000..0de28b9
--- /dev/null
+++ b/deploy/github-prs-digest.service
@@ -0,0 +1,36 @@
+[Unit]
+Description=Daily digest of contributor pull requests
+Documentation=https://github.com/session-foundation/session-shared-scripts
+After=network-online.target
+Wants=network-online.target
+OnFailure=github-prs-alert@%n.service
+
+[Service]
+Type=oneshot
+User=ghdigest
+Group=ghdigest
+WorkingDirectory=/opt/zendesk/github_prs
+EnvironmentFile=/etc/github-prs/env
+
+ExecStart=/opt/github-prs/venv/bin/python digest.py --window-hours 72 \
+    --state /var/lib/github-prs/seen.json
+
+# A dozen API calls. Anything near this is GitHub rate-limiting the run rather than
+# a large result set.
+TimeoutStartSec=15min
+
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+RestrictAddressFamilies=AF_INET AF_INET6
+RestrictNamespaces=yes
+LockPersonality=yes
+# The dedup state, and the only thing this job writes anywhere. Losing it re-reports
+# the window once: noisy, never wrong, so it needs persisting rather than backing up.
+ReadWritePaths=/var/lib/github-prs
+StateDirectory=github-prs
diff --git a/deploy/github-prs-digest.timer b/deploy/github-prs-digest.timer
new file mode 100644
index 0000000..ca94aa7
--- /dev/null
+++ b/deploy/github-prs-digest.timer
@@ -0,0 +1,20 @@
+[Unit]
+Description=Post the contributor pull request digest
+Documentation=https://github.com/session-foundation/session-shared-scripts
+
+[Timer]
+# Half an hour before the Zendesk digest, where the people reading it are, tracking
+# daylight saving. Weekdays only, which is what makes the 72h window load-bearing:
+# Monday's run has to reach back over the weekend.
+#
+# The timezone belongs in the expression: there is no Timezone= key in [Timer], and
+# systemd *ignores* one silently. Check any change with:
+#     systemd-analyze calendar "Mon..Fri 09:30 Australia/Melbourne"
+OnCalendar=Mon..Fri 09:30 Australia/Melbourne
+Persistent=yes
+# The window overlaps by two days and the state file absorbs the overlap, so drift
+# here costs nothing at all.
+RandomizedDelaySec=2min
+
+[Install]
+WantedBy=timers.target
diff --git a/github_prs/digest.py b/github_prs/digest.py
new file mode 100755
index 0000000..e6be7e8
--- /dev/null
+++ b/github_prs/digest.py
@@ -0,0 +1,405 @@
+#!/usr/bin/env python3
+"""
+Daily digest of contributor pull requests across the session-foundation org.
+
+Lists every open PR in the org's own repositories whose author is not a maintainer,
+grouped by repository, marking the ones never reported before and the ones that have
+moved since they were. Posted to a Discord webhook of its own.
+
+--state is what makes the second of those answerable: it records what reached Discord
+and what each PR's updated_at was at the time, so a PR the digest already showed stays
+out until something happens to it. Without it every PR in the window reads as new.
+
+Who is a maintainer comes from maintainers.txt, one login per line; bot accounts are
+dropped on GitHub's own account type rather than by name. Forks, archived and private
+repos are excluded by checking the search results against the org's repository list,
+so a repo created today is covered today.
+
+One search fetches every open PR in the org, and the window is applied to the result
+here rather than in the query. That is what lets the header carry the total open
+contributor backlog alongside the day's changes for the cost of a single query.
+
+Config (env vars, or flags for local runs):
+    GITHUB_PRS_TOKEN      GitHub token, read-only. Needs no scope at all: the
+                          digest reads public repositories only.
+    GITHUB_PRS_DISCORD_WEBHOOK_URL
+                          Discord incoming webhook for the channel this posts to
+                          (not needed with --dry-run)
+    GITHUB_PRS_ORG        (optional) org to scan; defaults to session-foundation
+
+Usage:
+    # real run (what the timer does)
+    python digest.py
+
+    # fetch and render, print the payload, post nothing
+    python digest.py --dry-run
+
+    # what the weekday timer does: a window covering the weekend, deduped
+    python digest.py --window-hours 72 --state /var/lib/github-prs/seen.json
+"""
+import argparse
+import json
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from operator import itemgetter
+
+import requests
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from shared import discord, state as dedup  # noqa: E402
+from shared.discord import MAX_MESSAGE_TEXT_CHARS, clip  # noqa: E402
+from shared.env import get_env  # noqa: E402
+from shared.retry import request_with_retry  # noqa: E402
+
+API = "https://api.github.com"
+DEFAULT_ORG = "session-foundation"
+# Three days, because the timer runs on weekdays: Monday's window has to reach back
+# over the weekend. Overlap between consecutive runs is what --state absorbs.
+DEFAULT_WINDOW_HOURS = 72
+DEFAULT_RETENTION_DAYS = 30
+STATE_VERSION = 1
+# The Search API caps a query at 1000 results and returns 422 for any page past it
+# (at per_page=100 that is page 11). Past the cap the digest reports truncation
+# rather than failing the run.
+# https://docs.github.com/en/rest/search#about-search
+SEARCH_RESULT_LIMIT = 1000
+PER_PAGE = 100
+MAINTAINERS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                                "maintainers.txt")
+
+
+def load_maintainers(path):
+    """Logins from maintainers.txt, lowercased. `#` comments and blanks ignored."""
+    logins = set()
+    with open(path, encoding="utf-8") as handle:
+        for line in handle:
+            login = line.split("#", 1)[0].strip()
+            if login:
+                logins.add(login.lower())
+    return frozenset(logins)
+
+
+def github_session(token):
+    session = requests.Session()
+    session.headers.update({
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/vnd.github+json",
+        "X-GitHub-Api-Version": "2022-11-28",
+    })
+    return session
+
+
+def fetch_json(session, url, **kwargs):
+    resp = request_with_retry(session, "GET", url, **kwargs)
+    if resp.status_code >= 400:
+        sys.exit(f"GitHub {resp.status_code} on {url}: {resp.text[:300]}")
+    return resp.json()
+
+
+def fetch_repos(session, org):
+    """Names of the org's own, live, public repositories.
+
+    Private repositories are never reported, whatever the token can see: the digest
+    posts to Discord, and nothing about them belongs there.
+    """
+    names, page = set(), 1
+    while True:
+        batch = fetch_json(session, f"{API}/orgs/{org}/repos",
+                           params={"per_page": PER_PAGE, "page": page, "type": "all"})
+        for repo in batch:
+            if repo.get("private") or repo.get("fork") or repo.get("archived"):
+                continue
+            names.add(repo["name"])
+        if len(batch) < PER_PAGE:
+            return names
+        page += 1
+
+
+def search_open_prs(session, org, max_results=SEARCH_RESULT_LIMIT):
+    """Every open PR in the org, newest activity first.
+
+    Returns (items, truncated). `truncated` is the caller's cue that the counts are
+    a floor rather than a total.
+    """
+    items, page = [], 1
+    limit = min(max_results, SEARCH_RESULT_LIMIT)
+    while len(items) < limit:
+        payload = fetch_json(session, f"{API}/search/issues", params={
+            "q": f"org:{org} is:pr is:open",
+            "sort": "updated",
+            "order": "desc",
+            "per_page": PER_PAGE,
+            "page": page,
+            # The legacy issue-search syntax is gone; without this the endpoint
+            # rejects the query rather than falling back to it.
+            "advanced_search": "true",
+        })
+        batch = payload.get("items", [])
+        items.extend(batch)
+        total = payload.get("total_count", len(items))
+        if len(batch) < PER_PAGE or len(items) >= total:
+            return items[:limit], total > len(items[:limit])
+        page += 1
+    return items[:limit], True
+
+
+def repo_name(item):
+    """Repo name for a search result — the API gives only the repository's API URL."""
+    return item.get("repository_url", "").rsplit("/", 1)[-1]
+
+
+def is_bot(item):
+    return (item.get("user") or {}).get("type") == "Bot"
+
+
+def author(item):
+    return (item.get("user") or {}).get("login") or "?"
+
+
+def parse_time(value):
+    return datetime.fromisoformat(value)
+
+
+def contributor_prs(items, repos, maintainers):
+    """The open PRs worth reporting: org repos, human authors, not maintainers."""
+    keep = []
+    for item in items:
+        if repo_name(item) not in repos or is_bot(item):
+            continue
+        if author(item).lower() in maintainers:
+            continue
+        keep.append(item)
+    return keep
+
+
+def in_window(prs, cutoff):
+    """The PRs that have moved at all since the cutoff. A PR nobody has touched in
+    three days is not news, whatever its state here."""
+    return [pr for pr in prs if parse_time(pr["updated_at"]) >= cutoff]
+
+
+def pr_id(pr):
+    return str(pr.get("id"))
+
+
+def activity_key(pr):
+    """The value a re-report is judged against.
+
+    `updated_at` moves on any change at all, so an edit that touches several PRs at
+    once — a label sweep, a base branch renamed — resurfaces every one of them. The
+    accurate alternative is the head SHA and the comment counts, which are not in the
+    search result and cost a request per PR that moved; this is the deliberate cheaper
+    half of that trade.
+    """
+    return pr.get("updated_at")
+
+
+# ---- Dedup state -----------------------------------------------------------
+
+
+def empty_state():
+    return dedup.empty_state(STATE_VERSION)
+
+
+def load_state(path):
+    return dedup.load_state(path, STATE_VERSION, "PR")
+
+
+def partition_by_state(prs, state):
+    """Split into (new, changed, unchanged) against what was last reported."""
+    seen = state.get("seen", {})
+    new, changed, unchanged = [], [], []
+    for pr in prs:
+        previous = seen.get(pr_id(pr))
+        if previous is None:
+            new.append(pr)
+        elif previous.get("updated_at") != activity_key(pr):
+            changed.append(pr)
+        else:
+            unchanged.append(pr)
+    return new, changed, unchanged
+
+
+def save_state(path, state, reported, retention_days=DEFAULT_RETENTION_DAYS):
+    """Record `reported` as seen. Returns (kept, pruned)."""
+    records = {pr_id(pr): {"updated_at": activity_key(pr),
+                           # Not read back. The file is the first thing anyone opens
+                           # when the digest reports the wrong thing, and an id
+                           # alone identifies nothing.
+                           "pr": f"{repo_name(pr)}#{pr.get('number')}"}
+               for pr in reported}
+    return dedup.save_state(path, state, records, retention_days, STATE_VERSION)
+
+
+# ---- Discord rendering -----------------------------------------------------
+#
+# A header, then one block per repository whose PRs changed. Components V2 so that
+# each repository is its own component: a long day splits between repositories
+# rather than mid-list, unless one repository alone outgrows a message.
+MAX_COMPONENTS_PER_MESSAGE = 10
+TITLE_CHARS = 90
+
+NEW_MARKER = "🟢"
+UPDATED_MARKER = "✏️"
+
+
+def age(then, now):
+    """Compact age, coarsening as it grows: 40m, 6h, 3d, 5w."""
+    minutes = max(0, int((now - then).total_seconds() // 60))
+    if minutes < 60:
+        return f"{minutes}m"
+    hours = minutes // 60
+    if hours < 48:
+        return f"{hours}h"
+    days = hours // 24
+    if days < 14:
+        return f"{days}d"
+    return f"{days // 7}w"
+
+
+def build_pr_line(pr, now, is_new):
+    marker = NEW_MARKER if is_new else UPDATED_MARKER
+    stamp = age(parse_time(pr["created_at"] if is_new else pr["updated_at"]), now)
+    title = clip(pr.get("title"), TITLE_CHARS)
+    if pr.get("draft"):
+        title = f"[draft] {title}"
+    comments = pr.get("comments") or 0
+    replies = f" · 💬{comments}" if comments else ""
+    return (f"{marker} [#{pr['number']}]({pr['html_url']}) @{author(pr)} · "
+            f"{stamp}{replies} · {title}")
+
+
+def split_block(heading, entries, max_chars):
+    """(text, ids) blocks of at most `max_chars`, each under its own copy of `heading`.
+
+    A block over the budget is rejected by Discord, and since nothing in it is then
+    recorded, the same block would be rebuilt every run until the window moved on.
+    """
+    blocks, lines, ids, used = [], [heading], set(), len(heading)
+    for line, key in entries:
+        if ids and used + 1 + len(line) > max_chars:
+            blocks.append(("\n".join(lines), ids))
+            lines, ids, used = [heading], set(), len(heading)
+        lines.append(line)
+        ids.add(key)
+        used += 1 + len(line)
+    blocks.append(("\n".join(lines), ids))
+    return blocks
+
+
+def group_by_repo(new, updated, now, max_chars=MAX_MESSAGE_TEXT_CHARS):
+    """(text, ids) per repository, new PRs above updated ones.
+
+    Repositories are ordered by how much changed, so the busiest is read first.
+    Within a group each PR sorts on the timestamp its line shows — the search
+    returns them in update order, which reads as no order at all next to an age
+    taken from the creation date.
+    """
+    repos = []
+    for repo in sorted({repo_name(pr) for pr in new + updated}):
+        entries = []
+        for prs, field, is_new in ((new, "created_at", True),
+                                   (updated, "updated_at", False)):
+            group = sorted((pr for pr in prs if repo_name(pr) == repo),
+                           key=itemgetter(field), reverse=True)
+            entries += [(build_pr_line(pr, now, is_new), pr_id(pr)) for pr in group]
+        repos.append((split_block(f"**{repo}**", entries, max_chars), len(entries)))
+    repos.sort(key=lambda item: -item[1])
+    return [block for blocks, _ in repos for block in blocks]
+
+
+def window_label(hours):
+    if hours % 24 == 0 and hours >= 24:
+        days = hours // 24
+        return f"{days} day{'s' if days > 1 else ''}"
+    return f"{hours}h"
+
+
+def build_header(new, updated, backlog, window_hours, truncated):
+    lines = [f"**Contributor pull requests** · last {window_label(window_hours)}"]
+    if new or updated:
+        lines.append(f"{NEW_MARKER} **{len(new)}** new · "
+                     f"{UPDATED_MARKER} **{len(updated)}** updated")
+    else:
+        lines.append("Nothing opened or updated.")
+    lines.append(f"**{backlog}** open from contributors across the org.")
+    if truncated:
+        lines.append("_GitHub capped the search at 1000 results; the counts are a floor._")
+    return "\n".join(lines)
+
+
+def build_messages(new, updated, backlog, window_hours, now, truncated=False):
+    """Return (messages, coverage), as shared.discord.messages_from_entries does."""
+    header = build_header(new, updated, backlog, window_hours, truncated)
+    # Every block is sized to fit beside the header, though only the first message
+    # carries it: simpler than sizing the first block differently.
+    blocks = group_by_repo(new, updated, now, MAX_MESSAGE_TEXT_CHARS - len(header))
+    return discord.messages_from_entries(header, blocks, MAX_COMPONENTS_PER_MESSAGE)
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Post a daily digest of contributor pull requests to Discord.")
+    parser.add_argument("--org", help=f"GitHub org to scan (else GITHUB_PRS_ORG, default {DEFAULT_ORG}).")
+    parser.add_argument("--token", help="GitHub token (else GITHUB_PRS_TOKEN).")
+    parser.add_argument("--webhook", help="Discord webhook URL (else GITHUB_PRS_DISCORD_WEBHOOK_URL).")
+    parser.add_argument("--maintainers", default=MAINTAINERS_FILE,
+                        help="Logins to treat as maintainers, one per line.")
+    parser.add_argument("--window-hours", type=int, default=DEFAULT_WINDOW_HOURS,
+                        help=f"How far back a PR must have moved to be considered "
+                             f"(default {DEFAULT_WINDOW_HOURS}).")
+    parser.add_argument("--state", metavar="PATH",
+                        help="Dedup state: without it every PR in the window is new.")
+    parser.add_argument("--state-retention-days", type=int,
+                        default=DEFAULT_RETENTION_DAYS, metavar="N",
+                        help=f"Drop state entries older than this "
+                             f"(default {DEFAULT_RETENTION_DAYS}).")
+    parser.add_argument("--dry-run", action="store_true",
+                        help="Print the Discord payload instead of posting it.")
+    args = parser.parse_args()
+
+    if args.window_hours < 1:
+        sys.exit("--window-hours must be at least 1.")
+
+    org = args.org or os.environ.get("GITHUB_PRS_ORG") or DEFAULT_ORG
+    token = get_env("GITHUB_PRS_TOKEN", args.token)
+    webhook = get_env("GITHUB_PRS_DISCORD_WEBHOOK_URL", args.webhook,
+                      required=not args.dry_run)
+    maintainers = load_maintainers(args.maintainers)
+
+    session = github_session(token)
+    repos = fetch_repos(session, org)
+    items, truncated = search_open_prs(session, org)
+    prs = contributor_prs(items, repos, maintainers)
+
+    now = datetime.now(timezone.utc)
+    state = load_state(args.state)
+    new, changed, unchanged = partition_by_state(
+        in_window(prs, now - timedelta(hours=args.window_hours)), state)
+    print(f"{len(items)} open PRs in {org}, {len(prs)} from contributors across "
+          f"{len(repos)} repos: {len(new)} new, {len(changed)} changed since last "
+          f"reported, {len(unchanged)} unchanged (skipped).")
+
+    messages, coverage = build_messages(new, changed, len(prs), args.window_hours,
+                                        now, truncated)
+    if args.dry_run:
+        print(json.dumps(messages, indent=2, ensure_ascii=False))
+        return
+
+    posted = discord.post_to_discord(requests.Session(),
+                                     discord.components_webhook_url(webhook), messages)
+    # Only what Discord accepted. A PR in a message that never landed stays eligible.
+    if args.state:
+        landed = set().union(*coverage[:posted]) if posted else set()
+        reported = [pr for pr in new + changed if pr_id(pr) in landed]
+        kept, pruned = save_state(args.state, state, reported,
+                                  args.state_retention_days)
+        print(f"State: {len(reported)} recorded, {kept} tracked "
+              f"({pruned} pruned beyond {args.state_retention_days} days).")
+    if posted < len(messages):
+        sys.exit(f"Posted {posted} of {len(messages)} messages.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/github_prs/maintainers.txt b/github_prs/maintainers.txt
new file mode 100644
index 0000000..b9c0ac5
--- /dev/null
+++ b/github_prs/maintainers.txt
@@ -0,0 +1,20 @@
+# GitHub logins whose pull requests the digest does not report: the people already in
+# the review loop. Everyone else opening a PR in session-foundation's repos is a
+# contributor and gets a line.
+#
+# Matched case-insensitively against the PR author's login. Bot accounts need no
+# entry — the digest drops every account GitHub types as a Bot.
+#
+# Neither org membership nor push access is a usable substitute for this list. The org
+# has six members, two of whom are not in the review loop, and push access is held by
+# a dozen more as outside collaborators. Both would report a maintainer's PR as a
+# stranger's, or leave a contractor's PR out of the digest entirely.
+#
+# The list goes stale silently: until a new maintainer is added, their PRs are
+# reported as if they came from an outside contributor. That is the failure mode to
+# watch for when someone joins.
+
+Bilb
+jagerman
+mpretty-cyro
+stfsession
diff --git a/github_prs/requirements.txt b/github_prs/requirements.txt
new file mode 100644
index 0000000..70429a6
--- /dev/null
+++ b/github_prs/requirements.txt
@@ -0,0 +1,2 @@
+# What the deployment runs. The suite is stdlib unittest and imports nothing else.
+requests==2.34.2
diff --git a/github_prs/test_digest.py b/github_prs/test_digest.py
new file mode 100644
index 0000000..0cc2b82
--- /dev/null
+++ b/github_prs/test_digest.py
@@ -0,0 +1,338 @@
+"""
+    python -m unittest discover        # from github_prs/
+"""
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import digest  # noqa: E402
+from shared import discord  # noqa: E402
+
+NOW = datetime(2026, 9, 24, 12, 0, tzinfo=timezone.utc)
+CUTOFF = NOW - timedelta(hours=25)
+
+
+def pr(number=1, login="octocat", repo="session-android", created="2026-09-24T08:00:00Z",
+       updated=None, title="Fix a thing", user_type="User", **extra):
+    item = {
+        "id": 10_000 + number,
+        "number": number,
+        "title": title,
+        "html_url": f"https://github.com/session-foundation/{repo}/pull/{number}",
+        "repository_url": f"https://api.github.com/repos/session-foundation/{repo}",
+        "user": {"login": login, "type": user_type},
+        "created_at": created,
+        "updated_at": updated or created,
+    }
+    item.update(extra)
+    return item
+
+
+class TestMaintainers(unittest.TestCase):
+    def test_comments_and_blanks_are_ignored(self):
+        with mock.patch("builtins.open", mock.mock_open(read_data=
+                "# a comment\n\nBilb\njagerman  # trailing\n")):
+            self.assertEqual(digest.load_maintainers("x"), {"bilb", "jagerman"})
+
+    def test_the_shipped_list_parses_and_is_unique(self):
+        with open(digest.MAINTAINERS_FILE, encoding="utf-8") as handle:
+            lines = [line.split("#", 1)[0].strip() for line in handle]
+        logins = [line for line in lines if line]
+        self.assertTrue(logins)
+        self.assertEqual(len(logins), len(set(login.lower() for login in logins)))
+
+
+class TestSelection(unittest.TestCase):
+    repos = {"session-android", "session-desktop"}
+    maintainers = frozenset({"bilb", "mpretty-cyro"})
+
+    def select(self, items):
+        return digest.contributor_prs(items, self.repos, self.maintainers)
+
+    def test_maintainers_are_dropped_whatever_the_case(self):
+        self.assertEqual(self.select([pr(login="BiLb")]), [])
+
+    def test_bots_are_dropped_on_account_type(self):
+        self.assertEqual(self.select([pr(login="dependabot[bot]", user_type="Bot")]), [])
+
+    def test_repos_outside_the_allowed_set_are_dropped(self):
+        self.assertEqual(self.select([pr(repo="session-pysogs")]), [])
+
+    def test_a_contributor_pr_is_kept(self):
+        self.assertEqual(len(self.select([pr(login="someone")])), 1)
+
+
+class TestWindow(unittest.TestCase):
+    def test_only_prs_that_moved_since_the_cutoff_are_considered(self):
+        fresh = pr(1, updated="2026-09-24T09:00:00Z")
+        stale = pr(2, created="2026-08-01T08:00:00Z", updated="2026-08-02T08:00:00Z")
+        self.assertEqual([p["number"] for p in digest.in_window([fresh, stale], CUTOFF)],
+                         [1])
+
+    def test_the_cutoff_itself_counts_as_inside_the_window(self):
+        edge = pr(1, updated=CUTOFF.strftime("%Y-%m-%dT%H:%M:%SZ"))
+        self.assertEqual(len(digest.in_window([edge], CUTOFF)), 1)
+
+    def test_an_old_pr_that_just_moved_is_in(self):
+        """New means never reported, not recently opened — a year-old PR that picks up
+        a comment is exactly what the digest is for."""
+        old = pr(1, created="2025-01-01T00:00:00Z", updated="2026-09-24T09:00:00Z")
+        self.assertEqual(len(digest.in_window([old], CUTOFF)), 1)
+
+
+class TestPartitionByState(unittest.TestCase):
+    def state(self, *records):
+        return {"version": digest.STATE_VERSION,
+                "seen": {key: {"updated_at": stamp} for key, stamp in records}}
+
+    def test_a_pr_never_reported_is_new(self):
+        new, changed, unchanged = digest.partition_by_state([pr(1)], digest.empty_state())
+        self.assertEqual([p["number"] for p in new], [1])
+        self.assertEqual((changed, unchanged), ([], []))
+
+    def test_a_pr_that_moved_since_it_was_reported_is_changed(self):
+        item = pr(1, updated="2026-09-24T09:00:00Z")
+        new, changed, unchanged = digest.partition_by_state(
+            [item], self.state((digest.pr_id(item), "2026-09-20T09:00:00Z")))
+        self.assertEqual([p["number"] for p in changed], [1])
+        self.assertEqual((new, unchanged), ([], []))
+
+    def test_a_pr_that_has_not_moved_is_dropped(self):
+        item = pr(1, updated="2026-09-24T09:00:00Z")
+        new, changed, unchanged = digest.partition_by_state(
+            [item], self.state((digest.pr_id(item), "2026-09-24T09:00:00Z")))
+        self.assertEqual([p["number"] for p in unchanged], [1])
+        self.assertEqual((new, changed), ([], []))
+
+
+class TestStateFile(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.path = os.path.join(self.directory.name, "sub", "seen.json")
+
+    def save(self, *prs, **kwargs):
+        with contextlib.redirect_stdout(io.StringIO()):
+            return digest.save_state(self.path, digest.empty_state(), list(prs), **kwargs)
+
+    def load(self):
+        with contextlib.redirect_stdout(io.StringIO()):
+            return digest.load_state(self.path)
+
+    def test_a_reported_pr_comes_back_unchanged_next_run(self):
+        item = pr(1, updated="2026-09-24T09:00:00Z")
+        self.save(item)
+        _, _, unchanged = digest.partition_by_state([item], self.load())
+        self.assertEqual(len(unchanged), 1)
+
+    def test_the_same_pr_moved_comes_back_changed(self):
+        self.save(pr(1, updated="2026-09-24T09:00:00Z"))
+        _, changed, _ = digest.partition_by_state(
+            [pr(1, updated="2026-09-24T11:00:00Z")], self.load())
+        self.assertEqual(len(changed), 1)
+
+    def test_the_file_names_the_pr_for_whoever_opens_it(self):
+        self.save(pr(1958, repo="session-desktop"))
+        with open(self.path, encoding="utf-8") as handle:
+            self.assertIn("session-desktop#1958", handle.read())
+
+    def test_no_path_means_no_state_and_no_complaint(self):
+        self.assertEqual(digest.load_state(None), digest.empty_state())
+
+
+class TestAge(unittest.TestCase):
+    def test_coarsens_as_it_grows(self):
+        for delta, expected in [(timedelta(minutes=40), "40m"),
+                                (timedelta(hours=6), "6h"),
+                                (timedelta(hours=47), "47h"),
+                                (timedelta(days=3), "3d"),
+                                (timedelta(days=35), "5w")]:
+            self.assertEqual(digest.age(NOW - delta, NOW), expected)
+
+    def test_a_clock_skewed_future_timestamp_reads_as_zero(self):
+        self.assertEqual(digest.age(NOW + timedelta(minutes=5), NOW), "0m")
+
+
+class TestLines(unittest.TestCase):
+    def test_a_new_pr_line_carries_the_link_author_and_age(self):
+        line = digest.build_pr_line(pr(2151, login="octocat"), NOW, True)
+        self.assertIn(digest.NEW_MARKER, line)
+        self.assertIn("[#2151](https://github.com/session-foundation/session-android/pull/2151)", line)
+        self.assertIn("@octocat", line)
+        self.assertIn("4h", line)
+
+    def test_an_updated_line_ages_from_the_update_not_the_creation(self):
+        item = pr(7, created="2026-09-01T12:00:00Z", updated="2026-09-24T09:00:00Z")
+        line = digest.build_pr_line(item, NOW, False)
+        self.assertIn(digest.UPDATED_MARKER, line)
+        self.assertIn("3h", line)
+
+    def test_drafts_and_comment_counts_are_marked(self):
+        line = digest.build_pr_line(pr(9, draft=True, comments=3), NOW, True)
+        self.assertIn("[draft]", line)
+        self.assertIn("💬3", line)
+
+    def test_a_pr_with_no_comments_says_nothing_about_them(self):
+        self.assertNotIn("💬", digest.build_pr_line(pr(9, comments=0), NOW, True))
+
+    def test_long_titles_are_clipped(self):
+        line = digest.build_pr_line(pr(9, title="x" * 200), NOW, True)
+        self.assertIn("…", line)
+        self.assertLess(len(line), 200)
+
+
+class TestGrouping(unittest.TestCase):
+    def test_the_busiest_repo_comes_first_and_new_leads_its_block(self):
+        new = [pr(1, repo="session-ios"), pr(2, repo="session-android"),
+               pr(3, repo="session-android")]
+        updated = [pr(4, repo="session-android")]
+        blocks = digest.group_by_repo(new, updated, NOW)
+        self.assertTrue(blocks[0][0].startswith("**session-android**"))
+        android = blocks[0][0].splitlines()
+        self.assertEqual(len(android), 4)
+        self.assertEqual(android.count(""), 0)
+        self.assertIn(digest.NEW_MARKER, android[1])
+        self.assertIn(digest.UPDATED_MARKER, android[3])
+
+    def test_each_group_sorts_on_the_age_it_shows(self):
+        new = [pr(1, created="2026-09-01T00:00:00Z", updated="2026-09-24T00:00:00Z"),
+               pr(2, created="2026-09-20T00:00:00Z", updated="2026-09-21T00:00:00Z")]
+        updated = [pr(3, created="2026-01-01T00:00:00Z", updated="2026-09-22T00:00:00Z"),
+                   pr(4, created="2026-01-01T00:00:00Z", updated="2026-09-23T00:00:00Z")]
+        lines = digest.group_by_repo(new, updated, NOW)[0][0].splitlines()
+        self.assertEqual([line.split("[#")[1].split("]")[0] for line in lines[1:]],
+                         ["2", "1", "4", "3"])
+
+    def test_a_repo_appears_once_however_many_prs_it_has(self):
+        blocks = digest.group_by_repo([pr(1), pr(2), pr(3)], [], NOW)
+        self.assertEqual(len(blocks), 1)
+
+    def test_a_repo_that_outgrows_a_message_is_split_under_repeated_headings(self):
+        prs = [pr(n, title="t" * 80) for n in range(60)]
+        blocks = digest.group_by_repo(prs, [], NOW, max_chars=1000)
+        self.assertGreater(len(blocks), 1)
+        for text, ids in blocks:
+            self.assertLessEqual(len(text), 1000)
+            self.assertTrue(text.startswith("**session-android**\n"))
+            self.assertEqual(len(text.splitlines()) - 1, len(ids))
+        self.assertEqual(sum(len(ids) for _, ids in blocks), len(prs))
+        self.assertEqual(set().union(*(ids for _, ids in blocks)),
+                         {digest.pr_id(p) for p in prs})
+
+    def test_a_split_repo_keeps_its_blocks_together_and_first(self):
+        busy = [pr(n, repo="busy", title="t" * 80) for n in range(30)]
+        quiet = [pr(100, repo="quiet")]
+        headings = [text.splitlines()[0]
+                    for text, _ in digest.group_by_repo(busy + quiet, [], NOW, max_chars=1000)]
+        self.assertGreater(len(headings), 2)
+        self.assertEqual(headings[-1], "**quiet**")
+        self.assertEqual(set(headings[:-1]), {"**busy**"})
+
+    def test_a_block_accounts_for_every_pr_it_shows(self):
+        """What reached Discord is recorded per message, so a partial post cannot
+        suppress the PRs that never went out."""
+        prs = [pr(1), pr(2)]
+        _, ids = digest.group_by_repo(prs, [], NOW)[0]
+        self.assertEqual(ids, {digest.pr_id(p) for p in prs})
+
+
+class TestHeader(unittest.TestCase):
+    def test_counts_both_kinds_and_the_backlog(self):
+        header = digest.build_header([pr(1)], [pr(2), pr(3)], 12, 25, False)
+        self.assertIn("**1** new", header)
+        self.assertIn("**2** updated", header)
+        self.assertIn("**12** open from contributors", header)
+        self.assertIn("last 25h", header)
+
+    def test_a_quiet_day_says_so(self):
+        header = digest.build_header([], [], 12, 25, False)
+        self.assertIn("Nothing opened or updated.", header)
+        self.assertIn("**12** open", header)
+
+    def test_a_long_window_reads_in_days(self):
+        self.assertIn("last 3 days", digest.build_header([], [], 0, 72, False))
+        self.assertIn("last 1 day", digest.build_header([], [], 0, 24, False))
+        self.assertIn("last 25h", digest.build_header([], [], 0, 25, False))
+
+    def test_truncation_is_declared(self):
+        self.assertIn("floor", digest.build_header([], [], 1000, 25, True))
+
+
+class TestMessages(unittest.TestCase):
+    def test_a_quiet_day_still_posts_the_header(self):
+        messages, coverage = digest.build_messages([], [], 4, 25, NOW)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(coverage, [set()])
+        blocks = messages[0]["components"][0]["components"]
+        self.assertEqual(len(blocks), 1)
+        self.assertIn("Nothing opened", blocks[0]["content"])
+
+    def test_the_payload_is_components_v2(self):
+        message = digest.build_messages([pr(1)], [], 1, 25, NOW)[0][0]
+        self.assertEqual(message["flags"], discord.COMPONENTS_V2_FLAG)
+        self.assertEqual(message["components"][0]["type"], discord.CONTAINER)
+
+    def test_one_busy_repo_never_exceeds_the_message_budget(self):
+        new = [pr(n, title="y" * 80) for n in range(120)]
+        messages, coverage = digest.build_messages(new, [], 120, 72, NOW)
+        self.assertGreater(len(messages), 1)
+        for message in messages:
+            text = sum(len(c.get("content", ""))
+                       for c in message["components"][0]["components"])
+            self.assertLessEqual(text, discord.MAX_MESSAGE_TEXT_CHARS)
+        self.assertEqual(set().union(*coverage), {digest.pr_id(p) for p in new})
+
+    def test_only_the_first_message_carries_the_header(self):
+        new = [pr(n, repo=f"repo-{n}", title="y" * 80) for n in range(40)]
+        messages, coverage = digest.build_messages(new, [], 40, 25, NOW)
+        self.assertGreater(len(messages), 1)
+        self.assertEqual(set().union(*coverage), {digest.pr_id(p) for p in new})
+        for message in messages[1:]:
+            rendered = json.dumps(message, ensure_ascii=False)
+            self.assertNotIn("Contributor pull requests", rendered)
+            self.assertLess(len(rendered), discord.MAX_MESSAGE_TEXT_CHARS * 2)
+
+
+class TestFetching(unittest.TestCase):
+    def test_forks_archived_and_private_repos_are_left_out(self):
+        page = [{"name": "session-android", "fork": False, "archived": False},
+                {"name": "session-pysogs", "fork": True, "archived": False},
+                {"name": "retired", "fork": False, "archived": True},
+                {"name": "internal", "fork": False, "archived": False, "private": True}]
+        with mock.patch.object(digest, "fetch_json", return_value=page):
+            self.assertEqual(digest.fetch_repos(None, "org"), {"session-android"})
+
+    def test_repo_pagination_follows_full_pages(self):
+        pages = [[{"name": f"r{n}", "fork": False, "archived": False}
+                  for n in range(digest.PER_PAGE)],
+                 [{"name": "last", "fork": False, "archived": False}]]
+        with mock.patch.object(digest, "fetch_json", side_effect=pages):
+            self.assertEqual(len(digest.fetch_repos(None, "org")), digest.PER_PAGE + 1)
+
+    def test_search_stops_when_the_results_run_out(self):
+        payload = {"total_count": 2, "items": [pr(1), pr(2)]}
+        with mock.patch.object(digest, "fetch_json", return_value=payload) as fetch:
+            items, truncated = digest.search_open_prs(None, "org")
+        self.assertEqual(len(items), 2)
+        self.assertFalse(truncated)
+        self.assertEqual(fetch.call_count, 1)
+
+    def test_search_stops_at_the_result_ceiling_rather_than_erroring(self):
+        full = {"total_count": 1500,
+                "items": [pr(n) for n in range(digest.PER_PAGE)]}
+        with mock.patch.object(digest, "fetch_json", return_value=full) as fetch:
+            items, truncated = digest.search_open_prs(None, "org")
+        self.assertEqual(len(items), digest.SEARCH_RESULT_LIMIT)
+        self.assertTrue(truncated)
+        self.assertEqual(fetch.call_count, digest.SEARCH_RESULT_LIMIT // digest.PER_PAGE)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/shared/__init__.py b/shared/__init__.py
new file mode 100644
index 0000000..cd9e60c
--- /dev/null
+++ b/shared/__init__.py
@@ -0,0 +1,6 @@
+"""Helpers shared by the scripts in this repository.
+
+A script inserts the repository root on sys.path and imports from here. Every job
+runs out of the same clone, so no install step is involved; the only dependency
+outside the standard library is `requests`, which every consumer already pins.
+"""
diff --git a/shared/discord.py b/shared/discord.py
new file mode 100644
index 0000000..d9662e1
--- /dev/null
+++ b/shared/discord.py
@@ -0,0 +1,124 @@
+"""Components V2 messages over a plain incoming webhook.
+
+A digest is a Container of Text Displays: one block per entry, so a reader skims
+lines rather than a wall, and each message records which entries it accounts for.
+
+Every component here is non-interactive, which is what lets a plain incoming
+webhook carry it: Discord allows a webhook that no application owns only those.
+Adding an interactive one would need the transport moved to a bot token.
+https://docs.discord.com/developers/components/reference
+"""
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import requests
+
+from shared.retry import request_with_retry
+
+COMPONENTS_V2_FLAG = 1 << 15
+CONTAINER = 17
+TEXT_DISPLAY = 10
+SEPARATOR = 14
+# Discord's ceiling on all the text in one Components V2 message.
+MAX_MESSAGE_TEXT_CHARS = 4000
+
+
+def clip(text, limit):
+    text = (text or "").strip()
+    return text if len(text) <= limit else text[: limit - 1] + "…"
+
+
+def text_display(content):
+    return {"type": TEXT_DISPLAY, "content": content}
+
+
+def separator():
+    return {"type": SEPARATOR}
+
+
+def container_message(blocks):
+    return {"flags": COMPONENTS_V2_FLAG,
+            "components": [{"type": CONTAINER, "components": blocks}]}
+
+
+def chunk_entries(entries, max_items, max_chars=MAX_MESSAGE_TEXT_CHARS, first_used=0):
+    """Group (text, ids) pairs into messages within Discord's budgets.
+
+    Whichever limit binds first splits the message. Entries are separate components
+    rather than joined text, so nothing is spent on the newlines between them.
+
+    `first_used` is what the caller has already spent on the first message before
+    any entry goes in: the header. Without it the header rides on top of a full
+    budget of entries and a busy day's first message goes over.
+
+    An entry longer than the character budget still gets its own message rather
+    than being dropped; callers pre-clip so that should not arise.
+
+    Entries are passed through, not rebuilt, so a caller can still find one by
+    identity inside a chunk.
+    """
+    chunks, current, current_chars = [], [], first_used
+    for entry in entries:
+        text, _ = entry
+        if current and (len(current) >= max_items
+                        or current_chars + len(text) > max_chars):
+            chunks.append(current)
+            current, current_chars = [], 0
+        current.append(entry)
+        current_chars += len(text)
+    if current:
+        chunks.append(current)
+    return chunks
+
+
+def messages_from_entries(header, entries, max_items, max_chars=MAX_MESSAGE_TEXT_CHARS):
+    """Return (messages, coverage): the header leads message one, entries follow.
+
+    coverage[i] is the union of the ids message i carries, so a run that fails
+    partway can record exactly what reached Discord. Message one's header is not
+    credited with any ids; a caller whose header accounts for entries adds them.
+
+    A quiet day still owes the channel its header, so no entries yields one
+    message.
+    """
+    messages, coverage = [], []
+    chunks = chunk_entries(entries, max_items, max_chars, first_used=len(header)) or [[]]
+    for index, chunk in enumerate(chunks):
+        blocks = []
+        if index == 0:
+            blocks.append(text_display(header))
+            if chunk:
+                blocks.append(separator())
+        blocks += [text_display(text) for text, _ in chunk]
+        messages.append(container_message(blocks))
+        coverage.append(set().union(*(ids for _, ids in chunk)) if chunk else set())
+    return messages, coverage
+
+
+def components_webhook_url(webhook_url):
+    """The webhook, told to respect the components field, which it ignores without."""
+    parts = urlsplit(webhook_url)
+    query = dict(parse_qsl(parts.query))
+    query["with_components"] = "true"
+    return urlunsplit(parts._replace(query=urlencode(query)))
+
+
+def post_to_discord(session, url, messages):
+    """POST each message in order; return how many Discord accepted.
+
+    Stops at the first failure and returns the count instead of exiting, so the
+    caller can record what did land before signalling the failure. Letting the
+    exhausted retry propagate would skip that recording, and the messages that
+    landed would be reposted on the next run.
+    """
+    for index, payload in enumerate(messages):
+        try:
+            resp = request_with_retry(session, "POST", url, json=payload)
+        except requests.RequestException as exc:
+            print(f"Discord unreachable on message {index + 1}/{len(messages)} "
+                  f"({exc}).")
+            return index
+        if resp.status_code >= 400:
+            print(f"Discord rejected message {index + 1}/{len(messages)} "
+                  f"({resp.status_code}): {resp.text[:300]}")
+            return index
+    return len(messages)
diff --git a/shared/env.py b/shared/env.py
new file mode 100644
index 0000000..4d6fcbf
--- /dev/null
+++ b/shared/env.py
@@ -0,0 +1,13 @@
+import os
+import sys
+
+
+def get_env(name, cli_value=None, required=True):
+    if cli_value:
+        return cli_value
+    value = os.environ.get(name)
+    if value:
+        return value
+    if required:
+        sys.exit(f"Missing required config: set the {name} environment variable (or pass the matching flag).")
+    return None
diff --git a/shared/retry.py b/shared/retry.py
new file mode 100644
index 0000000..82059f3
--- /dev/null
+++ b/shared/retry.py
@@ -0,0 +1,84 @@
+import email.utils
+import math
+import time
+from datetime import datetime, timezone
+
+import requests
+
+
+def retry_after_seconds(resp, default):
+    """Seconds to wait per the response's rate-limit headers, else `default`.
+
+    Retry-After first, in either form RFC 9110 allows: a delay in seconds or an
+    HTTP-date. GitHub answers a primary rate limit with x-ratelimit-reset as an
+    epoch second and no Retry-After at all, so that is read when the header is
+    absent.
+
+    Anything unparseable falls back rather than crashing the run. Negative, NaN
+    and infinite values fall back too: time.sleep() rejects the first two
+    outright, so a hostile or buggy proxy sending `Retry-After: -30` would
+    otherwise take the run down with a ValueError.
+    """
+    raw = resp.headers.get("retry-after")
+    if raw is None:
+        reset = resp.headers.get("x-ratelimit-reset")
+        if reset is None:
+            return default
+        try:
+            return max(0.0, float(reset) - time.time())
+        except (TypeError, ValueError):
+            return default
+    try:
+        seconds = float(raw)
+    except (TypeError, ValueError):
+        seconds = _seconds_until_http_date(raw)
+        if seconds is None:
+            return default
+    if not math.isfinite(seconds) or seconds < 0:
+        return default
+    return seconds
+
+
+def _seconds_until_http_date(value):
+    try:
+        when = email.utils.parsedate_to_datetime(value)
+    except (TypeError, ValueError):
+        return None
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=timezone.utc)
+    return (when - datetime.now(timezone.utc)).total_seconds()
+
+
+def request_with_retry(session, method, url, attempts=6, timeout=30, **kwargs):
+    """GET/POST with backoff on 429 and 5xx. Returns the response, whatever its status.
+
+    Lower `attempts` for calls whose result is nice-to-have: the full budget can
+    burn ~60s of backoff, which is not worth spending on optional data.
+    """
+    if attempts <= 0:
+        raise ValueError("attempts must be at least 1")
+
+    delay = 1.0
+    last_exc = None
+    resp = None
+    for attempt in range(attempts):
+        final = attempt == attempts - 1
+        try:
+            resp = session.request(method, url, timeout=timeout, **kwargs)
+        except requests.RequestException as exc:
+            last_exc = exc
+            if final:
+                break
+            time.sleep(delay)
+            delay = min(delay * 2, 30)
+            continue
+        if resp.status_code == 429 or resp.status_code >= 500:
+            if final:
+                break
+            time.sleep(min(retry_after_seconds(resp, delay), 60))
+            delay = min(delay * 2, 30)
+            continue
+        return resp
+    if last_exc:
+        raise last_exc
+    return resp
diff --git a/shared/state.py b/shared/state.py
new file mode 100644
index 0000000..90a8bb5
--- /dev/null
+++ b/shared/state.py
@@ -0,0 +1,73 @@
+"""The dedup state a digest keeps between runs: what it reported, and when.
+
+Every failure to read it is a cache miss rather than an error. Losing the file
+re-reports one window, which is noisy and never wrong, and that is what makes it
+a cache rather than something to back up.
+"""
+import json
+import os
+from datetime import datetime, timedelta, timezone
+
+STAMP = "%Y-%m-%dT%H:%M:%SZ"
+
+
+def empty_state(version):
+    return {"version": version, "seen": {}}
+
+
+def load_state(path, version, noun):
+    """The state at `path`, or an empty one for any file that cannot be trusted.
+
+    `noun` names what the caller dedups, for the log line.
+    """
+    if not path:
+        return empty_state(version)
+    if not os.path.exists(path):
+        print(f"No state file at {path}; treating every {noun} in the window as new.")
+        return empty_state(version)
+    try:
+        with open(path, encoding="utf-8") as handle:
+            data = json.load(handle)
+    except (OSError, json.JSONDecodeError) as exc:
+        print(f"Note: unreadable state file {path} ({exc}); treating every {noun} as new.")
+        return empty_state(version)
+    if not isinstance(data, dict) or not isinstance(data.get("seen"), dict):
+        print(f"Note: unexpected shape in {path}; treating every {noun} as new.")
+        return empty_state(version)
+    # A file written by another schema version cannot be trusted field by field.
+    if data.get("version") != version:
+        print(f"Note: {path} is version {data.get('version')!r}, expected {version}; "
+              f"treating every {noun} as new.")
+        return empty_state(version)
+    print(f"Loaded state for {len(data['seen'])} previously reported {noun}s.")
+    return data
+
+
+def save_state(path, state, records, retention_days, version):
+    """Merge `records` ({key: fields}) into the state as reported now, prune entries
+    older than `retention_days`, write atomically. Returns (kept, pruned)."""
+    now = datetime.now(timezone.utc)
+    stamp = now.strftime(STAMP)
+    seen = dict(state.get("seen", {}))
+    for key, fields in records.items():
+        seen[key] = {**fields, "last_reported": stamp}
+
+    cutoff = now - timedelta(days=retention_days)
+    kept = {}
+    for key, record in seen.items():
+        try:
+            last = datetime.strptime(record.get("last_reported", ""), STAMP) \
+                .replace(tzinfo=timezone.utc)
+        except (TypeError, ValueError):
+            continue  # malformed entry: drop it rather than keep it forever
+        if last >= cutoff:
+            kept[key] = record
+
+    directory = os.path.dirname(path)
+    if directory:
+        os.makedirs(directory, exist_ok=True)
+    temporary = f"{path}.tmp"
+    with open(temporary, "w", encoding="utf-8") as handle:
+        json.dump({"version": version, "updated_at": stamp, "seen": kept}, handle, indent=2)
+    os.replace(temporary, path)  # atomic: a crash mid-write cannot corrupt the state
+    return len(kept), len(seen) - len(kept)
diff --git a/shared/test_discord.py b/shared/test_discord.py
new file mode 100644
index 0000000..41330c4
--- /dev/null
+++ b/shared/test_discord.py
@@ -0,0 +1,130 @@
+import contextlib
+import io
+import os
+import sys
+import unittest
+from urllib.parse import parse_qsl, urlsplit
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from shared import discord  # noqa: E402
+from shared.testing import FakeResponse, FakeSession, NoSleep  # noqa: E402
+
+
+class TestComponentsWebhookUrl(unittest.TestCase):
+    """Discord ignores `components` on a webhook post without this param, and a
+    digest is nothing but components."""
+
+    def test_adds_the_param(self):
+        self.assertEqual(
+            discord.components_webhook_url("https://discord.com/api/webhooks/1/tok"),
+            "https://discord.com/api/webhooks/1/tok?with_components=true")
+
+    def test_keeps_an_existing_query(self):
+        url = discord.components_webhook_url(
+            "https://discord.com/api/webhooks/1/tok?thread_id=42")
+        self.assertEqual(dict(parse_qsl(urlsplit(url).query)),
+                         {"thread_id": "42", "with_components": "true"})
+
+    def test_does_not_duplicate_the_param(self):
+        once = discord.components_webhook_url("https://discord.com/api/webhooks/1/tok")
+        self.assertEqual(discord.components_webhook_url(once), once)
+
+
+class TestPostToDiscord(unittest.TestCase):
+    """Returns the accepted count rather than exiting, so a caller can record exactly
+    what landed before signalling the failure."""
+
+    def post(self, session, messages):
+        with contextlib.redirect_stdout(io.StringIO()):
+            return discord.post_to_discord(session, "https://hook", messages)
+
+    def test_all_accepted(self):
+        session = FakeSession([FakeResponse({}, status_code=204)] * 3)
+        self.assertEqual(self.post(session, [{}, {}, {}]), 3)
+        self.assertEqual(len(session.calls), 3)
+
+    def test_stops_at_the_first_failure_and_reports_the_prefix(self):
+        session = FakeSession([
+            FakeResponse({}, status_code=204),
+            FakeResponse({}, status_code=400),
+        ])
+        self.assertEqual(self.post(session, [{}, {}, {}]), 1)
+
+    def test_does_not_post_after_a_failure(self):
+        session = FakeSession([FakeResponse({}, status_code=404)])
+        self.post(session, [{}, {}, {}])
+        self.assertEqual(len(session.calls), 1)
+
+    def test_first_message_failing_reports_zero(self):
+        session = FakeSession([FakeResponse({}, status_code=500)] * 6)
+        with NoSleep():
+            self.assertEqual(self.post(session, [{}]), 0)
+
+    def test_an_unreachable_webhook_reports_the_prefix_rather_than_raising(self):
+        import requests
+        session = FakeSession([FakeResponse({}, status_code=204)]
+                              + [requests.ConnectionError("down")] * 6)
+        with NoSleep():
+            self.assertEqual(self.post(session, [{}, {}]), 1)
+
+    def test_no_messages_is_zero(self):
+        self.assertEqual(self.post(FakeSession([]), []), 0)
+
+
+class TestChunkEntries(unittest.TestCase):
+    def test_splits_on_the_entry_count(self):
+        entries = [("x", {i}) for i in range(11)]
+        self.assertEqual([len(c) for c in discord.chunk_entries(entries, 10)], [10, 1])
+
+    def test_splits_on_the_character_budget(self):
+        entries = [("x" * 2500, {1}), ("y" * 2500, {2})]
+        self.assertEqual(len(discord.chunk_entries(entries, 10)), 2)
+
+    def test_the_header_is_charged_to_the_first_message_only(self):
+        entries = [("x" * 1900, {1}), ("x" * 1900, {2})]
+        self.assertEqual(len(discord.chunk_entries(entries, 10, first_used=0)), 1)
+        self.assertEqual(len(discord.chunk_entries(entries, 10, first_used=1000)), 2)
+
+    def test_an_oversized_entry_still_gets_a_message(self):
+        entries = [("x" * (discord.MAX_MESSAGE_TEXT_CHARS + 100), {1})]
+        self.assertEqual(len(discord.chunk_entries(entries, 10)), 1)
+
+    def test_entries_come_back_by_identity(self):
+        entry = ("x", {1})
+        self.assertIs(discord.chunk_entries([entry], 10)[0][0], entry)
+
+
+class TestMessagesFromEntries(unittest.TestCase):
+    def texts(self, message):
+        return [c["content"] for c in message["components"][0]["components"]
+                if c["type"] == discord.TEXT_DISPLAY]
+
+    def test_no_entries_still_yields_the_header(self):
+        messages, coverage = discord.messages_from_entries("header", [], 10)
+        self.assertEqual(len(messages), 1)
+        self.assertEqual(self.texts(messages[0]), ["header"])
+        self.assertEqual(coverage, [set()])
+        self.assertEqual(len(messages[0]["components"][0]["components"]), 1)  # no separator
+
+    def test_the_payload_is_a_components_v2_container(self):
+        message = discord.messages_from_entries("h", [("a", {1})], 10)[0][0]
+        self.assertEqual(message["flags"], discord.COMPONENTS_V2_FLAG)
+        self.assertEqual(message["components"][0]["type"], discord.CONTAINER)
+        types = [c["type"] for c in message["components"][0]["components"]]
+        self.assertEqual(types, [discord.TEXT_DISPLAY, discord.SEPARATOR, discord.TEXT_DISPLAY])
+
+    def test_only_the_first_message_carries_the_header(self):
+        entries = [(f"e{i}", {i}) for i in range(3)]
+        messages, coverage = discord.messages_from_entries("header", entries, 2)
+        self.assertEqual(self.texts(messages[0]), ["header", "e0", "e1"])
+        self.assertEqual(self.texts(messages[1]), ["e2"])
+        self.assertEqual(coverage, [{0, 1}, {2}])
+
+    def test_the_header_counts_against_the_first_budget(self):
+        entries = [("x" * 2000, {1}), ("y" * 1500, {2})]
+        messages, _ = discord.messages_from_entries("h" * 1000, entries, 10)
+        self.assertEqual(len(messages), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/shared/test_retry.py b/shared/test_retry.py
new file mode 100644
index 0000000..f1fa3df
--- /dev/null
+++ b/shared/test_retry.py
@@ -0,0 +1,171 @@
+"""
+    cd shared && python -m unittest discover
+"""
+import os
+import sys
+import time
+import unittest
+from email.utils import format_datetime
+from datetime import datetime, timedelta, timezone
+
+import requests
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from shared import retry  # noqa: E402
+from shared.testing import FakeResponse, FakeSession, NoSleep  # noqa: E402
+
+
+class TestRequestWithRetry(unittest.TestCase):
+    def test_returns_the_first_success_without_retrying(self):
+        session = FakeSession([FakeResponse({"ok": True})])
+        resp = retry.request_with_retry(session, "GET", "https://x")
+        self.assertEqual(resp.json(), {"ok": True})
+        self.assertEqual(len(session.calls), 1)
+
+    def test_retries_a_server_error_then_succeeds(self):
+        session = FakeSession([
+            FakeResponse({}, status_code=500),
+            FakeResponse({"ok": True}),
+        ])
+        resp = retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(resp.json(), {"ok": True})
+        self.assertEqual(len(session.calls), 2)
+
+    def test_does_not_retry_a_client_error(self):
+        session = FakeSession([FakeResponse({}, status_code=404)])
+        resp = retry.request_with_retry(session, "GET", "https://x")
+        self.assertEqual(resp.status_code, 404)
+        self.assertEqual(len(session.calls), 1)
+
+    def test_gives_up_after_the_attempt_budget(self):
+        session = FakeSession([FakeResponse({}, status_code=503)] * 3)
+        resp = retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(resp.status_code, 503)
+        self.assertEqual(len(session.calls), 3)
+
+    def test_transport_failures_retry_then_raise_when_exhausted(self):
+        session = FakeSession([requests.ConnectionError("boom")] * 3)
+        with NoSleep():
+            with self.assertRaises(requests.ConnectionError):
+                retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(len(session.calls), 3)
+
+    def test_a_transport_failure_can_recover_on_a_later_attempt(self):
+        session = FakeSession([requests.Timeout("slow"), FakeResponse({"ok": True})])
+        with NoSleep():
+            resp = retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(resp.json(), {"ok": True})
+        self.assertEqual(len(session.calls), 2)
+
+    def test_no_sleep_after_the_final_attempt(self):
+        """Sleeping after the last try only delays the caller — nothing follows it."""
+        session = FakeSession([FakeResponse({}, status_code=503)] * 3)
+        with NoSleep() as clock:
+            retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(len(clock.slept), 2)  # 3 attempts, 2 gaps
+
+    def test_numeric_retry_after_is_honoured(self):
+        session = FakeSession([
+            FakeResponse({}, status_code=429, retry_after="7"),
+            FakeResponse({"ok": True}),
+        ])
+        with NoSleep() as clock:
+            retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(clock.slept, [7.0])
+
+    def test_retry_after_is_capped(self):
+        session = FakeSession([
+            FakeResponse({}, status_code=429, retry_after="9999"),
+            FakeResponse({"ok": True}),
+        ])
+        with NoSleep() as clock:
+            retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(clock.slept, [60])
+
+    def test_an_http_date_retry_after_is_honoured(self):
+        """RFC 9110 allows an HTTP-date here as well as a count of seconds."""
+        soon = format_datetime(datetime.now(timezone.utc) + timedelta(seconds=20), usegmt=True)
+        session = FakeSession([
+            FakeResponse({}, status_code=503, retry_after=soon),
+            FakeResponse({"ok": True}),
+        ])
+        with NoSleep() as clock:
+            resp = retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(resp.json(), {"ok": True})
+        self.assertEqual(len(clock.slept), 1)
+        self.assertTrue(15 <= clock.slept[0] <= 21, clock.slept)
+
+    def test_the_timeout_is_forwarded_and_defaults_to_thirty_seconds(self):
+        session = FakeSession([FakeResponse({}), FakeResponse({})])
+        retry.request_with_retry(session, "GET", "https://x")
+        retry.request_with_retry(session, "GET", "https://x", timeout=60)
+        self.assertEqual([kw["timeout"] for _, _, kw in session.calls], [30, 60])
+
+    def test_zero_attempts_is_rejected_rather_than_unbound(self):
+        with self.assertRaises(ValueError):
+            retry.request_with_retry(FakeSession([]), "GET", "https://x", attempts=0)
+
+    def test_a_negative_retry_after_does_not_crash_a_real_retry_loop(self):
+        session = FakeSession([
+            FakeResponse({}, status_code=503, retry_after="-30"),
+            FakeResponse({"ok": True}),
+        ])
+        with NoSleep() as clock:
+            resp = retry.request_with_retry(session, "GET", "https://x", attempts=3)
+        self.assertEqual(resp.json(), {"ok": True})
+        self.assertTrue(all(s >= 0 for s in clock.slept), clock.slept)
+
+
+class TestRetryAfterSeconds(unittest.TestCase):
+    def seconds(self, headers, default=4.0):
+        resp = FakeResponse({}, retry_after=None)
+        resp.headers = headers
+        return retry.retry_after_seconds(resp, default)
+
+    def test_missing_header_uses_the_default(self):
+        self.assertEqual(self.seconds({}), 4.0)
+
+    def test_numeric_header_wins(self):
+        self.assertEqual(self.seconds({"retry-after": "12"}), 12.0)
+
+    def test_unparseable_header_uses_the_default(self):
+        for raw in ("", "soon", "12s", "Wed, 32 Oct 2026 07:28:00 GMT"):
+            self.assertEqual(self.seconds({"retry-after": raw}), 4.0, msg=f"retry-after={raw!r}")
+
+    def test_an_http_date_in_the_past_uses_the_default(self):
+        """A date already gone means a negative wait, which time.sleep() rejects."""
+        self.assertEqual(self.seconds({"retry-after": "Wed, 21 Oct 2015 07:28:00 GMT"}), 4.0)
+
+    def test_an_http_date_is_the_seconds_until_it(self):
+        soon = format_datetime(datetime.now(timezone.utc) + timedelta(seconds=30), usegmt=True)
+        self.assertTrue(25 <= self.seconds({"retry-after": soon}) <= 31)
+
+    def test_negative_and_non_finite_values_use_the_default(self):
+        """time.sleep() rejects a negative or NaN duration, so passing one through
+        would crash the run on a hostile or buggy Retry-After header."""
+        for raw in ("-30", "-0.5", "nan", "inf", "-inf"):
+            self.assertEqual(self.seconds({"retry-after": raw}), 4.0, msg=f"retry-after={raw!r}")
+
+    def test_zero_is_honoured_rather_than_replaced(self):
+        """Zero is a valid instruction to retry immediately, not a missing value."""
+        self.assertEqual(self.seconds({"retry-after": "0"}), 0.0)
+
+    def test_a_primary_rate_limit_falls_back_to_the_reset_epoch(self):
+        """GitHub's primary limit sends x-ratelimit-reset and no Retry-After."""
+        reset = str(int(time.time()) + 30)
+        seconds = self.seconds({"x-ratelimit-reset": reset})
+        self.assertTrue(25 <= seconds <= 31, seconds)
+
+    def test_a_reset_epoch_in_the_past_means_no_wait(self):
+        self.assertEqual(self.seconds({"x-ratelimit-reset": "1"}), 0.0)
+
+    def test_retry_after_beats_the_reset_epoch(self):
+        reset = str(int(time.time()) + 3000)
+        self.assertEqual(self.seconds({"retry-after": "5", "x-ratelimit-reset": reset}), 5.0)
+
+    def test_an_unparseable_reset_epoch_uses_the_default(self):
+        self.assertEqual(self.seconds({"x-ratelimit-reset": "?"}), 4.0)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/shared/test_state.py b/shared/test_state.py
new file mode 100644
index 0000000..1097085
--- /dev/null
+++ b/shared/test_state.py
@@ -0,0 +1,85 @@
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from shared import state as dedup  # noqa: E402
+
+VERSION = 7
+
+
+class TestStateFile(unittest.TestCase):
+    def setUp(self):
+        self.directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self.directory.cleanup)
+        self.path = os.path.join(self.directory.name, "sub", "seen.json")
+
+    def load(self, path=None):
+        with contextlib.redirect_stdout(io.StringIO()):
+            return dedup.load_state(path or self.path, VERSION, "thing")
+
+    def save(self, records, state=None, retention_days=30):
+        with contextlib.redirect_stdout(io.StringIO()):
+            return dedup.save_state(self.path, state or dedup.empty_state(VERSION),
+                                    records, retention_days, VERSION)
+
+    def test_a_record_comes_back_with_its_fields_and_a_stamp(self):
+        self.save({"1": {"updated_at": "A"}})
+        record = self.load()["seen"]["1"]
+        self.assertEqual(record["updated_at"], "A")
+        self.assertIn("last_reported", record)
+
+    def test_a_later_save_replaces_the_record(self):
+        self.save({"1": {"updated_at": "A"}})
+        self.save({"1": {"updated_at": "B"}}, state=self.load())
+        self.assertEqual(self.load()["seen"]["1"]["updated_at"], "B")
+
+    def test_no_path_means_no_state_and_no_complaint(self):
+        self.assertEqual(dedup.load_state(None, VERSION, "thing"), dedup.empty_state(VERSION))
+
+    def test_a_missing_file_is_a_cache_miss(self):
+        self.assertEqual(self.load(), dedup.empty_state(VERSION))
+
+    def test_every_unreadable_state_is_a_cache_miss_not_an_error(self):
+        """Losing it re-reports the window once. Failing the run instead would mean a
+        corrupt cache stops the digest entirely."""
+        path = os.path.join(self.directory.name, "flat.json")
+        for content in ("{ not json", '{"version": %d}' % VERSION, "[]",
+                        '{"version": 99, "seen": {}}', '{"seen": []}'):
+            with self.subTest(content=content):
+                with open(path, "w", encoding="utf-8") as handle:
+                    handle.write(content)
+                self.assertEqual(self.load(path), dedup.empty_state(VERSION))
+
+    def test_the_right_version_is_loaded(self):
+        path = os.path.join(self.directory.name, "flat.json")
+        with open(path, "w", encoding="utf-8") as handle:
+            json.dump({"version": VERSION, "seen": {"1": {"updated_at": "A"}}}, handle)
+        self.assertEqual(list(self.load(path)["seen"]), ["1"])
+
+    def test_entries_are_pruned_past_the_retention(self):
+        self.save({"1": {}})
+        state = self.load()
+        for record in state["seen"].values():
+            record["last_reported"] = "2026-01-01T00:00:00Z"
+        self.assertEqual(self.save({}, state=state), (0, 1))
+
+    def test_a_fresh_entry_is_kept(self):
+        self.assertEqual(self.save({"1": {}, "2": {}}), (2, 0))
+
+    def test_a_malformed_entry_is_dropped_rather_than_kept_forever(self):
+        state = {"version": VERSION, "seen": {"1": {"last_reported": "never"}}}
+        kept, _ = self.save({}, state=state)
+        self.assertEqual(kept, 0)
+
+    def test_the_directory_is_created_and_the_write_is_atomic(self):
+        self.save({"1": {}})
+        self.assertEqual(sorted(os.listdir(os.path.dirname(self.path))), ["seen.json"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/shared/testing.py b/shared/testing.py
new file mode 100644
index 0000000..c5e3677
--- /dev/null
+++ b/shared/testing.py
@@ -0,0 +1,88 @@
+"""Fakes for the tests of every script that talks HTTP through `shared`."""
+import json
+import time
+
+import requests
+
+
+class FakeResponse:
+    # retry-after: 0 keeps the retry tests instant instead of sleeping through
+    # the real backoff, and exercises the header-honoring path while it's at it.
+    def __init__(self, payload, status_code=200, retry_after="0"):
+        self._payload = payload
+        self.status_code = status_code
+        self.headers = {"retry-after": retry_after}
+        self.text = json.dumps(payload)
+
+    def json(self):
+        return self._payload
+
+
+class NonJsonResponse(FakeResponse):
+    """A 200 whose body isn't JSON — a proxy error page, say."""
+
+    def __init__(self):
+        super().__init__({})
+        self.text = "<html>maintenance</html>"
+
+    def json(self):
+        raise requests.exceptions.JSONDecodeError("Expecting value", self.text, 0)
+
+
+class FakeSession:
+    """Returns queued responses in order and records the requests made.
+
+    A queued Exception is raised instead of returned, so transport failures can be
+    exercised alongside HTTP status codes.
+    """
+
+    def __init__(self, responses):
+        self._responses = list(responses)
+        self.calls = []
+
+    def request(self, method, url, **kwargs):
+        self.calls.append((method, url, kwargs))
+        item = self._responses.pop(0)
+        if isinstance(item, Exception):
+            raise item
+        return item
+
+
+class Patched:
+    """Swap module attributes for the duration of a block, then put them back."""
+
+    def __init__(self, module, **attrs):
+        self.module, self.attrs, self.saved = module, attrs, {}
+
+    def __enter__(self):
+        for name, value in self.attrs.items():
+            try:
+                self.saved[name] = getattr(self.module, name)
+            except AttributeError:
+                # Roll back what is already swapped. Without this, a typo'd or
+                # since-removed attribute leaves earlier patches applied and
+                # __exit__ never runs — every later test in the file then fails
+                # against a module the failing test quietly rewrote.
+                self.__exit__()
+                raise
+            setattr(self.module, name, value)
+        return self
+
+    def __exit__(self, *exc):
+        for name, value in self.saved.items():
+            setattr(self.module, name, value)
+        return False
+
+
+class NoSleep:
+    """Patch out time.sleep so retry tests assert on delays without waiting."""
+
+    def __enter__(self):
+        self.slept = []
+        self._real = time.sleep
+        time.sleep = self.slept.append
+        return self
+
+    def __exit__(self, *exc):
+        time.sleep = self._real
+        return False
diff --git a/zendesk_triage/test_triage.py b/zendesk_triage/test_triage.py
index 1eb854c..f9e3151 100644
--- a/zendesk_triage/test_triage.py
+++ b/zendesk_triage/test_triage.py
@@ -18,12 +18,15 @@
 import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
-from urllib.parse import parse_qsl, urlsplit
 
 import requests
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 import triage  # noqa: E402  (needs the path insert above)
+from shared import discord  # noqa: E402
+from shared.testing import (  # noqa: E402
+    FakeResponse, FakeSession, NoSleep, NonJsonResponse, Patched)
 
 
 STAMP = "%Y-%m-%dT%H:%M:%SZ"
@@ -74,7 +77,7 @@ def text_displays(node):
         return [t for item in node for t in text_displays(item)]
     if not isinstance(node, dict):
         return []
-    found = [node["content"]] if node.get("type") == triage.TEXT_DISPLAY else []
+    found = [node["content"]] if node.get("type") == discord.TEXT_DISPLAY else []
     for key in ("components", "accessory"):
         found += text_displays(node.get(key))
     return found
@@ -99,94 +102,6 @@ def all_text(message):
     return sum(len(t) for t in text_displays(message["components"]))
 
 
-class FakeResponse:
-    # retry-after: 0 keeps the retry tests instant instead of sleeping through
-    # the real backoff, and exercises the header-honoring path while it's at it.
-    def __init__(self, payload, status_code=200, retry_after="0"):
-        self._payload = payload
-        self.status_code = status_code
-        self.headers = {"retry-after": retry_after}
-        self.text = json.dumps(payload)
-
-    def json(self):
-        return self._payload
-
-
-class NonJsonResponse(FakeResponse):
-    """A 200 whose body isn't JSON — a proxy error page, say."""
-
-    def __init__(self):
-        super().__init__({})
-        self.text = "<html>maintenance</html>"
-
-    def json(self):
-        raise requests.exceptions.JSONDecodeError("Expecting value", self.text, 0)
-
-
-class FakeSession:
-    """Returns queued responses in order and records the requests made.
-
-    A queued Exception is raised instead of returned, so transport failures can be
-    exercised alongside HTTP status codes.
-    """
-
-    def __init__(self, responses):
-        self._responses = list(responses)
-        self.calls = []
-
-    def request(self, method, url, **kwargs):
-        self.calls.append((method, url, kwargs))
-        item = self._responses.pop(0)
-        if isinstance(item, Exception):
-            raise item
-        return item
-
-
-class Patched:
-    """Swap module attributes for the duration of a block, then put them back.
-
-    Lives here rather than in each test file because test_relay and test_reply both
-    need it, and two copies of a helper that restores state is two chances for one of
-    them to stop doing so.
-    """
-
-    def __init__(self, module, **attrs):
-        self.module, self.attrs, self.saved = module, attrs, {}
-
-    def __enter__(self):
-        for name, value in self.attrs.items():
-            try:
-                self.saved[name] = getattr(self.module, name)
-            except AttributeError:
-                # Roll back what is already swapped. Without this, a typo'd or
-                # since-removed attribute leaves earlier patches applied and
-                # __exit__ never runs — every later test in the file then fails
-                # against a module the failing test quietly rewrote.
-                self.__exit__()
-                raise
-            setattr(self.module, name, value)
-        return self
-
-    def __exit__(self, *exc):
-        for name, value in self.saved.items():
-            setattr(self.module, name, value)
-        return False
-
-
-class NoSleep:
-    """Patch out time.sleep so retry tests assert on delays without waiting."""
-
-    def __enter__(self):
-        self.slept = []
-        self._real = triage.time.sleep
-        triage.time.sleep = self.slept.append
-        return self
-
-    def __exit__(self, *exc):
-        triage.time.sleep = self._real
-        return False
-
-
 # ---- Window construction ---------------------------------------------------
 
 
@@ -687,10 +602,10 @@ def test_messages_are_components_v2(self):
         """content and embeds stop working once the flag is set, so a payload still
         carrying either would be silently rendered empty."""
         for message in build_messages([finding(1)], "acme"):
-            self.assertEqual(message["flags"], triage.COMPONENTS_V2_FLAG)
+            self.assertEqual(message["flags"], discord.COMPONENTS_V2_FLAG)
             self.assertNotIn("content", message)
             self.assertNotIn("embeds", message)
-            self.assertEqual(message["components"][0]["type"], triage.CONTAINER)
+            self.assertEqual(message["components"][0]["type"], discord.CONTAINER)
 
     def test_each_ticket_gets_its_own_block(self):
         """One Text Display per ticket, so a reader skims lines rather than a wall."""
@@ -1178,7 +1093,7 @@ def test_a_full_message_stays_inside_the_character_ceiling(self):
         component rather than three."""
         findings = [self.fat(i) for i in range(triage.MAX_HIGHLIGHTS)]
         for message in build_messages(findings, "acme"):
-            self.assertLessEqual(all_text(message), triage.MAX_MESSAGE_TEXT_CHARS)
+            self.assertLessEqual(all_text(message), discord.MAX_MESSAGE_TEXT_CHARS)
 
     def test_the_card_count_splits_a_busy_day(self):
         """Short lines never reach the character budget, so without the component
@@ -1196,12 +1111,13 @@ def test_chunking_splits_on_whichever_limit_binds_first(self):
         # Together these overrun MAX_COMPONENT_CHARS, so characters bind before the
         # entry count does.
         entries = [("x" * 2100, {1}), ("y" * 2100, {2})]
-        self.assertEqual(len(triage.chunk_entries(entries)), 2)   # characters
+        self.assertEqual(len(discord.chunk_entries(entries, triage.MAX_ENTRIES_PER_MESSAGE)), 2)   # characters
         lean = [("x", {i}) for i in range(triage.MAX_ENTRIES_PER_MESSAGE + 1)]
-        self.assertEqual(len(triage.chunk_entries(lean)), 2)      # card count
+        self.assertEqual(len(discord.chunk_entries(lean, triage.MAX_ENTRIES_PER_MESSAGE)), 2)      # card count
 
     def test_an_oversized_entry_still_gets_a_message(self):
-        chunks = triage.chunk_entries([("x" * (triage.MAX_COMPONENT_CHARS + 50), {1})])
+        chunks = discord.chunk_entries([("x" * (triage.MAX_COMPONENT_CHARS + 50), {1})],
+                                       triage.MAX_ENTRIES_PER_MESSAGE)
         self.assertEqual(len(chunks), 1)
 
     def test_a_full_digest_of_fat_lines_and_abuse_reports_stays_within_the_limit(self):
@@ -1284,56 +1200,6 @@ def test_abuse_reports_dropped_from_the_link_list_are_still_covered(self):
         self.assertIn("more", digest_text(messages))
 
 
-class TestDigestWebhookUrl(unittest.TestCase):
-    """Discord ignores `components` on a webhook post without this param, and the
-    digest is nothing but components."""
-
-    def test_adds_the_param(self):
-        self.assertEqual(
-            triage.digest_webhook_url("https://discord.com/api/webhooks/1/tok"),
-            "https://discord.com/api/webhooks/1/tok?with_components=true")
-
-    def test_keeps_an_existing_query(self):
-        url = triage.digest_webhook_url(
-            "https://discord.com/api/webhooks/1/tok?thread_id=42")
-        self.assertEqual(dict(parse_qsl(urlsplit(url).query)),
-                         {"thread_id": "42", "with_components": "true"})
-
-    def test_does_not_duplicate_the_param(self):
-        once = triage.digest_webhook_url("https://discord.com/api/webhooks/1/tok")
-        self.assertEqual(triage.digest_webhook_url(once), once)
-
-
-class TestPostToDiscord(unittest.TestCase):
-    """Returns the accepted count rather than exiting, so main can record exactly the
-    tickets that landed before signalling the failure."""
-
-    def test_all_accepted(self):
-        session = FakeSession([FakeResponse({}, status_code=204)] * 3)
-        self.assertEqual(triage.post_to_discord(session, "https://hook", [{}, {}, {}]), 3)
-        self.assertEqual(len(session.calls), 3)
-
-    def test_stops_at_the_first_failure_and_reports_the_prefix(self):
-        session = FakeSession([
-            FakeResponse({}, status_code=204),
-            FakeResponse({}, status_code=400),
-        ])
-        self.assertEqual(triage.post_to_discord(session, "https://hook", [{}, {}, {}]), 1)
-
-    def test_does_not_post_after_a_failure(self):
-        session = FakeSession([FakeResponse({}, status_code=404)])
-        triage.post_to_discord(session, "https://hook", [{}, {}, {}])
-        self.assertEqual(len(session.calls), 1)
-
-    def test_first_message_failing_reports_zero(self):
-        session = FakeSession([FakeResponse({}, status_code=500)] * 6)
-        with NoSleep():
-            self.assertEqual(triage.post_to_discord(session, "https://hook", [{}]), 0)
-
-    def test_no_messages_is_zero(self):
-        self.assertEqual(triage.post_to_discord(FakeSession([]), "https://hook", []), 0)
-
-
 class TestSelectHighlights(unittest.TestCase):
     def test_splits_at_the_display_cap(self):
         findings = [finding(i, priority_rank=i) for i in range(triage.MAX_HIGHLIGHTS + 4)]
@@ -1702,122 +1568,6 @@ def test_a_non_json_body_is_non_fatal(self):
         self.assertIsNone(triage.fetch_total_unsolved(session, "acme"))
 
 
-class TestRequestWithRetry(unittest.TestCase):
-    def test_returns_the_first_success_without_retrying(self):
-        session = FakeSession([FakeResponse({"ok": True})])
-        resp = triage.request_with_retry(session, "GET", "https://x")
-        self.assertEqual(resp.json(), {"ok": True})
-        self.assertEqual(len(session.calls), 1)
-
-    def test_retries_a_server_error_then_succeeds(self):
-        session = FakeSession([
-            FakeResponse({}, status_code=500),
-            FakeResponse({"ok": True}),
-        ])
-        resp = triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(resp.json(), {"ok": True})
-        self.assertEqual(len(session.calls), 2)
-
-    def test_does_not_retry_a_client_error(self):
-        session = FakeSession([FakeResponse({}, status_code=404)])
-        resp = triage.request_with_retry(session, "GET", "https://x")
-        self.assertEqual(resp.status_code, 404)
-        self.assertEqual(len(session.calls), 1)
-
-    def test_gives_up_after_the_attempt_budget(self):
-        session = FakeSession([FakeResponse({}, status_code=503)] * 3)
-        resp = triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(resp.status_code, 503)
-        self.assertEqual(len(session.calls), 3)
-
-    def test_transport_failures_retry_then_raise_when_exhausted(self):
-        session = FakeSession([requests.ConnectionError("boom")] * 3)
-        with NoSleep():
-            with self.assertRaises(requests.ConnectionError):
-                triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(len(session.calls), 3)
-
-    def test_a_transport_failure_can_recover_on_a_later_attempt(self):
-        session = FakeSession([requests.Timeout("slow"), FakeResponse({"ok": True})])
-        with NoSleep():
-            resp = triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(resp.json(), {"ok": True})
-        self.assertEqual(len(session.calls), 2)
-
-    def test_no_sleep_after_the_final_attempt(self):
-        """Sleeping after the last try only delays the caller — nothing follows it."""
-        session = FakeSession([FakeResponse({}, status_code=503)] * 3)
-        with NoSleep() as clock:
-            triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(len(clock.slept), 2)  # 3 attempts, 2 gaps
-
-    def test_numeric_retry_after_is_honoured(self):
-        session = FakeSession([
-            FakeResponse({}, status_code=429, retry_after="7"),
-            FakeResponse({"ok": True}),
-        ])
-        with NoSleep() as clock:
-            triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(clock.slept, [7.0])
-
-    def test_retry_after_is_capped(self):
-        session = FakeSession([
-            FakeResponse({}, status_code=429, retry_after="9999"),
-            FakeResponse({"ok": True}),
-        ])
-        with NoSleep() as clock:
-            triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(clock.slept, [60])
-
-    def test_http_date_retry_after_falls_back_instead_of_crashing(self):
-        """RFC 9110 allows an HTTP-date here; float() on it used to raise ValueError."""
-        session = FakeSession([
-            FakeResponse({}, status_code=503, retry_after="Wed, 21 Oct 2026 07:28:00 GMT"),
-            FakeResponse({"ok": True}),
-        ])
-        with NoSleep() as clock:
-            resp = triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(resp.json(), {"ok": True})
-        self.assertEqual(clock.slept, [1.0])  # fell back to the backoff delay
-
-    def test_zero_attempts_is_rejected_rather_than_unbound(self):
-        with self.assertRaises(ValueError):
-            triage.request_with_retry(FakeSession([]), "GET", "https://x", attempts=0)
-
-
-class TestRetryAfterSeconds(unittest.TestCase):
-    def test_missing_header_uses_the_default(self):
-        self.assertEqual(triage.retry_after_seconds(FakeResponse({}, retry_after=None), 4.0), 4.0)
-
-    def test_numeric_header_wins(self):
-        self.assertEqual(triage.retry_after_seconds(FakeResponse({}, retry_after="12"), 4.0), 12.0)
-
-    def test_unparseable_header_uses_the_default(self):
-        for raw in ("Wed, 21 Oct 2026 07:28:00 GMT", "", "soon", "12s"):
-            self.assertEqual(triage.retry_after_seconds(FakeResponse({}, retry_after=raw), 4.0), 4.0)
-
-    def test_negative_and_non_finite_values_use_the_default(self):
-        """time.sleep() rejects a negative or NaN duration, so passing one through
-        would crash the run on a hostile or buggy Retry-After header."""
-        for raw in ("-30", "-0.5", "nan", "inf", "-inf"):
-            self.assertEqual(triage.retry_after_seconds(FakeResponse({}, retry_after=raw), 4.0),
-                             4.0, msg=f"retry-after={raw!r}")
-
-    def test_zero_is_honoured_rather_than_replaced(self):
-        """Zero is a valid instruction to retry immediately, not a missing value."""
-        self.assertEqual(triage.retry_after_seconds(FakeResponse({}, retry_after="0"), 4.0), 0.0)
-
-    def test_a_negative_retry_after_does_not_crash_a_real_retry_loop(self):
-        session = FakeSession([
-            FakeResponse({}, status_code=503, retry_after="-30"),
-            FakeResponse({"ok": True}),
-        ])
-        with NoSleep() as clock:
-            resp = triage.request_with_retry(session, "GET", "https://x", attempts=3)
-        self.assertEqual(resp.json(), {"ok": True})
-        self.assertTrue(all(s >= 0 for s in clock.slept), clock.slept)
-
-
 # ---- The Claude Code CLI ---------------------------------------------------
 
 
@@ -1952,6 +1702,17 @@ def test_a_reported_failure_stops_the_run(self):
                 with self.assertRaises(SystemExit):
                     self.run_cli(response=response)
 
+    def test_a_zero_exit_that_reports_is_error_names_the_reason(self):
+        """The envelope's `result` is where the CLI puts the reason, whatever the exit
+        code; the subtype and status alone do not say what went wrong."""
+        with self.assertRaises(SystemExit) as caught:
+            self.run_cli(response={"subtype": "success", "is_error": True,
+                                   "api_error_status": 404,
+                                   "result": "There's an issue with the selected model."})
+        message = str(caught.exception)
+        self.assertIn("api_error_status=404", message)
+        self.assertIn("selected model", message)
+
     def test_a_non_zero_exit_prefers_stderr_over_stdout(self):
         """A non-zero exit means there is no JSON to read, and the CLI could echo the
         prompt back — which this repo's public run logs must not carry."""
diff --git a/zendesk_triage/triage.py b/zendesk_triage/triage.py
index 6edbd7d..9956dce 100644
--- a/zendesk_triage/triage.py
+++ b/zendesk_triage/triage.py
@@ -71,19 +71,22 @@
 """
 import argparse
 import json
-import math
 import os
 import re
 import subprocess
 import sys
 import textwrap
-import time
 from datetime import datetime, timedelta, timezone
 from functools import partial
-from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
 import requests
 
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from shared import discord, state as dedup  # noqa: E402
+from shared.discord import clip, post_to_discord  # noqa: E402
+from shared.env import get_env  # noqa: E402
+from shared.retry import request_with_retry  # noqa: E402
+
 # The channel AppFollow imports app-store reviews on. Identified reviews with no
 # false positives in a 3,662-ticket sample; tags did not (only 287 carried one).
 REVIEW_CHANNEL = "any_channel"
@@ -384,17 +387,6 @@ def window_label(hours):
 SYSTEM_PROMPT = _SYSTEM_PROMPT_TEMPLATE.replace("__CATEGORIES__", CATEGORY_GUIDANCE)
 
 
-def get_env(name, cli_value=None, required=True):
-    if cli_value:
-        return cli_value
-    value = os.environ.get(name)
-    if value:
-        return value
-    if required:
-        sys.exit(f"Missing required config: set the {name} environment variable (or pass the matching flag).")
-    return None
-
-
 def zendesk_session(email, token):
     session = requests.Session()
     # Zendesk API-token auth: username is "{email}/token", password is the token.
@@ -403,63 +395,6 @@ def zendesk_session(email, token):
     return session
 
 
-def retry_after_seconds(resp, default):
-    """Seconds to wait per the Retry-After header, falling back to `default`.
-
-    RFC 9110 allows either a delay in seconds or an HTTP-date; float() on the date
-    form raises, so anything unparseable falls back rather than crashing the run.
-
-    Negative, NaN, and infinite values fall back too: time.sleep() rejects the first
-    two outright, so a hostile or buggy proxy sending `Retry-After: -30` would
-    otherwise take the run down with a ValueError.
-    """
-    raw = resp.headers.get("retry-after")
-    if raw is None:
-        return default
-    try:
-        seconds = float(raw)
-    except (TypeError, ValueError):
-        return default
-    if not math.isfinite(seconds) or seconds < 0:
-        return default
-    return seconds
-
-
-def request_with_retry(session, method, url, attempts=6, **kwargs):
-    """GET/POST with backoff on 429 and 5xx.
-
-    Lower `attempts` for calls whose result is nice-to-have: the full budget can
-    burn ~60s of backoff, which is not worth spending on optional data.
-    """
-    if attempts <= 0:
-        raise ValueError("attempts must be at least 1")
-
-    delay = 1.0
-    last_exc = None
-    resp = None
-    for attempt in range(attempts):
-        final = attempt == attempts - 1
-        try:
-            resp = session.request(method, url, timeout=30, **kwargs)
-        except requests.RequestException as exc:
-            last_exc = exc
-            if final:
-                break
-            time.sleep(delay)
-            delay = min(delay * 2, 30)
-            continue
-        if resp.status_code == 429 or resp.status_code >= 500:
-            if final:
-                break
-            time.sleep(min(retry_after_seconds(resp, delay), 60))
-            delay = min(delay * 2, 30)
-            continue
-        return resp
-    if last_exc:
-        raise last_exc
-    return resp
-
-
 def fetch_every_ticket(session, subdomain, query, max_tickets):
     """Fetch past the Search API's 1000-result ceiling, in created_at slices.
 
@@ -606,30 +541,11 @@ def fetch_total_unsolved(session, subdomain, query=BACKLOG_QUERY):
 
 
 def empty_state():
-    return {"version": STATE_VERSION, "seen": {}}
+    return dedup.empty_state(STATE_VERSION)
 
 
 def load_state(path):
-    if not os.path.exists(path):
-        print(f"No state file at {path}; treating every ticket in the window as new.")
-        return empty_state()
-    try:
-        with open(path, encoding="utf-8") as fh:
-            data = json.load(fh)
-    except (OSError, json.JSONDecodeError) as exc:
-        print(f"Note: unreadable state file {path} ({exc}); treating every ticket as new.")
-        return empty_state()
-    if not isinstance(data, dict) or not isinstance(data.get("seen"), dict):
-        print(f"Note: unexpected shape in {path}; treating every ticket as new.")
-        return empty_state()
-    # A state file written by a different schema version can't be trusted field by
-    # field, so treat it as a cache miss rather than misreading it.
-    if data.get("version") != STATE_VERSION:
-        print(f"Note: {path} is version {data.get('version')!r}, expected {STATE_VERSION}; "
-              f"treating every ticket as new.")
-        return empty_state()
-    print(f"Loaded state for {len(data['seen'])} previously reported tickets.")
-    return data
+    return dedup.load_state(path, STATE_VERSION, "ticket")
 
 
 def activity_key(ticket):
@@ -702,40 +618,10 @@ def partition_by_state(tickets, state):
 
 
 def save_state(path, state, reported, retention_days):
-    """Record `reported` as seen, prune old entries, write atomically.
-
-    Returns (kept, pruned).
-    """
-    now = datetime.now(timezone.utc)
-    stamp = now.strftime("%Y-%m-%dT%H:%M:%SZ")
-    seen = dict(state.get("seen", {}))
-    for ticket in reported:
-        seen[str(ticket.get("id"))] = {
-            "requester_updated_at": activity_key(ticket),
-            "last_reported": stamp,
-        }
-
-    # Bound the file: the window is 72h, so anything older than retention is moot.
-    cutoff = now - timedelta(days=retention_days)
-    kept = {}
-    for ticket_id, record in seen.items():
-        try:
-            last = datetime.strptime(
-                record.get("last_reported", ""), "%Y-%m-%dT%H:%M:%SZ"
-            ).replace(tzinfo=timezone.utc)
-        except (TypeError, ValueError):
-            continue  # malformed entry — drop it rather than keep it forever
-        if last >= cutoff:
-            kept[ticket_id] = record
-
-    directory = os.path.dirname(path)
-    if directory:
-        os.makedirs(directory, exist_ok=True)
-    temporary = f"{path}.tmp"
-    with open(temporary, "w", encoding="utf-8") as fh:
-        json.dump({"version": STATE_VERSION, "updated_at": stamp, "seen": kept}, fh, indent=2)
-    os.replace(temporary, path)  # atomic: a crash mid-write can't corrupt the state
-    return len(kept), len(seen) - len(kept)
+    """Record `reported` as seen. Returns (kept, pruned)."""
+    records = {str(t.get("id")): {"requester_updated_at": activity_key(t)}
+               for t in reported}
+    return dedup.save_state(path, state, records, retention_days, STATE_VERSION)
 
 
 # ---- App-store review filtering --------------------------------------------
@@ -1469,9 +1355,11 @@ def claude_cli_json(model, effort, system_prompt, schema, prompt, timeout, label
     # is not — a successful structured-output run reports "tool_use", because that is
     # how the schema is enforced underneath.
     if response.get("is_error") or response.get("subtype") != "success":
+        detail = cli_failure_detail(done.stdout, done.stderr)
         sys.exit(f"{CLAUDE_CLI} reported failure on {label} "
                  f"(subtype={response.get('subtype')!r}, "
-                 f"api_error_status={response.get('api_error_status')!r}).")
+                 f"api_error_status={response.get('api_error_status')!r})"
+                 f"{': ' + detail if detail else '.'}")
     # stop_reason is worth reading for this one value. There is no --max-tokens to
     # raise, so an answer too long to finish comes back as JSON that stops mid-object,
     # and the parse below would report a baffling syntax error for something whose
@@ -1555,40 +1443,23 @@ def analyze(model, effort, compact_tickets):
 
 # ---- Components V2 ---------------------------------------------------------
 #
-# The digest is a Container of Text Displays: one block per ticket, so a reader skims
-# lines rather than a wall, and each message records which ticket ids it accounts for.
-#
-# Every component here is non-interactive, which is what lets a plain incoming webhook
-# carry it: Discord allows a webhook that no application owns only those. Adding an
-# interactive one would need the transport moved back to a bot token — see
-# test_the_digest_carries_no_interactive_components.
-#
-# https://docs.discord.com/developers/components/reference
-COMPONENTS_V2_FLAG = 1 << 15
-CONTAINER = 17
-TEXT_DISPLAY = 10
-SEPARATOR = 14
+# The digest is a Container of Text Displays, one per ticket, built by
+# shared.discord — which also documents why a plain webhook can carry it.
 
 # Discord allows 40 components in one message, and a ticket now costs one Text
 # Display, so that ceiling no longer binds — the character budget below does. Ten is
 # kept because it is a readable message, not because it is the limit.
 MAX_ENTRIES_PER_MESSAGE = 10
-# Discord's ceiling on all the text in one Components V2 message, and the constraint
-# that actually binds. Ten clipped ticket lines plus a header come to roughly 3,500,
+# Discord's ceiling on all the text in one message, and the constraint that
+# actually binds. Ten clipped ticket lines plus a header come to roughly 3,500,
 # so this is a guard rather than a routine constraint.
-MAX_MESSAGE_TEXT_CHARS = 4000
-MAX_COMPONENT_CHARS = MAX_MESSAGE_TEXT_CHARS
+MAX_COMPONENT_CHARS = discord.MAX_MESSAGE_TEXT_CHARS
 
 
 def ticket_url(subdomain, ticket_id):
     return f"https://{subdomain}.zendesk.com/agent/tickets/{ticket_id}"
 
 
-def clip(text, limit):
-    text = (text or "").strip()
-    return text if len(text) <= limit else text[: limit - 1] + "…"
-
-
 def is_urgent(finding):
     return finding.get("category") in URGENT_CATEGORIES
 
@@ -1735,42 +1606,6 @@ def build_header(findings, highlights, stats=None):
     return "\n".join(lines)
 
 
-def chunk_entries(entries, max_items=MAX_ENTRIES_PER_MESSAGE,
-                  max_chars=MAX_COMPONENT_CHARS, first_used=0):
-    """Group (line, ticket_ids) pairs into messages within Discord's budgets.
-
-    Two limits rather than one, and whichever binds first splits the message: a
-    Components V2 message allows 40 components, of which a ticket costs three, and a
-    character budget that clipped lines rarely approach. Sections are separate
-    components rather than joined text, so unlike the old plain-content digest
-    nothing is spent on the newlines between them.
-
-    `first_used` is what the caller has already spent on the first message before any
-    ticket goes in — the header. Without it the header rides on top of a full budget
-    of ticket lines, and a busy day's accounting lines are enough to put message one
-    over the limit.
-
-    An entry longer than the character budget still gets its own message rather than
-    being dropped; the pieces are pre-clipped so that shouldn't arise.
-
-    Entries are passed through, not rebuilt, so the caller can still tell which one
-    it is looking at by identity — build_messages needs that to find the collapsed
-    line again once its entry is somewhere inside a chunk.
-    """
-    chunks, current, current_chars = [], [], first_used
-    for entry in entries:
-        text, _ = entry
-        if current and (len(current) >= max_items
-                        or current_chars + len(text) > max_chars):
-            chunks.append(current)
-            current, current_chars = [], 0
-        current.append(entry)
-        current_chars += len(text)
-    if current:
-        chunks.append(current)
-    return chunks
-
-
 def select_highlights(findings):
     """Ordered highlights split into (shown, omitted) by the display cap.
 
@@ -1820,70 +1655,12 @@ def build_messages(findings, subdomain, stats=None, updated_ids=None):
     if collapsed:
         entries.append((build_collapsed_line(collapsed, subdomain), collapsed_ids))
 
-    messages, coverage = [], []
-    # A quiet day still owes the channel the header — chunk_entries has nothing to
-    # chunk when no ticket is worth looking into, so seed one empty chunk.
-    # The header only lands on message one, so only message one's budget pays for
-    # it. chunk_entries resets to zero for every chunk after the first.
-    chunks = chunk_entries(entries, first_used=len(header)) or [[]]
-    for index, chunk in enumerate(chunks):
-        blocks = []
-        covered = set()
-        if index == 0:
-            blocks.append({"type": TEXT_DISPLAY, "content": header})
-            if chunk:
-                blocks.append({"type": SEPARATOR})
-            # The header accounts for every classified ticket except the highlights
-            # that didn't fit; those are covered by no message and stay eligible.
-            covered |= header_ids
-        for text, ids in chunk:
-            blocks.append({"type": TEXT_DISPLAY, "content": text})
-            covered |= ids
-        messages.append({
-            "flags": COMPONENTS_V2_FLAG,
-            "components": [{"type": CONTAINER, "components": blocks}],
-        })
-        coverage.append(covered)
+    messages, coverage = discord.messages_from_entries(
+        header, entries, MAX_ENTRIES_PER_MESSAGE, MAX_COMPONENT_CHARS)
+    coverage[0] |= header_ids
     return messages, coverage
 
 
-def digest_webhook_url(webhook_url):
-    """The webhook, told to respect the components field.
-
-    Discord ignores `components` on a webhook post without it, and the digest is
-    nothing but components.
-    """
-    parts = urlsplit(webhook_url)
-    query = dict(parse_qsl(parts.query))
-    query["with_components"] = "true"
-    return urlunsplit(parts._replace(query=urlencode(query)))
-
-
-def post_to_discord(session, url, messages):
-    """POST each message in order; return how many Discord accepted.
-
-    Stops at the first failure and returns the accepted count instead of exiting, so
-    the caller can record the tickets that did land before signalling the failure —
-    otherwise a failure on message 3 of 3 reposts messages 1 and 2 on the next run.
-    """
-    for index, payload in enumerate(messages):
-        try:
-            resp = request_with_retry(session, "POST", url, json=payload)
-        except requests.RequestException as exc:
-            # request_with_retry re-raises once its budget is spent. Letting that
-            # propagate would skip save_state entirely, so the messages that already
-            # landed would be reposted on the next run — the exact thing returning a
-            # count exists to prevent.
-            print(f"Discord unreachable on message {index + 1}/{len(messages)} "
-                  f"({exc}).")
-            return index
-        if resp.status_code >= 400:
-            print(f"Discord rejected message {index + 1}/{len(messages)} "
-                  f"({resp.status_code}): {resp.text[:300]}")
-            return index
-    return len(messages)
-
-
 def main():
     parser = argparse.ArgumentParser(description="Triage open Zendesk tickets with Claude and post a Discord summary.")
     parser.add_argument("--subdomain", help="Zendesk subdomain (else ZENDESK_SUBDOMAIN).")
@@ -2114,7 +1891,8 @@ def main():
         return
 
     # A fresh session, never the Zendesk one: that carries the API-token auth header.
-    posted = post_to_discord(requests.Session(), digest_webhook_url(webhook), messages)
+    posted = post_to_discord(requests.Session(), discord.components_webhook_url(webhook),
+                             messages)
     print(f"Posted {posted} of {len(messages)} Discord message(s).")
 
     # Record only tickets covered by messages Discord actually accepted, so a partial