refactor(webex): converge on the shared nonce minter and byte chunker

src/kiro_crew/webex/attachments.py

@@ -17,8 +17,6 @@
 from __future__ import annotations
 
 import logging
-import mimetypes
-import os
 from typing import TYPE_CHECKING
 
 from kiro_crew.messaging.attachments import (
@@ -80,15 +78,3 @@ async def _download(url: str, dest: str) -> None:
         handle_audio=True,
     )
     return await transcribe_audio_attachments(result, "Webex")
-
-
-def outbound_mimetype(path: str, fallback: str = "application/octet-stream") -> str:
-    """A content type for an outbound upload, from the extension.
-
-    Webex needs one on the multipart part, and only a handful of types render an
-    inline preview — everything else still uploads, it just shows as a file. So a
-    wrong guess costs a preview, not the delivery, which is why the fallback is a
-    generic binary type rather than a refusal.
-    """
-    guessed, _ = mimetypes.guess_type(os.path.basename(path))
-    return guessed or fallback

src/kiro_crew/webex/client.py

@@ -58,40 +58,14 @@ def truncate_utf8(text: str, max_bytes: int = WEBEX_MAX_TEXT) -> str:
     """Byte-exact truncation, defaulted to Webex's own cap.
 
     The implementation is the shared one in ``messaging.split``; this wrapper
-    exists only to keep ``WEBEX_MAX_TEXT`` as the default for Webex's call sites.
-    It loses the tail, so it is the last-resort guard for a SINGLE send —
-    multi-message content is split losslessly first by :func:`chunk_utf8`.
+    exists only to keep ``WEBEX_MAX_TEXT`` as the default for this module's three
+    call sites. It loses the tail, so it is the last-resort guard for a SINGLE
+    send — multi-message content is split losslessly first, by the shared
+    ``messaging.split.chunk_utf8_bytes`` at the renderer's answer path.
     """
     return _truncate_utf8(text, max_bytes)
 
 
-def chunk_utf8(text: str, max_bytes: int = WEBEX_MAX_TEXT) -> list[str]:
-    """Split ``text`` into chunks of at most *max_bytes* UTF-8 bytes each,
-    never splitting a code point and never dropping content.
-
-    The neutral ``chunk_text`` helper splits by CHARACTERS, but Webex limits
-    BYTES — a multibyte-heavy chunk under the character cap could exceed the
-    byte limit and be silently tail-truncated by the send path, losing the
-    remainder. Splitting on the encoded bytes and re-decoding with
-    ``errors="ignore"`` finds the largest whole-code-point prefix per chunk;
-    the loop then resumes from exactly the characters consumed, so the
-    concatenation of all chunks always equals the input.
-    """
-    if not text:
-        return []
-    chunks: list[str] = []
-    remaining = text
-    while remaining:
-        encoded = remaining.encode("utf-8")
-        if len(encoded) <= max_bytes:
-            chunks.append(remaining)
-            break
-        piece = encoded[:max_bytes].decode("utf-8", errors="ignore")
-        chunks.append(piece)
-        remaining = remaining[len(piece) :]
-    return chunks
-
-
 # A WS connection must live at least this long to count as "healthy" and reset
 # the reconnect backoff. A connect->immediate-close (bad token) stays on the
 # backoff curve so it cannot hot-loop with zero delay. Mirrors WeComClient.

src/kiro_crew/webex/renderer.py

@@ -37,7 +37,6 @@
 import logging
 import os
 import re
-import secrets
 import time
 from dataclasses import replace
 from typing import TYPE_CHECKING, Any, Callable
@@ -49,12 +48,18 @@
     hide_local_refs,
     upload_filename,
 )
-from kiro_crew.messaging.renderer import Renderer, apply_options_cap, split_options_trailer
+from kiro_crew.messaging.renderer import (
+    Renderer,
+    apply_options_cap,
+    new_approval_nonce,
+    split_options_trailer,
+)
+from kiro_crew.messaging.split import chunk_utf8_bytes
 from kiro_crew.messaging.tables import TABLE_POLICY_CARDS
 from kiro_crew.messaging.transport import TransportCapabilities
 from kiro_crew.security import redact_credentials, redact_exfiltration_urls
 from kiro_crew.webex.cards import approval_card, options_card, usable_choices
-from kiro_crew.webex.client import WEBEX_MAX_TEXT, chunk_utf8
+from kiro_crew.webex.client import WEBEX_MAX_TEXT
 
 if TYPE_CHECKING:
     from kiro_crew.webex.client import WebexClient
@@ -372,7 +377,13 @@ async def on_done(self, stop_reason: str = "") -> None:
         # here and must reassemble exactly, which a line-oriented splitter cannot
         # promise because it consumes the boundary whitespace. Pinned by
         # test_channel_table_rendering.py::TestDeliveryFraming.
-        chunks = chunk_utf8(content) or ["…"]
+        #
+        # The SHARED primitive, not a local copy: it carries two termination
+        # guards (a non-positive budget, and a single code point wider than the
+        # budget) that a hand-rolled copy of this loop spins forever on. And
+        # deliberately the FENCE-BLIND one, not ``split_markdown_bytes``: the
+        # answer path above re-seals its own fences.
+        chunks = chunk_utf8_bytes(content, WEBEX_MAX_TEXT) or ["…"]
         first, rest = chunks[0], chunks[1:]
         delivered = False
         if self._placeholder_id is not None:
@@ -695,6 +706,12 @@ def _options_card(self, choices: list[str]) -> dict[str, Any] | None:
         usable = usable_choices(choices)
         if not usable:
             return None
-        nonce = secrets.token_hex(8)
+        # The SHARED minter, for the reason stated at ``new_approval_nonce``: an
+        # options press and an approval press are the same hazard (a control left
+        # in a chat from an earlier turn naming indexes that are live again), so a
+        # second generator with its own alphabet is what eventually diverges. The
+        # approval card on this same renderer already reaches it through
+        # ``PendingApprovals.reserve``.
+        nonce = new_approval_nonce()
         self._publish_choices(nonce, usable)
         return options_card(usable, nonce=nonce)

test/test_webex_attachments.py

@@ -3,23 +3,38 @@
 Everything channel-neutral (caps, classification, signature sniffing, temp-file
 ownership, the SEL audit) belongs to ``messaging/attachments.py`` and is tested
 there. What is Webex-specific and tested here: an opaque content URL becomes a
-described :class:`Attachment` via a HEAD probe, and a probe that fails still
-yields an entry so the failure surfaces to the user instead of the file
-disappearing.
+described :class:`Attachment` via a HEAD probe, a probe that fails still yields an
+entry so the failure surfaces to the user instead of the file disappearing, and
+``process_webex_attachments`` wires the probe, the download and the audio
+transcription into the shared ingest.
 """
 
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 import pytest
 
-from kiro_crew.webex.attachments import outbound_mimetype, to_attachments
+from kiro_crew.messaging.attachments import cleanup
+from kiro_crew.webex.attachments import process_webex_attachments, to_attachments
 
 
 class FakeClient:
-    """A client whose HEAD answers are scripted per URL."""
-
-    def __init__(self, answers: dict[str, tuple[str, str, int]] | None = None) -> None:
+    """A client whose HEAD answers are scripted per URL.
+
+    ``bodies`` makes a download actually write bytes, which the shared ingest
+    needs: it classifies on the file's SIGNATURE, so a download that writes
+    nothing is indistinguishable from a corrupt file and never reaches the
+    interesting paths.
+    """
+
+    def __init__(
+        self,
+        answers: dict[str, tuple[str, str, int]] | None = None,
+        bodies: dict[str, bytes] | None = None,
+    ) -> None:
         self.answers = answers or {}
+        self.bodies = bodies or {}
         self.head_calls: list[str] = []
         self.downloads: list[tuple[str, str]] = []
 
@@ -29,6 +44,10 @@ async def head_content(self, url: str) -> tuple[str, str, int]:
 
     async def download_content(self, url: str, dest: str) -> None:
         self.downloads.append((url, dest))
+        body = self.bodies.get(url)
+        if body is not None:
+            with open(dest, "wb") as fh:
+                fh.write(body)
 
 
 class TestToAttachments:
@@ -95,26 +114,73 @@ async def test_a_hostile_filename_cannot_steer_the_temp_path(self) -> None:
             assert hostile not in hint
 
 
-class TestOutboundMimetype:
-    @pytest.mark.parametrize(
-        "path,expected",
-        [
-            ("/tmp/chart.png", "image/png"),
-            ("/tmp/a.jpg", "image/jpeg"),
-            ("/tmp/notes.txt", "text/plain"),
-        ],
-    )
-    def test_a_known_extension_maps_to_its_type(self, path: str, expected: str) -> None:
-        assert outbound_mimetype(path) == expected
+class TestProcessWebexAttachments:
+    """The Webex half of ingest: probe -> download -> shared ingest -> transcribe."""
 
-    def test_an_unknown_extension_falls_back_to_binary(self) -> None:
-        """A wrong guess costs a preview, not the delivery.
+    @pytest.mark.asyncio
+    async def test_a_text_file_becomes_prompt_material(self) -> None:
+        url = "https://webexapis.com/v1/contents/C1"
+        client = FakeClient(
+            {url: ("notes.txt", "text/plain", 11)},
+            {url: b"hello there"},
+        )
+        inbound = SimpleNamespace(file_urls=(url,))
+
+        result = await process_webex_attachments(client, inbound)  # type: ignore[arg-type]
+
+        try:
+            # The download went through the client's own content endpoint, which is
+            # what carries the bot's Authorization header — a plain fetch of that URL
+            # is unauthenticated and would 401.
+            assert client.downloads and client.downloads[0][0] == url
+            assert any("hello there" in block for block in result.text_blocks)
+        finally:
+            # ``finally`` so a failed assertion still drops the downloaded bytes:
+            # a bypassed cleanup leaves temp residue that trips the suite's
+            # residue reporting and mis-attributes it to the next test.
+            cleanup(result.temp_paths)
+
+    @pytest.mark.asyncio
+    async def test_no_files_does_no_work(self) -> None:
+        client = FakeClient()
+        inbound = SimpleNamespace(file_urls=())
 
-        Webex previews only a handful of types; everything else still uploads, so
-        a generic binary type is the right fallback rather than a refusal.
+        result = await process_webex_attachments(client, inbound)  # type: ignore[arg-type]
+
+        assert client.head_calls == []
+        assert client.downloads == []
+        assert result.text_blocks == []
+
+    @pytest.mark.asyncio
+    async def test_audio_is_transcribed_here(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """``handle_audio=True`` because Webex offers no server-side transcript.
+
+        iLink voice clips arrive with one and are preferred; a Webex clip has to be
+        downloaded and run through local STT, so this is the channel that must ask
+        the shared ingest to keep the audio bytes.
         """
-        assert outbound_mimetype("/tmp/thing.qqq") == "application/octet-stream"
+        import os
 
-    def test_only_the_basename_is_consulted(self) -> None:
-        # A directory called "x.png" must not decide the type of a file that is not.
-        assert outbound_mimetype("/tmp/x.png/data.txt") == "text/plain"
+        url = "https://webexapis.com/v1/contents/C1"
+        client = FakeClient(
+            {url: ("voice.ogg", "audio/ogg", 36)},
+            {url: b"OggS" + b"\x00" * 32},
+        )
+        transcribed: list[str] = []
+
+        async def _transcribe(path: str) -> str:
+            assert os.path.exists(path), "STT must run against the downloaded bytes"
+            transcribed.append(path)
+            return "spoken words"
+
+        monkeypatch.setattr("kiro_crew.transcribe.is_available", lambda: True)
+        monkeypatch.setattr("kiro_crew.transcribe.transcribe_audio", _transcribe)
+        inbound = SimpleNamespace(file_urls=(url,))
+
+        result = await process_webex_attachments(client, inbound)  # type: ignore[arg-type]
+
+        try:
+            assert transcribed == result.audio_paths
+            assert any("spoken words" in block for block in result.text_blocks)
+        finally:
+            cleanup(result.temp_paths)