From 95dfd0fe36851a48a9fe54236cfd92cf0dd516a3 Mon Sep 17 00:00:00 2001 From: "Ara (Claude)" Date: Tue, 18 Aug 2026 11:09:46 +0000 Subject: [PATCH] feat(gateways): store inbound media as HTTP reference, not inline RDF MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inbound messenger attachments (voice notes, images) are persisted as a durable on-disk blob and referenced from the message record by a kb:attachment IRI resolved over the gateway's own token-gated GET /media/ — never embedded inline in RDF (no data: URI), regardless of size. Consistency over data-in-graph: one uniform mechanism for every attachment type keeps the life store pure triples. - inbound_store.py: kb:attachment predicate (multi-valued), store_media/ load_media (hex-keyed, traversal-safe, no RDF extension so qlever-dir ignores the blobs), write_message/undelivered carry attachment_urls. - signal/whatsapp/telegram gateways: read media bytes once, persist the durable reference BEFORE transcription so a failed/garbled transcript never costs the recording; add the original audio to the forwarded files payload (size-capped) so it rides into the dashboard conversation alongside its transcript; serve GET /media/ (token-gated). Co-Authored-By: Claude --- scripts/inbound_store.py | 96 +++++++++++++++++++- scripts/signal-gateway.py | 120 ++++++++++++++++++++----- scripts/telegram-gateway.py | 132 +++++++++++++++++++++++----- scripts/whatsapp-gateway.py | 127 +++++++++++++++++++++----- tests/test_inbound_image_forward.py | 63 ++++++++----- 5 files changed, 448 insertions(+), 90 deletions(-) diff --git a/scripts/inbound_store.py b/scripts/inbound_store.py index 89d1d49..b554be0 100644 --- a/scripts/inbound_store.py +++ b/scripts/inbound_store.py @@ -57,12 +57,32 @@ P_TEXT = KB + "text" P_MESSAGE_ID = KB + "messageId" P_DELIVERED = KB + "delivered" +# A message's media (voice note, image) is NOT embedded in the graph: consistency +# over data-in-graph means every attachment — regardless of size — is a *reference* +# resolved over HTTP, never an inline data-URI literal. This predicate carries that +# reference as an IRI object (the gateway's own token-gated GET /media/ URL); +# it is multi-valued, so a message with several images gets one triple each. The +# attachment's media type is not stored here on purpose — it is returned by the +# HTTP response's Content-Type header when the reference is resolved, which is +# where a media type belongs once the payload lives behind a URL. +P_ATTACHMENT = KB + "attachment" # Subdirectory (under the gateway's store dir) that holds the per-message files. # The gateway owns this folder read-write; the life store mounts it read-only. MESSAGES_SUBDIR = "messages" +# Subdirectory holding the durable media blobs referenced by P_ATTACHMENT. Blobs +# are named by a server-generated hex id (never an untrusted filename) with no +# RDF extension, so qlever-dir — which indexes only .nt/.ttl/.n3 plus declared +# converters — ignores them: the binaries sit on the same volume as the message +# .nt files without ever entering the triple store. +MEDIA_SUBDIR = "media" + _SLUG_RE = re.compile(r"[^a-z0-9]+") +# A stored media id is exactly token_hex(16) — 32 lowercase hex chars. Validating +# against this on read makes the GET /media/ path traversal-safe: an id that +# is not pure hex can never resolve to a file outside the media dir. +_MEDIA_ID_RE = re.compile(r"^[0-9a-f]{32}$") def _slug(value: str) -> str: @@ -74,6 +94,10 @@ def messages_dir(store_dir: str | Path) -> Path: return Path(store_dir) / MESSAGES_SUBDIR +def media_dir(store_dir: str | Path) -> Path: + return Path(store_dir) / MEDIA_SUBDIR + + # -- N-Triples serialization -------------------------------------------------- # A tiny, self-contained N-Triples reader/writer. It supports exactly the three # object shapes this store uses: an IRI object (rdf:type), a plain string @@ -138,12 +162,18 @@ def _render(fields: dict) -> str: lines.append(_lit(subj, P_GROUP, fields["group"])) if fields.get("message_id"): lines.append(_lit(subj, P_MESSAGE_ID, fields["message_id"])) + # Multi-valued: one IRI triple per attachment reference (deduped, order-free + # since the whole record is sorted before serialization). + for url in dict.fromkeys(fields.get("attachments") or []): + if url: + lines.append(_iri(subj, P_ATTACHMENT, url)) return "".join(l + "\n" for l in sorted(lines)) def _parse(text: str) -> dict | None: """Read a message file back into a ``fields`` dict, or None if unparseable.""" - fields: dict = {"delivered": False, "group": None, "message_id": None} + fields: dict = {"delivered": False, "group": None, "message_id": None, + "attachments": []} subject = None for line in text.splitlines(): line = line.strip() @@ -169,6 +199,10 @@ def _parse(text: str) -> dict | None: fields["text"] = value elif pred == P_MESSAGE_ID: fields["message_id"] = value + elif pred == P_ATTACHMENT: + # IRI object → obj_iri is set; append the reference URL. + if value: + fields["attachments"].append(value) elif pred == P_DELIVERED: fields["delivered"] = value.strip().lower() == "true" if subject is None or "channel" not in fields: @@ -238,6 +272,7 @@ def write_message( message_id: str | None = None, timestamp: float | None = None, delivered: bool = False, + attachment_urls: list[str] | None = None, ) -> tuple[str, Path]: """Persist one inbound message as a deterministic N-Triples file. @@ -245,6 +280,10 @@ def write_message( message as still owed to triage; pass ``delivered=True`` for a message the gateway is deliberately *not* forwarding (blacklisted, group-blocked or no-action-class) so the daily drain never re-surfaces it. + + ``attachment_urls`` are HTTP-resolvable references to this message's media + (voice note, image), each emitted as a ``kb:attachment`` IRI. The bytes are + never inlined into the graph — see :func:`store_media`. """ ts = time.time() if timestamp is None else float(timestamp) token = secrets.token_hex(8) @@ -258,6 +297,7 @@ def write_message( "message_id": message_id or None, "received_at": _iso(ts), "delivered": bool(delivered), + "attachments": [u for u in (attachment_urls or []) if u], } # Filename: zero-padded epoch millis (sortable) + token (unique, IRI-safe). fname = f"{int(ts * 1000):016d}-{token}.nt" @@ -266,6 +306,56 @@ def write_message( return subject, path +def store_media(store_dir: str | Path, data: bytes, content_type: str | None) -> str: + """Persist one inbound media blob durably and return its server-generated id. + + The blob is keyed by ``token_hex(16)`` — never by an untrusted filename — so + the id is path-safe by construction and reveals nothing about the sender. The + ``content_type`` is written to a ``.type`` sidecar so the serving endpoint + can set the right ``Content-Type`` without trusting anything client-supplied. + Neither file carries an RDF extension, so the life store never indexes them. + + The caller builds the HTTP reference (``…/media/``) and passes it to + :func:`write_message` as an ``attachment_urls`` entry; the bytes stay on disk + and out of the graph. + """ + media_id = secrets.token_hex(16) + d = media_dir(store_dir) + d.mkdir(parents=True, exist_ok=True) + blob = d / media_id + tmp = blob.with_suffix(".tmp") + tmp.write_bytes(data or b"") + os.replace(tmp, blob) + ct = (content_type or "application/octet-stream").strip() or "application/octet-stream" + _atomic_write(ct + "\n", d / (media_id + ".type")) + return media_id + + +def load_media(store_dir: str | Path, media_id: str) -> tuple[bytes, str] | None: + """Return ``(bytes, content_type)`` for a stored media id, or None. + + Validates the id against :data:`_MEDIA_ID_RE` before touching the filesystem, + so a crafted ``media_id`` can never escape the media dir (path traversal). The + content type falls back to ``application/octet-stream`` if the sidecar is + missing or unreadable. + """ + if not _MEDIA_ID_RE.match(media_id or ""): + return None + d = media_dir(store_dir) + try: + data = (d / media_id).read_bytes() + except OSError: + return None + ct = "application/octet-stream" + try: + sidecar = (d / (media_id + ".type")).read_text(encoding="utf-8").strip() + if sidecar: + ct = sidecar + except OSError: + pass + return data, ct + + def undelivered( store_dir: str | Path, since: str | float | None = None, @@ -280,7 +370,8 @@ def undelivered( anything; only the daily triage drain does. Each returned dict has: ``subject``, ``channel``, ``sender``, ``group``, - ``message_id``, ``received_at`` (ISO-8601), ``text``. + ``message_id``, ``received_at`` (ISO-8601), ``text``, ``attachments`` (a + possibly-empty list of HTTP media reference URLs). """ mdir = messages_dir(store_dir) if not mdir.is_dir(): @@ -311,5 +402,6 @@ def undelivered( "message_id": fields.get("message_id"), "received_at": fields["received_at"], "text": fields["text"], + "attachments": fields.get("attachments") or [], }) return out diff --git a/scripts/signal-gateway.py b/scripts/signal-gateway.py index 1ea3395..5f0f86b 100644 --- a/scripts/signal-gateway.py +++ b/scripts/signal-gateway.py @@ -201,13 +201,23 @@ def _inbound_gate_decision(sender: str, group_id: str | None) -> dict: return {"forward": True, "flagged_unknown": False, "delivered_if_held": True, "reason": "policy-error"} +# This gateway's own base URL on the internal Docker network, used to build the +# HTTP references stored for inbound media (GET /media/ below). Defaults to +# the compose service name + HTTP port; a deployment running more than one Signal +# identity (e.g. the user's personal account) overrides it per container. +GATEWAY_SELF_URL = os.environ.get( + "SIGNAL_GATEWAY_SELF_URL", f"http://signal-gateway:{HTTP_PORT}" +).rstrip("/") + + def _persist_inbound(question: str, sender: str, group_id: str | None, - delivered: bool) -> None: + delivered: bool, attachment_urls: list[str] | None = None) -> None: """Best-effort persist of one inbound message to the store; never raises.""" try: _ibstore.write_message( INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown", text=question, group=group_id or None, delivered=delivered, + attachment_urls=attachment_urls or None, ) except Exception as exc: print(f"[signal-gateway] could not persist inbound message: {exc}", flush=True) @@ -223,6 +233,23 @@ def _forward_news(question: str, source: str, group_id: str | None, lang: str) - print(f"[signal-gateway] forwarded news-flagged message from {source}", flush=True) +def _store_media_ref(data: bytes, content_type: str | None) -> str | None: + """Persist inbound media durably and return its HTTP-resolvable reference URL. + + Best-effort: any failure returns None so the message still forwards and + persists with its transcript — only the media link is skipped. The bytes go + to disk (out of the graph); the returned URL is what lands in the message's + ``kb:attachment`` triple.""" + if not data: + return None + try: + media_id = _ibstore.store_media(INBOUND_STORE_DIR, data, content_type) + except Exception as exc: + print(f"[signal-gateway] could not store inbound media: {exc}", flush=True) + return None + return f"{GATEWAY_SELF_URL}/media/{media_id}" + + # Outbound send-control policy — the messenger analogue of EMAIL_SEND_POLICY. # Keyed by the *sending* account number (this gateway's own SIGNAL_ACCOUNT), NOT # the recipient: the category is resolved for the identity a message goes out as, @@ -385,19 +412,27 @@ def _attachment_path(att: dict) -> Path | None: return None -def _split_attachments(event: dict) -> tuple[Path | None, list[dict]]: - """Partition inbound attachments into (voice_note_path, image_files). +def _split_attachments(event: dict) -> tuple[Path | None, list[dict], list[str]]: + """Partition inbound attachments into (voice_note_path, files, attachment_urls). signal-cli labels each attachment with its contentType: ``audio/*`` is a voice note to transcribe, ``image/*`` is forwarded to the agent as a file payload (``{"filename", "content_type", "data"(base64)}`` — the shape the retinue gateway's POST /message accepts as ``files``). An attachment with no contentType keeps the legacy voice-note treatment, since before this - split every attachment was handed to the transcriber.""" + split every attachment was handed to the transcriber. + + Every attachment (image or voice note) is ALSO persisted durably and its + HTTP reference collected in ``attachment_urls`` — the ``kb:attachment`` triple + on the stored message. The voice note is additionally added to ``files`` so + the original audio rides into the conversation alongside its transcript. The + durable reference is stored regardless of size (consistency: a reference, + never inline); only the transient ``files`` payload honours the size cap.""" msg = event.get("envelope", {}).get("dataMessage") or {} attachments = msg.get("attachments") or [] voice: Path | None = None - images: list[dict] = [] + files: list[dict] = [] + attachment_urls: list[str] = [] for att in attachments: if not isinstance(att, dict): continue @@ -406,26 +441,46 @@ def _split_attachments(event: dict) -> tuple[Path | None, list[dict]]: print(f"[signal-gateway] attachment metadata present but file not found: {att}", flush=True) continue content_type = str(att.get("contentType") or "").lower() + try: + data = path.read_bytes() + except OSError as exc: + print(f"[signal-gateway] could not read inbound attachment {path}: {exc}", flush=True) + data = b"" if content_type.startswith("image/"): - try: - data = path.read_bytes() - except OSError as exc: - print(f"[signal-gateway] could not read inbound image {path}: {exc}", flush=True) - continue if not data: continue + ref = _store_media_ref(data, content_type) + if ref: + attachment_urls.append(ref) if len(data) > MAX_INBOUND_FILE_BYTES: print(f"[signal-gateway] inbound image too large to forward ({len(data)} bytes)", flush=True) continue suffix = path.suffix or mimetypes.guess_extension(content_type) or ".jpg" - images.append({ + files.append({ "filename": f"signal-image{suffix}", "content_type": content_type, "data": base64.b64encode(data).decode("ascii"), }) - elif voice is None: - voice = path - return voice, images + else: + # Voice note (audio/*) or an unlabeled attachment: transcribe the + # first one. Persist it durably as a reference, and — when it fits — + # attach the audio itself so the conversation carries it. + mime = content_type or "audio/ogg" + if data: + ref = _store_media_ref(data, mime) + if ref: + attachment_urls.append(ref) + if len(data) <= MAX_INBOUND_FILE_BYTES: + suffix = path.suffix or mimetypes.guess_extension(mime) or ".ogg" + label = "voice" if mime.startswith("audio/") else "attachment" + files.append({ + "filename": f"signal-{label}{suffix}", + "content_type": mime, + "data": base64.b64encode(data).decode("ascii"), + }) + if voice is None: + voice = path + return voice, files, attachment_urls def _extract_sender(event: dict) -> str | None: @@ -1053,7 +1108,7 @@ def _handle_event(event: dict) -> None: except Exception as exc: print(f"[signal-gateway] could not record recent sender: {exc}", flush=True) - voice, files = _split_attachments(event) + voice, files, attachment_urls = _split_attachments(event) if voice is not None: print(f"[signal-gateway] processing voice message from {sender}", flush=True) question, lang = _transcribe(voice) @@ -1079,7 +1134,7 @@ def _handle_event(event: dict) -> None: # account hands it to the user's triage and stays silent towards the sender. if SIGNAL_GATEWAY_MODE == "inbox": _forward_to_inbox(question, lang, sender, group_id=_extract_group_id(event), - files=files) + files=files, attachment_urls=attachment_urls) else: _handle_control_message(question, lang, sender, files=files) @@ -1122,7 +1177,8 @@ def _handle_control_message(question: str, lang: str, sender: str, def _forward_to_inbox(question: str, lang: str, sender: str, group_id: str | None = None, - files: list[dict] | None = None) -> None: + files: list[dict] | None = None, + attachment_urls: list[str] | None = None) -> None: """Hand an inbox-account message to the user's triage, notifying the user. The account is one of the user's own message sources, so the message is the @@ -1150,7 +1206,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str, if gate.get("news"): _forward_news(question, group_id if is_group else sender, group_id, lang) if not gate["forward"]: - _persist_inbound(question, sender, group_id, delivered=gate["delivered_if_held"]) + _persist_inbound(question, sender, group_id, delivered=gate["delivered_if_held"], + attachment_urls=attachment_urls) print( f"[signal-gateway] gate held inbox message from {sender_label} " f"({gate['reason']}); no model turn", @@ -1188,8 +1245,11 @@ def _forward_to_inbox(question: str, lang: str, sender: str, if gate["flagged_unknown"] else "" ) attachment_line = ( - (f"\nThe message includes {len(files)} attached image(s), forwarded " - f"with this prompt; their saved on-disk paths are listed at the end.\n") + (f"\nThe message includes {len(files)} attached file(s) (image(s) and/or " + f"the original voice note), forwarded with this prompt; their saved " + f"on-disk paths are listed at the end. When a voice note is attached, " + f"include the audio itself in the dashboard conversation (not only its " + f"transcript).\n") if files else "" ) prompt = ( @@ -1234,7 +1294,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str, # Persist AFTER forwarding so the delivered flag reflects reality: a message # handed to triage is delivered; one whose forward failed stays undelivered # and the daily drain retries it. - _persist_inbound(question, sender, group_id, delivered=forwarded) + _persist_inbound(question, sender, group_id, delivered=forwarded, + attachment_urls=attachment_urls) @@ -1673,6 +1734,23 @@ def do_GET(self): else: self._reply(status, body) return + if self.path.split("?", 1)[0].rstrip("/").startswith("/media/"): + # Resolve a durable inbound-media reference (kb:attachment). The bytes + # live on the store volume, out of the graph; this serves them back + # over HTTP. Token-gated like /qr — it is the user's private inbound + # content. load_media validates the id, so a crafted path cannot + # escape the media dir. + if not self._authorized(): + self._reply(401, {"error": "unauthorized"}) + return + media_id = self.path.split("?", 1)[0].rstrip("/")[len("/media/"):] + loaded = _ibstore.load_media(INBOUND_STORE_DIR, media_id) + if loaded is None: + self._reply(404, {"error": "not found"}) + return + data, content_type = loaded + self._reply_raw(200, data, content_type) + return if self.path.rstrip("/") == "/pending-sends": if not self._authorized(): self._reply(401, {"error": "unauthorized"}) diff --git a/scripts/telegram-gateway.py b/scripts/telegram-gateway.py index c49f7c0..add39fa 100644 --- a/scripts/telegram-gateway.py +++ b/scripts/telegram-gateway.py @@ -113,6 +113,13 @@ HTTP_PORT = int(os.environ.get("TELEGRAM_GATEWAY_HTTP_PORT", "8093")) DEFAULT_RECIPIENT = os.environ.get("TELEGRAM_DEFAULT_RECIPIENT", "").strip() GATEWAY_TOKEN = os.environ.get("TELEGRAM_GATEWAY_TOKEN", "").strip() +# The base URL other services (the life-store emitter, the dashboard) resolve a +# durable inbound-media reference against — i.e. where this gateway serves its own +# token-gated GET /media/. Defaults to the in-cluster service name; a +# deployment overrides it when the gateway is reachable at a different host. +GATEWAY_SELF_URL = os.environ.get( + "TELEGRAM_GATEWAY_SELF_URL", f"http://telegram-gateway:{HTTP_PORT}" +).rstrip("/") MAX_PUSH_BODY_BYTES = int(os.environ.get("TELEGRAM_GATEWAY_MAX_BODY_BYTES", str(25 * 1024 * 1024))) # Cap the decoded size of an inbound image forwarded to the agent (it travels # base64-encoded inside the POST /message JSON). Matches the retinue gateway's @@ -206,12 +213,13 @@ def _inbound_gate_decision(sender: str, group_id: str | None) -> dict: def _persist_inbound(question: str, sender: str, group_id: str | None, - delivered: bool) -> None: + delivered: bool, attachment_urls: list[str] | None = None) -> None: """Best-effort persist of one inbound message to the store; never raises.""" try: _ibstore.write_message( INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown", text=question, group=group_id or None, delivered=delivered, + attachment_urls=attachment_urls, ) except Exception as exc: print(f"[telegram-gateway] could not persist inbound message: {exc}", flush=True) @@ -227,6 +235,24 @@ def _forward_news(question: str, source: str, group_id: str | None, lang: str) - print(f"[telegram-gateway] forwarded news-flagged message from {source}", flush=True) +def _store_media_ref(data: bytes, content_type: str | None) -> str | None: + """Persist one inbound media blob durably and return its HTTP-resolvable URL. + + Best-effort: the durable reference is what keeps the original audio/image out + of the graph while still recoverable, so a failure here must never cost the + message — it just means this attachment has no reference. The bytes are stored + under the gateway's media dir (never inline in RDF) and served back by the + token-gated GET /media/.""" + if not data: + return None + try: + media_id = _ibstore.store_media(INBOUND_STORE_DIR, data, content_type) + return f"{GATEWAY_SELF_URL}/media/{media_id}" + except Exception as exc: + print(f"[telegram-gateway] could not store inbound media: {exc}", flush=True) + return None + + SEND_APPROVAL_BASE_URL = os.environ.get("SEND_APPROVAL_BASE_URL", "").rstrip("/") # Optional override for the /sends// segment of approval links. # Normally UNSET: the slug is derived per request from the Host header — the @@ -547,40 +573,49 @@ def _list_contacts() -> list: return fut.result(timeout=30) -def _inbound_image_files(image_path, image_mime: str | None) -> list[dict]: - """Read a downloaded inbound image as forward-ready file payloads. +def _inbound_image_files(image_path, image_mime: str | None) -> tuple[list[dict], list[str]]: + """Read a downloaded inbound image as forward-ready files + a durable ref. - Returns ``[{"filename", "content_type", "data"(base64)}, ...]`` — the shape - the retinue gateway's POST /message accepts as ``files``. Best-effort: any - failure (or an oversized image) forwards the message without its image - rather than dropping it. The temp file is always removed.""" + Returns ``(files, attachment_urls)`` where ``files`` is + ``[{"filename", "content_type", "data"(base64)}, ...]`` — the shape the + retinue gateway's POST /message accepts — and ``attachment_urls`` are + HTTP-resolvable references stored on this gateway's volume (never inlined in + RDF). Best-effort: any failure forwards the message without its image rather + than dropping it. The durable reference is stored regardless of size (it is a + plain on-disk blob); only the forwarded ``files`` payload honours the size + cap, since that one travels base64-encoded through the triage POST. The temp + file is always removed.""" if not image_path: - return [] + return [], [] path = Path(image_path) try: data = path.read_bytes() except OSError as exc: print(f"[telegram-gateway] could not read inbound image {path}: {exc}", flush=True) - return [] + return [], [] finally: path.unlink(missing_ok=True) if not data: - return [] + return [], [] + mime = image_mime or "image/jpeg" + ref = _store_media_ref(data, mime) + attachment_urls = [ref] if ref else [] if len(data) > MAX_INBOUND_FILE_BYTES: print(f"[telegram-gateway] inbound image too large to forward ({len(data)} bytes)", flush=True) - return [] - mime = image_mime or "image/jpeg" + return [], attachment_urls suffix = path.suffix or mimetypes.guess_extension(mime) or ".jpg" - return [{ + files = [{ "filename": f"telegram-image{suffix}", "content_type": mime, "data": base64.b64encode(data).decode("ascii"), }] + return files, attachment_urls def _handle_inbound(text: str, lang: str, chat_id: str, sender: str, is_group: bool, sender_name: str | None, - files: list[dict] | None = None) -> None: + files: list[dict] | None = None, + attachment_urls: list[str] | None = None) -> None: """Blocking dispatch — runs in a worker thread, off the asyncio loop.""" _record_recent_sender(str(chat_id), sender_name, None, is_group) if not text and not files: @@ -590,7 +625,8 @@ def _handle_inbound(text: str, lang: str, chat_id: str, sender: str, lang = _detect_text_language(text) if TELEGRAM_GATEWAY_MODE == "inbox": _forward_to_inbox(text, lang, str(chat_id), is_group=is_group, - sender_name=sender_name, files=files) + sender_name=sender_name, files=files, + attachment_urls=attachment_urls) else: _handle_control_message(text, lang, str(chat_id), sender, files=files) @@ -649,17 +685,42 @@ async def _on_new_message(event) -> None: def _work(): nonlocal text, lang + attachment_urls: list[str] = [] + voice_files: list[dict] = [] if media_path: + vpath = Path(media_path) + # Read the audio once and persist a durable reference BEFORE + # transcribing: the recording is the source of truth, so a failed + # or garbled transcription must never cost it. The bytes go to an + # on-disk blob (any size) plus the forward-ready files payload + # (size-capped, since that one travels base64 through triage). + try: + audio_bytes = vpath.read_bytes() + except OSError: + audio_bytes = b"" + vmime = mimetypes.guess_type(str(vpath))[0] or "audio/ogg" + ref = _store_media_ref(audio_bytes, vmime) if audio_bytes else None + if ref: + attachment_urls.append(ref) + if audio_bytes and len(audio_bytes) <= MAX_INBOUND_FILE_BYTES: + suffix = vpath.suffix or mimetypes.guess_extension(vmime) or ".ogg" + voice_files.append({ + "filename": f"telegram-voice{suffix}", + "content_type": vmime, + "data": base64.b64encode(audio_bytes).decode("ascii"), + }) try: print(f"[telegram-gateway] transcribing voice note from {sender}", flush=True) - text, lang = _transcribe(Path(media_path)) + text, lang = _transcribe(vpath) except Exception as exc: # noqa: BLE001 - degrade to placeholder print(f"[telegram-gateway] transcription failed: {exc}", flush=True) finally: - Path(media_path).unlink(missing_ok=True) - files = _inbound_image_files(image_path, image_mime) + vpath.unlink(missing_ok=True) + image_files, image_urls = _inbound_image_files(image_path, image_mime) + attachment_urls.extend(image_urls) + files = voice_files + image_files _handle_inbound(text, lang, str(chat_id), sender, is_group, sender_name, - files=files) + files=files, attachment_urls=attachment_urls) _LOOP.run_in_executor(None, _work) except Exception as exc: # noqa: BLE001 - one bad message must not stall the loop @@ -870,7 +931,8 @@ def _handle_control_message(question: str, lang: str, chat_id: str, sender: str, def _forward_to_inbox(question: str, lang: str, chat_id: str, is_group: bool = False, sender_name: str | None = None, - files: list[dict] | None = None) -> None: + files: list[dict] | None = None, + attachment_urls: list[str] | None = None) -> None: """Hand an inbox-account message to the user's triage, notifying the user.""" sender_label = sender_name or chat_id if sender_name: @@ -891,7 +953,8 @@ def _forward_to_inbox(question: str, lang: str, chat_id: str, if gate.get("news"): _forward_news(question, sender_name or handle, group_id, lang) if not gate["forward"]: - _persist_inbound(question, handle, group_id, delivered=gate["delivered_if_held"]) + _persist_inbound(question, handle, group_id, delivered=gate["delivered_if_held"], + attachment_urls=attachment_urls) print( f"[telegram-gateway] gate held inbox message from {sender_label} " f"({gate['reason']}); no model turn", @@ -928,8 +991,11 @@ def _forward_to_inbox(question: str, lang: str, chat_id: str, if gate["flagged_unknown"] else "" ) attachment_line = ( - (f"\nThe message includes {len(files)} attached image(s), forwarded " - f"with this prompt; their saved on-disk paths are listed at the end.\n") + (f"\nThe message includes {len(files)} attachment(s) — a voice note's " + f"audio and/or image(s) — forwarded with this prompt; their saved " + f"on-disk paths are listed at the end. When you raise the dashboard " + f"conversation, include the audio/media itself (not only its " + f"transcript), so the user can listen to or view the original.\n") if files else "" ) prompt = ( @@ -966,7 +1032,8 @@ def _forward_to_inbox(question: str, lang: str, chat_id: str, # Persist AFTER forwarding so the delivered flag reflects reality: a failed # forward stays undelivered and the daily drain retries it. - _persist_inbound(question, handle, group_id, delivered=forwarded) + _persist_inbound(question, handle, group_id, delivered=forwarded, + attachment_urls=attachment_urls) # ── Recent-senders store ────────────────────────────────────────────────────── @@ -1317,6 +1384,23 @@ def do_GET(self): else: self._reply(status, body) return + media_match = re.match(r"^/media/([^/?]+)/?$", self.path.split("?", 1)[0]) + if media_match: + # Resolve a durable inbound-media reference (kb:attachment). Token-gated + # like /qr — the blob is the user's own inbound audio/image, never + # inlined into the graph but served back here on demand. load_media + # validates the id (traversal-safe) and returns None for anything that + # is not a stored blob. + if not self._authorized(): + self._reply(401, {"error": "unauthorized"}) + return + got = _ibstore.load_media(INBOUND_STORE_DIR, media_match.group(1)) + if got is None: + self._reply(404, {"error": "not found"}) + return + data, content_type = got + self._reply_raw(200, data, content_type) + return if self.path.rstrip("/") == "/pending-sends": if not self._authorized(): self._reply(401, {"error": "unauthorized"}) diff --git a/scripts/whatsapp-gateway.py b/scripts/whatsapp-gateway.py index 5a1a879..c7c1a68 100644 --- a/scripts/whatsapp-gateway.py +++ b/scripts/whatsapp-gateway.py @@ -221,13 +221,23 @@ def _inbound_gate_decision(sender: str, group_id: str | None) -> dict: return {"forward": True, "flagged_unknown": False, "delivered_if_held": True, "reason": "policy-error"} +# This gateway's own base URL on the internal Docker network, used to build the +# HTTP references stored for inbound media (GET /media/ below). Defaults to +# the compose service name + HTTP port; a deployment running more than one +# WhatsApp identity overrides it per container (as it does SEND_APPROVAL_*). +GATEWAY_SELF_URL = os.environ.get( + "WHATSAPP_GATEWAY_SELF_URL", f"http://whatsapp-gateway:{HTTP_PORT}" +).rstrip("/") + + def _persist_inbound(question: str, sender: str, group_id: str | None, - delivered: bool) -> None: + delivered: bool, attachment_urls: list[str] | None = None) -> None: """Best-effort persist of one inbound message to the store; never raises.""" try: _ibstore.write_message( INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown", text=question, group=group_id or None, delivered=delivered, + attachment_urls=attachment_urls or None, ) except Exception as exc: print(f"[whatsapp-gateway] could not persist inbound message: {exc}", flush=True) @@ -243,6 +253,23 @@ def _forward_news(question: str, source: str, group_id: str | None, lang: str) - print(f"[whatsapp-gateway] forwarded news-flagged message from {source}", flush=True) +def _store_media_ref(data: bytes, content_type: str | None) -> str | None: + """Persist inbound media durably and return its HTTP-resolvable reference URL. + + Best-effort: any failure returns None so the message still forwards and + persists with its transcript — only the audio/image link is skipped, never + the message. The bytes go to disk (out of the graph); the returned URL is + what lands in the message's ``kb:attachment`` triple.""" + if not data: + return None + try: + media_id = _ibstore.store_media(INBOUND_STORE_DIR, data, content_type) + except Exception as exc: + print(f"[whatsapp-gateway] could not store inbound media: {exc}", flush=True) + return None + return f"{GATEWAY_SELF_URL}/media/{media_id}" + + # Public base URL used to build approval links returned to the caller. SEND_APPROVAL_BASE_URL = os.environ.get("SEND_APPROVAL_BASE_URL", "").rstrip("/") # Optional override for the /sends// segment of approval links. @@ -800,36 +827,43 @@ def _extract_image(message): return None -def _inbound_image_files(message) -> list[dict]: - """Download this message's image, if any, as forward-ready file payloads. +def _inbound_image_files(message) -> tuple[list[dict], list[str]]: + """Download this message's image, if any, as forward-ready files + a ref. - Returns ``[{"filename", "content_type", "data"(base64)}, ...]`` — the shape - the retinue gateway's POST /message accepts as ``files``, where each file is - materialized to disk for the answering session. Best-effort: any failure - (or an oversized image) forwards the message without its image rather than - dropping it.""" + Returns ``(files, attachment_urls)`` where ``files`` is + ``[{"filename", "content_type", "data"(base64)}, ...]`` — the shape the + retinue gateway's POST /message accepts as ``files``, each materialized to + disk for the answering session — and ``attachment_urls`` are the durable + HTTP references stored for the same image (its ``kb:attachment`` triple). + The durable reference is stored regardless of size (a plain on-disk blob); + only the forwarded ``files`` payload honours the size cap, since that one + travels base64-encoded through the triage POST. Best-effort: any failure + forwards the message without its image rather than dropping it.""" image = _extract_image(message) if image is None: - return [] + return [], [] media = _download_media(message) if media is None: - return [] + return [], [] try: data = media.read_bytes() finally: media.unlink(missing_ok=True) if not data: - return [] + return [], [] + mime = str(_attr(image, "mimetype", "Mimetype") or "image/jpeg") + ref = _store_media_ref(data, mime) + attachment_urls = [ref] if ref else [] if len(data) > MAX_INBOUND_FILE_BYTES: print(f"[whatsapp-gateway] inbound image too large to forward ({len(data)} bytes)", flush=True) - return [] - mime = str(_attr(image, "mimetype", "Mimetype") or "image/jpeg") + return [], attachment_urls suffix = mimetypes.guess_extension(mime) or ".jpg" - return [{ + files = [{ "filename": f"whatsapp-image{suffix}", "content_type": mime, "data": base64.b64encode(data).decode("ascii"), }] + return files, attachment_urls def _download_media(message) -> Path | None: @@ -1285,7 +1319,12 @@ def _handle_message_event(event) -> None: # An included image is forwarded alongside the text (which, for an image # message, is its caption). Status posts are excluded: they are gated to a # no-model-turn path anyway, so their media is never downloaded. - files = [] if is_broadcast else _inbound_image_files(message) + # attachment_urls collect the durable HTTP references (kb:attachment) for + # every piece of media on this message — image(s) here, the voice note below. + if is_broadcast: + files, attachment_urls = [], [] + else: + files, attachment_urls = _inbound_image_files(message) if not text and not files: # No text — try a voice note (download + transcribe via the STT service). @@ -1293,6 +1332,28 @@ def _handle_message_event(event) -> None: if audio is not None: media = _download_media(message) if media is not None: + # Read the bytes once: they feed the durable media reference, the + # files payload (so the original audio rides into the conversation + # alongside its transcript), AND transcription. Persist the audio + # BEFORE transcribing so a garbled or failed transcript never costs + # the recording — the audio is the source of truth here. + mime = str(_attr(audio, "mimetype", "Mimetype") or "audio/ogg; codecs=opus") + try: + audio_bytes = media.read_bytes() + except OSError as exc: + audio_bytes = b"" + print(f"[whatsapp-gateway] could not read voice note: {exc}", flush=True) + if audio_bytes: + ref = _store_media_ref(audio_bytes, mime) + if ref: + attachment_urls.append(ref) + if len(audio_bytes) <= MAX_INBOUND_FILE_BYTES: + suffix = mimetypes.guess_extension(mime.split(";", 1)[0].strip()) or ".ogg" + files.append({ + "filename": f"whatsapp-voice{suffix}", + "content_type": mime, + "data": base64.b64encode(audio_bytes).decode("ascii"), + }) try: print(f"[whatsapp-gateway] transcribing voice note from {sender}", flush=True) text, lang = _transcribe(media) @@ -1333,7 +1394,8 @@ def _handle_message_event(event) -> None: # through the normal send-approval policy, so a group send is not silent. origin = _jid_addr(chat_jid) or _jid_addr(sender_jid) _forward_to_inbox(text, lang, sender, is_group=is_group, - sender_name=push_name, origin=origin, files=files) + sender_name=push_name, origin=origin, files=files, + attachment_urls=attachment_urls) else: _handle_control_message(text, lang, sender, files=files) @@ -1371,7 +1433,8 @@ def _handle_control_message(question: str, lang: str, sender: str, def _forward_to_inbox(question: str, lang: str, sender: str, is_group: bool = False, sender_name: str | None = None, origin: str | None = None, - files: list[dict] | None = None) -> None: + files: list[dict] | None = None, + attachment_urls: list[str] | None = None) -> None: """Hand an inbox-account message to the user's triage, notifying the user. The account is one of the user's own message sources, so the message is the @@ -1402,7 +1465,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str, if gate.get("news"): _forward_news(question, sender_name or (group_id if is_group else sender), group_id, lang) if not gate["forward"]: - _persist_inbound(question, sender, group_id, delivered=gate["delivered_if_held"]) + _persist_inbound(question, sender, group_id, delivered=gate["delivered_if_held"], + attachment_urls=attachment_urls) print( f"[whatsapp-gateway] gate held inbox message from {sender_label} " f"({gate['reason']}); no model turn", @@ -1434,8 +1498,11 @@ def _forward_to_inbox(question: str, lang: str, sender: str, if gate["flagged_unknown"] else "" ) attachment_line = ( - (f"\nThe message includes {len(files)} attached image(s), forwarded " - f"with this prompt; their saved on-disk paths are listed at the end.\n") + (f"\nThe message includes {len(files)} attached file(s) (image(s) and/or " + f"the original voice note), forwarded with this prompt; their saved " + f"on-disk paths are listed at the end. When a voice note is attached, " + f"include the audio itself in the dashboard conversation (not only its " + f"transcript).\n") if files else "" ) prompt = ( @@ -1472,7 +1539,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str, # Persist AFTER forwarding so the delivered flag reflects reality: a failed # forward stays undelivered and the daily drain retries it. - _persist_inbound(question, sender, group_id, delivered=forwarded) + _persist_inbound(question, sender, group_id, delivered=forwarded, + attachment_urls=attachment_urls) def _forward_status_to_inbox(text: str, lang: str, sender: str, @@ -1926,6 +1994,23 @@ def do_GET(self): return self._reply_raw(200, png, "image/png") return + if self.path.split("?", 1)[0].rstrip("/").startswith("/media/"): + # Resolve a durable inbound-media reference (kb:attachment). The + # bytes live on the store volume, out of the graph; this serves them + # back over HTTP. Token-gated like /qr — it is the user's private + # inbound content. load_media validates the id, so a crafted path + # cannot escape the media dir. + if not self._authorized(): + self._reply(401, {"error": "unauthorized"}) + return + media_id = self.path.split("?", 1)[0].rstrip("/")[len("/media/"):] + loaded = _ibstore.load_media(INBOUND_STORE_DIR, media_id) + if loaded is None: + self._reply(404, {"error": "not found"}) + return + data, content_type = loaded + self._reply_raw(200, data, content_type) + return if self.path.rstrip("/") == "/pending-sends": if not self._authorized(): self._reply(401, {"error": "unauthorized"}) diff --git a/tests/test_inbound_image_forward.py b/tests/test_inbound_image_forward.py index ede6611..41d213a 100644 --- a/tests/test_inbound_image_forward.py +++ b/tests/test_inbound_image_forward.py @@ -182,20 +182,24 @@ def test_whatsapp_inbound_image_files(): media = Path(tmp) / "downloaded" media.write_bytes(PNG_BYTES) wg._download_media = lambda m: media - files = wg._inbound_image_files(message) + files, urls = wg._inbound_image_files(message) assert len(files) == 1, files assert files[0]["content_type"] == "image/png" assert files[0]["filename"].endswith(".png") assert base64.b64decode(files[0]["data"]) == PNG_BYTES assert not media.exists() # temp file cleaned up + # The durable HTTP reference is stored and points at GET /media/. + assert len(urls) == 1 and "/media/" in urls[0], urls - # Oversized image → message forwarded without it. + # Oversized image → forwarded without the base64 payload, but the durable + # reference is stored regardless of size (consistency over data-in-graph). media.write_bytes(PNG_BYTES) wg.MAX_INBOUND_FILE_BYTES = 4 - assert wg._inbound_image_files(message) == [] - # No image in the message → no download attempted. - assert wg._inbound_image_files(types.SimpleNamespace()) == [] - print("ok: whatsapp inbound image becomes a files payload") + files, urls = wg._inbound_image_files(message) + assert files == [] and len(urls) == 1 and "/media/" in urls[0], (files, urls) + # No image in the message → no download attempted, no ref stored. + assert wg._inbound_image_files(types.SimpleNamespace()) == ([], []) + print("ok: whatsapp inbound image becomes a files payload + durable ref") def test_whatsapp_forward_includes_files(): @@ -208,7 +212,7 @@ def test_whatsapp_forward_includes_files(): assert len(calls) == 1, calls payload = calls[0]["json"] assert payload["files"] == files - assert "1 attached image(s)" in payload["message"] + assert "1 attached file(s)" in payload["message"] # A message without images must not carry the key at all. wg._forward_to_inbox("plain text", "en", "+15551234567") assert "files" not in calls[1]["json"] @@ -241,30 +245,42 @@ def test_signal_split_attachments(): voice_file = att_dir / "note.ogg" voice_file.write_bytes(b"fake-ogg") - voice, images = sg._split_attachments(_signal_event([ + voice, files, urls = sg._split_attachments(_signal_event([ {"contentType": "image/jpeg", "file": str(image_file)}, {"contentType": "audio/ogg", "file": str(voice_file)}, {"contentType": "image/png", "file": str(att_dir / "missing.png")}, ])) assert voice == voice_file + images = [f for f in files if f["filename"].startswith("signal-image")] assert len(images) == 1, images assert images[0]["content_type"] == "image/jpeg" assert images[0]["filename"].endswith(".jpg") assert base64.b64decode(images[0]["data"]) == PNG_BYTES - - # Legacy: an attachment with no contentType keeps the voice-note path. - voice, images = sg._split_attachments(_signal_event([ + # The voice note now also rides in `files` as its own audio payload, so + # the original recording reaches the conversation alongside its transcript. + audio = [f for f in files if f["filename"].startswith("signal-voice")] + assert len(audio) == 1 and base64.b64decode(audio[0]["data"]) == b"fake-ogg", audio + # Both attachments (image + voice) get a durable HTTP reference. + assert len(urls) == 2 and all("/media/" in u for u in urls), urls + + # Legacy: an attachment with no contentType keeps the voice-note path — + # still transcribed, still referenced, and now carried as audio too. + voice, files, urls = sg._split_attachments(_signal_event([ {"file": str(voice_file)}, ])) - assert voice == voice_file and images == [] + assert voice == voice_file + assert [f for f in files if f["filename"].startswith("signal-image")] == [] + assert len(urls) == 1 and "/media/" in urls[0], urls - # Oversized image → dropped from the payload. + # Oversized image → dropped from the forwarded payload, but its durable + # reference is stored regardless of size (consistency over data-in-graph). sg.MAX_INBOUND_FILE_BYTES = 4 - voice, images = sg._split_attachments(_signal_event([ + voice, files, urls = sg._split_attachments(_signal_event([ {"contentType": "image/jpeg", "file": str(image_file)}, ])) - assert voice is None and images == [] - print("ok: signal attachments split into voice vs. image payloads") + assert voice is None and files == [] + assert len(urls) == 1 and "/media/" in urls[0], urls + print("ok: signal attachments split into voice/image payloads + durable refs") def test_signal_forward_includes_files(): @@ -276,7 +292,7 @@ def test_signal_forward_includes_files(): assert len(calls) == 1, calls payload = calls[0]["json"] assert payload["files"] == files - assert "1 attached image(s)" in payload["message"] + assert "1 attached file(s)" in payload["message"] sg._forward_to_inbox("plain", "en", "+15551234567") assert "files" not in calls[1]["json"] print("ok: signal forward carries files in the POST /message payload") @@ -299,18 +315,21 @@ def test_telegram_inbound_image_files(): tg = _load_telegram_gateway(Path(tmp)) image = Path(tmp) / "tg-img" image.write_bytes(PNG_BYTES) - files = tg._inbound_image_files(str(image), "image/png") + files, urls = tg._inbound_image_files(str(image), "image/png") assert len(files) == 1, files assert files[0]["content_type"] == "image/png" assert files[0]["filename"].endswith(".png") assert base64.b64decode(files[0]["data"]) == PNG_BYTES assert not image.exists() # temp file cleaned up + assert len(urls) == 1 and "/media/" in urls[0], urls - assert tg._inbound_image_files(None, None) == [] + assert tg._inbound_image_files(None, None) == ([], []) + # Oversized → no forwarded payload, but the durable reference is kept. image.write_bytes(PNG_BYTES) tg.MAX_INBOUND_FILE_BYTES = 4 - assert tg._inbound_image_files(str(image), "image/png") == [] - print("ok: telegram inbound image becomes a files payload") + files, urls = tg._inbound_image_files(str(image), "image/png") + assert files == [] and len(urls) == 1 and "/media/" in urls[0], (files, urls) + print("ok: telegram inbound image becomes a files payload + durable ref") def test_telegram_forward_includes_files(): @@ -322,7 +341,7 @@ def test_telegram_forward_includes_files(): assert len(calls) == 1, calls payload = calls[0]["json"] assert payload["files"] == files - assert "1 attached image(s)" in payload["message"] + assert "1 attachment(s)" in payload["message"] tg._forward_to_inbox("plain", "en", "12345") assert "files" not in calls[1]["json"] print("ok: telegram forward carries files in the POST /message payload")