diff --git a/scripts/inbound_store.py b/scripts/inbound_store.py index dba7158..af2319d 100644 --- a/scripts/inbound_store.py +++ b/scripts/inbound_store.py @@ -62,23 +62,45 @@ 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" # Optional reference to a retained raw-media file (e.g. a voice note's audio), # recorded when a message is persisted *before* transcription so a failed or # crashed STT run leaves a re-transcribable artifact instead of a silent drop. -# Cleared once the message is accounted for (transcribed and forwarded). +# Cleared once the message is accounted for (transcribed and forwarded). This is +# a *local file path*, not a reference for the reader: unlike P_ATTACHMENT (the +# message's durable, permanent media) it is bookkeeping for the re-transcribe +# retry and disappears the moment the transcript lands. P_MEDIA = KB + "media" # 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 raw media (voice-note audio) retained for a message that -# was persisted before transcription. It lives beside the messages so it shares -# the gateway's durable data volume; the reference is recorded via P_MEDIA and -# the file is unlinked once the message is transcribed and accounted for. +# 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. +# +# It doubles as the spool for raw media (voice-note audio) retained for a message +# persisted before transcription — same volume, same durability, but referenced +# via P_MEDIA and unlinked once the message is transcribed and accounted for. + 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: @@ -158,6 +180,11 @@ 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)) if fields.get("media"): lines.append(_lit(subj, P_MEDIA, fields["media"])) return "".join(l + "\n" for l in sorted(lines)) @@ -166,7 +193,7 @@ def _render(fields: dict) -> str: 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, - "media": None} + "attachments": [], "media": None} subject = None for line in text.splitlines(): line = line.strip() @@ -192,6 +219,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_MEDIA: fields["media"] = value elif pred == P_DELIVERED: @@ -263,6 +294,7 @@ def write_message( message_id: str | None = None, timestamp: float | None = None, delivered: bool = False, + attachment_urls: list[str] | None = None, media: str | None = None, ) -> tuple[str, Path]: """Persist one inbound message as a deterministic N-Triples file. @@ -272,9 +304,15 @@ def write_message( 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`. + ``media`` optionally records a reference (a durable file path) to raw media retained alongside this message — used by the persist-before-transcribe path - so a voice note survives a failed or crashed STT run. + so a voice note survives a failed or crashed STT run. Unlike + ``attachment_urls`` it is transient bookkeeping, cleared by + :func:`update_message` once the transcript is in. """ ts = time.time() if timestamp is None else float(timestamp) token = secrets.token_hex(8) @@ -288,6 +326,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], "media": media or None, } # Filename: zero-padded epoch millis (sortable) + token (unique, IRI-safe). @@ -297,6 +336,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, @@ -311,7 +400,9 @@ 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) and ``media`` (the local + path of an as-yet-untranscribed voice note, else None). """ mdir = messages_dir(store_dir) if not mdir.is_dir(): @@ -342,6 +433,7 @@ def undelivered( "message_id": fields.get("message_id"), "received_at": fields["received_at"], "text": fields["text"], + "attachments": fields.get("attachments") or [], "media": fields.get("media"), }) return out diff --git a/scripts/signal-gateway.py b/scripts/signal-gateway.py index 6bc65dd..9d481a5 100644 --- a/scripts/signal-gateway.py +++ b/scripts/signal-gateway.py @@ -203,19 +203,31 @@ 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, media: str | None = None): + delivered: bool, media: str | None = None, + attachment_urls: list[str] | None = None): """Best-effort persist of one inbound message to the store; never raises. Returns the store ``Path`` (so the caller can later flip the delivered flag with :func:`_mark_delivered`) or ``None`` if persistence failed. ``media`` records a retained raw-audio file for a voice note persisted before - transcription (see :func:`_retain_media`). + transcription (see :func:`_retain_media`); ``attachment_urls`` are the + durable HTTP references to this message's media (see :func:`_store_media_ref`). """ try: _, path = _ibstore.write_message( INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown", text=question, group=group_id or None, delivered=delivered, media=media, + attachment_urls=attachment_urls or None, ) return path except Exception as exc: @@ -276,6 +288,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, @@ -438,19 +467,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 @@ -459,26 +496,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: @@ -1106,7 +1163,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) # A voice note is persisted BEFORE transcription (never-drop): if the pre- # persist happened, this holds its store Path so the forward below reuses the # same record instead of writing a second one. @@ -1120,9 +1177,14 @@ def _handle_event(event: dict) -> None: # _forward_to_inbox persists. Only in inbox mode: a control account has no # triage drain that would pick a persisted record back up, so the never- # drop ledger is an inbox-mode concept and persisting there would leak. + # + # The retained copy is the *retry* artifact and is dropped once the + # transcript lands; it is distinct from the durable kb:attachment blob + # _split_attachments already wrote, which stays for good. durable = _retain_media(voice) or voice voice_store_path = _persist_inbound( "", sender, _extract_group_id(event), delivered=False, media=str(durable), + attachment_urls=attachment_urls, ) try: question, lang = _transcribe(durable) @@ -1170,7 +1232,8 @@ 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, store_path=voice_store_path) + files=files, attachment_urls=attachment_urls, + store_path=voice_store_path) else: _handle_control_message(question, lang, sender, files=files) @@ -1214,6 +1277,7 @@ 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, + attachment_urls: list[str] | None = None, store_path=None) -> None: """Hand an inbox-account message to the user's triage, notifying the user. @@ -1243,7 +1307,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str, # fully-resolved class). A voice note was already persisted before # transcription; reuse that record instead of writing a second one. if store_path is None: - store_path = _persist_inbound(question, sender, group_id, delivered=False) + store_path = _persist_inbound(question, sender, group_id, delivered=False, + attachment_urls=attachment_urls) # Delivery gate: decide whether this sender is worth a model turn now. A # held message is already persisted above; no `claude -p` session is spawned. @@ -1296,8 +1361,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 = ( @@ -1784,6 +1852,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 f59be45..beecaac 100644 --- a/scripts/telegram-gateway.py +++ b/scripts/telegram-gateway.py @@ -115,6 +115,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 @@ -208,18 +215,21 @@ def _inbound_gate_decision(sender: str, group_id: str | None) -> dict: def _persist_inbound(question: str, sender: str, group_id: str | None, - delivered: bool, media: str | None = None): + delivered: bool, media: str | None = None, + attachment_urls: list[str] | None = None): """Best-effort persist of one inbound message to the store; never raises. Returns the store ``Path`` (so the caller can later flip the delivered flag with :func:`_mark_delivered`) or ``None`` if persistence failed. ``media`` records a retained raw-audio file for a voice note persisted before - transcription (see :func:`_retain_media`). + transcription (see :func:`_retain_media`); ``attachment_urls`` are the + durable HTTP references to this message's media (see :func:`_store_media_ref`). """ try: _, path = _ibstore.write_message( INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown", text=question, group=group_id or None, delivered=delivered, media=media, + attachment_urls=attachment_urls, ) return path except Exception as exc: @@ -280,6 +290,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 @@ -600,40 +628,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, + attachment_urls: list[str] | None = None, store_path=None) -> None: """Blocking dispatch — runs in a worker thread, off the asyncio loop. @@ -657,6 +694,7 @@ def _handle_inbound(text: str, lang: str, chat_id: str, sender: str, if TELEGRAM_GATEWAY_MODE == "inbox": _forward_to_inbox(text, lang, str(chat_id), is_group=is_group, sender_name=sender_name, files=files, + attachment_urls=attachment_urls, store_path=store_path) else: _handle_control_message(text, lang, str(chat_id), sender, files=files) @@ -716,11 +754,42 @@ async def _on_new_message(event) -> None: def _work(): nonlocal text, lang + # Resolve the image attachment first so its durable reference is + # already in attachment_urls by the time the voice-note branch below + # pre-persists the message (a Telegram message carries one media, so + # in practice only one of the two ever fires — this just makes the + # record complete whichever it is). + image_files, image_urls = _inbound_image_files(image_path, image_mime) + attachment_urls: list[str] = list(image_urls) + voice_files: list[dict] = [] # A voice note is persisted BEFORE transcription (never-drop): if the # pre-persist happened, this holds its store Path so the forward reuses # the same record instead of writing a second one. voice_store_path = None 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). + # This read must precede _retain_media below, which *moves* the + # temp file away. + 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"), + }) if TELEGRAM_GATEWAY_MODE == "inbox": # Never-drop: retain the audio and persist the message up front, # THEN transcribe. A failed or crashed STT run leaves a durable, @@ -729,10 +798,15 @@ def _work(): # _handle_inbound, downstream of where _forward_to_inbox # persists. Only in inbox mode: a control account has no triage # drain that would pick a persisted record back up. + # + # The retained copy is the *retry* artifact and is dropped once + # the transcript lands; the kb:attachment blob stored above is + # the message's permanent media and stays. durable = _retain_media(media_path) or media_path grp = str(chat_id) if is_group else None voice_store_path = _persist_inbound( "", sender, grp, delivered=False, media=str(durable), + attachment_urls=attachment_urls, ) try: print(f"[telegram-gateway] transcribing voice note from {sender}", flush=True) @@ -750,14 +824,15 @@ def _work(): # Control mode: transient handling (no durable spool, no retry). 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) + files = voice_files + image_files _handle_inbound(text, lang, str(chat_id), sender, is_group, sender_name, - files=files, store_path=voice_store_path) + files=files, attachment_urls=attachment_urls, + store_path=voice_store_path) _LOOP.run_in_executor(None, _work) except Exception as exc: # noqa: BLE001 - one bad message must not stall the loop @@ -969,6 +1044,7 @@ 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, + attachment_urls: list[str] | None = None, store_path=None) -> None: """Hand an inbox-account message to the user's triage, notifying the user. @@ -996,7 +1072,8 @@ def _forward_to_inbox(question: str, lang: str, chat_id: str, # silently dropping it. The flag is flipped to true below once the message is # accounted for (forwarded to triage, or held in a fully-resolved class). if store_path is None: - store_path = _persist_inbound(question, handle, group_id, delivered=False) + store_path = _persist_inbound(question, handle, group_id, delivered=False, + attachment_urls=attachment_urls) # Delivery gate: only whitelisted / unknown senders get a model turn now. gate = _inbound_gate_decision(handle, group_id) @@ -1046,8 +1123,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 = ( @@ -1438,6 +1518,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 269e1f4..c8c9f8e 100644 --- a/scripts/whatsapp-gateway.py +++ b/scripts/whatsapp-gateway.py @@ -223,19 +223,31 @@ 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, media: str | None = None): + delivered: bool, media: str | None = None, + attachment_urls: list[str] | None = None): """Best-effort persist of one inbound message to the store; never raises. Returns the store ``Path`` (so the caller can later flip the delivered flag with :func:`_mark_delivered`) or ``None`` if persistence failed. ``media`` records a retained raw-audio file for a voice note persisted before - transcription (see :func:`_retain_media`). + transcription (see :func:`_retain_media`); ``attachment_urls`` are the + durable HTTP references to this message's media (see :func:`_store_media_ref`). """ try: _, path = _ibstore.write_message( INBOUND_STORE_DIR, channel=INBOUND_CHANNEL, sender=sender or "unknown", text=question, group=group_id or None, delivered=delivered, media=media, + attachment_urls=attachment_urls or None, ) return path except Exception as exc: @@ -296,6 +308,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. @@ -853,36 +882,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: @@ -1338,7 +1374,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) # A voice note is persisted BEFORE transcription (never-drop): if the pre- # persist happened, this holds its store Path so the forward below reuses the @@ -1350,6 +1391,29 @@ 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. This read + # must precede _retain_media below, which *moves* the temp file. + 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"), + }) if is_broadcast or WHATSAPP_GATEWAY_MODE != "inbox": # Transient handling (no durable spool, no retry): a status # post is gated to a no-model-turn path anyway, and a @@ -1369,10 +1433,15 @@ def _handle_message_event(event) -> None: # durable, re-transcribable record (delivered=False, media set) # for the daily drain — instead of vanishing at the skip-return # below, downstream of where _forward_to_inbox persists. + # + # The retained copy is the *retry* artifact and is dropped once + # the transcript lands; the kb:attachment blob stored above is + # the message's permanent media and stays. durable = _retain_media(media) or media grp = _jid_addr(chat_jid) if is_group else None voice_store_path = _persist_inbound( "", sender, grp, delivered=False, media=str(durable), + attachment_urls=attachment_urls, ) try: print(f"[whatsapp-gateway] transcribing voice note from {sender}", flush=True) @@ -1426,6 +1495,7 @@ def _handle_message_event(event) -> None: 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, + attachment_urls=attachment_urls, store_path=voice_store_path) else: _handle_control_message(text, lang, sender, files=files) @@ -1465,6 +1535,7 @@ 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, + attachment_urls: list[str] | None = None, store_path=None) -> None: """Hand an inbox-account message to the user's triage, notifying the user. @@ -1502,7 +1573,8 @@ def _forward_to_inbox(question: str, lang: str, sender: str, # accounted for (forwarded to triage, or held in a fully-resolved class). # A voice note was already persisted before transcription; reuse that record. if store_path is None: - store_path = _persist_inbound(question, sender, group_id, delivered=False) + store_path = _persist_inbound(question, sender, group_id, delivered=False, + attachment_urls=attachment_urls) # Delivery gate: only whitelisted / unknown senders get a model turn now. gate = _inbound_gate_decision(sender, group_id) @@ -1547,8 +1619,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 = ( @@ -2092,6 +2167,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")