From 6d85a23cd291aa3e249670a103b874b9673c0018 Mon Sep 17 00:00:00 2001
From: Audric Ackermann {html.escape(text)}` 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
+ if add_tags:
+ fields["additional_tags"] = list(add_tags)
+ if drop_tags:
+ fields["remove_tags"] = list(drop_tags)
+ 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}).")
+
+
+# ---- 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, comments):
+ """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(reply.customer_text(ticket, comments), 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 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"(p|pre)>", "\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 = FakeSession([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 = FakeSession([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)
+ ticket = session.calls[0][2]["json"]["ticket"]
+ self.assertEqual(ticket.get("additional_tags", []), [])
+ self.assertIn(note_reply.TAG_ERROR, ticket["remove_tags"])
+
+ def test_the_transcript_note_is_never_public(self):
+ turns = [{"index": 0, "who": "Customer", "when": "2026-09-03 10:00 UTC",
+ "body": "Hallo"}]
+ session = FakeSession([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 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 = FakeSession([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"]
+ self.assertEqual(puts[0][2]["json"]["ticket"]["comment"],
+ {"body": "zweite", "public": True})
+
+ def test_an_ambiguous_reply_writes_no_public_comment(self):
+ session = FakeSession([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 _, _, kwargs in session.calls:
+ 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 = FakeSession([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)
+ ticket = session.calls[0][2]["json"]["ticket"]
+ self.assertEqual(ticket.get("additional_tags", []), [])
+
+ 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 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 FakeSession([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, FakeSession([]), "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 = FakeSession([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 = [call for call in session.calls if call[0] == "PUT"]
+ 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 = FakeSession([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 _, _, kwargs in session.calls:
+ self.assertIs(kwargs["json"]["ticket"]["comment"]["public"], False)
+
+ def test_a_dry_run_writes_nothing(self):
+ session = FakeSession([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([call for call in session.calls if call[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 = FakeSession([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_move_the_ticket_out_of_the_queue(self):
+ session = FakeSession([FakeResponse({"ticket": {}})])
+ note_reply.write_to_ticket(session, "sub", 7, "note", public=False,
+ add_tags=[note_reply.TAG_DRAFTED],
+ drop_tags=[note_reply.TAG_QUEUED])
+ ticket = session.calls[0][2]["json"]["ticket"]
+ self.assertEqual(ticket["additional_tags"], [note_reply.TAG_DRAFTED])
+ self.assertEqual(ticket["remove_tags"], [note_reply.TAG_QUEUED])
+
+
+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 = FakeSession([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 = FakeSession([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, [])
From 4ace08bad30400ee44209378241c7cbe029e47b2 Mon Sep 17 00:00:00 2001
From: Audric Ackermann
Date: Thu, 3 Sep 2026 12:15:34 +1000
Subject: [PATCH 2/7] feat: add claude: explain
---
zendesk_triage/note_reply.py | 154 ++++++++++++++++++++++++++++--
zendesk_triage/test_note_reply.py | 134 ++++++++++++++++++++++++++
2 files changed, 281 insertions(+), 7 deletions(-)
diff --git a/zendesk_triage/note_reply.py b/zendesk_triage/note_reply.py
index 81cb19a..59b7468 100644
--- a/zendesk_triage/note_reply.py
+++ b/zendesk_triage/note_reply.py
@@ -29,6 +29,10 @@
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.
+ 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
@@ -151,8 +155,9 @@ def english_marker(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)\b[\s\-–—:.]*(.*)$",
- re.IGNORECASE)
+COMMAND = re.compile(
+ r"^[\s>*_]*claude\s*:\s*(draft|reply|english|explain)\b[\s\-–—:.]*(.*)$",
+ re.IGNORECASE)
def parse_command(text):
@@ -223,11 +228,61 @@ def may_command(user):
return not allowed or str(user.get("id")) in allowed
+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
+ url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
+ resp = triage.request_with_retry(session, "PUT", url,
+ json={"ticket": {"remove_tags": [TAG_QUEUED]}})
+ if resp.status_code >= 400:
+ # Never worth failing a run over: the tag is a dashboard light, not the work.
+ print(f"Note: could not clear {TAG_QUEUED} on #{ticket_id} "
+ f"({resp.status_code}).")
+
+
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.
@@ -732,6 +787,58 @@ def run_draft(session, subdomain, model, ticket, comments, command, api_user, dr
drop_tags=[TAG_QUEUED, TAG_ERROR])
+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, 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.
@@ -765,6 +872,30 @@ def run_reply(session, subdomain, ticket, comments, command, api_user, dry_run):
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.
@@ -788,24 +919,28 @@ def run_english(session, subdomain, model, ticket, comments, command, dry_run):
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}")
- english = triage.clip(triage.render_transcript(turns, rendered.get("turns")),
- triage.TRANSCRIPT_CHARS)
+ 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
- # One paragraph per turn, split on the blank line render_transcript puts between
- # them, so a long conversation reads as a conversation rather than a wall.
note = "".join(
[para(f"This conversation in English — {len(turns)} turn(s), both sides."),
para("Translated for reading; the customer has not seen this.")]
- + [verbatim(block) for block in english.split("\n\n") if block.strip()]
+ + 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])
@@ -883,14 +1018,19 @@ def main():
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"] == "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:
diff --git a/zendesk_triage/test_note_reply.py b/zendesk_triage/test_note_reply.py
index 9b9be1b..a2133af 100644
--- a/zendesk_triage/test_note_reply.py
+++ b/zendesk_triage/test_note_reply.py
@@ -149,6 +149,67 @@ def test_nothing_new_is_not_an_error(self):
self.assertEqual(ticket.get("additional_tags", []), [])
self.assertIn(note_reply.TAG_ERROR, ticket["remove_tags"])
+ 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 = FakeSession([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"}]
@@ -368,6 +429,79 @@ def test_the_prompt_asks_for_genuinely_different_options(self):
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 = FakeSession([FakeResponse({"ticket": {}})])
+ note_reply.clear_queued(session, "sub", 7)
+ ticket = session.calls[0][2]["json"]["ticket"]
+ self.assertEqual(ticket["remove_tags"], [note_reply.TAG_QUEUED])
+ self.assertNotIn("comment", ticket)
+
+ def test_a_dry_run_clears_nothing(self):
+ session = FakeSession([])
+ 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 = FakeSession([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 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 = FakeSession([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 = FakeSession([FakeResponse({"ticket": {}})])
+ note_reply.run_explain(session, "sub", "model", {"id": 7, "tags": []}, [],
+ {"id": 9, "author": AGENT, "action": "explain",
+ "brief": ""}, dry_run=False)
+ ticket = session.calls[0][2]["json"]["ticket"]
+ self.assertIn("no precedent", ticket["comment"]["html_body"])
+ self.assertEqual(ticket.get("additional_tags", []), [])
+
+ def test_no_house_answers_configured_is_said_plainly(self):
+ with Patched(note_reply, load_house=lambda *a: None):
+ session = FakeSession([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"):
From f9ac794faff4c5ecc964b2bd346e7f8581773333 Mon Sep 17 00:00:00 2001
From: Audric Ackermann
Date: Thu, 3 Sep 2026 12:24:30 +1000
Subject: [PATCH 3/7] fix: guess better the customer language
---
zendesk_triage/note_reply.py | 81 ++++++++++++---
zendesk_triage/test_note_reply.py | 167 ++++++++++++++++++++++--------
2 files changed, 190 insertions(+), 58 deletions(-)
diff --git a/zendesk_triage/note_reply.py b/zendesk_triage/note_reply.py
index 59b7468..6587b5a 100644
--- a/zendesk_triage/note_reply.py
+++ b/zendesk_triage/note_reply.py
@@ -213,6 +213,39 @@ def fetch_user(session, subdomain, user_id):
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.
@@ -228,6 +261,29 @@ def may_command(user):
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.
@@ -240,13 +296,7 @@ def clear_queued(session, subdomain, ticket_id, dry_run=False):
"""
if dry_run:
return
- url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
- resp = triage.request_with_retry(session, "PUT", url,
- json={"ticket": {"remove_tags": [TAG_QUEUED]}})
- if resp.status_code >= 400:
- # Never worth failing a run over: the tag is a dashboard light, not the work.
- print(f"Note: could not clear {TAG_QUEUED} on #{ticket_id} "
- f"({resp.status_code}).")
+ change_tags(session, subdomain, ticket_id, drop=[TAG_QUEUED])
def para(text):
@@ -309,15 +359,13 @@ def write_to_ticket(session, subdomain, ticket_id, body, public,
fields = {"comment": {("html_body" if as_html else "body"): body, "public": public}}
if status:
fields["status"] = status
- if add_tags:
- fields["additional_tags"] = list(add_tags)
- if drop_tags:
- fields["remove_tags"] = list(drop_tags)
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 ---------------------------------------------------
@@ -387,10 +435,10 @@ def tagged_placement(ticket):
).strip()
-def place_ticket(model, book, ticket, comments):
+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(reply.customer_text(ticket, comments), CUSTOMER_SAMPLE_CHARS)
+ body = triage.clip(sample, CUSTOMER_SAMPLE_CHARS)
try:
found = triage.claude_cli_json(
model, "medium", PLACEMENT_SYSTEM, PLACEMENT_SCHEMA,
@@ -753,11 +801,12 @@ def run_draft(session, subdomain, model, ticket, comments, command, api_user, dr
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, comments)
+ 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 []) + \
@@ -770,7 +819,6 @@ def run_draft(session, subdomain, model, ticket, comments, command, api_user, dr
print(f"#{ticket_id}: grounded in {group}/{covering} "
f"({cell['n']} solved, {cell['consistency']} consistency).")
- sample = reply.customer_text(ticket, comments)
result = compose(model, triage.clip(sample, CUSTOMER_SAMPLE_CHARS), brief,
previous, precedent)
print(f"#{ticket_id}: {'revised' if previous else 'drafted'} "
@@ -806,7 +854,8 @@ def run_explain(session, subdomain, model, ticket, comments, command, dry_run):
group, platform = tagged_placement(ticket)
new_tags = []
if not group:
- group, platform = place_ticket(model, book, ticket, comments)
+ 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)
diff --git a/zendesk_triage/test_note_reply.py b/zendesk_triage/test_note_reply.py
index a2133af..34bb634 100644
--- a/zendesk_triage/test_note_reply.py
+++ b/zendesk_triage/test_note_reply.py
@@ -40,6 +40,15 @@ def option(text="Anhänge werden 14 Tage lang gespeichert.", approach="explain a
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}
@@ -127,7 +136,7 @@ def test_an_up_to_date_transcript_is_not_re_rendered(self):
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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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)
@@ -140,19 +149,24 @@ def test_nothing_new_is_not_an_error(self):
author=API_USER, cid=8)
comments = [comment("claude: english", cid=9),
dict(comment("hallo", author=42, cid=7), public=True), prior]
- session = FakeSession([FakeResponse({"ticket": {}})])
+ 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)
- ticket = session.calls[0][2]["json"]["ticket"]
- self.assertEqual(ticket.get("additional_tags", []), [])
- self.assertIn(note_reply.TAG_ERROR, ticket["remove_tags"])
+ # 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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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"}]}):
@@ -213,7 +227,7 @@ def test_each_turn_gets_its_speaker_line(self):
def test_the_transcript_note_is_never_public(self):
turns = [{"index": 0, "who": "Customer", "when": "2026-09-03 10:00 UTC",
"body": "Hallo"}]
- session = FakeSession([FakeResponse({"ticket": {}})])
+ session = fake_session(*[FakeResponse({"ticket": {}})])
with Patched(triage, conversation_turns=lambda *a: turns,
claude_cli_json=lambda *a, **k: {"turns": [{"index": 0,
"english": "Hello"}]}):
@@ -363,6 +377,53 @@ def test_the_stored_markup_escapes_what_the_draft_contains(self):
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."""
@@ -391,37 +452,40 @@ def test_reads_the_number_off_the_command(self):
self.assertEqual(note_reply.asked_option(text), want)
def test_reply_two_sends_the_second_option_verbatim(self):
- session = FakeSession([FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+ 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"]
+ 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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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 _, _, kwargs in session.calls:
+ 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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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)
- ticket = session.calls[0][2]["json"]["ticket"]
- self.assertEqual(ticket.get("additional_tags", []), [])
+ 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())
@@ -436,20 +500,20 @@ class QueueTag(unittest.TestCase):
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 = FakeSession([FakeResponse({"ticket": {}})])
+ session = fake_session(*[FakeResponse({})])
note_reply.clear_queued(session, "sub", 7)
- ticket = session.calls[0][2]["json"]["ticket"]
- self.assertEqual(ticket["remove_tags"], [note_reply.TAG_QUEUED])
- self.assertNotIn("comment", ticket)
+ 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 = FakeSession([])
+ 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 = FakeSession([FakeResponse({}, status_code=500),
+ session = fake_session(*[FakeResponse({}, status_code=500),
FakeResponse({}, status_code=500),
FakeResponse({}, status_code=500),
FakeResponse({}, status_code=500),
@@ -469,7 +533,7 @@ def test_it_shows_what_was_actually_done(self):
actions=["Fix shipped in v2.15.3 (case 27896)"])
book = {"groups": BOOK["groups"],
"cells": dict(BOOK["cells"], **{"attachments|android": cell})}
- session = FakeSession([FakeResponse({"ticket": {}})])
+ session = fake_session(*[FakeResponse({"ticket": {}})])
with Patched(note_reply, load_house=lambda *a: book):
note_reply.run_explain(
session, "sub", "model",
@@ -484,17 +548,19 @@ def test_it_shows_what_was_actually_done(self):
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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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)
- ticket = session.calls[0][2]["json"]["ticket"]
- self.assertIn("no precedent", ticket["comment"]["html_body"])
- self.assertEqual(ticket.get("additional_tags", []), [])
+ 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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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)
@@ -522,7 +588,7 @@ def test_allowlist_does_not_override_the_role_check(self):
class LatestCommand(unittest.TestCase):
def session_for(self, role="agent"):
- return FakeSession([FakeResponse({"user": {"id": AGENT, "role": role}})])
+ 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)]
@@ -551,7 +617,7 @@ def test_an_unauthorised_author_stops_the_search(self):
def test_no_command_present(self):
self.assertIsNone(note_reply.latest_command(
- [comment("just a note")], API_USER, FakeSession([]), "sub"))
+ [comment("just a note")], API_USER, fake_session(*[]), "sub"))
class Idempotency(unittest.TestCase):
@@ -569,13 +635,13 @@ def test_a_handled_command_is_recognised(self):
class Writes(unittest.TestCase):
def test_reply_sends_the_draft_verbatim_and_sets_pending(self):
- session = FakeSession([FakeResponse({"user": {"id": AGENT, "name": "Audric"}}),
+ 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 = [call for call in session.calls if call[0] == "PUT"]
+ 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})
@@ -583,39 +649,56 @@ def test_reply_sends_the_draft_verbatim_and_sets_pending(self):
self.assertIs(puts[1][2]["json"]["ticket"]["comment"]["public"], False)
def test_reply_without_a_draft_writes_no_public_comment(self):
- session = FakeSession([FakeResponse({"ticket": {}})])
+ 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 _, _, kwargs in session.calls:
+ 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 = FakeSession([FakeResponse({"user": {"id": AGENT, "name": "Audric"}})])
+ 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([call for call in session.calls if call[0] == "PUT"], [])
+ 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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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_move_the_ticket_out_of_the_queue(self):
- session = FakeSession([FakeResponse({"ticket": {}})])
+ 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])
- ticket = session.calls[0][2]["json"]["ticket"]
- self.assertEqual(ticket["additional_tags"], [note_reply.TAG_DRAFTED])
- self.assertEqual(ticket["remove_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):
@@ -692,7 +775,7 @@ 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 = FakeSession([FakeResponse({"ticket": {}})])
+ 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),
@@ -709,7 +792,7 @@ def fake(model, sample, brief, previous=None, precedent=None):
seen.update(previous=previous, precedent=precedent)
return GERMAN
with Patched(note_reply, compose=fake):
- session = FakeSession([FakeResponse({"ticket": {}})])
+ session = fake_session(*[FakeResponse({"ticket": {}})])
note_reply.run_draft(
session, "sub", "model", {"id": 7, "requester_id": 42},
[comment("claude: draft - x", cid=11)],
From e560eb630d9e6519e18bda2c1a65c5165b2241a7 Mon Sep 17 00:00:00 2001
From: Audric Ackermann
Date: Thu, 3 Sep 2026 12:51:54 +1000
Subject: [PATCH 4/7] feat: add claude: solve to silently resolve a ticket
---
README.md | 19 +++++++++--
zendesk_triage/note_reply.py | 42 ++++++++++++++++++++++-
zendesk_triage/test_note_reply.py | 57 +++++++++++++++++++++++++++++++
3 files changed, 114 insertions(+), 4 deletions(-)
diff --git a/README.md b/README.md
index 9610c58..3a5f5d7 100644
--- a/README.md
+++ b/README.md
@@ -504,13 +504,26 @@ claude: draft - attachments are only kept on the server for 14 days. A second
```
Claude replies with a private note carrying the drafted reply, a back-translation,
-and the brief it was written from. The agent reads it and writes:
+and the brief it was written from. A draft usually offers two or three genuinely
+different approaches, numbered, so the agent reads:
```
-claude: reply
+claude: reply 2
```
-which publishes the draft **verbatim** and moves the ticket to `pending`.
+which publishes that option **verbatim** and moves the ticket to `pending`. A bare
+`claude: reply` sends the only option when there is one, and refuses to guess when
+there are several.
+
+### The commands
+
+| Command | What it does | Touches the customer |
+| --- | --- | --- |
+| `claude: draft - ` | 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
diff --git a/zendesk_triage/note_reply.py b/zendesk_triage/note_reply.py
index 6587b5a..6df7928 100644
--- a/zendesk_triage/note_reply.py
+++ b/zendesk_triage/note_reply.py
@@ -29,6 +29,8 @@
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`
@@ -83,10 +85,14 @@
# 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.
@@ -156,7 +162,7 @@ def english_marker(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)\b[\s\-–—:.]*(.*)$",
+ r"^[\s>*_]*claude\s*:\s*(draft|reply|english|explain|solve)\b[\s\-–—:.]*(.*)$",
re.IGNORECASE)
@@ -835,6 +841,38 @@ def run_draft(session, subdomain, model, ticket, comments, command, api_user, dr
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.
@@ -1077,6 +1115,8 @@ def main():
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)
diff --git a/zendesk_triage/test_note_reply.py b/zendesk_triage/test_note_reply.py
index 34bb634..69fb656 100644
--- a/zendesk_triage/test_note_reply.py
+++ b/zendesk_triage/test_note_reply.py
@@ -522,6 +522,63 @@ def test_a_failure_to_clear_is_not_fatal(self):
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."""
From 5740557a2b0e87d1cec73910815c647785abea64 Mon Sep 17 00:00:00 2001
From: Audric Ackermann
Date: Fri, 4 Sep 2026 10:51:52 +1000
Subject: [PATCH 5/7] fix: ignore tickets pending as they are awaiting a user
msg
---
README.md | 2 +-
zendesk_triage/test_triage.py | 26 ++++++++++++++++++++++----
zendesk_triage/triage.py | 27 +++++++++++++++++----------
3 files changed, 40 insertions(+), 15 deletions(-)
diff --git a/README.md b/README.md
index 3a5f5d7..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")[0]
+ self.assertTrue(triage.BACKLOG_QUERY.startswith(analysed))
+
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 +536,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 +1930,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{cutoff} order_by:updated_at sort:desc"
+ return (f"type:ticket status{cutoff} "
+ f"order_by:updated_at sort:desc")
def window_label(hours):
@@ -1473,10 +1480,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"
From 17249e1190d4d9c6b819f8e79b8160ea01e07b77 Mon Sep 17 00:00:00 2001
From: Audric Ackermann
Date: Fri, 4 Sep 2026 11:59:18 +1000
Subject: [PATCH 6/7] fix: ignore tickets that were updated by ourselves only
---
zendesk_triage/test_triage.py | 31 ++++++++++++++++
zendesk_triage/triage.py | 66 +++++++++++++++++++++++++++++------
2 files changed, 87 insertions(+), 10 deletions(-)
diff --git a/zendesk_triage/test_triage.py b/zendesk_triage/test_triage.py
index 13ea0fd..527a0ef 100644
--- a/zendesk_triage/test_triage.py
+++ b/zendesk_triage/test_triage.py
@@ -258,6 +258,37 @@ def test_the_backlog_count_is_scoped_like_the_analysis(self):
analysed = triage.build_window_query(72).split(" updated>")[0]
self.assertTrue(triage.BACKLOG_QUERY.startswith(analysed))
+ 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]
diff --git a/zendesk_triage/triage.py b/zendesk_triage/triage.py
index 94d9dd6..e032d0d 100644
--- a/zendesk_triage/triage.py
+++ b/zendesk_triage/triage.py
@@ -104,7 +104,7 @@
STATE_VERSION = 2
-def build_window_query(hours):
+def build_window_query(hours, cutoff=None):
"""Query for unsolved tickets touched in the last `hours`, most recent first.
`updated>`, not `created>`: a ticket the requester adds detail to days after
@@ -116,13 +116,45 @@ 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} "
+ return (f"type:ticket status{cutoff or window_cutoff(hours)} "
f"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):
if hours % 24 == 0 and hours >= 24:
days = hours // 24
@@ -1761,13 +1793,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
@@ -1805,11 +1839,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, "
From 6f3ce04761ee0ab4940fbeceead2fe29ad020c9a Mon Sep 17 00:00:00 2001
From: Audric Ackermann
Date: Fri, 4 Sep 2026 12:27:41 +1000
Subject: [PATCH 7/7] fix: ios ignore app reviews from processed tickets
---
zendesk_triage/test_triage.py | 13 +++++++++++--
zendesk_triage/triage.py | 16 ++++++++++++----
2 files changed, 23 insertions(+), 6 deletions(-)
diff --git a/zendesk_triage/test_triage.py b/zendesk_triage/test_triage.py
index 527a0ef..a174f3b 100644
--- a/zendesk_triage/test_triage.py
+++ b/zendesk_triage/test_triage.py
@@ -244,6 +244,13 @@ def test_query_keeps_unsolved_filter_and_newest_first_ordering(self):
self.assertIn("order_by:updated_at", query)
self.assertIn("sort:desc", query)
+ def test_store_reviews_are_never_in_scope(self):
+ """A review takes one developer response, replacing any previous one, and
+ cannot be asked a follow-up. It is not work a digest can queue up."""
+ for query in (triage.build_window_query(72), triage.DEFAULT_QUERY):
+ with self.subTest(query=query):
+ self.assertIn(f"-via:{triage.REVIEW_CHANNEL}", query)
+
def test_pending_tickets_are_out_of_scope(self):
"""A pending ticket is one somebody already answered. It leaves the queue on
its own after 72h, so listing it asks for attention that is not needed."""
@@ -254,9 +261,11 @@ def test_pending_tickets_are_out_of_scope(self):
def test_the_backlog_count_is_scoped_like_the_analysis(self):
"""The header number and the tickets below it must mean the same thing, or
- the digest reports a backlog it is not showing."""
+ the digest reports a backlog it is not showing. The headline figure is the
+ review-excluded one, which is what the analysis now covers."""
analysed = triage.build_window_query(72).split(" updated>")[0]
- self.assertTrue(triage.BACKLOG_QUERY.startswith(analysed))
+ 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
diff --git a/zendesk_triage/triage.py b/zendesk_triage/triage.py
index e032d0d..868477e 100644
--- a/zendesk_triage/triage.py
+++ b/zendesk_triage/triage.py
@@ -82,6 +82,15 @@
import requests
+# The channel AppFollow imports app-store reviews on. Identified reviews with no
+# false positives in a 3,662-ticket sample; tags did not (only 287 carried one).
+REVIEW_CHANNEL = "any_channel"
+# Never analyzed. A store review cannot be answered the way a ticket can: it takes
+# one developer response, replacing any previous one, with no way to ask a follow-up
+# question — so it is not work a digest can queue up for someone. The volume stays
+# visible in the header's review count.
+NO_REVIEWS = f"-via:{REVIEW_CHANNEL}"
+
# New and open tickets, newest first. Broad on purpose within that: we want bug
# reports AND low-star reviews, legal requests, security/legislation questions, and
# non-English tickets — Claude does the categorising, so we don't filter to a single
@@ -91,7 +100,7 @@
# the ball is with the customer. The "Pending to Solved" automation resolves those on
# its own after 72h, so putting them in a digest asks a human to look at work that is
# already done. On-hold is included in neither — this account has never used it.
-DEFAULT_QUERY = "type:ticket status72hours` form, so the exact window lands in the run log.
"""
- return (f"type:ticket status{cutoff or window_cutoff(hours)} "
- f"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):
@@ -698,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}"