diff --git a/README.md b/README.md index 453d1ac..7bd8fb3 100644 --- a/README.md +++ b/README.md @@ -213,7 +213,7 @@ Runs **Monday to Friday at 10:00 Melbourne** over a 72h window (~70 tickets) — The window is on `updated>`, not `created>`, so a ticket the requester adds detail to days after opening it is fetched again — a created-window would never see it. 72h rather than the 24h between runs so a failed run doesn't drop a day and Monday still reaches back past the weekend. Neither the overlap nor the wider net duplicates posts, because of the dedup state above. -[Zendesk Resolve Positive Reviews](#zendesk-resolve-positive-reviews) runs first, as the unit's first `ExecStart`. Order matters: the triage query is `status` | Compose the reply from the brief, in the requester's language. A second `draft` amends the one already there rather than starting over | no | +| `claude: reply [n]` | Publish the chosen option verbatim, status -> `pending` | **yes** | +| `claude: english` | Post the conversation, both sides, in English. Says so and writes nothing when the ticket is already English | no | +| `claude: explain` | Post what support usually replied to this kind of ticket, what was actually done about it, and the caveats | no | +| `claude: solve [reason]` | Solve without writing to the customer, for tickets that need no reply. The note records who decided and why | no comment, but **solving fires the CSAT automation** | + +### Why a draft is always reviewed + +[reply.py](zendesk_triage/reply.py) sends an English ticket immediately, because the +agent typed the exact words and there is nothing to check. Here Claude *composes* the +reply from a brief, so nobody has read that wording yet — every draft is reviewed, +English included. `reply` never re-composes: what was reviewed is what goes out, or +the review means nothing. To change a draft, write a new brief. + +### What stops it drafting against itself + +Claude's own draft note names both commands in its instructions. If those parsed as +commands, every draft would trigger another one, forever. Two independent guards: + +- `COMMAND` only matches at the **start of a line**, and the instructions in a draft + note are written mid-line on purpose. `test_a_generated_draft_note_is_not_a_command` + asserts it. +- The command search skips notes authored by the API user, and the Zendesk trigger + should exclude that same user so a draft never reaches the webhook at all. + +Give the automation its own Zendesk user rather than reusing an account a human signs +into — otherwise excluding it in the trigger also excludes that person's notes, and +the tool silently stops working for them. + +### Who may command it + +Only private comments count, so a customer typing `claude:` into a public reply is +ignored. The author must be an agent or admin — the set of people who can write a +private note at all. `ZENDESK_NOTE_AUTHORS` narrows that to named user ids; the role +check still applies, so an id on the list that is not an agent is still refused. + +An unauthorised author **stops** the search rather than falling through to an older +command. Their note is the most recent instruction on the ticket, and quietly acting +on a previous one instead would be a surprising thing to do. + +### What the model may write + +The brief is the only source of facts. The system prompt forbids adding a version +number, a date, a retention period, a link or a timeline the brief does not contain — +and forbids claiming an action was taken unless the brief says it was. That second +rule is the important one: 183 solved tickets in this account tell a reporter their +Account ID "has been banned from communities we operate", and a reply asserting +something nobody did is the worst thing this can produce. Both rules are asserted by +`test_the_prompt_forbids_inventing_facts_and_actions`, so a prompt edit cannot +quietly drop them. + +### Idempotency + +Zendesk retries a webhook that does not answer cleanly, and the reply is written +before the run finishes — so without a guard, a slow run emails the customer twice. +Every outcome note carries `[claude:done:]`, keyed on the commanding +comment rather than the ticket, because two briefs on one ticket are two commands and +the second must not be swallowed by the first one's marker. Refusals carry it too: a +command that cannot be satisfied is still a command that was answered. + +### Tags + +| Tag | Set by | Cleared by | +| --- | --- | --- | +| `claude-queued` | the Zendesk trigger, when the note lands | a successful run | +| `claude-drafted` | a draft being posted | the reply going out | +| `claude-sent` | the reply going out | — | +| `claude-error` | a refusal, with the reason in the note | the next successful run | + +The tag is the durable queue and the webhook is only a latency optimisation. A relay +that is down leaves `claude-queued` on the ticket, so `tags:claude-queued` older than +a few minutes is the list of dropped jobs — a webhook-only design would lose them +silently. Two views are worth making: `tags:claude-queued` for what did not run, and +`tags:claude-drafted` for what is waiting on a human. + +### Zendesk setup + +A trigger, and a webhook it calls: + +- **Webhook** — POST to `https:///zendesk/notes`, JSON body `{"ticket_id": + "{{ticket.id}}"}`, signed. Put the signing secret in `ZENDESK_WEBHOOK_SECRET`; + without it the route refuses everything, because a URL that writes to customers + must not default to open. +- **Trigger** — conditions: *Ticket is Updated*, *Comment is Private*, *Comment text + contains `claude:`*, and *Current user is not* the automation user. Actions: notify + the webhook, and add the tag `claude-queued`. + +### Required Secrets + +| Secret | Description | +| --- | --- | +| `ZENDESK_WEBHOOK_SECRET` | Shared secret Zendesk signs the webhook with. Unset refuses every request | +| `ZENDESK_NOTE_AUTHORS` | *(optional)* Comma-separated Zendesk user ids allowed to command it. Unset means any agent or admin | +| `ZENDESK_NOTE_MODEL` | *(optional)* Overrides the model | + +The Zendesk credentials and Claude authentication are the ones the digest already +uses. `RELAY_DRY_RUN` covers this path too: the whole run happens and nothing is +written. + +### Local Testing + +``` +# what the webhook does, against a real ticket, writing nothing +python zendesk_triage/note_reply.py --ticket 27603 --dry-run +``` + +A ticket with no command note prints `no command note to act on` and stops, so this +is safe to point at anything. + ## 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/deploy/README.md b/deploy/README.md index 7ed1cc1..1ce9c1d 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -135,6 +135,15 @@ DISCORD_GUILD_ID= ALLOWED_USER_IDS= ALLOWED_ROLE_IDS= +# Verifies Zendesk's webhook signatures, for the `claude:` private-note route. +# Empty refuses every note webhook: a URL that writes public comments must not +# default to open. Same value as the secret on the Zendesk webhook itself. +ZENDESK_WEBHOOK_SECRET= +# Optional. Comma-separated Zendesk user ids allowed to drive `claude:` notes. +# Empty means any agent or admin, which is already everybody who can write a +# private note. The agent/admin role check applies either way. +#ZENDESK_NOTE_AUTHORS= + # Optional, and read by both units. The numeric id of a multi-line text ticket # field in Zendesk; the digest renders every non-English ticket it is about to post # into English there — both sides of the conversation, timestamped — and the compose @@ -143,7 +152,8 @@ ALLOWED_ROLE_IDS= # -> Multi-line text, not customer-visible, then read the id off the field's URL. #ZENDESK_ENGLISH_FIELD_ID= -# Uncomment to run the whole path and write nothing to Zendesk. +# Uncomment to run the whole path and write nothing to Zendesk. Covers both the +# Discord reply flow and the `claude:` note flow. #RELAY_DRY_RUN=1 ``` diff --git a/deploy/nginx-webhooks.conf b/deploy/nginx-webhooks.conf index 1499ded..02c69ef 100644 --- a/deploy/nginx-webhooks.conf +++ b/deploy/nginx-webhooks.conf @@ -79,6 +79,21 @@ server { proxy_connect_timeout 5s; } + # Zendesk's turn: a `claude:` private note on a ticket. Same shape as above and + # for the same reasons — exact path, nothing that touches the body, since the + # HMAC signature covers the raw bytes too. + location = /zendesk/notes { + proxy_pass http://127.0.0.1:8080; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + # The relay answers straight away and does the work in the background, so + # this waits on an acknowledgement rather than on Claude. + proxy_read_timeout 10s; + proxy_connect_timeout 5s; + } + location = /healthz { proxy_pass http://127.0.0.1:8080; proxy_set_header Host $host; diff --git a/zendesk_triage/note_reply.py b/zendesk_triage/note_reply.py new file mode 100644 index 0000000..6df7928 --- /dev/null +++ b/zendesk_triage/note_reply.py @@ -0,0 +1,1130 @@ +#!/usr/bin/env python3 +""" +Draft and send Zendesk replies from private notes on the ticket itself. + +An agent writes a private note saying what the answer is; Claude writes it properly, +in the language the requester uses, and posts it back as a private note. The agent +reads it and writes a second note to send it. + + claude: draft - attachments are only kept on the server for 14 days. A second + device that was offline for longer cannot fetch them. + + claude: reply + +`claude: english` is the read-only one: it puts the conversation in English on the +ticket so an agent can read a ticket in a language they do not speak. + + ⚠️ `reply` writes a public comment, which emails the requester. That is not + reversible. --dry-run does everything except the writes. + +Unlike reply.py — where the agent types the exact English text and Claude only +translates it — here Claude *composes* the reply from a brief. So every draft is +reviewed before it goes out, English ones included: nobody has read that wording yet. + +Two actions, both read from the ticket's own comments: + + draft Compose the reply from the brief, translate it into the requester's + language, and post it as a private note with a back-translation. + reply Send the most recent draft, verbatim. It never re-composes: what was + reviewed is what goes out, or the review means nothing. + english Post the whole conversation, both sides, in English as a private note. + Reads only; the customer never sees it. + solve Solve the ticket, writing nothing to the customer. For the ones that + need no reply at all. The private note records who asked and why. + explain Post what support usually replies to this kind of ticket, and what was + actually done about it before — fixes shipped, bugs filed, escalations. + Reads only. This is where known fixes are surfaced, because `draft` + refuses to assert one that is not in the brief. + +The action comes from the newest private note that parses as a command, so the +webhook only has to say which ticket changed. Notes written by the API user are +skipped, which is what stops Claude's own drafts from re-triggering it. + +Because this repo is public, nothing here prints ticket content, briefs, or reply +text. stdout gets ticket ids and outcomes. + +Config (env vars, or flags for local runs): + ZENDESK_SUBDOMAIN e.g. "mycompany" -> https://mycompany.zendesk.com + ZENDESK_EMAIL agent email for API token auth; authors every comment + ZENDESK_API_TOKEN Zendesk API token + ZENDESK_NOTE_AUTHORS (optional) comma-separated Zendesk user ids allowed to + command it. Unset means any agent or admin on the account, + which is already the set of people who can write a private + note at all. + ZENDESK_NOTE_MODEL (optional) overrides the model + ZENDESK_HOUSE_ANSWERS (optional) path to the house-answer file — what support + usually replied to each kind of problem, per platform. + Absent, drafting works exactly as it does without it + +Usage: + note_reply.py --ticket 12345 [--dry-run] +""" +import argparse +import html +import json +import os +import re +import sys +import textwrap + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import reply # noqa: E402 (needs the path insert above) +import triage # noqa: E402 + +DEFAULT_MODEL = "claude-sonnet-5" +COMPOSE_TIMEOUT_SECONDS = 240 +PLACEMENT_TIMEOUT_SECONDS = 120 + +# What Claude is given of the customer's own words, to answer in the right language +# and at the right level of detail. +CUSTOMER_SAMPLE_CHARS = 2000 +# A brief is a sentence or two. Anything longer is a pasted reply, which is +# reply.py's job, not this one. +BRIEF_CHARS = 2000 + +# Status the ticket moves to once the reply is out: the ball is with the customer. +# Same convention as reply.py, so `open` keeps meaning "ours". +REPLIED_STATUS = "pending" +# `claude: solve` sets this. Not "closed": Zendesk refuses closed over the API, and +# the account's own automation closes a solved ticket four days later anyway. +SOLVED_STATUS = "solved" + +TAG_QUEUED = "claude-queued" +TAG_DRAFTED = "claude-drafted" +TAG_SENT = "claude-sent" +TAG_SOLVED = "claude-solved" +TAG_ERROR = "claude-error" +# Where a ticket was filed in the taxonomy, cached on the ticket so a second draft +# does not pay for the classification again. Also what a future `group` verb writes. +TAG_GROUP_PREFIX = "grp-" +TAG_PLATFORM_PREFIX = "plat-" + +# The house answers: what support actually replied to this kind of problem before, +# per platform. Off unless ZENDESK_HOUSE_ANSWERS points at the file — the drafting +# path works exactly as it did without it. +# +# Deliberately NOT in this repo. It is built from real solved tickets and carries +# their ids and their content, and this repository is public. It belongs beside the +# dedup state on the host, for the same reason that does. +HOUSE_ENV = "ZENDESK_HOUSE_ANSWERS" +PLATFORMS_COLLAPSED = ("android", "ios", "desktop", "multiple", "unknown") + +# Wraps the exact text that will be sent. `reply` copies what is between these lines +# and nothing else, so the draft note can carry back-translations and instructions +# around them without any of that reaching the customer. Numbered, because a draft +# offers the agent a choice: `claude: reply 2` sends the second. +BEGIN = "-----BEGIN REPLY" +END = "-----END REPLY" +MAX_OPTIONS = 3 + + +def begin_marker(n): + return f"{BEGIN} {n}-----" + + +def end_marker(n): + return f"{END} {n}-----" + + +# The number in the closing line has to match the opening one, so a stray delimiter +# inside a reply cannot silently truncate the block that gets sent. +OPTION_BLOCK = re.compile( + re.escape(BEGIN) + r"\s+(\d+)-----(.*?)" + re.escape(END) + r"\s+\1-----", + re.DOTALL) + + +def done_marker(comment_id): + """Written into the outcome note so a replayed webhook cannot act twice. + + Zendesk retries a webhook that does not answer cleanly, and this runs after the + comment is already posted — so without this, a slow run means the customer is + emailed twice. Keyed on the commanding comment because that is what is unique + per instruction; the ticket id is not. + """ + return f"[claude:done:{comment_id}]" + + +def draft_marker(comment_id): + """Marks a note as carrying a sendable draft, and says which brief produced it.""" + return f"[claude:draft:{comment_id}]" + + +def english_marker(latest_public_id): + """Records how far a transcript note read, so re-running is cheap. + + Keyed on the newest public comment it covered rather than on the command: asking + twice with nothing said in between should cost nothing, and asking again after + the customer writes back should produce a fresh transcript. + """ + return f"[claude:english:{latest_public_id}]" + + +# Anchored to the start of a line so that prose mentioning the command in passing — +# including the instructions in Claude's own draft notes — is not a command. +COMMAND = re.compile( + r"^[\s>*_]*claude\s*:\s*(draft|reply|english|explain|solve)\b[\s\-–—:.]*(.*)$", + re.IGNORECASE) + + +def parse_command(text): + """(action, brief) from a note, or None when it is not a command. + + The brief runs from the command line to the end of the note, so it can be + several lines without needing quoting. + """ + lines = (text or "").splitlines() + for index, line in enumerate(lines): + found = COMMAND.match(line) + if not found: + continue + rest = [found.group(2)] + lines[index + 1:] + return found.group(1).lower(), "\n".join(rest).strip()[:BRIEF_CHARS] + return None + + +def comment_text(comment): + """The note as text, as it was typed. + + `plain_body` because Zendesk stores comments as HTML, and unescaped because + plain_body strips the tags without touching the entities — an ampersand posted + as `&` comes back as `&`. Since notes are written escaped, unescaping + here is what makes the round trip exact, which is what lets `reply` send the + reviewed draft byte for byte. + """ + return html.unescape(comment.get("plain_body") or comment.get("body") or "") + + +# ---- Zendesk ---------------------------------------------------------------- + + +def api_user_id(session, subdomain): + """The user the API token authenticates as. + + Its own notes are skipped when looking for a command, which is the in-code half + of the loop guard. The other half is the Zendesk trigger, which should exclude + this same user so a draft never fires the webhook at all — see the README. + """ + url = f"https://{subdomain}.zendesk.com/api/v2/users/me.json" + resp = triage.request_with_retry(session, "GET", url) + if resp.status_code >= 400: + sys.exit(f"Zendesk refused to identify the API user ({resp.status_code}).") + return ((resp.json() or {}).get("user") or {}).get("id") + + +def fetch_user(session, subdomain, user_id): + url = f"https://{subdomain}.zendesk.com/api/v2/users/{user_id}.json" + resp = triage.request_with_retry(session, "GET", url) + if resp.status_code >= 400: + return {} + return (resp.json() or {}).get("user") or {} + + +def customer_sample(session, subdomain, ticket, comments): + """The customer's own words, for deciding which language to reply in. + + reply.customer_text takes only comments the REQUESTER authored, which is right + for email and web tickets. On a Twitter or Sunshine DM the integration authors + the customer's message under its own id, so that filter drops everything they + wrote and leaves the ticket's "Conversation with " description — and the + reply goes out in English to somebody writing Chinese. + + So: the requester's own words when the ticket carries any, and otherwise every + public comment written by someone who is not an agent on this account. Roles are + looked up rather than guessed from the id, because the integration's id is an + account detail and an unknown author is a customer, not an agent. + """ + if not triage.is_content_free(ticket): + return reply.customer_text(ticket, comments) + roles, parts = {}, [] + subject = triage.squash(ticket.get("subject")) + for comment in reversed(comments): # oldest first, so it reads in order + if not comment.get("public"): + continue + author = comment.get("author_id") + if author not in roles: + roles[author] = (fetch_user(session, subdomain, author) or {}).get("role") + if roles[author] in ("agent", "admin"): + continue + body = triage.squash(comment.get("body")) + if body and body != subject: + parts.append(body) + return triage.clip("\n\n".join(parts), + CUSTOMER_SAMPLE_CHARS) or reply.customer_text(ticket, comments) + + +def may_command(user): + """Whether this Zendesk user may drive the command. + + Default is any agent or admin: only they can write a private note in the first + place, so an allowlist is a narrowing, not the gate. Setting ZENDESK_NOTE_AUTHORS + narrows it to named ids — and an id that is not on the account fails closed, + because the role check is applied either way. + """ + if user.get("role") not in ("agent", "admin"): + return False + allowed = [item.strip() for item in + (os.environ.get("ZENDESK_NOTE_AUTHORS") or "").split(",") if item.strip()] + return not allowed or str(user.get("id")) in allowed + + +def change_tags(session, subdomain, ticket_id, add=(), drop=()): + """Add and remove tags, through the tags sub-resource. + + NOT `additional_tags`/`remove_tags` on the ticket update: those are update_many + fields. A single-ticket update accepts them with a 200 and silently ignores them, + which is how every tag this tool set went missing while every call reported + success. Measured against the live API, not assumed. + + The sub-resource is also additive rather than read-modify-write, so two runs on + one ticket cannot clobber each other's tags. + """ + url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}/tags.json" + for method, names in (("PUT", [t for t in add if t]), + ("DELETE", [t for t in drop if t])): + if not names: + continue + resp = triage.request_with_retry(session, method, url, json={"tags": names}) + if resp.status_code >= 400: + # Never worth failing a run over: tags are a dashboard light, not the work. + print(f"Note: could not {method.lower()} tags on #{ticket_id} " + f"({resp.status_code}).") + + +def clear_queued(session, subdomain, ticket_id, dry_run=False): + """Take the ticket out of the "waiting on Claude" queue, writing no comment. + + Every run that finishes servicing a ticket clears it, including one that decides + there is nothing to do. Otherwise the tag accumulates: Claude's own notes name + the commands, so posting one re-fires the trigger, and that second run finds only + its own note and writes nothing — leaving `claude-queued` behind on every ticket + it ever touched, which is precisely the signal that is supposed to mean a job was + dropped. + """ + if dry_run: + return + change_tags(session, subdomain, ticket_id, drop=[TAG_QUEUED]) + + +def para(text): + """One paragraph of the note, escaped.""" + return f"

{html.escape(text)}

" + + +def bold_para(text): + """A paragraph that leads a section — a speaker line in a transcript.""" + return f"

{html.escape(text)}

" + + +def transcript_blocks(turns, translated): + """One turn at a time, as readable paragraphs. + + Deliberately not triage.render_transcript's output re-split on blank lines: a + turn whose own text contains a blank line gets torn into several pieces that + way, which is what made the first version render as a row of disconnected code + boxes. The speaker line leads each turn and the body follows as prose — nothing + extracts this text, so it wants readability, not byte fidelity. + """ + english = {} + for item in translated or []: + try: + english[int(item.get("index"))] = (item.get("english") or "").strip() + except (TypeError, ValueError): + continue + out = [] + for turn in turns: + out.append(bold_para(" ".join(part for part in (turn["when"], f'{turn["who"]}:') + if part))) + body = english.get(turn["index"]) or turn["body"] + out += [para(chunk.strip()) for chunk in body.split("\n\n") if chunk.strip()] + return out + + +def verbatim(text): + """A block whose whitespace is preserved exactly. + + `
` rather than paragraphs because this is what `reply` will publish: line
+    breaks, blank lines and indentation have to survive the round trip through
+    Zendesk unchanged, and paragraph markup silently reflows them.
+    """
+    return f"
{html.escape(text)}
"
+
+
+def write_to_ticket(session, subdomain, ticket_id, body, public,
+                    status=None, add_tags=(), drop_tags=(), as_html=False):
+    """One PUT carrying a comment and any tag or status change.
+
+    `additional_tags`/`remove_tags` rather than writing the whole tag list: two runs
+    on one ticket would otherwise race and one would drop the other's tag.
+
+    The response body is never printed. Zendesk echoes the submitted comment back in
+    a 422, and that comment is the reply — which this repo's public logs must not
+    carry.
+    """
+    # Notes go as html_body so their structure survives; the public reply goes as
+    # plain body, the way reply.py has always sent one — it is prose, not a document.
+    fields = {"comment": {("html_body" if as_html else "body"): body, "public": public}}
+    if status:
+        fields["status"] = status
+    url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
+    resp = triage.request_with_retry(session, "PUT", url, json={"ticket": fields})
+    if resp.status_code >= 400:
+        sys.exit(f"Zendesk rejected the {'reply' if public else 'note'} on "
+                 f"#{ticket_id} ({resp.status_code}).")
+    # After the comment, so a tag failure cannot lose the thing that mattered.
+    change_tags(session, subdomain, ticket_id, add_tags, drop_tags)
+
+
+# ---- What we usually reply ---------------------------------------------------
+
+
+def load_house(path=None):
+    """The house answers, or None when the feature is off or the file is unusable.
+
+    Degrades rather than fails: a missing or corrupt knowledge file must cost a
+    slightly thinner draft, never the ability to answer a customer at all.
+    """
+    path = path or os.environ.get(HOUSE_ENV)
+    if not path:
+        return None
+    try:
+        with open(path, encoding="utf-8") as handle:
+            book = json.load(handle)
+    except (OSError, ValueError) as exc:
+        print(f"Note: could not read the house answers at {path} ({exc}); "
+              f"drafting without them.")
+        return None
+    if not isinstance(book, dict) or not book.get("cells"):
+        print(f"Note: {path} carries no house answers; drafting without them.")
+        return None
+    return book
+
+
+def tagged_placement(ticket):
+    """(group, platform) already recorded on the ticket, or (None, None).
+
+    Read before classifying so a revision, or a ticket a `group` command has already
+    filed, does not pay for the same model call twice.
+    """
+    group = platform = None
+    for tag in ticket.get("tags") or []:
+        if tag.startswith(TAG_GROUP_PREFIX):
+            group = tag[len(TAG_GROUP_PREFIX):]
+        elif tag.startswith(TAG_PLATFORM_PREFIX):
+            platform = tag[len(TAG_PLATFORM_PREFIX):]
+    return group, platform
+
+
+PLACEMENT_SCHEMA = {
+    "type": "object", "additionalProperties": False,
+    "required": ["group", "platform"],
+    "properties": {
+        "group": {"type": "string", "description": "a group key from the catalogue, "
+                                                   "or 'none' when nothing fits"},
+        "platform": {"type": "string", "enum": list(PLATFORMS_COLLAPSED)},
+    },
+}
+
+PLACEMENT_SYSTEM = textwrap.dedent(
+    """
+    You file one Zendesk ticket for Session, a private messenger, into an existing
+    taxonomy, so that what support usually replies to this kind of problem can be
+    looked up.
+
+    `group` is the key whose problem this ticket describes. Judge by what the
+    customer needs answered. Answer `none` rather than forcing a fit — a wrong group
+    hands the agent someone else's answer, which is worse than handing them none.
+
+    `platform` is the platform the ticket is about. Answer `unknown` unless the text
+    actually says: a guessed platform produces per-platform advice that was never
+    about this customer's device.
+    """
+).strip()
+
+
+def place_ticket(model, book, ticket, sample):
+    """Which group and platform this ticket belongs to. (None, None) if unplaceable."""
+    catalogue = "\n".join(f"- {g['key']}: {g['title']}" for g in book["groups"])
+    body = triage.clip(sample, CUSTOMER_SAMPLE_CHARS)
+    try:
+        found = triage.claude_cli_json(
+            model, "medium", PLACEMENT_SYSTEM, PLACEMENT_SCHEMA,
+            f"GROUP CATALOGUE:\n{catalogue}\n\nTHE TICKET:\n"
+            f"{(ticket.get('subject') or '')[:200]}\n\n{body}",
+            PLACEMENT_TIMEOUT_SECONDS, f"the placement of #{ticket['id']}")
+    except SystemExit as exc:
+        # Grounding is an enrichment. A failed classification costs a thinner draft,
+        # not the draft — the same call triage.py makes about its transcripts.
+        print(f"Note: could not place #{ticket['id']} ({exc}); drafting without "
+              f"the house answer.")
+        return None, None
+    group = found.get("group")
+    if group == "none" or not any(g["key"] == group for g in book["groups"]):
+        return None, found.get("platform")
+    return group, found.get("platform")
+
+
+def house_cell(book, group, platform):
+    """The house answer for this group and platform, falling back to all platforms.
+
+    A group with only three solved tickets has no per-platform answer, and the
+    all-platform one is still better than nothing. Returns (cell, which_platform).
+    """
+    if not (book and group):
+        return None, None
+    for candidate in (platform, "any"):
+        cell = book["cells"].get(f"{group}|{candidate}")
+        if cell:
+            return cell, candidate
+    return None, None
+
+
+def render_precedent(cell, title, platform):
+    """The house answer as prompt text.
+
+    Its version numbers and fix claims are deliberately not offered as facts to
+    repeat — see the PRECEDENT rules in COMPOSE_SYSTEM. What the model is meant to
+    take is the shape: what support covers for this problem, and in what order.
+    """
+    lines = [f"PROBLEM AS PREVIOUSLY FILED: {title}",
+             f"PLATFORM THIS PRECEDENT COVERS: {platform}",
+             f"BUILT FROM {cell['n']} SOLVED TICKETS ({cell['consistency']} consistency)",
+             "", "WHAT SUPPORT USUALLY SAYS:", cell["answer"]]
+    if cell.get("steps"):
+        lines += ["", "STEPS USUALLY GIVEN:"] + [f"- {s}" for s in cell["steps"]]
+    return "\n".join(lines)
+
+
+# ---- Composing --------------------------------------------------------------
+
+OPTION_PROPERTIES = {
+    "approach": {"type": "string",
+                 "description": "What this option does, in English, a few words — "
+                                "'explain and close', 'ask which device was online first'. "
+                                "It is how the agent tells the options apart."},
+    "reply_en": {"type": "string", "description": "The reply, in English."},
+    "translated": {"type": "string",
+                   "description": "`reply_en` in the customer's language. Identical to "
+                                  "`reply_en` when `is_english` is true."},
+    "back_translation": {"type": "string",
+                         "description": "`translated` rendered literally back into English. "
+                                        "Empty when `is_english` is true."},
+}
+COMPOSE_PROPERTIES = {
+    "language": {"type": "string",
+                 "description": "Language the customer writes in, in English, e.g. 'German'."},
+    "language_code": {"type": "string", "description": "BCP-47 code, e.g. 'de', 'pt-BR'."},
+    "is_english": {"type": "boolean",
+                   "description": "True only if the customer already writes in English."},
+    "options": {
+        "type": "array",
+        "description": "The candidate replies, best first. Two or three on a first "
+                       "draft; usually one when amending.",
+        "items": {"type": "object", "additionalProperties": False,
+                  "required": list(OPTION_PROPERTIES.keys()),
+                  "properties": OPTION_PROPERTIES},
+    },
+}
+COMPOSE_SCHEMA = {
+    "type": "object",
+    "additionalProperties": False,
+    "required": list(COMPOSE_PROPERTIES.keys()),
+    "properties": COMPOSE_PROPERTIES,
+}
+
+COMPOSE_SYSTEM = textwrap.dedent(
+    """
+    You write support replies for Session, a private messenger. A support agent has
+    read the ticket and written you a brief — the substance of the answer, in
+    shorthand. You turn it into the reply the customer receives.
+
+    Two halves, and they have different rules. The facts are the agent's and you may
+    not touch them. The writing is yours and you are expected to do it well.
+
+    FACTS — THE BRIEF IS THE ONLY SOURCE, and the customer's own message.
+
+    - Never state a fact neither of them contains: no version numbers, no dates, no
+      timelines, no retention periods, no links, no "our team is working on it". If
+      the brief says attachments last 14 days, say 14 days; do not explain the
+      mechanism behind it and do not guess what happens after.
+    - Never offer a workaround, a next step or a "try this" the brief does not give
+      you. This is the rule most often broken by trying to be helpful, and a
+      confident wrong instruction costs the customer more than a short answer does.
+    - NEVER claim an action was taken unless the brief says it was taken. Do not
+      write that an account was banned, a bug was filed, a refund was issued or a
+      case was escalated on your own initiative. A reply asserting something nobody
+      did is the worst thing you can produce here.
+    - If the brief is too thin to answer what they actually asked, answer the part it
+      covers and stop. Do not fill the gap.
+
+    WRITING — this half is yours, and a reply that is correct and cold is a worse
+    reply. You may and generally should:
+
+    - open by acknowledging what happened to them, in their own terms
+    - say plainly that it is frustrating or disappointing, where it plainly is
+    - restate their situation back to them, drawn from THEIR message, so they can
+      see they were understood
+    - draw out what a fact from the brief means for them, where that follows directly
+      from it — "so they are no longer on the server" follows from a 14-day limit
+    - close by inviting them back if something is still unclear
+
+    None of that introduces a fact, so none of it is forbidden. The line is simple:
+    "I'm sorry your photos are gone" is writing. "You can get them back by X" is a
+    fact, and needs the brief behind it.
+
+    Do not address the customer by name — you have not been given it. Never emit a
+    placeholder like (User name) or [Name]: on this account those have been sent to
+    real customers literally, and it is the single most visible way a reply looks
+    machine-made.
+
+    `language` and `language_code` describe the language the CUSTOMER writes in,
+    judged from their words alone, never the agent's brief.
+
+    OPTIONS. Return two or three, and make them GENUINELY DIFFERENT — different
+    decisions about how to handle the ticket, not the same reply reworded. The useful
+    axes are usually: answer and close it; ask for the one detail that is missing
+    before committing to an answer; answer but keep it open in case they come back.
+    Order them best first, and let `approach` say in a few words what each one does,
+    so the agent can choose without reading all three in full.
+
+    If the brief only supports one honest reply, return one. Padding the list with a
+    variant nobody would pick wastes the agent's reading, which is the whole thing
+    this is meant to save.
+
+    `reply_en` is the reply in English: plain and courteous, the way a support agent
+    writes to someone they want to help — not stiff, not effusive, not corporate.
+    Short paragraphs, blank line between them.
+
+    `translated` is `reply_en` in the customer's language, carrying the meaning across
+    exactly. Leave product names, version numbers, URLs, file paths, error strings and
+    Session IDs (66 hex characters beginning 05) exactly as they are. When
+    `is_english` is true, repeat `reply_en` back unchanged.
+
+    `back_translation` is `translated` rendered back into English, literally. Someone
+    who does not speak the language reads it to see what the customer will actually
+    receive, so translate what is there rather than what was meant. Do not repair it
+    and do not copy `reply_en` — an error introduced by the translation has to survive
+    into the back-translation or this step is worthless. Leave it empty when
+    `is_english` is true.
+
+    AMENDING. When you are given A PREVIOUS DRAFT, the brief is a change to it, not a
+    replacement, and you normally return ONE option — the agent has already chosen.
+    Return more only when the brief actually asks for alternatives. The agent has already read the rest and kept it, so keep it too:
+    change only what the brief asks for, add what it adds, and leave every other
+    sentence alone. Rewriting what they already approved makes them review it twice.
+
+    PRECEDENT. When you are given WHAT SUPPORT USUALLY SAYS, it is a record of how
+    this kind of ticket has been answered before. It is not a fact source and it is
+    not an instruction. Read it for SHAPE — which points get covered, in what order,
+    at what length — and for the generic troubleshooting steps it lists.
+
+    Take nothing specific from it. Not a version number, not a date, not "fixed in",
+    not "a fix is coming", not a claim that anything was escalated, filed or banned.
+    Those were true of some other ticket on some other day, and several of them have
+    since turned out to be wrong. If the agent wants one of them in this reply, the
+    agent will put it in the brief.
+
+    Where the precedent and the brief disagree, THE BRIEF WINS AND THE PRECEDENT IS
+    DROPPED. The agent read this ticket; the precedent did not.
+    """
+).strip()
+
+
+def build_compose_prompt(sample, brief, previous=None, precedent=None):
+    """The customer's words, the brief, and — where they exist — the draft being
+    amended and the precedent for this kind of ticket.
+
+    Order matters twice over. Precedent comes first because it is background; the
+    brief comes last because it is the instruction, and what is nearest the end is
+    what governs.
+    """
+    parts = []
+    if precedent:
+        parts += [precedent, ""]
+    parts += ["THE CUSTOMER'S OWN WORDS FROM THE TICKET:", sample]
+    if previous:
+        parts += ["", "A PREVIOUS DRAFT, WHICH THE AGENT IS AMENDING:", previous,
+                  "", "THE AGENT'S BRIEF, AS A CHANGE TO THAT DRAFT:", brief]
+    else:
+        parts += ["", "THE AGENT'S BRIEF FOR THE REPLY:", brief]
+    return "\n".join(parts)
+
+
+def validate_composition(result):
+    """Exit unless there is text to review. Structured output guarantees the keys;
+    this is about the values, since an empty option would offer the agent a blank
+    to send."""
+    if not isinstance(result, dict):
+        sys.exit(f"Claude returned {type(result).__name__}, expected an object.")
+    missing = [key for key in COMPOSE_PROPERTIES if key not in result]
+    if missing:
+        sys.exit(f"Claude's draft is missing: {', '.join(missing)}.")
+    options = [o for o in (result.get("options") or [])
+               if isinstance(o, dict) and (o.get("translated") or "").strip()]
+    if not options:
+        sys.exit("Claude returned no usable reply option.")
+    result["options"] = options[:MAX_OPTIONS]
+    return result
+
+
+def compose(model, sample, brief, previous=None, precedent=None):
+    return validate_composition(triage.claude_cli_json(
+        model, "medium", COMPOSE_SYSTEM, COMPOSE_SCHEMA,
+        build_compose_prompt(sample, brief, previous, precedent),
+        COMPOSE_TIMEOUT_SECONDS, "the reply draft"))
+
+
+# ---- Notes ------------------------------------------------------------------
+
+
+def build_draft_note(result, brief, comment_id, amended=False,
+                     cell=None, title=None, covering=None):
+    """The note the agent reviews, as HTML.
+
+    Each option sits in its own numbered verbatim block: those are what `reply` will
+    publish, so their whitespace has to survive Zendesk unchanged. Everything around
+    them is prose and gets paragraphs.
+
+    The instructions name the commands, so they are written mid-line on purpose:
+    COMMAND only matches at the start of a line, so this note cannot command itself
+    even if the trigger is misconfigured.
+    """
+    options = result["options"]
+    what = "revised the reply" if amended else "drafted a reply"
+    lead = (f"Claude {what}. It will be sent in {result['language']}.") if len(options) == 1 \
+        else (f"Claude {what} — {len(options)} options, in {result['language']}. "
+              f"Pick one with a private note reading claude: reply 2.")
+    out = [para(lead)]
+    for number, option in enumerate(options, 1):
+        out.append(para(f"Option {number} — {option.get('approach') or 'reply'}"))
+        out += [para(begin_marker(number)), verbatim(option["translated"].strip()),
+                para(end_marker(number))]
+        if not result["is_english"] and (option.get("back_translation") or "").strip():
+            out += [para(f"Option {number}, back in English:"),
+                    verbatim(option["back_translation"].strip())]
+    out += [para("Brief this was written from:" if not amended
+                 else "Change this revision was asked for:"),
+            verbatim(brief.strip())]
+    if cell:
+        out.append(para(f"Shaped by what we usually reply to: {title} ({covering}) — "
+                        f"{cell['n']} solved tickets, {cell['consistency']} consistency."))
+        if cell.get("examples"):
+            out.append(para("Past tickets: " + ", ".join(f"#{i}" for i in cell["examples"])))
+        if (cell.get("caveat") or "").strip():
+            out.append(para("Careful, from those past replies: " + cell["caveat"].strip()))
+    send = "add a private note — claude: reply" if len(options) == 1 \
+        else "add a private note — claude: reply "
+    out += [para(f"To send it exactly as above, {send}"),
+            para("To change it, add a private note — claude: draft "),
+            para(f"{done_marker(comment_id)} {draft_marker(comment_id)}")]
+    return "".join(out)
+
+
+def find_draft(comments, api_user):
+    """The options carried by the newest draft note, as {number: text}. {} if none.
+
+    Newest wins: an agent who did not like the first draft writes a new brief, and
+    `reply` must offer what they last looked at. Restricted to notes this tool wrote,
+    so a human pasting the delimiters into a note of their own cannot smuggle text
+    past the review.
+
+    reply.fetch_comments returns newest first, so this walks the list as it comes.
+    """
+    for comment in comments:
+        if comment.get("public") or comment.get("author_id") != api_user:
+            continue
+        found = {int(number): body.strip()
+                 for number, body in OPTION_BLOCK.findall(comment_text(comment))
+                 if body.strip()}
+        if found:
+            return found
+    return {}
+
+
+def choose_option(options, asked):
+    """(text, complaint) for the option the agent asked for.
+
+    A bare `claude: reply` sends the only option there is. With more than one it
+    refuses and lists them rather than guessing: picking for them would send a
+    customer a reply they did not choose.
+    """
+    numbers = sorted(options)
+    if asked is None:
+        if len(numbers) == 1:
+            return options[numbers[0]], None
+        return None, ("This draft offers " + str(len(numbers)) + " options ("
+                      + ", ".join(str(n) for n in numbers) + "). Say which one, "
+                      "like — claude: reply " + str(numbers[0]))
+    if asked not in options:
+        return None, (f"There is no option {asked} on this draft. It offers "
+                      + ", ".join(str(n) for n in numbers) + ".")
+    return options[asked], None
+
+
+OPTION_NUMBER = re.compile(r"^\s*#?(\d+)\b")
+
+
+def asked_option(brief):
+    """The option number in `claude: reply 2`, or None for a bare `claude: reply`."""
+    found = OPTION_NUMBER.match(brief or "")
+    return int(found.group(1)) if found else None
+
+
+def first_line(text, limit=90):
+    """A short quotation of what went out, for the audit note.
+
+    Enough to recognise the draft that was sent when two were drafted; never the
+    whole reply, which is already on the ticket one comment above.
+    """
+    line = next((part.strip() for part in (text or "").splitlines() if part.strip()), "")
+    return triage.clip(line, limit)
+
+
+def build_sent_note(user, comment_id, sent):
+    return "".join([
+        para(f"Reply sent by {user}, from their note on this ticket."),
+        para(f'Sent draft beginning: "{first_line(sent)}"'),
+        para(done_marker(comment_id)),
+    ])
+
+
+# ---- Actions ----------------------------------------------------------------
+
+
+def run_draft(session, subdomain, model, ticket, comments, command, api_user, dry_run):
+    """Compose the reply, or revise the one already on the ticket.
+
+    A second `draft` on a ticket that already has one is an amendment, not a fresh
+    start: an agent writing "also mention X" wants the draft they just read plus X,
+    and regenerating from the new brief alone would throw away the wording they kept.
+    """
+    ticket_id = ticket["id"]
+    comment_id, brief = command["id"], command["brief"]
+    if not brief:
+        say(session, subdomain, ticket_id, comment_id,
+            "The brief was empty, so there was nothing to write. Add a private note "
+            "like — claude: draft ", dry_run)
+        return
+    shown = find_draft(comments, api_user)
+    previous = "\n\n".join(f"Option {n}:\n{shown[n]}" for n in sorted(shown)) or None
+
+    sample = customer_sample(session, subdomain, ticket, comments)
+    book = load_house()
+    group, platform = tagged_placement(ticket)
+    new_tags = []
+    if book and not group:
+        group, platform = place_ticket(model, book, ticket, sample)
+        # Cached on the ticket so a revision does not pay for the same call again,
+        # and so the placement is visible to a human who disagrees with it.
+        new_tags = ([f"{TAG_GROUP_PREFIX}{group}"] if group else []) + \
+                   ([f"{TAG_PLATFORM_PREFIX}{platform}"] if platform else [])
+    cell, covering = house_cell(book, group, platform)
+    title = next((g["title"] for g in (book or {}).get("groups", [])
+                  if g["key"] == group), group)
+    precedent = render_precedent(cell, title, covering) if cell else None
+    if cell:
+        print(f"#{ticket_id}: grounded in {group}/{covering} "
+              f"({cell['n']} solved, {cell['consistency']} consistency).")
+
+    result = compose(model, triage.clip(sample, CUSTOMER_SAMPLE_CHARS), brief,
+                     previous, precedent)
+    print(f"#{ticket_id}: {'revised' if previous else 'drafted'} "
+          f"{len(result['options'])} option(s); requester writes "
+          f"{result['language']!r} ({result['language_code']}).")
+    if dry_run:
+        print(f"#{ticket_id}: dry run, nothing written.")
+        return
+    write_to_ticket(session, subdomain, ticket_id,
+                    build_draft_note(result, brief, comment_id, bool(previous),
+                                     cell, title, covering),
+                    public=False, as_html=True,
+                    add_tags=[TAG_DRAFTED] + new_tags,
+                    drop_tags=[TAG_QUEUED, TAG_ERROR])
+
+
+def run_solve(session, subdomain, ticket, command, dry_run):
+    """Solve the ticket without writing anything to the customer.
+
+    For the ones that need no reply — spam, an abuse report with nothing actionable,
+    a duplicate, a question already answered elsewhere. There were 197 of those in
+    the backlog when this was written.
+
+    The private note is the point: a status change on its own leaves nothing on the
+    ticket saying who decided that or why, which is exactly what somebody reopening
+    it in three months needs to know.
+    """
+    ticket_id = ticket["id"]
+    if ticket.get("status") == SOLVED_STATUS:
+        say(session, subdomain, ticket_id, command["id"],
+            "This ticket is already solved.", dry_run, error=False)
+        return
+    author = fetch_user(session, subdomain, command["author"])
+    who = author.get("name") or f"user {command['author']}"
+    if dry_run:
+        print(f"#{ticket_id}: dry run, would solve on behalf of {who}.")
+        return
+    note = [para(f"Solved by {who}, from their note on this ticket. "
+                 f"No reply was sent to the customer.")]
+    if command["brief"]:
+        note.append(para("Reason given: " + command["brief"]))
+    note.append(para(done_marker(command["id"])))
+    write_to_ticket(session, subdomain, ticket_id, "".join(note), public=False,
+                    status=SOLVED_STATUS, as_html=True, add_tags=[TAG_SOLVED],
+                    drop_tags=[TAG_QUEUED, TAG_ERROR])
+    print(f"#{ticket_id}: solved, no reply sent.")
+
+
+def run_explain(session, subdomain, model, ticket, comments, command, dry_run):
+    """Say what we already know about this kind of ticket, without writing a reply.
+
+    The house answer, the steps, and — the reason this verb exists — what was
+    actually DONE before: fixes confirmed shipped, bugs filed, escalations. Those are
+    the things `draft` deliberately refuses to put in a reply, because they were true
+    of another ticket on another day. Here they are shown to a human, who can decide
+    whether one still applies and put it in the brief.
+    """
+    ticket_id = ticket["id"]
+    book = load_house()
+    if not book:
+        say(session, subdomain, ticket_id, command["id"],
+            "No house answers are configured on this relay, so there is nothing to "
+            "look up.", dry_run)
+        return
+    group, platform = tagged_placement(ticket)
+    new_tags = []
+    if not group:
+        group, platform = place_ticket(
+            model, book, ticket, customer_sample(session, subdomain, ticket, comments))
+        new_tags = ([f"{TAG_GROUP_PREFIX}{group}"] if group else []) + \
+                   ([f"{TAG_PLATFORM_PREFIX}{platform}"] if platform else [])
+    cell, covering = house_cell(book, group, platform)
+    if not cell:
+        say(session, subdomain, ticket_id, command["id"],
+            "This ticket does not match anything in the house answers, so there is "
+            "no precedent to show. Write the brief yourself.", dry_run, error=False)
+        return
+    title = next((g["title"] for g in book["groups"] if g["key"] == group), group)
+    print(f"#{ticket_id}: {group}/{covering}, {cell['n']} solved.")
+    if dry_run:
+        print(f"#{ticket_id}: dry run, nothing written.")
+        return
+    out = [para(f"What we usually reply to: {title} ({covering}) — "
+                f"{cell['n']} solved tickets, {cell['consistency']} consistency."),
+           para(cell["answer"])]
+    if cell.get("steps"):
+        out.append(para("Steps usually given: " + "; ".join(cell["steps"])))
+    if cell.get("actions"):
+        out.append(bold_para("What was actually done on those tickets:"))
+        out += [para(action) for action in cell["actions"]]
+    if (cell.get("caveat") or "").strip():
+        out.append(para("Careful: " + cell["caveat"].strip()))
+    if cell.get("examples"):
+        out.append(para("Verify against " + ", ".join(f"#{i}" for i in cell["examples"])))
+    out.append(para("Nothing was written to the customer. To answer, add a private "
+                    "note — claude: draft "))
+    out.append(para(done_marker(command["id"])))
+    write_to_ticket(session, subdomain, ticket_id, "".join(out), public=False,
+                    as_html=True, add_tags=new_tags, drop_tags=[TAG_QUEUED, TAG_ERROR])
+
+
+def run_reply(session, subdomain, ticket, comments, command, api_user, dry_run):
+    """Publish the chosen option, exactly as it was reviewed.
+
+    It never re-composes. What the agent read in the draft note is what the customer
+    receives, or the review step is theatre — so a change means a new brief, not a
+    different send.
+    """
+    ticket_id = ticket["id"]
+    comment_id = command["id"]
+    options = find_draft(comments, api_user)
+    if not options:
+        say(session, subdomain, ticket_id, comment_id,
+            "There is no draft on this ticket to send. Add a private note like — "
+            "claude: draft ", dry_run)
+        return
+    sending, complaint = choose_option(options, asked_option(command["brief"]))
+    if complaint:
+        say(session, subdomain, ticket_id, comment_id, complaint, dry_run, error=False)
+        return
+    author = fetch_user(session, subdomain, command["author"])
+    who = author.get("name") or f"user {command['author']}"
+    if dry_run:
+        print(f"#{ticket_id}: dry run, would send an option on behalf of {who}.")
+        return
+    write_to_ticket(session, subdomain, ticket_id, sending, public=True,
+                    status=REPLIED_STATUS)
+    write_to_ticket(session, subdomain, ticket_id,
+                    build_sent_note(who, comment_id, sending), public=False,
+                    as_html=True, add_tags=[TAG_SENT],
+                    drop_tags=[TAG_QUEUED, TAG_DRAFTED, TAG_ERROR])
+    print(f"#{ticket_id}: reply sent and status -> {REPLIED_STATUS}.")
+
+
+def already_english(turns, translated):
+    """Whether the translation came back as the text it was given.
+
+    Asked after the call rather than guessed before it. A character test looked
+    cheaper, but "Hallo, ich habe ein Problem" is pure ASCII — it would have called
+    every unaccented German ticket English and left the agent unable to read the
+    thing they asked to read. Judging by the output costs one model call and cannot
+    make that mistake.
+
+    Any turn the model did not return, or returned changed, means translating
+    happened — so this fails towards posting the transcript.
+    """
+    english = {}
+    for item in translated or []:
+        try:
+            english[int(item.get("index"))] = (item.get("english") or "").strip()
+        except (TypeError, ValueError):
+            continue
+    squash = lambda text: " ".join((text or "").split()).lower()
+    return all(english.get(turn["index"])
+               and squash(english[turn["index"]]) == squash(turn["body"])
+               for turn in turns)
+
+
+def run_english(session, subdomain, model, ticket, comments, command, dry_run):
+    """Put the conversation on the ticket in English, as a private note.
+
+    Reads both sides, not just the customer's: their second message is usually an
+    answer to a reply, and without the reply it reads as a complaint about nothing.
+
+    Python owns the speaker labels and timestamps and the model only translates —
+    the same split triage.py makes, for the same reason: a model asked to format the
+    transcript can drop a turn, merge two, or date one it was never given, and each
+    of those is invisible in the output.
+    """
+    ticket_id = ticket["id"]
+    latest = next((c.get("id") for c in comments if c.get("public")), None)
+    if latest is not None and reply.already_replied(comments, english_marker(latest)):
+        say(session, subdomain, ticket_id, command["id"],
+            "The English transcript on this ticket is already up to date — nothing "
+            "has been said since it was written.", dry_run, error=False)
+        return
+    turns = triage.conversation_turns(session, subdomain, ticket)
+    if not turns:
+        say(session, subdomain, ticket_id, command["id"],
+            "There are no public comments on this ticket to translate.", dry_run)
+        return
+
+    payload = json.dumps([{"index": t["index"], "speaker": t["who"], "text": t["body"]}
+                          for t in turns], ensure_ascii=False)
+    rendered = triage.claude_cli_json(
+        model, "medium", triage.TRANSCRIPT_SYSTEM_PROMPT, triage.TRANSCRIPT_SCHEMA,
+        triage.clip(payload, triage.TRANSCRIPT_INPUT_CHARS),
+        triage.ENGLISH_TIMEOUT_SECONDS, f"the English transcript of #{ticket_id}")
+    if already_english(turns, rendered.get("turns")):
+        # A transcript of English text repeats what is already a few comments above
+        # it. Say so rather than posting the same words back.
+        say(session, subdomain, ticket_id, command["id"],
+            "This conversation is already in English, so there is nothing to "
+            "translate.", dry_run, error=False)
+        return
+    print(f"#{ticket_id}: rendered {len(turns)} turn(s) in English.")
+    if dry_run:
+        print(f"#{ticket_id}: dry run, nothing written.")
+        return
+    note = "".join(
+        [para(f"This conversation in English — {len(turns)} turn(s), both sides."),
+         para("Translated for reading; the customer has not seen this.")]
+        + transcript_blocks(turns, rendered.get("turns"))
+        + [para(f"{done_marker(command['id'])} {english_marker(latest)}")])
+    write_to_ticket(session, subdomain, ticket_id, note, public=False, as_html=True,
+                    drop_tags=[TAG_QUEUED, TAG_ERROR])
+
+
+def say(session, subdomain, ticket_id, comment_id, text, dry_run, error=True):
+    """Answer the agent on the ticket, and record that this command was handled.
+
+    Every refusal carries the done marker: a command that cannot be satisfied is
+    still a command that was answered, and without the marker a webhook retry would
+    ask again and post the same complaint a second time.
+
+    `error=False` for an answer that is not a failure — "there is nothing new to
+    translate" is the command working. Tagging that `claude-error` would put a
+    working ticket in the queue of broken ones.
+    """
+    print(f"#{ticket_id}: {text.splitlines()[0]}")
+    if dry_run:
+        return
+    write_to_ticket(session, subdomain, ticket_id,
+                    para(text) + para(done_marker(comment_id)), public=False,
+                    as_html=True, add_tags=[TAG_ERROR] if error else [],
+                    drop_tags=[TAG_QUEUED] + ([] if error else [TAG_ERROR]))
+
+
+# ---- Entry point ------------------------------------------------------------
+
+
+def latest_command(comments, api_user, session, subdomain):
+    """The newest private note that is a command from someone allowed to give one.
+
+    reply.fetch_comments returns newest first, so the first match is the newest
+    command: older ones have already been handled and carry their own done markers.
+
+    An unauthorised author stops the search rather than falling through to an older
+    command. Their note is the most recent instruction on the ticket, and quietly
+    acting on a previous one instead would be a surprising thing to do.
+    """
+    for comment in comments:
+        if comment.get("public") or comment.get("author_id") == api_user:
+            continue
+        parsed = parse_command(comment_text(comment))
+        if not parsed:
+            continue
+        user = fetch_user(session, subdomain, comment.get("author_id"))
+        if not may_command(user):
+            print(f"Ignoring a command from {user.get('role') or 'an unknown user'}.")
+            return None
+        return {"id": comment.get("id"), "author": comment.get("author_id"),
+                "action": parsed[0], "brief": parsed[1]}
+    return None
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Act on claude: notes on a Zendesk ticket.")
+    parser.add_argument("--ticket", type=int, required=True)
+    parser.add_argument("--model", default=os.environ.get("ZENDESK_NOTE_MODEL", DEFAULT_MODEL))
+    parser.add_argument("--dry-run", action="store_true",
+                        help="do everything except write to Zendesk")
+    args = parser.parse_args()
+
+    subdomain = triage.get_env("ZENDESK_SUBDOMAIN")
+    session = triage.zendesk_session(triage.get_env("ZENDESK_EMAIL"),
+                                     triage.get_env("ZENDESK_API_TOKEN"))
+    api_user = api_user_id(session, subdomain)
+
+    ticket = reply.fetch_ticket(session, subdomain, args.ticket)
+    if ticket.get("status") == "closed":
+        # Closed is irreversible and takes no comments at all, so there is nowhere to
+        # even report the refusal. Say it to the journal and stop.
+        print(f"#{args.ticket}: closed, so Zendesk takes no comments. Nothing done.")
+        return
+    comments = reply.fetch_comments(session, subdomain, args.ticket)
+
+    command = latest_command(comments, api_user, session, subdomain)
+    if not command:
+        print(f"#{args.ticket}: no command note to act on.")
+        clear_queued(session, subdomain, args.ticket, args.dry_run)
+        return
+    if reply.already_replied(comments, done_marker(command["id"])):
+        print(f"#{args.ticket}: this command was already handled; nothing written.")
+        clear_queued(session, subdomain, args.ticket, args.dry_run)
+        return
+
+    if command["action"] == "draft":
+        run_draft(session, subdomain, args.model, ticket, comments, command,
+                  api_user, args.dry_run)
+    elif command["action"] == "solve":
+        run_solve(session, subdomain, ticket, command, args.dry_run)
+    elif command["action"] == "explain":
+        run_explain(session, subdomain, args.model, ticket, comments, command,
+                    args.dry_run)
+    elif command["action"] == "english":
+        run_english(session, subdomain, args.model, ticket, comments, command, args.dry_run)
+    else:
+        run_reply(session, subdomain, ticket, comments, command, api_user, args.dry_run)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/zendesk_triage/relay.py b/zendesk_triage/relay.py
index 813ff40..ea14d8b 100644
--- a/zendesk_triage/relay.py
+++ b/zendesk_triage/relay.py
@@ -1,9 +1,12 @@
 #!/usr/bin/env python3
 """
-Discord interactions endpoint for replying to Zendesk tickets.
+Webhook endpoints for replying to Zendesk tickets.
 
-Serves POST /discord/interactions — vendor-namespaced so the same host and
-certificate can carry other integrations later.
+Two routes, vendor-namespaced so the same host and certificate carry both:
+
+  POST /discord/interactions  a Comment button on a digest card, handed to reply.py
+  POST /zendesk/notes         a `claude:` private note on a ticket, handed to
+                              note_reply.py
 
 Discord pushes every interaction to one HTTPS endpoint and wants an answer inside
 three seconds. This is that endpoint: it verifies the signature, decides who is
@@ -36,8 +39,11 @@
     ZENDESK_SUBDOMAIN     e.g. "mycompany"
     ZENDESK_EMAIL         agent email for API token auth
     ZENDESK_API_TOKEN     Zendesk API token
-    RELAY_DRY_RUN         (optional) "1" passes --dry-run to reply.py, so the whole
-                          path runs and nothing is written to Zendesk
+    ZENDESK_WEBHOOK_SECRET  shared secret Zendesk signs its webhooks with, for the
+                          /zendesk/notes route. Unset refuses every note webhook
+    RELAY_DRY_RUN         (optional) "1" passes --dry-run to reply.py and
+                          note_reply.py, so the whole path runs and nothing is
+                          written to Zendesk
 
 Requests are rejected unless their signature is valid and their timestamp is within
 MAX_SIGNATURE_AGE_SECONDS, so a captured request cannot be replayed later. A timestamp
@@ -48,6 +54,10 @@
     uvicorn relay:app --host 127.0.0.1 --port 8080
 """
 import asyncio
+import base64
+import datetime
+import hashlib
+import hmac
 import json
 import os
 import subprocess
@@ -64,6 +74,7 @@
 import triage  # noqa: E402  (needs the path insert above)
 
 REPLY_SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reply.py")
+NOTE_SCRIPT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "note_reply.py")
 
 # Bounds the reply so its translation and back-translation still fit Discord's
 # 6,000-character budget across the preview's embeds. reply.py refuses anything that
@@ -93,6 +104,14 @@
 # apiece, so a legitimate run can outlast this — which is why overrunning it reports
 # to the agent rather than only to the journal. See tell_discord.
 REPLY_TIMEOUT_SECONDS = 300
+# How long one note_reply.py run may take. Longer than the Discord path's need for a
+# 15-minute interaction token allows, because nothing is waiting on it: Zendesk gets
+# its 200 immediately and the outcome is written to the ticket whenever it lands.
+NOTE_TIMEOUT_SECONDS = 420
+# Zendesk signs a webhook over the timestamp followed by the raw body. Its clock is
+# not ours, so the same skew allowance as the Discord path applies.
+ZENDESK_SIGNATURE_HEADER = "x-zendesk-webhook-signature"
+ZENDESK_TIMESTAMP_HEADER = "x-zendesk-webhook-signature-timestamp"
 
 INTERACTION_PING = 1
 INTERACTION_COMPONENT = 3
@@ -489,6 +508,57 @@ def dry_run_requested():
     return bool(value) and value.split()[0] not in ("0", "false", "no", "off")
 
 
+def run_note_reply(ticket_id):
+    """Run note_reply.py over one ticket. Never raises.
+
+    A subprocess for the same reason reply.py is one: its error paths are sys.exit
+    calls, and SystemExit would otherwise escape into a long-running service.
+
+    There is nobody to report a failure to here — the agent is looking at a Zendesk
+    ticket, not at an open dialog — so the outcome goes to the journal and, where
+    note_reply.py got far enough to write one, to the ticket itself.
+    """
+    command = [sys.executable, NOTE_SCRIPT, "--ticket", str(ticket_id)]
+    if dry_run_requested():
+        command.append("--dry-run")
+    try:
+        done = subprocess.run(command, check=False, timeout=NOTE_TIMEOUT_SECONDS,
+                              capture_output=True, text=True)
+    except subprocess.TimeoutExpired:
+        print(f"note_reply.py timed out on #{ticket_id}", flush=True)
+        return
+    for line in (done.stdout or "").splitlines():
+        print(line, flush=True)
+    if done.returncode != 0:
+        print(f"note_reply.py exited {done.returncode} on #{ticket_id}: "
+              f"{(done.stderr or '').strip()[:300]}", flush=True)
+
+
+def zendesk_signature_ok(raw, signature, timestamp, secret):
+    """Whether Zendesk signed this body, recently.
+
+    HMAC-SHA256 over the timestamp followed by the raw body, base64 encoded. The
+    timestamp is ISO 8601 rather than the epoch seconds the Discord path uses, and it
+    is covered by the signature — so the age check bounds replay of a genuinely
+    signed request, exactly as it does there.
+    """
+    if not (signature and timestamp and secret):
+        return False
+    try:
+        when = datetime.datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+    except (AttributeError, TypeError, ValueError):
+        return False
+    age = time.time() - when.timestamp()
+    if age < -MAX_CLOCK_SKEW_SECONDS or age > MAX_SIGNATURE_AGE_SECONDS:
+        return False
+    expected = hmac.new(secret.encode(), timestamp.encode() + raw, hashlib.sha256).digest()
+    try:
+        given = base64.b64decode(signature, validate=True)
+    except (ValueError, TypeError):
+        return False
+    return hmac.compare_digest(expected, given)
+
+
 def interaction_payload(interaction, action, **extra):
     return {
         "action": action,
@@ -578,6 +648,36 @@ def healthz():
     return {"ok": True}
 
 
+@app.post("/zendesk/notes")
+async def zendesk_notes(request: Request, background: BackgroundTasks):
+    """A private note on a ticket asked Claude to do something.
+
+    The Zendesk trigger that calls this is the first gate — it fires only on private
+    comments, and must exclude the API user so a draft cannot trigger another draft.
+    This is the second: without ZENDESK_WEBHOOK_SECRET set, nothing is accepted, so
+    an unconfigured relay refuses rather than trusting anything that reaches the URL.
+    Which Zendesk user may command it is note_reply.py's decision, on the note.
+
+    Answers immediately and works in the background. Zendesk retries a webhook that
+    does not answer quickly, and a retry that arrives mid-run would be a second reply
+    to the customer — note_reply.py's done marker covers that, but not needing the
+    cover is better.
+    """
+    raw = await request.body()
+    if not zendesk_signature_ok(raw, request.headers.get(ZENDESK_SIGNATURE_HEADER),
+                                request.headers.get(ZENDESK_TIMESTAMP_HEADER),
+                                env("ZENDESK_WEBHOOK_SECRET")):
+        return Response("bad signature", status_code=401)
+    try:
+        ticket_id = ticket_number((json.loads(raw) or {}).get("ticket_id"))
+    except (ValueError, TypeError):
+        ticket_id = None
+    if not ticket_id:
+        return Response("no ticket id", status_code=400)
+    background.add_task(run_in_threadpool, run_note_reply, ticket_id)
+    return {"ok": True}
+
+
 @app.post("/discord/interactions")
 async def interactions(request: Request, background: BackgroundTasks):
     raw = await request.body()
diff --git a/zendesk_triage/test_note_reply.py b/zendesk_triage/test_note_reply.py
new file mode 100644
index 0000000..69fb656
--- /dev/null
+++ b/zendesk_triage/test_note_reply.py
@@ -0,0 +1,884 @@
+#!/usr/bin/env python3
+"""Tests for driving a Zendesk reply from private notes on the ticket.
+
+Stdlib unittest, offline, same stubs as the rest. Run from anywhere:
+
+    python -m unittest discover -s zendesk_triage -v
+
+This publishes comments that email a customer, so the tests are about the guards
+rather than the happy path. Four of them exist because getting them wrong is how this
+sends the wrong thing to a real person:
+
+  * a draft note must not read as a command, or Claude drafts against itself forever
+  * `reply` must send the newest draft, not the oldest — the comment feed is newest
+    first and reversing it is an easy and silent mistake
+  * a replayed webhook must not send twice
+  * `reply` must send what was reviewed, byte for byte
+"""
+import os
+import re
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import note_reply  # noqa: E402
+import triage  # noqa: E402
+from test_triage import FakeResponse, FakeSession, Patched  # noqa: E402
+
+API_USER = 901790886886
+AGENT = 555
+def option(text="Anhänge werden 14 Tage lang gespeichert.", approach="explain and close"):
+    return {"approach": approach, "reply_en": "Attachments are kept for 14 days.",
+            "translated": text, "back_translation": "Attachments are stored for 14 days."}
+
+
+GERMAN = {"language": "German", "language_code": "de", "is_english": False,
+          "options": [option()]}
+GERMAN_THREE = {"language": "German", "language_code": "de", "is_english": False,
+                "options": [option("Erste Antwort", "explain and close"),
+                            option("Zweite Antwort", "ask which device was online"),
+                            option("Dritte Antwort", "answer and keep it open")]}
+
+
+def fake_session(*responses):
+    """A stub session with room for the tag sub-resource calls a write now makes.
+
+    Tag changes go through PUT/DELETE /tickets/{id}/tags.json rather than fields on
+    the ticket update, so every write costs up to two extra requests.
+    """
+    return FakeSession(list(responses) + [FakeResponse({}) for _ in range(6)])
+
+
+def comment(body, author=AGENT, public=False, cid=1):
+    return {"id": cid, "author_id": author, "public": public,
+            "body": body, "plain_body": body}
+
+
+def as_zendesk_plain(markup):
+    """`plain_body` as Zendesk actually returns it for a comment posted as html_body.
+
+    Measured against the live API rather than assumed: block tags become line breaks,
+    every other tag is dropped, whitespace inside 
 survives exactly — and
+    entities are NOT unescaped, which is why note_reply.comment_text unescapes. A
+    fixture that skipped this passed on raw markup and would have shipped a
+    find_draft that returned "
…" to a customer.
+    """
+    text = re.sub(r"", "\n", markup)
+    return re.sub(r"<[^>]+>", "", text)
+
+
+def posted_note(markup, author=API_USER, cid=9):
+    """A note this tool wrote, as it reads back off the ticket."""
+    return {"id": cid, "author_id": author, "public": False,
+            "body": markup, "plain_body": as_zendesk_plain(markup)}
+
+
+def draft_note(*texts, cid=9, marker_for=1):
+    """A draft note as it reads back, carrying one numbered option per text."""
+    blocks = [note_reply.para("Claude drafted a reply.")]
+    for number, text in enumerate(texts, 1):
+        blocks += [note_reply.para(note_reply.begin_marker(number)),
+                   note_reply.verbatim(text),
+                   note_reply.para(note_reply.end_marker(number))]
+    blocks.append(note_reply.para(f"{note_reply.done_marker(marker_for)} "
+                                  f"{note_reply.draft_marker(marker_for)}"))
+    return posted_note("".join(blocks), cid=cid)
+
+
+class ParseCommand(unittest.TestCase):
+    def test_reads_action_and_brief(self):
+        self.assertEqual(note_reply.parse_command("claude: draft - keeps 14 days"),
+                         ("draft", "keeps 14 days"))
+
+    def test_tolerates_spacing_case_and_dashes(self):
+        for text in ("claude:draft x", "Claude : DRAFT — x", "  claude: draft: x",
+                     "*claude: draft* x".replace("*", "", 1)):
+            with self.subTest(text=text):
+                self.assertEqual(note_reply.parse_command(text)[0], "draft")
+
+    def test_brief_runs_to_the_end_of_the_note(self):
+        action, brief = note_reply.parse_command("claude: draft - one\ntwo\nthree")
+        self.assertEqual((action, brief), ("draft", "one\ntwo\nthree"))
+
+    def test_reply_takes_no_brief(self):
+        self.assertEqual(note_reply.parse_command("claude: reply"), ("reply", ""))
+
+    def test_ignores_a_note_that_is_not_a_command(self):
+        self.assertIsNone(note_reply.parse_command("we should tell them 14 days"))
+
+    def test_only_matches_at_the_start_of_a_line(self):
+        """The loop guard. Claude's own draft note names both commands in its
+        instructions; if those parsed, every draft would command another draft."""
+        self.assertIsNone(note_reply.parse_command(
+            "To send it exactly as above, add a private note — claude: reply"))
+
+    def test_a_generated_draft_note_is_not_a_command(self):
+        note = note_reply.build_draft_note(GERMAN, "keeps 14 days", 42)
+        self.assertIsNone(note_reply.parse_command(
+            note_reply.comment_text(posted_note(note))))
+
+
+class EnglishTranscript(unittest.TestCase):
+    def test_english_is_a_command(self):
+        self.assertEqual(note_reply.parse_command("claude: english"), ("english", ""))
+
+    def test_the_marker_tracks_the_newest_public_comment(self):
+        """Keyed on the conversation, not the command: asking twice with nothing said
+        in between must cost nothing, and asking after a reply must re-render."""
+        self.assertNotEqual(note_reply.english_marker(1), note_reply.english_marker(2))
+
+    def test_an_up_to_date_transcript_is_not_re_rendered(self):
+        """The expensive half is the model call. A repeat ask with no new comment
+        must not reach it."""
+        called = []
+        prior = comment(f"transcript\n\n{note_reply.english_marker(7)}",
+                        author=API_USER, cid=8)
+        comments = [comment("claude: english", cid=9),
+                    dict(comment("hallo", author=42, cid=7), public=True), prior]
+        with Patched(triage, conversation_turns=lambda *a: called.append(a)):
+            session = fake_session(*[FakeResponse({"ticket": {}})])
+            note_reply.run_english(session, "sub", "model", {"id": 7}, comments,
+                                   {"id": 9, "author": AGENT, "action": "english",
+                                    "brief": ""}, dry_run=False)
+        self.assertEqual(called, [])
+
+    def test_nothing_new_is_not_an_error(self):
+        """`claude-error` is the queue of broken tickets. A working command that had
+        nothing to do does not belong in it."""
+        prior = comment(f"transcript\n\n{note_reply.english_marker(7)}",
+                        author=API_USER, cid=8)
+        comments = [comment("claude: english", cid=9),
+                    dict(comment("hallo", author=42, cid=7), public=True), prior]
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        with Patched(triage, conversation_turns=lambda *a: None):
+            note_reply.run_english(session, "sub", "model", {"id": 7}, comments,
+                                   {"id": 9, "author": AGENT, "action": "english",
+                                    "brief": ""}, dry_run=False)
+        # the note itself carries no tag fields; the tag work is separate calls
+        self.assertNotIn("additional_tags", session.calls[0][2]["json"]["ticket"])
+        dropped = [kw["json"]["tags"] for m, u, kw in session.calls
+                   if m == "DELETE" and "/tags.json" in u]
+        self.assertTrue(any(note_reply.TAG_ERROR in names for names in dropped))
+        added = [kw["json"]["tags"] for m, u, kw in session.calls
+                 if m == "PUT" and "/tags.json" in u]
+        self.assertEqual(added, [], "a working command must not be tagged an error")
+
+    def test_an_english_ticket_gets_no_transcript(self):
+        """A transcript of English text repeats what is already on the ticket."""
+        turns = [{"index": 0, "who": "Customer", "when": "t", "body": "My app crashes"}]
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        with Patched(triage, conversation_turns=lambda *a: turns,
+                     claude_cli_json=lambda *a, **k: {
+                         "turns": [{"index": 0, "english": "My app crashes"}]}):
+            note_reply.run_english(session, "sub", "model", {"id": 7},
+                                   [comment("claude: english", cid=9)],
+                                   {"id": 9, "author": AGENT, "action": "english",
+                                    "brief": ""}, dry_run=False)
+        body = session.calls[0][2]["json"]["ticket"]["comment"]["html_body"]
+        self.assertIn("already in English", body)
+        self.assertEqual(session.calls[0][2]["json"]["ticket"].get("additional_tags", []), [])
+
+    def test_unaccented_german_is_still_translated(self):
+        """The reason this is checked after the call, not guessed before it:
+        "Hallo, ich habe ein Problem" is pure ASCII, and a character test would have
+        called it English and left the agent unable to read it."""
+        turns = [{"index": 0, "who": "Customer", "when": "t",
+                  "body": "Hallo, ich habe ein Problem"}]
+        self.assertFalse(note_reply.already_english(
+            turns, [{"index": 0, "english": "Hello, I have a problem"}]))
+
+    def test_a_turn_the_model_dropped_counts_as_needing_translation(self):
+        turns = [{"index": 0, "who": "Customer", "when": "t", "body": "Hallo"},
+                 {"index": 1, "who": "Support", "when": "t", "body": "Hi"}]
+        self.assertFalse(note_reply.already_english(
+            turns, [{"index": 0, "english": "Hallo"}]))
+
+    def test_whitespace_and_case_do_not_count_as_a_translation(self):
+        turns = [{"index": 0, "who": "Customer", "when": "t", "body": "My  app\ncrashes"}]
+        self.assertTrue(note_reply.already_english(
+            turns, [{"index": 0, "english": "my app crashes"}]))
+
+    def test_a_turn_with_blank_lines_stays_one_turn(self):
+        """The first version split render_transcript's output on blank lines, which
+        tore a single multi-paragraph message into a row of disconnected boxes."""
+        turns = [{"index": 0, "who": "Customer", "when": "2026-09-03 10:00 UTC",
+                  "body": "Hallo,\n\nzweiter Absatz.\n\ndritter Absatz."}]
+        blocks = note_reply.transcript_blocks(turns, [])
+        self.assertEqual(sum(1 for b in blocks if "" in b), 1,
+                         "one speaker line per turn, not one per paragraph")
+        self.assertEqual(len(blocks), 4)  # speaker line + three paragraphs
+
+    def test_the_transcript_is_prose_not_code_blocks(self):
+        """
 is for text that gets extracted and sent byte for byte. Nothing
+        extracts a transcript, and a code box is the wrong shape for prose."""
+        turns = [{"index": 0, "who": "Customer", "when": "", "body": "Hallo"}]
+        self.assertNotIn("
", "".join(note_reply.transcript_blocks(turns, [])))
+
+    def test_each_turn_gets_its_speaker_line(self):
+        turns = [{"index": 0, "who": "Customer", "when": "t1", "body": "a"},
+                 {"index": 1, "who": "Support", "when": "t2", "body": "b"}]
+        blocks = note_reply.transcript_blocks(turns, [{"index": 1, "english": "B"}])
+        joined = "".join(blocks)
+        self.assertIn("Customer:", joined)
+        self.assertIn("Support:", joined)
+        self.assertIn("B", joined)          # translated turn used
+        self.assertIn("a", joined)          # untranslated turn falls back to original
+
+    def test_the_transcript_note_is_never_public(self):
+        turns = [{"index": 0, "who": "Customer", "when": "2026-09-03 10:00 UTC",
+                  "body": "Hallo"}]
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        with Patched(triage, conversation_turns=lambda *a: turns,
+                     claude_cli_json=lambda *a, **k: {"turns": [{"index": 0,
+                                                                "english": "Hello"}]}):
+            note_reply.run_english(session, "sub", "model", {"id": 7},
+                                   [dict(comment("Hallo", author=42, cid=3), public=True)],
+                                   {"id": 9, "author": AGENT, "action": "english",
+                                    "brief": ""}, dry_run=False)
+        ticket = session.calls[0][2]["json"]["ticket"]
+        self.assertIs(ticket["comment"]["public"], False)
+        self.assertIn("Hello", ticket["comment"]["html_body"])
+        self.assertIn(note_reply.english_marker(3), ticket["comment"]["html_body"])
+
+
+BOOK = {
+    "groups": [{"key": "attachments", "title": "Attachments fail", "platform_sensitive": True}],
+    "cells": {
+        "attachments|android": {"answer": "We usually explain the 14-day window.",
+                                "steps": ["Ask for the app version"], "actions": [],
+                                "caveat": "A fix was promised in Nov 2025 and never shipped.",
+                                "consistency": "high", "n": 12, "examples": [111, 222]},
+        "attachments|any": {"answer": "All-platform version.", "steps": [], "actions": [],
+                            "caveat": "", "consistency": "medium", "n": 30, "examples": []},
+    },
+}
+
+
+class HouseAnswers(unittest.TestCase):
+    def test_absent_config_turns_the_feature_off(self):
+        """Drafting must work exactly as before on a host with no knowledge file."""
+        with Patched(os, environ={k: v for k, v in os.environ.items()
+                                  if k != note_reply.HOUSE_ENV}):
+            self.assertIsNone(note_reply.load_house())
+
+    def test_a_corrupt_file_degrades_rather_than_fails(self):
+        """A bad knowledge file must cost a thinner draft, never the ability to
+        answer a customer."""
+        import tempfile
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
+            handle.write("{not json")
+        self.assertIsNone(note_reply.load_house(handle.name))
+        os.unlink(handle.name)
+
+    def test_falls_back_to_the_all_platform_answer(self):
+        """A small group has no per-platform answer, and the general one still beats
+        nothing."""
+        cell, covering = note_reply.house_cell(BOOK, "attachments", "ios")
+        self.assertEqual(covering, "any")
+        self.assertEqual(cell["n"], 30)
+
+    def test_prefers_the_platform_specific_answer(self):
+        cell, covering = note_reply.house_cell(BOOK, "attachments", "android")
+        self.assertEqual((covering, cell["n"]), ("android", 12))
+
+    def test_an_unplaced_ticket_has_no_precedent(self):
+        self.assertEqual(note_reply.house_cell(BOOK, None, "android"), (None, None))
+
+    def test_a_cached_placement_skips_the_classifier(self):
+        """The tags are written on the first draft so a revision does not pay for the
+        same model call twice."""
+        ticket = {"id": 7, "tags": ["grp-attachments", "plat-android", "relay-test"]}
+        self.assertEqual(note_reply.tagged_placement(ticket), ("attachments", "android"))
+
+    def test_no_placement_tags_means_no_cache(self):
+        self.assertEqual(note_reply.tagged_placement({"id": 7, "tags": ["relay-test"]}),
+                         (None, None))
+
+    def test_the_precedent_carries_shape_not_claims(self):
+        cell, _ = note_reply.house_cell(BOOK, "attachments", "android")
+        text = note_reply.render_precedent(cell, "Attachments fail", "android")
+        self.assertIn("We usually explain the 14-day window.", text)
+        self.assertIn("Ask for the app version", text)
+        # The caveat is for the agent reviewing the draft, never for the model
+        # writing it: it is a note about which past promises went stale.
+        self.assertNotIn("never shipped", text)
+
+    def test_the_brief_sits_after_the_precedent_in_the_prompt(self):
+        """Precedent is background, the brief is the instruction, and what is nearest
+        the end is what governs."""
+        built = note_reply.build_compose_prompt("their words", "the brief",
+                                                precedent="PRECEDENT BLOCK")
+        self.assertLess(built.index("PRECEDENT BLOCK"), built.index("the brief"))
+
+    def test_the_prompt_forbids_repeating_specifics_from_precedent(self):
+        """The caveats say fixes were declared shipped and recurred. Repeating one
+        into a live reply re-promises something nobody delivered."""
+        prompt = " ".join(note_reply.COMPOSE_SYSTEM.lower().split())
+        self.assertIn("take nothing specific from it", prompt)
+        self.assertIn("the brief wins and the precedent is dropped", prompt)
+
+    def test_the_draft_note_shows_the_grounding_and_the_caveat(self):
+        cell, covering = note_reply.house_cell(BOOK, "attachments", "android")
+        note = note_reply.build_draft_note(GERMAN, "brief", 42, False, cell,
+                                           "Attachments fail", covering)
+        self.assertIn("Attachments fail", note)
+        self.assertIn("#111", note)
+        self.assertIn("never shipped", note)
+        # and it still must not be able to command itself
+        self.assertIsNone(note_reply.parse_command(
+            note_reply.comment_text(posted_note(note))))
+
+
+class FindDraft(unittest.TestCase):
+    def test_finds_the_text_between_the_delimiters(self):
+        self.assertEqual(note_reply.find_draft([draft_note("Hallo")], API_USER),
+                         {1: "Hallo"})
+
+    def test_finds_every_numbered_option(self):
+        found = note_reply.find_draft([draft_note("eins", "zwei", "drei")], API_USER)
+        self.assertEqual(found, {1: "eins", 2: "zwei", 3: "drei"})
+
+    def test_newest_draft_wins(self):
+        """fetch_comments returns newest first, so the first match is the newest.
+        An agent who rejected a draft and rewrote the brief must get the new one."""
+        comments = [draft_note("second", cid=10), draft_note("first", cid=9)]
+        self.assertEqual(note_reply.find_draft(comments, API_USER), {1: "second"})
+
+    def test_ignores_a_draft_shaped_note_from_a_human(self):
+        """Otherwise an agent could paste the delimiters into a note and have
+        `reply` publish text nobody generated or reviewed."""
+        forged = dict(posted_note(f"{note_reply.para(note_reply.begin_marker(1))}"
+                                  f"{note_reply.verbatim('send me')}"
+                                  f"{note_reply.para(note_reply.end_marker(1))}"),
+                      author_id=AGENT)
+        self.assertEqual(note_reply.find_draft([forged], API_USER), {})
+
+    def test_ignores_public_comments(self):
+        public = dict(draft_note("x"), public=True)
+        self.assertEqual(note_reply.find_draft([public], API_USER), {})
+
+    def test_no_draft_at_all(self):
+        self.assertEqual(note_reply.find_draft([comment("claude: reply")], API_USER), {})
+
+    def test_preserves_the_reviewed_text_exactly(self):
+        """What was reviewed is what goes out. The note is HTML, so the draft has to
+        survive escaping, Zendesk's tag stripping and unescaping and come back
+        identical — blank lines, indentation, ampersands and angle brackets included.
+        Anything less and the customer receives something nobody read."""
+        body = ('Hallo,\n\nAnhänge werden 14 Tage gespeichert & danach gelöscht.\n'
+                'Zeile mit  und „Anführungszeichen".\n\n'
+                '  eingerückte Zeile\n\n05ab & Grüße')
+        note = draft_note(body)
+        self.assertEqual(note_reply.find_draft([note], API_USER), {1: body})
+
+    def test_the_stored_markup_escapes_what_the_draft_contains(self):
+        """The other half: unescaping on read is only safe because writing escapes."""
+        note = draft_note("a & b ")
+        self.assertIn("a & b <c>", note["body"])
+
+
+class CustomerSample(unittest.TestCase):
+    """Which text decides the language the customer is answered in."""
+
+    TWEET = {"id": 1, "subject": "Conversation with 我命由我不由天",
+             "description": "Conversation with 我命由我不由天", "requester_id": 999}
+    EMAIL = {"id": 2, "subject": "Cannot log in",
+             "description": "Ich kann mich nicht anmelden.", "requester_id": 999}
+
+    def test_an_ordinary_ticket_uses_the_requester_s_own_words(self):
+        session = fake_session()
+        got = note_reply.customer_sample(session, "sub", self.EMAIL, [])
+        self.assertIn("Ich kann mich nicht anmelden", got)
+        self.assertEqual(session.calls, [], "no lookups needed for a normal ticket")
+
+    def test_a_dm_falls_back_to_the_integration_authored_message(self):
+        """The bug this exists for: on a Twitter DM the integration authors the
+        customer's message under its own id, so filtering on requester_id drops
+        every word they wrote and the reply goes out in English to a Chinese
+        speaker."""
+        comments = [
+            {"id": 20, "author_id": 901790886886, "public": True,
+             "body": "Thanks for getting in touch."},
+            {"id": 10, "author_id": -1, "public": True,
+             "body": "(10:36:27) 我命由我不由天: 中国大陆可以使用吗？"},
+        ]
+        # oldest comment first, so the integration author is resolved before the agent
+        session = fake_session(FakeResponse({"user": {}}),
+                               FakeResponse({"user": {"id": 901790886886, "role": "admin"}}))
+        got = note_reply.customer_sample(session, "sub", self.TWEET, comments)
+        self.assertIn("中国大陆可以使用吗", got)
+        self.assertNotIn("Thanks for getting in touch", got,
+                         "an agent's English reply must not skew the detection")
+
+    def test_an_unknown_author_counts_as_the_customer(self):
+        """The integration's id is an account detail; a user we cannot resolve is a
+        customer, not an agent."""
+        comments = [{"id": 10, "author_id": -1, "public": True, "body": "中国大陆可以使用吗？"}]
+        session = fake_session(FakeResponse({}, status_code=404))
+        self.assertIn("中国大陆", note_reply.customer_sample(session, "sub", self.TWEET, comments))
+
+    def test_private_notes_never_reach_the_detector(self):
+        comments = [{"id": 10, "author_id": -1, "public": False, "body": "claude: draft - x"}]
+        session = fake_session(FakeResponse({"user": {}}))
+        got = note_reply.customer_sample(session, "sub", self.TWEET, comments)
+        self.assertNotIn("claude: draft", got)
+
+
+class ChoosingAnOption(unittest.TestCase):
+    """Which of the offered replies actually reaches the customer."""
+
+    def test_a_bare_reply_sends_the_only_option(self):
+        text, complaint = note_reply.choose_option({1: "only"}, None)
+        self.assertEqual((text, complaint), ("only", None))
+
+    def test_a_bare_reply_refuses_to_guess_between_options(self):
+        """Picking for them would send a customer a reply nobody chose."""
+        text, complaint = note_reply.choose_option({1: "a", 2: "b", 3: "c"}, None)
+        self.assertIsNone(text)
+        self.assertIn("3 options", complaint)
+
+    def test_a_number_selects_that_option(self):
+        self.assertEqual(note_reply.choose_option({1: "a", 2: "b"}, 2)[0], "b")
+
+    def test_an_option_that_does_not_exist_is_refused(self):
+        text, complaint = note_reply.choose_option({1: "a", 2: "b"}, 7)
+        self.assertIsNone(text)
+        self.assertIn("no option 7", complaint)
+
+    def test_reads_the_number_off_the_command(self):
+        for text, want in [("2", 2), (" 3 ", 3), ("#2", 2), ("", None),
+                           ("please send", None), ("2 but nicer", 2)]:
+            with self.subTest(text=text):
+                self.assertEqual(note_reply.asked_option(text), want)
+
+    def test_reply_two_sends_the_second_option_verbatim(self):
+        session = fake_session(*[FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+                               FakeResponse({"ticket": {}}), FakeResponse({"ticket": {}})])
+        comments = [comment("claude: reply 2", cid=3),
+                    draft_note("erste", "zweite", "dritte", cid=2)]
+        note_reply.run_reply(session, "sub", {"id": 7}, comments,
+                             {"id": 3, "author": AGENT, "action": "reply", "brief": "2"},
+                             API_USER, dry_run=False)
+        puts = [c for c in session.calls if c[0] == "PUT" and "/tags.json" not in c[1]]
+        self.assertEqual(puts[0][2]["json"]["ticket"]["comment"],
+                         {"body": "zweite", "public": True})
+
+    def test_an_ambiguous_reply_writes_no_public_comment(self):
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        comments = [comment("claude: reply", cid=3),
+                    draft_note("erste", "zweite", cid=2)]
+        note_reply.run_reply(session, "sub", {"id": 7}, comments,
+                             {"id": 3, "author": AGENT, "action": "reply", "brief": ""},
+                             API_USER, dry_run=False)
+        for _, url, kwargs in session.calls:
+            if "/tags.json" in url:
+                continue
+            self.assertIs(kwargs["json"]["ticket"]["comment"]["public"], False)
+
+    def test_being_asked_to_choose_is_not_an_error(self):
+        """`claude-error` is the queue of broken tickets, not of ordinary prompts."""
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        note_reply.run_reply(session, "sub", {"id": 7},
+                             [comment("claude: reply", cid=3),
+                              draft_note("a", "b", cid=2)],
+                             {"id": 3, "author": AGENT, "action": "reply", "brief": ""},
+                             API_USER, dry_run=False)
+        added = [kw["json"]["tags"] for m, u, kw in session.calls
+                 if m == "PUT" and "/tags.json" in u]
+        self.assertEqual(added, [], "being asked to choose is not an error")
+
+    def test_the_prompt_asks_for_genuinely_different_options(self):
+        prompt = " ".join(note_reply.COMPOSE_SYSTEM.lower().split())
+        self.assertIn("genuinely different", prompt)
+        self.assertIn("if the brief only supports one honest reply, return one", prompt)
+
+
+class QueueTag(unittest.TestCase):
+    """`claude-queued` means "a webhook fired and nobody serviced it". Anything else
+    left in it turns the dropped-job view into noise."""
+
+    def test_a_run_with_nothing_to_do_still_clears_the_tag(self):
+        """Claude's own notes name the commands, so posting one re-fires the trigger.
+        That second run finds only its own note — and must not leave the tag behind."""
+        session = fake_session(*[FakeResponse({})])
+        note_reply.clear_queued(session, "sub", 7)
+        method, url, kwargs = session.calls[0]
+        self.assertEqual((method, kwargs["json"]), ("DELETE", {"tags": [note_reply.TAG_QUEUED]}))
+        self.assertTrue(url.endswith("/tickets/7/tags.json"))
+
+    def test_a_dry_run_clears_nothing(self):
+        session = fake_session(*[])
+        note_reply.clear_queued(session, "sub", 7, dry_run=True)
+        self.assertEqual(session.calls, [])
+
+    def test_a_failure_to_clear_is_not_fatal(self):
+        """The tag is a dashboard light, not the work."""
+        session = fake_session(*[FakeResponse({}, status_code=500),
+                               FakeResponse({}, status_code=500),
+                               FakeResponse({}, status_code=500),
+                               FakeResponse({}, status_code=500),
+                               FakeResponse({}, status_code=500),
+                               FakeResponse({}, status_code=500)])
+        note_reply.clear_queued(session, "sub", 7)   # must not raise
+
+
+class Solve(unittest.TestCase):
+    """Solving writes to no customer, but it is still a state change on a real
+    ticket — and the account's CSAT automation fires on it."""
+
+    CMD = {"id": 9, "author": AGENT, "action": "solve", "brief": ""}
+
+    def test_solve_is_a_command(self):
+        self.assertEqual(note_reply.parse_command("claude: solve"), ("solve", ""))
+
+    def test_it_sets_solved_and_writes_no_public_comment(self):
+        session = fake_session(FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+                               FakeResponse({"ticket": {}}))
+        note_reply.run_solve(session, "sub", {"id": 7, "status": "open"}, self.CMD,
+                             dry_run=False)
+        ticket = next(kw["json"]["ticket"] for m, u, kw in session.calls
+                      if m == "PUT" and "/tags.json" not in u)
+        self.assertEqual(ticket["status"], note_reply.SOLVED_STATUS)
+        self.assertIs(ticket["comment"]["public"], False)
+
+    def test_the_note_records_who_decided_and_why(self):
+        """A status change on its own leaves nothing on the ticket explaining it."""
+        session = fake_session(FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+                               FakeResponse({"ticket": {}}))
+        note_reply.run_solve(session, "sub", {"id": 7, "status": "open"},
+                             dict(self.CMD, brief="spam, nothing actionable"),
+                             dry_run=False)
+        body = next(kw["json"]["ticket"]["comment"]["html_body"]
+                    for m, u, kw in session.calls if m == "PUT" and "/tags.json" not in u)
+        self.assertIn("Audric", body)
+        self.assertIn("spam, nothing actionable", body)
+
+    def test_it_is_tagged_so_a_batch_can_be_found_again(self):
+        session = fake_session(FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+                               FakeResponse({"ticket": {}}))
+        note_reply.run_solve(session, "sub", {"id": 7, "status": "open"}, self.CMD,
+                             dry_run=False)
+        added = [kw["json"]["tags"] for m, u, kw in session.calls
+                 if m == "PUT" and "/tags.json" in u]
+        self.assertEqual(added, [[note_reply.TAG_SOLVED]])
+
+    def test_an_already_solved_ticket_is_not_an_error(self):
+        session = fake_session(FakeResponse({"ticket": {}}))
+        note_reply.run_solve(session, "sub", {"id": 7, "status": "solved"}, self.CMD,
+                             dry_run=False)
+        body = next(kw["json"]["ticket"]["comment"]["html_body"]
+                    for m, u, kw in session.calls if m == "PUT" and "/tags.json" not in u)
+        self.assertIn("already solved", body)
+        self.assertEqual([kw["json"]["tags"] for m, u, kw in session.calls
+                          if m == "PUT" and "/tags.json" in u], [])
+
+    def test_a_dry_run_solves_nothing(self):
+        session = fake_session(FakeResponse({"user": {"id": AGENT, "name": "Audric"}}))
+        note_reply.run_solve(session, "sub", {"id": 7, "status": "open"}, self.CMD,
+                             dry_run=True)
+        self.assertEqual([c for c in session.calls if c[0] == "PUT"], [])
+
+
+class Explain(unittest.TestCase):
+    """The read-only verb that surfaces known fixes, which `draft` refuses to assert."""
+
+    def test_explain_is_a_command(self):
+        self.assertEqual(note_reply.parse_command("claude: explain"), ("explain", ""))
+
+    def test_it_shows_what_was_actually_done(self):
+        cell = dict(BOOK["cells"]["attachments|android"],
+                    actions=["Fix shipped in v2.15.3 (case 27896)"])
+        book = {"groups": BOOK["groups"],
+                "cells": dict(BOOK["cells"], **{"attachments|android": cell})}
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        with Patched(note_reply, load_house=lambda *a: book):
+            note_reply.run_explain(
+                session, "sub", "model",
+                {"id": 7, "tags": ["grp-attachments", "plat-android"]}, [],
+                {"id": 9, "author": AGENT, "action": "explain", "brief": ""},
+                dry_run=False)
+        body = session.calls[0][2]["json"]["ticket"]["comment"]["html_body"]
+        self.assertIn("v2.15.3", body)
+        self.assertIn("never shipped", body)      # the caveat travels with it
+        self.assertIs(session.calls[0][2]["json"]["ticket"]["comment"]["public"], False)
+
+    def test_it_writes_nothing_when_the_ticket_matches_nothing(self):
+        with Patched(note_reply, load_house=lambda *a: BOOK,
+                     place_ticket=lambda *a: (None, "android")):
+            session = fake_session(*[FakeResponse({"ticket": {}})])
+            note_reply.run_explain(session, "sub", "model", {"id": 7, "tags": []}, [],
+                                   {"id": 9, "author": AGENT, "action": "explain",
+                                    "brief": ""}, dry_run=False)
+        self.assertIn("no precedent",
+                      session.calls[0][2]["json"]["ticket"]["comment"]["html_body"])
+        added = [kw["json"]["tags"] for m, u, kw in session.calls
+                 if m == "PUT" and "/tags.json" in u]
+        self.assertEqual(added, [])
+
+    def test_no_house_answers_configured_is_said_plainly(self):
+        with Patched(note_reply, load_house=lambda *a: None):
+            session = fake_session(*[FakeResponse({"ticket": {}})])
+            note_reply.run_explain(session, "sub", "model", {"id": 7, "tags": []}, [],
+                                   {"id": 9, "author": AGENT, "action": "explain",
+                                    "brief": ""}, dry_run=False)
+        self.assertIn("No house answers",
+                      session.calls[0][2]["json"]["ticket"]["comment"]["html_body"])
+
+
+class Authorisation(unittest.TestCase):
+    def test_agents_and_admins_may_command(self):
+        for role in ("agent", "admin"):
+            self.assertTrue(note_reply.may_command({"id": AGENT, "role": role}))
+
+    def test_end_users_may_not(self):
+        self.assertFalse(note_reply.may_command({"id": AGENT, "role": "end-user"}))
+
+    def test_allowlist_narrows_further(self):
+        with Patched(os, environ={**os.environ, "ZENDESK_NOTE_AUTHORS": "1,2"}):
+            self.assertFalse(note_reply.may_command({"id": AGENT, "role": "agent"}))
+            self.assertTrue(note_reply.may_command({"id": 2, "role": "agent"}))
+
+    def test_allowlist_does_not_override_the_role_check(self):
+        with Patched(os, environ={**os.environ, "ZENDESK_NOTE_AUTHORS": str(AGENT)}):
+            self.assertFalse(note_reply.may_command({"id": AGENT, "role": "end-user"}))
+
+
+class LatestCommand(unittest.TestCase):
+    def session_for(self, role="agent"):
+        return fake_session(*[FakeResponse({"user": {"id": AGENT, "role": role}})])
+
+    def test_takes_the_newest_command(self):
+        comments = [comment("claude: reply", cid=3), comment("claude: draft - x", cid=2)]
+        found = note_reply.latest_command(comments, API_USER, self.session_for(), "sub")
+        self.assertEqual((found["action"], found["id"]), ("reply", 3))
+
+    def test_skips_notes_written_by_the_api_user(self):
+        """The in-code half of the loop guard, independent of the trigger's config."""
+        comments = [comment("claude: draft - mine", author=API_USER, cid=4),
+                    comment("claude: draft - theirs", cid=2)]
+        found = note_reply.latest_command(comments, API_USER, self.session_for(), "sub")
+        self.assertEqual(found["id"], 2)
+
+    def test_skips_public_comments(self):
+        """A customer can type the prefix into a public reply."""
+        comments = [comment("claude: reply", author=42, public=True, cid=5),
+                    comment("claude: draft - x", cid=2)]
+        found = note_reply.latest_command(comments, API_USER, self.session_for(), "sub")
+        self.assertEqual(found["id"], 2)
+
+    def test_an_unauthorised_author_stops_the_search(self):
+        comments = [comment("claude: reply", cid=3), comment("claude: draft - x", cid=2)]
+        found = note_reply.latest_command(comments, API_USER,
+                                          self.session_for("end-user"), "sub")
+        self.assertIsNone(found)
+
+    def test_no_command_present(self):
+        self.assertIsNone(note_reply.latest_command(
+            [comment("just a note")], API_USER, fake_session(*[]), "sub"))
+
+
+class Idempotency(unittest.TestCase):
+    def test_the_done_marker_is_keyed_on_the_command(self):
+        """Not on the ticket: two briefs on one ticket are two commands, and the
+        second must not be swallowed by the first one's marker."""
+        self.assertNotEqual(note_reply.done_marker(1), note_reply.done_marker(2))
+
+    def test_a_handled_command_is_recognised(self):
+        import reply
+        note = note_reply.build_draft_note(GERMAN, "x", 42)
+        self.assertTrue(reply.already_replied([comment(note)], note_reply.done_marker(42)))
+        self.assertFalse(reply.already_replied([comment(note)], note_reply.done_marker(43)))
+
+
+class Writes(unittest.TestCase):
+    def test_reply_sends_the_draft_verbatim_and_sets_pending(self):
+        session = fake_session(*[FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+                               FakeResponse({"ticket": {}}), FakeResponse({"ticket": {}})])
+        comments = [comment("claude: reply", cid=3), draft_note("Hallo Welt", cid=2)]
+        note_reply.run_reply(session, "sub", {"id": 7}, comments,
+                             {"id": 3, "author": AGENT, "action": "reply", "brief": ""},
+                             API_USER, dry_run=False)
+        puts = [c for c in session.calls if c[0] == "PUT" and "/tags.json" not in c[1]]
+        self.assertEqual(len(puts), 2)
+        public = puts[0][2]["json"]["ticket"]
+        self.assertEqual(public["comment"], {"body": "Hallo Welt", "public": True})
+        self.assertEqual(public["status"], note_reply.REPLIED_STATUS)
+        self.assertIs(puts[1][2]["json"]["ticket"]["comment"]["public"], False)
+
+    def test_reply_without_a_draft_writes_no_public_comment(self):
+        session = fake_session(*[FakeResponse({"ticket": {}})])
+        note_reply.run_reply(session, "sub", {"id": 7}, [comment("claude: reply", cid=3)],
+                             {"id": 3, "author": AGENT, "action": "reply", "brief": ""},
+                             API_USER, dry_run=False)
+        for _, url, kwargs in session.calls:
+            if "/tags.json" in url:
+                continue
+            self.assertIs(kwargs["json"]["ticket"]["comment"]["public"], False)
+
+    def test_a_dry_run_writes_nothing(self):
+        session = fake_session(*[FakeResponse({"user": {"id": AGENT, "name": "Audric"}})])
+        note_reply.run_reply(session, "sub", {"id": 7},
+                             [comment("claude: reply", cid=3), draft_note("Hallo", cid=2)],
+                             {"id": 3, "author": AGENT, "action": "reply", "brief": ""},
+                             API_USER, dry_run=True)
+        self.assertEqual([c for c in session.calls if c[0] == "PUT"], [])
+
+    def test_an_empty_brief_is_refused_without_calling_claude(self):
+        called = []
+        with Patched(note_reply, compose=lambda *a: called.append(a)):
+            session = fake_session(*[FakeResponse({"ticket": {}})])
+            note_reply.run_draft(session, "sub", "model", {"id": 7}, [],
+                                 {"id": 3, "author": AGENT, "action": "draft", "brief": ""},
+                                 API_USER, dry_run=False)
+        self.assertEqual(called, [])
+        self.assertIs(session.calls[0][2]["json"]["ticket"]["comment"]["public"], False)
+
+    def test_tags_go_through_the_sub_resource_not_the_ticket_update(self):
+        """`additional_tags`/`remove_tags` are update_many fields. A single-ticket
+        update takes them with a 200 and silently ignores them, which is how every
+        tag this tool set went missing while every call looked successful."""
+        session = fake_session(*[FakeResponse({"ticket": {}}), FakeResponse({}),
+                               FakeResponse({})])
+        note_reply.write_to_ticket(session, "sub", 7, "note", public=False,
+                                   add_tags=[note_reply.TAG_DRAFTED],
+                                   drop_tags=[note_reply.TAG_QUEUED])
+        comment_put = session.calls[0][2]["json"]["ticket"]
+        self.assertNotIn("additional_tags", comment_put)
+        self.assertNotIn("remove_tags", comment_put)
+        tag_calls = [(m, u.rsplit("/", 1)[-1], kw["json"]["tags"])
+                     for m, u, kw in session.calls[1:]]
+        self.assertEqual(tag_calls, [("PUT", "tags.json", [note_reply.TAG_DRAFTED]),
+                                     ("DELETE", "tags.json", [note_reply.TAG_QUEUED])])
+
+    def test_the_comment_is_written_before_the_tags(self):
+        """A tag failure must not lose the note."""
+        session = fake_session(*[FakeResponse({"ticket": {}}), FakeResponse({})])
+        note_reply.write_to_ticket(session, "sub", 7, "note", public=False,
+                                   add_tags=[note_reply.TAG_SENT])
+        self.assertIn("comment", session.calls[0][2]["json"]["ticket"])
+
+
+class DraftNote(unittest.TestCase):
+    def test_carries_the_back_translation_for_a_foreign_ticket(self):
+        note = note_reply.build_draft_note(GERMAN, "keeps 14 days", 42)
+        self.assertIn(GERMAN["options"][0]["back_translation"], note)
+
+    def test_omits_the_back_translation_for_an_english_ticket(self):
+        text = "Attachments are kept for 14 days."
+        english = {"language": "English", "language_code": "en", "is_english": True,
+                   "options": [{"approach": "explain", "reply_en": text,
+                                "translated": text, "back_translation": ""}]}
+        note = note_reply.build_draft_note(english, "keeps 14 days", 42)
+        self.assertNotIn("back in English", note)
+        self.assertEqual(note_reply.find_draft([posted_note(note)], API_USER), {1: text})
+
+    def test_numbers_every_option_and_names_its_approach(self):
+        """The approach line is how the agent picks without reading all three."""
+        note = note_reply.build_draft_note(GERMAN_THREE, "keeps 14 days", 42)
+        for number in (1, 2, 3):
+            self.assertIn(f"Option {number}", note)
+        self.assertIn("ask which device was online", note)
+        self.assertEqual(
+            note_reply.find_draft([posted_note(note)], API_USER),
+            {1: "Erste Antwort", 2: "Zweite Antwort", 3: "Dritte Antwort"})
+
+    def test_the_sent_note_quotes_what_went_out(self):
+        note = note_reply.build_sent_note("Audric", 42, "Hallo Welt\nzweite Zeile")
+        self.assertIn("Hallo Welt", note)
+        self.assertNotIn("zweite Zeile", note)
+
+
+class Composition(unittest.TestCase):
+    def test_an_empty_translation_is_refused(self):
+        with self.assertRaises(SystemExit):
+            note_reply.validate_composition(dict(GERMAN, options=[option("  ")]))
+
+    def test_empty_options_are_dropped_not_offered(self):
+        """An empty block in the note is a blank the agent could send."""
+        result = note_reply.validate_composition(
+            dict(GERMAN, options=[option("keep me"), option("   "), option("also me")]))
+        self.assertEqual([o["translated"] for o in result["options"]],
+                         ["keep me", "also me"])
+
+    def test_more_options_than_the_cap_are_trimmed(self):
+        result = note_reply.validate_composition(
+            dict(GERMAN, options=[option(f"n{i}") for i in range(6)]))
+        self.assertEqual(len(result["options"]), note_reply.MAX_OPTIONS)
+
+    def test_a_missing_field_is_refused(self):
+        partial = {key: value for key, value in GERMAN.items() if key != "options"}
+        with self.assertRaises(SystemExit):
+            note_reply.validate_composition(partial)
+
+    @staticmethod
+    def prompt():
+        """The system prompt as one line, so a probe cannot fail merely because the
+        text was re-wrapped."""
+        return " ".join(note_reply.COMPOSE_SYSTEM.lower().split())
+
+    def test_the_prompt_forbids_inventing_facts_and_actions(self):
+        """The two rules that matter most, asserted so a prompt edit cannot quietly
+        drop them: 183 solved tickets say an account was banned, and a reply
+        claiming an action nobody took is the worst output this can produce."""
+        prompt = self.prompt()
+        self.assertIn("never state a fact neither of them contains", prompt)
+        self.assertIn("never claim an action was taken", prompt)
+
+    def test_a_second_draft_amends_the_first(self):
+        """An agent writing "also mention X" wants the draft they just read plus X.
+        Regenerating from the new brief alone throws away wording they kept."""
+        seen = {}
+        def fake(model, sample, brief, previous=None, precedent=None):
+            seen.update(brief=brief, previous=previous, precedent=precedent)
+            return GERMAN
+        with Patched(note_reply, compose=fake):
+            session = fake_session(*[FakeResponse({"ticket": {}})])
+            note_reply.run_draft(
+                session, "sub", "model", {"id": 7, "requester_id": 42},
+                [comment("claude: draft - also mention X", cid=11),
+                 draft_note("Erster Entwurf", cid=10)],
+                {"id": 11, "author": AGENT, "action": "draft",
+                 "brief": "also mention X"}, API_USER, dry_run=False)
+        self.assertIn("Erster Entwurf", seen["previous"])
+        self.assertIn("revised",
+                      session.calls[0][2]["json"]["ticket"]["comment"]["html_body"])
+
+    def test_a_first_draft_has_nothing_to_amend(self):
+        seen = {}
+        def fake(model, sample, brief, previous=None, precedent=None):
+            seen.update(previous=previous, precedent=precedent)
+            return GERMAN
+        with Patched(note_reply, compose=fake):
+            session = fake_session(*[FakeResponse({"ticket": {}})])
+            note_reply.run_draft(
+                session, "sub", "model", {"id": 7, "requester_id": 42},
+                [comment("claude: draft - x", cid=11)],
+                {"id": 11, "author": AGENT, "action": "draft", "brief": "x"},
+                API_USER, dry_run=False)
+        self.assertIsNone(seen["previous"])
+
+    def test_the_amend_prompt_puts_the_brief_after_the_draft(self):
+        """The brief reads as an instruction about the draft above it, which is how
+        the agent meant it."""
+        built = note_reply.build_compose_prompt("their words", "add X", "old draft")
+        self.assertLess(built.index("old draft"), built.index("add X"))
+
+    def test_the_prompt_still_forbids_workarounds_and_placeholder_names(self):
+        """Loosening the tone must not loosen the facts. A confident wrong
+        instruction costs more than a short answer, and 23 macros on this account
+        already send "(User name)" literally."""
+        prompt = self.prompt()
+        self.assertIn("never offer a workaround", prompt)
+        self.assertIn("(user name)", prompt)
+
+    def test_the_prompt_grants_tone_latitude(self):
+        """The other half: a reply that is correct and cold is a worse reply."""
+        self.assertIn("correct and cold is a worse reply", self.prompt())
+
+    def test_the_brief_is_bounded(self):
+        action, brief = note_reply.parse_command("claude: draft - " + "x" * 5000)
+        self.assertLessEqual(len(brief), note_reply.BRIEF_CHARS)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/zendesk_triage/test_relay.py b/zendesk_triage/test_relay.py
index 347e850..d5f18ee 100644
--- a/zendesk_triage/test_relay.py
+++ b/zendesk_triage/test_relay.py
@@ -12,6 +12,10 @@
 that an unlisted person is refused, that a dialog still opens when Zendesk is
 unreachable, and that an attachment URL never reaches a channel-visible message.
 """
+import base64
+import datetime
+import hashlib
+import hmac
 import json
 import os
 import sys
@@ -672,3 +676,98 @@ def test_reporting_a_failure_cannot_add_one(self):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+# ---- The Zendesk note webhook ----------------------------------------------
+
+
+ZENDESK_SECRET = "s3cret"
+
+
+def zendesk_post(body=None, *, secret=ZENDESK_SECRET, sign=True, age=0, tamper=False):
+    """Send a signed Zendesk webhook through the real endpoint."""
+    raw = json.dumps({"ticket_id": "27603"} if body is None else body).encode()
+    when = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=age)
+    stamp = when.strftime("%Y-%m-%dT%H:%M:%SZ")
+    headers = {"Content-Type": "application/json"}
+    if sign:
+        digest = hmac.new(secret.encode(), stamp.encode() + raw, hashlib.sha256).digest()
+        headers[relay.ZENDESK_SIGNATURE_HEADER] = base64.b64encode(digest).decode()
+        headers[relay.ZENDESK_TIMESTAMP_HEADER] = stamp
+    if tamper:
+        raw = raw.replace(b"27603", b"27604")
+    with TestClient(relay.app) as client:
+        return client.post("/zendesk/notes", content=raw, headers=headers)
+
+
+class TestZendeskWebhook(unittest.TestCase):
+    """The second gate on a path that publishes comments to customers. The Zendesk
+    trigger is the first, and note_reply.py decides who may command it."""
+
+    def setUp(self):
+        self.ran = []
+
+    def run_with(self, **env):
+        with Env(ZENDESK_WEBHOOK_SECRET=ZENDESK_SECRET, **env), \
+                Patched(relay, run_note_reply=self.ran.append):
+            return zendesk_post(**self.kwargs)
+
+    def test_a_signed_webhook_queues_the_ticket(self):
+        self.kwargs = {}
+        response = self.run_with()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(self.ran, ["27603"])
+
+    def test_an_unsigned_webhook_is_refused(self):
+        self.kwargs = {"sign": False}
+        self.assertEqual(self.run_with().status_code, 401)
+        self.assertEqual(self.ran, [])
+
+    def test_the_wrong_secret_is_refused(self):
+        self.kwargs = {"secret": "wrong"}
+        self.assertEqual(self.run_with().status_code, 401)
+        self.assertEqual(self.ran, [])
+
+    def test_a_tampered_body_is_refused(self):
+        """The signature covers the body, so swapping the ticket id invalidates it —
+        otherwise anyone who captured one webhook could redirect it at any ticket."""
+        self.kwargs = {"tamper": True}
+        self.assertEqual(self.run_with().status_code, 401)
+        self.assertEqual(self.ran, [])
+
+    def test_a_stale_webhook_is_refused(self):
+        self.kwargs = {"age": relay.MAX_SIGNATURE_AGE_SECONDS + 60}
+        self.assertEqual(self.run_with().status_code, 401)
+        self.assertEqual(self.ran, [])
+
+    def test_an_unconfigured_relay_refuses_everything(self):
+        """No secret must mean no, not yes. This URL writes to customers."""
+        self.kwargs = {}
+        with Env(ZENDESK_WEBHOOK_SECRET=None), Patched(relay, run_note_reply=self.ran.append):
+            self.assertEqual(zendesk_post().status_code, 401)
+        self.assertEqual(self.ran, [])
+
+    def test_a_missing_ticket_id_is_a_400(self):
+        self.kwargs = {"body": {"nothing": "here"}}
+        self.assertEqual(self.run_with().status_code, 400)
+        self.assertEqual(self.ran, [])
+
+    def test_a_bogus_ticket_id_is_refused(self):
+        self.kwargs = {"body": {"ticket_id": "27603; rm -rf /"}}
+        self.assertEqual(self.run_with().status_code, 400)
+        self.assertEqual(self.ran, [])
+
+    def test_a_non_json_body_is_refused_not_crashed(self):
+        with Env(ZENDESK_WEBHOOK_SECRET=ZENDESK_SECRET), \
+                Patched(relay, run_note_reply=self.ran.append):
+            raw = b"not json"
+            stamp = datetime.datetime.now(datetime.timezone.utc).strftime(
+                "%Y-%m-%dT%H:%M:%SZ")
+            digest = hmac.new(ZENDESK_SECRET.encode(), stamp.encode() + raw,
+                              hashlib.sha256).digest()
+            with TestClient(relay.app) as client:
+                response = client.post("/zendesk/notes", content=raw, headers={
+                    relay.ZENDESK_SIGNATURE_HEADER: base64.b64encode(digest).decode(),
+                    relay.ZENDESK_TIMESTAMP_HEADER: stamp})
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(self.ran, [])
diff --git a/zendesk_triage/test_triage.py b/zendesk_triage/test_triage.py
index a67bb48..a174f3b 100644
--- a/zendesk_triage/test_triage.py
+++ b/zendesk_triage/test_triage.py
@@ -234,12 +234,70 @@ def test_the_window_is_on_updated_at_not_created_at(self):
     def test_query_keeps_unsolved_filter_and_newest_first_ordering(self):
         query = triage.build_window_query(72)
         self.assertIn("type:ticket", query)
-        self.assertIn("status")[0]
+        self.assertEqual(sorted(analysed.split()),
+                         sorted(triage.BACKLOG_NON_REVIEW_QUERY.split()))
+
+    def test_a_ticket_only_we_touched_leaves_the_window(self):
+        """The bug this exists for: a `claude: explain` note bumps updated_at, and
+        the window query is on updated_at — so a ticket whose customer last wrote
+        100 days ago was appearing in a 72-hour digest because we touched it."""
+        cutoff = "2026-09-01T00:00:00Z"
+        ours = {"id": 1, "updated_at": "2026-09-03T02:35:00Z",
+                "requester_updated_at": "2026-05-26T06:55:00Z"}
+        theirs = {"id": 2, "updated_at": "2026-09-02T09:00:00Z",
+                  "requester_updated_at": "2026-09-02T09:00:00Z"}
+        fresh, quiet = triage.drop_quiet_tickets([ours, theirs], cutoff)
+        self.assertEqual([t["id"] for t in fresh], [2])
+        self.assertEqual([t["id"] for t in quiet], [1])
+
+    def test_a_missing_requester_stamp_keeps_the_ticket(self):
+        """A failed metric sideload must leave the digest noisy, never silent."""
+        blind = {"id": 1, "updated_at": "2026-09-03T02:35:00Z"}
+        fresh, quiet = triage.drop_quiet_tickets([blind], "2026-09-01T00:00:00Z")
+        self.assertEqual(fresh, [blind])
+        self.assertEqual(quiet, [])
+
+    def test_a_ticket_touched_exactly_at_the_cutoff_is_kept(self):
+        edge = {"id": 1, "requester_updated_at": "2026-09-01T00:00:00Z"}
+        fresh, _ = triage.drop_quiet_tickets([edge], "2026-09-01T00:00:00Z")
+        self.assertEqual(fresh, [edge])
+
+    def test_the_query_and_the_filter_share_one_cutoff(self):
+        """Computed twice they would sit seconds apart, which is enough to drop a
+        ticket that arrived mid-run."""
+        cutoff = triage.window_cutoff(72)
+        self.assertIn(f"updated>{cutoff}", triage.build_window_query(72, cutoff))
+
     def test_a_longer_window_reaches_further_back(self):
         short = triage.build_window_query(48).split("updated>")[1].split(" ")[0]
         long = triage.build_window_query(168).split("updated>")[1].split(" ")[0]
@@ -518,14 +576,14 @@ def test_reports_the_backlog_excluding_store_reviews(self):
         of unsolved tickets are AppFollow reviews."""
         text = self.header([finding(1)], {"total_unsolved": 5680,
                                           "total_unsolved_non_review": 428})
-        self.assertIn("Backlog: **428** unsolved excluding app-store reviews", text)
+        self.assertIn("Backlog: **428** awaiting a reply, excluding app-store reviews", text)
         self.assertIn("**5,252** more are reviews", text)
 
     def test_falls_back_to_the_total_when_the_review_count_is_unavailable(self):
         """Both counts are best-effort; losing one must not lose the whole line."""
         text = self.header([finding(1)], {"total_unsolved": 5609,
                                           "total_unsolved_non_review": None})
-        self.assertIn("Backlog: **5,609** unsolved tickets in total", text)
+        self.assertIn("Backlog: **5,609** tickets awaiting a reply", text)
 
     def test_omits_the_backlog_line_when_the_count_is_unavailable(self):
         self.assertNotIn("Backlog", self.header([finding(1)], {"total_unsolved": None}))
@@ -1912,7 +1970,7 @@ def unit_commands(unit="zendesk-digest.service"):
 
 class TestDigestOrdering(unittest.TestCase):
     """The digest is only correct if the positive-review resolver ran first: solved
-    reviews leave the triage's `status`, not `created>`: a ticket the requester adds detail to days after
@@ -110,10 +125,43 @@ def build_window_query(hours):
     The cutoff is an explicit UTC timestamp rather than Zendesk's relative
     `updated>72hours` form, so the exact window lands in the run log.
     """
-    cutoff = (datetime.now(timezone.utc) - timedelta(hours=hours)).strftime(
-        "%Y-%m-%dT%H:%M:%SZ"
-    )
-    return f"type:ticket status{cutoff} order_by:updated_at sort:desc"
+    return (f"type:ticket status{cutoff or window_cutoff(hours)} order_by:updated_at sort:desc")
+
+
+def window_cutoff(hours):
+    """The UTC timestamp bounding a window, as Zendesk search formats it.
+
+    Separate from build_window_query so the run computes it once and both the query
+    and the requester-activity filter judge against the same instant. Computing it
+    twice would put seconds between them, which is enough to drop a ticket that
+    arrived mid-run.
+    """
+    return (datetime.now(timezone.utc) - timedelta(hours=hours)).strftime(
+        "%Y-%m-%dT%H:%M:%SZ")
+
+
+def drop_quiet_tickets(tickets, cutoff):
+    """Split off tickets the requester has not touched inside the window.
+
+    The window query is on `updated_at`, which moves on ANY change — a tag edit, the
+    hourly automation that bumps tickets at :01, and our own private notes. So a
+    `claude: explain` on a ticket whose customer last wrote 100 days ago drags it
+    into today's digest, and the noise grows in proportion to how much the reply
+    tooling is used, which is backwards.
+
+    Zendesk search has no `requester_updated>`, so the narrowing happens here, over
+    the value hydrate_requester_activity has already fetched. Timestamps are Zendesk's
+    fixed-width UTC form, so a string compare is a chronological one.
+
+    A ticket whose requester_updated_at is missing is KEPT: a failed sideload should
+    leave the digest noisy, never silent.
+    """
+    fresh, quiet = [], []
+    for ticket in tickets:
+        stamp = ticket.get("requester_updated_at")
+        (quiet if stamp and stamp < cutoff else fresh).append(ticket)
+    return fresh, quiet
 
 
 def window_label(hours):
@@ -659,7 +707,6 @@ def save_state(path, state, reported, retention_days):
 # whereas the `app-store` tag was present on only 287 of them — so filter on the
 # channel, not on tags. 4-5 star reviews were 59% of *all* tickets and are never
 # actionable, so counting them beats paying tokens to classify them.
-REVIEW_CHANNEL = "any_channel"
 # The same backlog minus store reviews. 92% of unsolved tickets are AppFollow
 # reviews, so the unqualified number reads as ~13x the queue that needs a human.
 BACKLOG_NON_REVIEW_QUERY = f"{BACKLOG_QUERY} -via:{REVIEW_CHANNEL}"
@@ -1473,10 +1520,10 @@ def build_header(findings, highlights, stats=None):
 
     non_review = stats.get("total_unsolved_non_review")
     if backlog is not None and non_review is not None:
-        lines.append(f"Backlog: **{non_review:,}** unsolved excluding app-store reviews "
+        lines.append(f"Backlog: **{non_review:,}** awaiting a reply, excluding app-store reviews "
                      f"(**{backlog - non_review:,}** more are reviews, not triaged).")
     elif backlog is not None:
-        lines.append(f"Backlog: **{backlog:,}** unsolved tickets in total (not triaged).")
+        lines.append(f"Backlog: **{backlog:,}** tickets awaiting a reply (not triaged).")
 
     serious = by_severity.get("crash", 0) + by_severity.get("data_loss", 0)
     tail = f"**{len(highlights)}** worth looking into"
@@ -1754,13 +1801,15 @@ def main():
         api_token = get_env("ZENDESK_API_TOKEN", args.api_token)
 
         # An explicit query wins over --window-hours; warn rather than silently drop it.
+        window_start = None
         explicit_query = args.query or os.environ.get("ZENDESK_QUERY")
         if explicit_query:
             if args.window_hours:
                 print("Note: --window-hours ignored because an explicit query was given.")
             query = explicit_query
         elif args.window_hours:
-            query = build_window_query(args.window_hours)
+            window_start = window_cutoff(args.window_hours)
+            query = build_window_query(args.window_hours, window_start)
             stats["scope"] = window_label(args.window_hours)
         else:
             query = DEFAULT_QUERY
@@ -1798,11 +1847,23 @@ def main():
                 print("Only positive reviews in this window; nothing to report.")
                 return
 
-        if args.state:
-            # Before partition_by_state, which compares on requester_updated_at, and
-            # after the review filter, so the sideload only covers what can be
-            # reported. One request per 100 tickets.
+        # One sideload serves both the window filter and the dedup, so it runs
+        # whenever either needs it. After the review filter, so it only covers
+        # tickets that can still be reported. One request per 100 tickets.
+        if window_start or args.state:
             hydrate_requester_activity(zd, subdomain, tickets)
+
+        if window_start:
+            tickets, quiet = drop_quiet_tickets(tickets, window_start)
+            if quiet:
+                stats["skipped_quiet"] = len(quiet)
+                print(f"Skipped {len(quiet)} ticket(s) that only we touched in this "
+                      f"window; {len(tickets)} remain.")
+            if not tickets:
+                print("Nothing the requester touched in this window; nothing to report.")
+                return
+
+        if args.state:
             state = load_state(args.state)
             new, changed, unchanged = partition_by_state(tickets, state)
             print(f"{len(new)} new, {len(changed)} changed since last reported, "