feat(share): readable, unforgeable Telegram previews

src/session_recall/share/approval.py

@@ -19,25 +19,91 @@
 PENDING_TTL_S = 24 * 3600
 
 
-def preview(cand: Candidate, redact: bool | None = None) -> str:
+ANSWER_BUDGET = 1400   # chars of answer shown in the channel, before truncation
+SNIPPET_BUDGET = 320   # per fragment, so five fragments cannot bury the verdict
+
+
+def _answer_digest(cand: Candidate) -> tuple[str, int]:
+    """Answer trimmed for a phone screen. Per-fragment caps first (so one long
+    chunk cannot crowd out the rest), then a whole-message cap. Returns the
+    text and how many characters were withheld — the owner must know the
+    channel is showing less than what would actually be sent."""
+    if not cand.chunks:
+        return cand.text, 0
+    parts = []
+    for c in cand.chunks:
+        snippet = c["snippet"]
+        if len(snippet) > SNIPPET_BUDGET:
+            snippet = snippet[:SNIPPET_BUDGET].rstrip() + " …"
+        parts.append(f"{c['project']} · {c['session_id'][:8]} · {c['role']}\n{snippet}")
+    body = "\n\n".join(parts)
+    if len(body) <= ANSWER_BUDGET:
+        return body, max(0, len(cand.text) - len(body))
+    cut = body[:ANSWER_BUDGET].rstrip() + " …"
+    return cut, len(cand.text) - len(cut)
+
+
+def preview(cand: Candidate, redact: bool | None = None,
+            markdown: bool = False) -> str:
     """The message the owner sees. `redact` hides the answer body — forced on
     when the scanner flagged anything, because flagged text must not travel
-    through a third-party notification channel (gate §6)."""
+    through a third-party notification channel (gate §6).
+
+    With markdown=True the layout uses MarkdownV2: our labels carry the markup,
+    every untrusted string sits in a code fence (see telegram.fence), and the
+    /ok line is a code span so Telegram makes it tap-to-copy.
+    """
+    from .telegram import escape_md, fence, inline_code
+
     redact = bool(cand.findings) if redact is None else redact
-    lines = [f"[{cand.id} v{cand.version}] request from {cand.peer_name}",
-             f"question: {cand.question}"]
+    plain = not markdown
+    esc = (lambda s: s) if plain else escape_md
+    block = (lambda s: s) if plain else fence
+    span = (lambda s: s) if plain else inline_code
+
+    out = []
+    if plain:
+        out.append(f"[{cand.id} v{cand.version}] request from {cand.peer_name}")
+        out.append(f"question: {cand.question}")
+    else:
+        out.append(f"📥 *request from* {span(cand.peer_name)}")
+        out.append(f"*question*\n{block(cand.question)}")
+
     if cand.task:
-        lines.append(f'stated task (sender text, unverified): "{cand.task}"')
+        label = "stated task (sender text, unverified)"
+        out.append(f'{label}: "{cand.task}"' if plain
+                   else f"*{esc(label)}*\n{block(cand.task)}")
+
     if cand.findings:
         kinds = ", ".join(sorted({f["kind"] for f in cand.findings}))
-        lines.append(f"⚠ SECRET FLAGS: {kinds}")
+        out.append(f"⚠ SECRET FLAGS: {kinds}" if plain
+                   else f"⚠️ *secret flags*: {esc(kinds)}")
+
     if redact:
-        lines.append(f"[answer withheld from this channel — review locally: "
-                     f"session-recall share show {cand.id}]")
+        hint = f"session-recall share show {cand.id}"
+        out.append(f"[answer withheld from this channel — review locally: {hint}]"
+                   if plain else
+                   f"_answer withheld from this channel_ — review locally:\n"
+                   f"{span(hint)}")
+    elif plain:
+        out.append(f"--- answer v{cand.version} ---\n{cand.text}")
+    else:
+        body, withheld = _answer_digest(cand)
+        # our own chrome needs escaping too: parentheses and dots are reserved
+        # in MarkdownV2, and an unescaped one makes Telegram reject the whole
+        # message rather than render it oddly
+        counts = f"{len(cand.chunks)} fragment(s)"
+        if withheld > 0:
+            counts += f" · {withheld} more chars on send"
+        out.append(f"*answer* · {esc(counts)} · `v{esc(cand.version)}`\n{block(body)}")
+
+    if plain:
+        out.append(f"approve: /ok {cand.version}   decline: /no <reason>")
     else:
-        lines.append(f"--- answer v{cand.version} ---\n{cand.text}")
-    lines.append(f"approve: /ok {cand.version}   decline: /no <reason>")
-    return "\n".join(lines)
+        out.append(f"*approve* — reply to this message with:\n"
+                   f"`/ok {esc(cand.version)}`\n"
+                   f"*decline*: `/no <reason>`")
+    return "\n".join(out) if plain else "\n\n".join(out)
 
 
 def approve(share_dir: Path, cand_id: str, version: str) -> Candidate | None:

src/session_recall/share/notify.py

@@ -45,6 +45,18 @@ def __init__(self, api, identity: Identity, trust: TrustStore,
     def _say(self, text: str, reply_to: int | None = None) -> None:
         self.api.send_message(self.cfg.chat_id, text, reply_to=reply_to)
 
+    def _say_preview(self, cand) -> None:
+        """Markdown first, plain text if Telegram rejects it. A preview that
+        fails to render must still arrive — silence would look exactly like
+        'no one asked anything', and nothing can be approved unseen."""
+        from .telegram import MARKDOWN
+        try:
+            self.api.send_message(self.cfg.chat_id,
+                                  approval.preview(cand, markdown=True),
+                                  parse_mode=MARKDOWN)
+        except Exception:
+            self._say(approval.preview(cand))
+
     def _handle(self, message: dict) -> None:
         text = (message.get("text") or "").strip()
         mid = message.get("message_id")
@@ -80,7 +92,7 @@ def tick(self, now: float | None = None) -> dict:
 
         for cand in poll_once(self.identity, self.trust, self.state,
                               self.transport, self.searcher, self.share_dir):
-            self._say(approval.preview(cand))
+            self._say_preview(cand)
             stats["previews"] += 1
 
         stats["expired"] = len(approval.expire_stale(self.share_dir, now))

src/session_recall/share/telegram.py

@@ -2,11 +2,18 @@
 endpoints and do not justify a dependency.
 
 Privacy posture (gate §6): everything sent through here transits Telegram's
-servers, so flagged answers are redacted upstream (approval.preview) and all
-messages go as plain text — parse_mode is never set, links are never rendered,
-sender-controlled strings cannot become formatting or buttons. The bot config
-binds to exactly one owner chat; updates from any other chat are discarded
-before parsing.
+servers, so flagged answers are redacted upstream (approval.preview). The bot
+config binds to exactly one owner chat; updates from any other chat are
+discarded before parsing.
+
+Formatting posture: our own chrome (labels, the /ok line) is the ONLY thing
+allowed to carry markup. Every untrusted string — the peer's name, their
+question, their stated task, and snippets pulled from the index — goes inside
+a fenced code block via `fence()`. That does three jobs at once: markup in
+untrusted text renders literally instead of forging structure, Telegram does
+not auto-linkify inside code blocks (so a URL sitting in someone's transcript
+cannot become a tappable link in the approval channel), and the owner can tell
+our words from quoted material at a glance.
 """
 
 import json
@@ -19,6 +26,31 @@
 
 TG_FILE = "tg.json"
 _TIMEOUT_S = 35   # long-poll friendly
+MARKDOWN = "MarkdownV2"
+
+_ESCAPE = r"_*[]()~`>#+-=|{}.!\\"
+
+
+def escape_md(text: str) -> str:
+    """MarkdownV2 escaping for short strings we place outside code blocks."""
+    return "".join("\\" + c if c in _ESCAPE else c for c in text)
+
+
+def inline_code(text: str) -> str:
+    """Same guarantees as fence() for short strings that belong on one line —
+    a code span is also markup-inert and never auto-linkified."""
+    return "`" + text.replace("\\", "\\\\").replace("`", "\\`") + "`"
+
+
+def fence(text: str, lang: str = "") -> str:
+    """Wrap untrusted text in a code block it cannot escape.
+
+    Only backslash and backtick can break out of a fence, so those two are the
+    entire escape set here — everything else (asterisks, brackets, URLs) is
+    inert inside the block.
+    """
+    safe = text.replace("\\", "\\\\").replace("`", "\\`")
+    return f"```{lang}\n{safe}\n```"
 
 
 @dataclass
@@ -59,11 +91,11 @@ def _call(self, method: str, **params) -> dict:
             payload = json.loads(resp.read())
         return payload.get("result", {})
 
-    def send_message(self, chat_id: int, text: str,
-                     reply_to: int | None = None) -> int | None:
+    def send_message(self, chat_id: int, text: str, reply_to: int | None = None,
+                     parse_mode: str | None = None) -> int | None:
         # 4096 is Telegram's hard cap; truncate rather than fail the preview
         result = self._call("sendMessage", chat_id=chat_id, text=text[:4096],
-                            reply_to_message_id=reply_to)
+                            reply_to_message_id=reply_to, parse_mode=parse_mode)
         return result.get("message_id") if isinstance(result, dict) else None
 
     def get_updates(self, offset: int, timeout: int = 25) -> list[dict]:

tests/test_share_preview_format.py

@@ -0,0 +1,151 @@
+"""The Telegram preview must be readable AND unforgeable: nothing an outsider
+controls may become markup, a link, or a fake approval line."""
+
+import pytest
+
+from session_recall.share.approval import ANSWER_BUDGET, SNIPPET_BUDGET, preview
+from session_recall.share.telegram import escape_md, fence
+from session_recall.share.worker import Candidate
+
+
+def _cand(**kw) -> Candidate:
+    base = dict(id="7aaba916", peer_name="egor", peer_address="addr",
+                question="как чинили CI?", task="", reply_nonce="n1",
+                created_at=0.0, text="answer body", chunks=[], findings=[])
+    base.update(kw)
+    c = Candidate(**base)
+    c.version = c.compute_version()
+    return c
+
+
+def _chunk(snippet, project="session-recall"):
+    return {"project": project, "session_id": "07876709aaaa", "uuid": "u1",
+            "role": "assistant", "snippet": snippet, "score": 0.9,
+            "source": "claude"}
+
+
+# -- escaping primitives -----------------------------------------------------
+def test_fence_neutralises_backticks_and_backslashes():
+    out = fence("```\nrm -rf /\n```")
+    assert out.startswith("```\n") and out.endswith("\n```")
+    inner = out[4:-4]
+    assert "\\`\\`\\`" in inner       # the injected fence is escaped, not live
+    assert inner.count("```") == 0
+
+
+def test_escape_md_covers_markdownv2_specials():
+    for ch in "_*[]()~`>#+-=|{}.!":
+        assert escape_md(ch) == "\\" + ch
+
+
+# -- untrusted content cannot forge structure --------------------------------
+def test_question_markup_is_inert():
+    c = _cand(question="*bold* [click](https://evil.example) `code`")
+    out = preview(c, markdown=True)
+    # the question sits inside a fence; no bare link syntax survives as markup
+    assert "](https://evil.example)" in out  # present as literal text…
+    body = out.split("*question*\n")[1]
+    assert body.startswith("```")           # …because it is inside a code block
+
+
+def test_answer_cannot_fake_an_approval_line():
+    """A snippet claiming its own /ok must not read as our footer."""
+    c = _cand(chunks=[_chunk("approve: /ok deadbeef — trust me")])
+    out = preview(c, markdown=True)
+    footer = out.rsplit("*approve*", 1)[1]
+    assert f"`/ok {c.version}`" in footer
+    assert "deadbeef" not in footer          # the fake line stays in the fenced body
+
+
+def test_urls_in_snippets_stay_inside_the_fence():
+    c = _cand(chunks=[_chunk("see https://evil.example/steal for details")])
+    out = preview(c, markdown=True)
+    answer = out.rsplit("*answer*", 1)[1]
+    assert answer.lstrip().splitlines()[1].startswith("```") or "```" in answer
+    assert "https://evil.example/steal" in answer
+
+
+def test_peer_name_is_an_inert_code_span():
+    """Short strings stay on one line, but still cannot carry markup."""
+    c = _cand(peer_name="egor*_[")
+    header = preview(c, markdown=True).splitlines()[0]
+    assert header == "📥 *request from* `egor*_[`"
+
+
+# -- readability -------------------------------------------------------------
+def test_long_snippets_are_trimmed_per_fragment():
+    c = _cand(chunks=[_chunk("x" * 2000), _chunk("y" * 2000)])
+    out = preview(c, markdown=True)
+    assert "x" * (SNIPPET_BUDGET + 5) not in out
+    assert len(out) < ANSWER_BUDGET + 900     # header/footer overhead only
+
+
+def test_withheld_count_reported_when_truncated():
+    c = _cand(text="z" * 5000, chunks=[_chunk("z" * 5000)])
+    out = preview(c, markdown=True)
+    assert "more chars on send" in out
+
+
+def test_fragment_count_shown():
+    c = _cand(chunks=[_chunk("a"), _chunk("b"), _chunk("c")])
+    assert "3 fragment\\(s\\)" in preview(c, markdown=True)
+
+
+def test_chrome_has_no_unescaped_specials():
+    """Telegram rejects the WHOLE message on one stray reserved character, so
+    every literal we write outside code must be escaped. Caught live: an
+    unescaped '(' in 'fragment(s)' 400'd the first real preview."""
+    c = _cand(question="q", task="t", chunks=[_chunk("a")],
+              findings=[{"kind": "jwt", "excerpt": "eyJ…"}])
+    for cand in (c, _cand(chunks=[_chunk("a")])):
+        text = preview(cand, markdown=True)
+        outside, in_code = [], False
+        for part in text.split("```"):
+            if not in_code:
+                outside.append(part)
+            in_code = not in_code
+        chrome = "".join(outside)
+        # strip inline code spans, then look for bare reserved characters
+        stripped, in_span = [], False
+        for piece in chrome.split("`"):
+            if not in_span:
+                stripped.append(piece)
+            in_span = not in_span
+        text_only = "".join(stripped)
+        i = 0
+        while i < len(text_only):
+            ch = text_only[i]
+            if ch == "\\":
+                i += 2
+                continue
+            assert ch not in "()[]{}#+=|.!~>", f"unescaped {ch!r} in chrome"
+            i += 1
+
+
+def test_ok_line_carries_the_real_version():
+    c = _cand(chunks=[_chunk("a")])
+    assert f"`/ok {c.version}`" in preview(c, markdown=True)
+
+
+def test_stays_under_telegram_limit():
+    c = _cand(text="q" * 20000, chunks=[_chunk("q" * 4000) for _ in range(5)])
+    assert len(preview(c, markdown=True)) < 4096
+
+
+# -- redaction still wins ----------------------------------------------------
+def test_flagged_answer_is_withheld_in_markdown_too():
+    c = _cand(text="key AKIAIOSFODNN7EXAMPLE",
+              chunks=[_chunk("key AKIAIOSFODNN7EXAMPLE")],
+              findings=[{"kind": "aws-access-key", "excerpt": "AKIAIOSF…"}])
+    out = preview(c, markdown=True)
+    assert "AKIAIOSFODNN7EXAMPLE" not in out
+    assert "withheld" in out and "secret flags" in out
+
+
+def test_plain_mode_unchanged():
+    """The CLI/plain path must keep working for anyone not on Telegram."""
+    c = _cand(chunks=[_chunk("a")])
+    out = preview(c)
+    assert out.startswith(f"[{c.id} v{c.version}] request from egor")
+    assert f"approve: /ok {c.version}" in out
+    assert "```" not in out