Continue AI features and improve protocol support

core/clients/__init__.py

@@ -21,19 +21,19 @@ class ClientBase(ABC):
     #     ...
 
     async def message_received(self, *args, **kwargs):
-        self.ur.message_received(self.service, *args, **kwargs)
+        await self.ur.message_received(self.service, *args, **kwargs)
 
     async def message_read(self, *args, **kwargs):
-        self.ur.message_read(self.service, *args, **kwargs)
+        await self.ur.message_read(self.service, *args, **kwargs)
 
     async def started_typing(self, *args, **kwargs):
-        self.ur.started_typing(self.service, *args, **kwargs)
+        await self.ur.started_typing(self.service, *args, **kwargs)
 
     async def stopped_typing(self, *args, **kwargs):
-        self.ur.stopped_typing(self.service, *args, **kwargs)
+        await self.ur.stopped_typing(self.service, *args, **kwargs)
 
     async def reacted(self, *args, **kwargs):
-        self.ur.reacted(self.service, *args, **kwargs)
+        await self.ur.reacted(self.service, *args, **kwargs)
 
     async def replied(self, *args, **kwargs):
-        self.ur.replied(self.service, *args, **kwargs)
+        await self.ur.replied(self.service, *args, **kwargs)

core/clients/gateway.py

@@ -0,0 +1,185 @@
+import asyncio
+import time
+
+import aiohttp
+from asgiref.sync import sync_to_async
+from django.conf import settings
+
+from core.clients import ClientBase, transport
+from core.messaging import history
+from core.modules.mixed_protocol import normalize_gateway_event
+from core.models import PersonIdentifier
+
+
+class GatewayClient(ClientBase):
+    """
+    Generic gateway-backed client for mixed protocol services.
+
+    Expected gateway contract:
+    - GET  /v1/events/next -> JSON event (or 204 for no events)
+    - POST /v1/send
+    - POST /v1/typing/start
+    - POST /v1/typing/stop
+    """
+
+    poll_interval_seconds = 1
+
+    def __init__(self, ur, loop, service):
+        super().__init__(ur, loop, service)
+        self._task = None
+        self._stopping = False
+        self._not_found_count = 0
+        self.base_url = str(
+            getattr(
+                settings,
+                f"{service.upper()}_HTTP_URL",
+                f"http://{service}:8080",
+            )
+        ).rstrip("/")
+        self.enabled = bool(
+            str(
+                getattr(settings, f"{service.upper()}_ENABLED", "true"),
+            ).lower()
+            in {"1", "true", "yes", "on"}
+        )
+
+    def start(self):
+        if not self.enabled:
+            self.log.info("%s gateway disabled by settings", self.service)
+            return
+        if self._task is None:
+            self.log.info("%s gateway client starting (%s)", self.service, self.base_url)
+            self._task = self.loop.create_task(self._poll_loop())
+
+    async def start_typing(self, identifier):
+        return await transport.start_typing(self.service, identifier)
+
+    async def stop_typing(self, identifier):
+        return await transport.stop_typing(self.service, identifier)
+
+    async def _gateway_next_event(self):
+        url = f"{self.base_url}/v1/events/next"
+        timeout = aiohttp.ClientTimeout(total=30)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.get(url) as response:
+                if response.status == 204:
+                    return "empty", None
+                if response.status == 404:
+                    return "not_found", None
+                if response.status != 200:
+                    return "error", None
+                try:
+                    return "ok", await response.json()
+                except Exception:
+                    return "error", None
+
+    async def _poll_loop(self):
+        while not self._stopping:
+            try:
+                status, event = await self._gateway_next_event()
+                if status == "ok" and event:
+                    self._not_found_count = 0
+                    await self._handle_event(event)
+                elif status == "not_found":
+                    self._not_found_count += 1
+                    if self._not_found_count >= 3:
+                        self.log.warning(
+                            "%s gateway endpoint /v1/events/next returned 404 repeatedly; stopping client. "
+                            "Set %s_ENABLED=false or configure %s_HTTP_URL.",
+                            self.service,
+                            self.service.upper(),
+                            self.service.upper(),
+                        )
+                        self._stopping = True
+                        break
+                    await asyncio.sleep(self.poll_interval_seconds)
+                elif status in {"empty", "error"}:
+                    await asyncio.sleep(self.poll_interval_seconds)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                self.log.warning("%s gateway poll error: %s", self.service, exc)
+                await asyncio.sleep(max(2, self.poll_interval_seconds))
+
+    async def _handle_event(self, event):
+        normalized = normalize_gateway_event(self.service, event)
+        event_type = normalized.event_type
+        if event_type == "message":
+            await self._handle_message(normalized)
+            return
+        if event_type == "read":
+            await self.ur.message_read(
+                self.service,
+                identifier=normalized.identifier,
+                message_timestamps=normalized.message_timestamps,
+                read_ts=normalized.payload.get("read_ts"),
+                read_by=normalized.payload.get("read_by")
+                or normalized.payload.get("reader")
+                or normalized.identifier,
+                payload=normalized.payload,
+            )
+            return
+        if event_type in {"typing_start", "typing_started"}:
+            await self.ur.started_typing(
+                self.service,
+                identifier=normalized.identifier,
+                payload=normalized.payload,
+            )
+            return
+        if event_type in {"typing_stop", "typing_stopped"}:
+            await self.ur.stopped_typing(
+                self.service,
+                identifier=normalized.identifier,
+                payload=normalized.payload,
+            )
+            return
+
+    async def _handle_message(self, event):
+        identifier_value = event.identifier
+        if not identifier_value:
+            return
+
+        text = event.text
+        ts = int(event.ts or int(time.time() * 1000))
+        attachments = event.attachments
+        identifiers = await sync_to_async(list)(
+            PersonIdentifier.objects.filter(
+                identifier=identifier_value,
+                service=self.service,
+            )
+        )
+        if not identifiers:
+            return
+
+        xmpp_attachments = []
+        if attachments:
+            fetched = await asyncio.gather(
+                *[transport.fetch_attachment(self.service, att) for att in attachments]
+            )
+            for row in fetched:
+                if row:
+                    xmpp_attachments.append(row)
+
+        for identifier in identifiers:
+            session = await history.get_chat_session(identifier.user, identifier)
+            await history.store_message(
+                session=session,
+                sender=identifier_value,
+                text=text,
+                ts=ts,
+                outgoing=False,
+            )
+            await self.ur.xmpp.client.send_from_external(
+                identifier.user,
+                identifier,
+                text,
+                is_outgoing_message=False,
+                attachments=xmpp_attachments,
+            )
+            await self.ur.message_received(
+                self.service,
+                identifier=identifier,
+                text=text,
+                ts=ts,
+                payload=event.payload,
+            )

core/clients/instagram.py

@@ -0,0 +1,6 @@
+from core.clients.gateway import GatewayClient
+
+
+class InstagramClient(GatewayClient):
+    def __init__(self, ur, loop, service="instagram"):
+        super().__init__(ur, loop, service)

core/clients/serviceapi.py

@@ -0,0 +1,8 @@
+"""
+Backward-compatible compatibility layer.
+
+Prefer importing from `core.clients.transport`.
+"""
+
+from core.clients.transport import *  # noqa: F401,F403
+

core/clients/signal.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+from urllib.parse import urlparse
 
 import aiohttp
 from asgiref.sync import sync_to_async
@@ -15,12 +16,19 @@ from core.util import logs
 log = logs.get_logger("signalF")
 
 
-if settings.DEBUG:
-    SIGNAL_HOST = "127.0.0.1"
+_signal_http_url = getattr(settings, "SIGNAL_HTTP_URL", "").strip()
+if _signal_http_url:
+    parsed = urlparse(
+        _signal_http_url if "://" in _signal_http_url else f"http://{_signal_http_url}"
+    )
+    SIGNAL_HOST = parsed.hostname or "signal"
+    SIGNAL_PORT = parsed.port or 8080
 else:
-    SIGNAL_HOST = "signal"
-
-SIGNAL_PORT = 8080
+    if settings.DEBUG:
+        SIGNAL_HOST = "127.0.0.1"
+    else:
+        SIGNAL_HOST = "signal"
+    SIGNAL_PORT = 8080
 
 SIGNAL_URL = f"{SIGNAL_HOST}:{SIGNAL_PORT}"
 
@@ -103,6 +111,36 @@ def _extract_attachments(raw_payload):
     return results
 
 
+def _extract_receipt_timestamps(receipt_payload):
+    raw_ts = receipt_payload.get("timestamp")
+    if raw_ts is None:
+        raw_ts = receipt_payload.get("timestamps")
+    if isinstance(raw_ts, list):
+        out = []
+        for item in raw_ts:
+            try:
+                out.append(int(item))
+            except Exception:
+                continue
+        return out
+    if raw_ts is not None:
+        try:
+            return [int(raw_ts)]
+        except Exception:
+            return []
+    return []
+
+
+def _typing_started(typing_payload):
+    action = str(typing_payload.get("action") or "").strip().lower()
+    if action in {"started", "start", "typing", "composing"}:
+        return True
+    explicit = typing_payload.get("isTyping")
+    if isinstance(explicit, bool):
+        return explicit
+    return True
+
+
 class NewSignalBot(SignalBot):
     def __init__(self, ur, service, config):
         self.ur = ur
@@ -221,16 +259,53 @@ class HandleMessage(Command):
             log.warning("No Signal identifier available for message routing.")
             return
 
-        # Handle attachments across multiple Signal payload variants.
-        attachment_list = _extract_attachments(raw)
-
-        # Get users/person identifiers for this Signal sender/recipient.
+        # Resolve person identifiers once for this event.
         identifiers = await sync_to_async(list)(
             PersonIdentifier.objects.filter(
                 identifier=identifier_uuid,
                 service=self.service,
             )
         )
+
+        envelope = raw.get("envelope", {})
+        typing_payload = envelope.get("typingMessage")
+        if isinstance(typing_payload, dict):
+            for identifier in identifiers:
+                if _typing_started(typing_payload):
+                    await self.ur.started_typing(
+                        self.service,
+                        identifier=identifier,
+                        payload=typing_payload,
+                    )
+                else:
+                    await self.ur.stopped_typing(
+                        self.service,
+                        identifier=identifier,
+                        payload=typing_payload,
+                    )
+            return
+
+        receipt_payload = envelope.get("receiptMessage")
+        if isinstance(receipt_payload, dict):
+            read_timestamps = _extract_receipt_timestamps(receipt_payload)
+            read_ts = (
+                envelope.get("timestamp")
+                or envelope.get("serverReceivedTimestamp")
+                or c.message.timestamp
+            )
+            for identifier in identifiers:
+                await self.ur.message_read(
+                    self.service,
+                    identifier=identifier,
+                    message_timestamps=read_timestamps,
+                    read_ts=read_ts,
+                    payload=receipt_payload,
+                    read_by=source_uuid,
+                )
+            return
+
+        # Handle attachments across multiple Signal payload variants.
+        attachment_list = _extract_attachments(raw)
         xmpp_attachments = []
 
         # Asynchronously fetch all attachments

core/clients/signalapi.py

@@ -5,7 +5,6 @@ import aiohttp
 import orjson
 import requests
 from django.conf import settings
-from requests.exceptions import RequestException
 from rest_framework import status
 
 
@@ -57,12 +56,12 @@ async def download_and_encode_base64(file_url, filename, content_type):
                     f"data:{content_type};filename={filename};base64,{base64_encoded}"
                 )
 
-    except aiohttp.ClientError as e:
+    except aiohttp.ClientError:
         # log.error(f"Failed to download file: {file_url}, error: {e}")
         return None
 
 
-async def send_message_raw(recipient_uuid, text=None, attachments=[]):
+async def send_message_raw(recipient_uuid, text=None, attachments=None):
     """
     Sends a message using the Signal REST API, ensuring attachment links are not included in the text body.
 
@@ -84,6 +83,7 @@ async def send_message_raw(recipient_uuid, text=None, attachments=[]):
     }
 
     # Asynchronously download and encode all attachments
+    attachments = attachments or []
     tasks = [
         download_and_encode_base64(att["url"], att["filename"], att["content_type"])
         for att in attachments
@@ -182,12 +182,12 @@ def download_and_encode_base64_sync(file_url, filename, content_type):
 
         # Format according to Signal's expected structure
         return f"data:{content_type};filename={filename};base64,{base64_encoded}"
-    except requests.RequestException as e:
+    except requests.RequestException:
         # log.error(f"Failed to download file: {file_url}, error: {e}")
         return None
 
 
-def send_message_raw_sync(recipient_uuid, text=None, attachments=[]):
+def send_message_raw_sync(recipient_uuid, text=None, attachments=None):
     """
     Sends a message using the Signal REST API, ensuring attachment links are not included in the text body.
 
@@ -199,13 +199,16 @@ def send_message_raw_sync(recipient_uuid, text=None, attachments=[]):
     Returns:
         int | bool: Timestamp if successful, False otherwise.
     """
-    url = "http://signal:8080/v2/send"
+    base = getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080").rstrip("/")
+    url = f"{base}/v2/send"
     data = {
         "recipients": [recipient_uuid],
         "number": settings.SIGNAL_NUMBER,
         "base64_attachments": [],
     }
 
+    attachments = attachments or []
+
     # Convert attachments to Base64
     for att in attachments:
         base64_data = download_and_encode_base64_sync(
@@ -225,7 +228,7 @@ def send_message_raw_sync(recipient_uuid, text=None, attachments=[]):
     try:
         response = requests.post(url, json=data, timeout=10)
         response.raise_for_status()
-    except requests.RequestException as e:
+    except requests.RequestException:
         # log.error(f"Failed to send Signal message: {e}")
         return False
 

core/clients/transport.py

@@ -0,0 +1,417 @@
+import asyncio
+import base64
+import io
+import secrets
+import time
+from typing import Any
+
+import aiohttp
+import orjson
+import qrcode
+from django.conf import settings
+from django.core.cache import cache
+
+from core.clients import signalapi
+from core.messaging import media_bridge
+from core.util import logs
+
+log = logs.get_logger("transport")
+
+_RUNTIME_STATE_TTL = 60 * 60 * 24
+_RUNTIME_CLIENTS: dict[str, Any] = {}
+
+
+def _service_key(service: str) -> str:
+    return str(service or "").strip().lower()
+
+
+def _runtime_key(service: str) -> str:
+    return f"gia:service:runtime:{_service_key(service)}"
+
+
+def _gateway_base(service: str) -> str:
+    key = f"{service.upper()}_HTTP_URL"
+    default = f"http://{service}:8080"
+    return str(getattr(settings, key, default)).rstrip("/")
+
+
+def _as_qr_png(data: str) -> bytes:
+    image = qrcode.make(data)
+    stream = io.BytesIO()
+    image.save(stream, format="PNG")
+    return stream.getvalue()
+
+
+def _parse_timestamp(data: Any):
+    if isinstance(data, dict):
+        ts = data.get("timestamp")
+        if ts:
+            return ts
+    return None
+
+
+def register_runtime_client(service: str, client: Any):
+    """
+    Register an in-process runtime client (UR process).
+    """
+    _RUNTIME_CLIENTS[_service_key(service)] = client
+
+
+def get_runtime_client(service: str):
+    return _RUNTIME_CLIENTS.get(_service_key(service))
+
+
+def get_runtime_state(service: str) -> dict[str, Any]:
+    return dict(cache.get(_runtime_key(service)) or {})
+
+
+def update_runtime_state(service: str, **updates):
+    """
+    Persist runtime state to shared cache so web/UI process can read it.
+    """
+    key = _runtime_key(service)
+    state = dict(cache.get(key) or {})
+    state.update(updates)
+    state["updated_at"] = int(time.time())
+    cache.set(key, state, timeout=_RUNTIME_STATE_TTL)
+    return state
+
+
+def list_accounts(service: str):
+    """
+    Return account identifiers for service UI list.
+    """
+    service_key = _service_key(service)
+    if service_key == "signal":
+        import requests
+
+        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip("/")
+        try:
+            response = requests.get(f"{base}/v1/accounts", timeout=20)
+            if not response.ok:
+                return []
+            payload = orjson.loads(response.text or "[]")
+            if isinstance(payload, list):
+                return payload
+        except Exception:
+            return []
+        return []
+
+    state = get_runtime_state(service_key)
+    accounts = state.get("accounts") or []
+    if isinstance(accounts, list):
+        return accounts
+    return []
+
+
+def get_service_warning(service: str) -> str:
+    service_key = _service_key(service)
+    if service_key == "signal":
+        return ""
+
+    state = get_runtime_state(service_key)
+    warning = str(state.get("warning") or "").strip()
+    if warning:
+        return warning
+
+    if not state.get("connected"):
+        return (
+            f"{service_key.title()} runtime is not connected yet. "
+            "Start UR with the service enabled, open Services -> "
+            f"{service_key.title()} -> Add Account, then scan the QR from "
+            "WhatsApp Linked Devices."
+        )
+    return ""
+
+
+async def _gateway_json(method: str, url: str, payload=None):
+    timeout = aiohttp.ClientTimeout(total=20)
+    async with aiohttp.ClientSession(timeout=timeout) as session:
+        request = getattr(session, method.lower())
+        async with request(url, json=payload) as response:
+            body = await response.read()
+            if not body:
+                return response.status, None
+            try:
+                return response.status, orjson.loads(body)
+            except Exception:
+                return response.status, None
+
+
+async def _normalize_gateway_attachment(service: str, row: dict, session):
+    normalized = dict(row or {})
+    content = normalized.get("content")
+    if isinstance(content, memoryview):
+        content = content.tobytes()
+    if isinstance(content, bytes):
+        blob_key = media_bridge.put_blob(
+            service=service,
+            content=content,
+            filename=normalized.get("filename") or "attachment.bin",
+            content_type=normalized.get("content_type") or "application/octet-stream",
+        )
+        return {
+            "blob_key": blob_key,
+            "filename": normalized.get("filename") or "attachment.bin",
+            "content_type": normalized.get("content_type")
+            or "application/octet-stream",
+            "size": normalized.get("size") or len(content),
+        }
+
+    if normalized.get("blob_key"):
+        return normalized
+
+    source_url = normalized.get("url")
+    if source_url:
+        try:
+            async with session.get(source_url) as response:
+                if response.status == 200:
+                    payload = await response.read()
+                    blob_key = media_bridge.put_blob(
+                        service=service,
+                        content=payload,
+                        filename=normalized.get("filename")
+                        or source_url.rstrip("/").split("/")[-1]
+                        or "attachment.bin",
+                        content_type=normalized.get("content_type")
+                        or response.headers.get(
+                            "Content-Type", "application/octet-stream"
+                        ),
+                    )
+                    return {
+                        "blob_key": blob_key,
+                        "filename": normalized.get("filename")
+                        or source_url.rstrip("/").split("/")[-1]
+                        or "attachment.bin",
+                        "content_type": normalized.get("content_type")
+                        or response.headers.get(
+                            "Content-Type", "application/octet-stream"
+                        ),
+                        "size": normalized.get("size") or len(payload),
+                    }
+        except Exception:
+            log.warning("%s attachment fetch failed for %s", service, source_url)
+    return normalized
+
+
+async def _gateway_send(service: str, recipient: str, text=None, attachments=None):
+    base = _gateway_base(service)
+    url = f"{base}/v1/send"
+    timeout = aiohttp.ClientTimeout(total=20)
+    async with aiohttp.ClientSession(timeout=timeout) as media_session:
+        normalized_attachments = await asyncio.gather(
+            *[
+                _normalize_gateway_attachment(service, dict(att or {}), media_session)
+                for att in (attachments or [])
+            ]
+        )
+
+    data = {
+        "recipient": recipient,
+        "text": text or "",
+        "attachments": normalized_attachments,
+    }
+    status, payload = await _gateway_json("post", url, data)
+    if 200 <= status < 300:
+        ts = _parse_timestamp(payload)
+        return ts if ts else True
+    log.warning("%s gateway send failed (%s): %s", service, status, payload)
+    return False
+
+
+async def _gateway_typing(service: str, recipient: str, started: bool):
+    base = _gateway_base(service)
+    action = "start" if started else "stop"
+    url = f"{base}/v1/typing/{action}"
+    payload = {"recipient": recipient}
+    status, _ = await _gateway_json("post", url, payload)
+    if 200 <= status < 300:
+        return True
+    return False
+
+
+async def send_message_raw(service: str, recipient: str, text=None, attachments=None):
+    """
+    Unified outbound send path used by models/views/UR.
+    """
+    service_key = _service_key(service)
+    if service_key == "signal":
+        return await signalapi.send_message_raw(recipient, text, attachments or [])
+
+    if service_key in {"whatsapp", "instagram"}:
+        runtime_client = get_runtime_client(service_key)
+        if runtime_client and hasattr(runtime_client, "send_message_raw"):
+            try:
+                runtime_result = await runtime_client.send_message_raw(
+                    recipient,
+                    text=text,
+                    attachments=attachments or [],
+                )
+                if runtime_result is not False and runtime_result is not None:
+                    return runtime_result
+            except Exception as exc:
+                log.warning("%s runtime send failed: %s", service_key, exc)
+        return await _gateway_send(
+            service_key,
+            recipient,
+            text=text,
+            attachments=attachments or [],
+        )
+
+    if service_key == "xmpp":
+        raise NotImplementedError("Direct XMPP send is handled by the XMPP client.")
+    raise NotImplementedError(f"Unsupported service: {service}")
+
+
+async def start_typing(service: str, recipient: str):
+    service_key = _service_key(service)
+    if service_key == "signal":
+        await signalapi.start_typing(recipient)
+        return True
+
+    if service_key in {"whatsapp", "instagram"}:
+        runtime_client = get_runtime_client(service_key)
+        if runtime_client and hasattr(runtime_client, "start_typing"):
+            try:
+                result = await runtime_client.start_typing(recipient)
+                if result:
+                    return True
+            except Exception as exc:
+                log.warning("%s runtime start_typing failed: %s", service_key, exc)
+        return await _gateway_typing(service_key, recipient, started=True)
+    return False
+
+
+async def stop_typing(service: str, recipient: str):
+    service_key = _service_key(service)
+    if service_key == "signal":
+        await signalapi.stop_typing(recipient)
+        return True
+
+    if service_key in {"whatsapp", "instagram"}:
+        runtime_client = get_runtime_client(service_key)
+        if runtime_client and hasattr(runtime_client, "stop_typing"):
+            try:
+                result = await runtime_client.stop_typing(recipient)
+                if result:
+                    return True
+            except Exception as exc:
+                log.warning("%s runtime stop_typing failed: %s", service_key, exc)
+        return await _gateway_typing(service_key, recipient, started=False)
+    return False
+
+
+async def fetch_attachment(service: str, attachment_ref: dict):
+    """
+    Fetch attachment bytes from a source service or URL.
+    """
+    service_key = _service_key(service)
+    if service_key == "signal":
+        attachment_id = attachment_ref.get("id") or attachment_ref.get("attachment_id")
+        if not attachment_id:
+            return None
+        return await signalapi.fetch_signal_attachment(attachment_id)
+
+    runtime_client = get_runtime_client(service_key)
+    if runtime_client and hasattr(runtime_client, "fetch_attachment"):
+        try:
+            from_runtime = await runtime_client.fetch_attachment(attachment_ref)
+            if from_runtime:
+                return from_runtime
+        except Exception as exc:
+            log.warning("%s runtime attachment fetch failed: %s", service_key, exc)
+
+    direct_url = attachment_ref.get("url")
+    blob_key = attachment_ref.get("blob_key")
+    if blob_key:
+        return media_bridge.get_blob(blob_key)
+    if direct_url:
+        timeout = aiohttp.ClientTimeout(total=20)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.get(direct_url) as response:
+                if response.status != 200:
+                    return None
+                content = await response.read()
+                return {
+                    "content": content,
+                    "content_type": response.headers.get(
+                        "Content-Type",
+                        attachment_ref.get("content_type", "application/octet-stream"),
+                    ),
+                    "filename": attachment_ref.get("filename")
+                    or direct_url.rstrip("/").split("/")[-1]
+                    or "attachment.bin",
+                    "size": len(content),
+                }
+    return None
+
+
+def _qr_from_runtime_state(service: str) -> bytes | None:
+    state = get_runtime_state(service)
+    qr_payload = str(state.get("pair_qr") or "").strip()
+    if not qr_payload:
+        return None
+    if qr_payload.startswith("data:image/") and "," in qr_payload:
+        _, b64_data = qr_payload.split(",", 1)
+        try:
+            return base64.b64decode(b64_data)
+        except Exception:
+            return None
+    return _as_qr_png(qr_payload)
+
+
+def get_link_qr(service: str, device_name: str):
+    """
+    Returns PNG bytes for account-linking QR.
+
+    - Signal: uses signal-cli REST endpoint.
+    - WhatsApp/Instagram: runtime QR from shared state when available.
+      Falls back to local pairing token QR in development.
+    """
+    service_key = _service_key(service)
+    device = (device_name or "GIA Device").strip()
+
+    if service_key == "signal":
+        import requests
+
+        base = str(getattr(settings, "SIGNAL_HTTP_URL", "http://signal:8080")).rstrip("/")
+        response = requests.get(
+            f"{base}/v1/qrcodelink",
+            params={"device_name": device},
+            timeout=20,
+        )
+        response.raise_for_status()
+        return response.content
+
+    if service_key in {"whatsapp", "instagram"}:
+        runtime_client = get_runtime_client(service_key)
+        if runtime_client and hasattr(runtime_client, "get_link_qr_png"):
+            try:
+                image_bytes = runtime_client.get_link_qr_png(device)
+                if image_bytes:
+                    return image_bytes
+            except Exception:
+                pass
+
+        cached = _qr_from_runtime_state(service_key)
+        if cached:
+            return cached
+
+        token = secrets.token_urlsafe(24)
+        uri = f"gia://{service_key}/link?device={device}&token={token}"
+        update_runtime_state(
+            service_key,
+            pair_device=device,
+            pair_requested_at=int(time.time()),
+            warning=(
+                "Waiting for runtime pairing QR. "
+                "If this persists, check UR logs and Neonize session state."
+            ),
+        )
+        return _as_qr_png(uri)
+
+    raise NotImplementedError(f"Unsupported service for QR linking: {service}")
+
+
+def image_bytes_to_base64(image_bytes: bytes) -> str:
+    return base64.b64encode(image_bytes).decode("utf-8")

core/clients/whatsapp.py

@@ -0,0 +1,627 @@
+import asyncio
+import re
+import time
+
+import aiohttp
+from asgiref.sync import sync_to_async
+from django.conf import settings
+
+from core.clients import ClientBase, transport
+from core.messaging import history, media_bridge
+from core.models import PersonIdentifier
+
+
+class WhatsAppClient(ClientBase):
+    """
+    Async WhatsApp transport backed by Neonize.
+
+    Design notes:
+    - Runs in UR process.
+    - Publishes runtime state to shared cache via transport.
+    - Degrades gracefully when Neonize/session is unavailable.
+    """
+
+    def __init__(self, ur, loop, service="whatsapp"):
+        super().__init__(ur, loop, service)
+        self._task = None
+        self._stopping = False
+        self._client = None
+        self._build_jid = None
+        self._connected = False
+        self._last_qr_payload = ""
+        self._accounts = []
+
+        self.enabled = bool(
+            str(getattr(settings, "WHATSAPP_ENABLED", "false")).lower()
+            in {"1", "true", "yes", "on"}
+        )
+        self.client_name = str(
+            getattr(settings, "WHATSAPP_CLIENT_NAME", "gia_whatsapp")
+        ).strip() or "gia_whatsapp"
+        self.database_url = str(
+            getattr(settings, "WHATSAPP_DATABASE_URL", "")
+        ).strip()
+
+        transport.register_runtime_client(self.service, self)
+        self._publish_state(
+            connected=False,
+            warning=(
+                "WhatsApp runtime is disabled by settings."
+                if not self.enabled
+                else ""
+            ),
+            accounts=[],
+        )
+
+    def _publish_state(self, **updates):
+        state = transport.update_runtime_state(self.service, **updates)
+        accounts = state.get("accounts")
+        if isinstance(accounts, list):
+            self._accounts = accounts
+
+    def start(self):
+        if not self.enabled:
+            self.log.info("whatsapp client disabled by settings")
+            return
+        if self._task is None:
+            self.log.info("whatsapp neonize client starting")
+            self._task = self.loop.create_task(self._run())
+
+    async def _run(self):
+        try:
+            from neonize.aioze.client import NewAClient
+            from neonize.aioze import events as wa_events
+            try:
+                from neonize.utils import build_jid as wa_build_jid
+            except Exception:
+                wa_build_jid = None
+        except Exception as exc:
+            self._publish_state(
+                connected=False,
+                warning=f"Neonize not available: {exc}",
+                accounts=[],
+            )
+            self.log.warning("whatsapp neonize import failed: %s", exc)
+            return
+
+        self._build_jid = wa_build_jid
+        self._client = self._build_client(NewAClient)
+        if self._client is None:
+            self._publish_state(
+                connected=False,
+                warning="Failed to initialize Neonize client.",
+                accounts=[],
+            )
+            return
+
+        self._register_event_handlers(wa_events)
+
+        try:
+            await self._maybe_await(self._client.connect())
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            self._publish_state(
+                connected=False,
+                warning=f"WhatsApp connect failed: {exc}",
+                accounts=[],
+            )
+            self.log.warning("whatsapp connect failed: %s", exc)
+            return
+
+        # Keep task alive so state/callbacks remain active.
+        while not self._stopping:
+            await asyncio.sleep(1)
+
+    def _build_client(self, cls):
+        candidates = []
+        if self.database_url:
+            candidates.append((self.client_name, self.database_url))
+        candidates.append((self.client_name,))
+        for args in candidates:
+            try:
+                return cls(*args)
+            except TypeError:
+                continue
+            except Exception as exc:
+                self.log.warning("whatsapp client init failed for args %s: %s", args, exc)
+        try:
+            if self.database_url:
+                return cls(name=self.client_name, database=self.database_url)
+            return cls(name=self.client_name)
+        except Exception as exc:
+            self.log.warning("whatsapp client init failed: %s", exc)
+            return None
+
+    def _register_event_handlers(self, wa_events):
+        connected_ev = getattr(wa_events, "ConnectedEv", None)
+        message_ev = getattr(wa_events, "MessageEv", None)
+        receipt_ev = getattr(wa_events, "ReceiptEv", None)
+        presence_ev = getattr(wa_events, "PresenceEv", None)
+        pair_ev = getattr(wa_events, "PairStatusEv", None)
+
+        if connected_ev is not None:
+
+            async def on_connected(client, event: connected_ev):
+                self._connected = True
+                account = await self._resolve_account_identifier()
+                self._publish_state(
+                    connected=True,
+                    warning="",
+                    accounts=[account] if account else [self.client_name],
+                )
+
+            self._client.event(on_connected)
+
+        if message_ev is not None:
+
+            async def on_message(client, event: message_ev):
+                await self._handle_message_event(event)
+
+            self._client.event(on_message)
+
+        if receipt_ev is not None:
+
+            async def on_receipt(client, event: receipt_ev):
+                await self._handle_receipt_event(event)
+
+            self._client.event(on_receipt)
+
+        if presence_ev is not None:
+
+            async def on_presence(client, event: presence_ev):
+                await self._handle_presence_event(event)
+
+            self._client.event(on_presence)
+
+        if pair_ev is not None:
+
+            async def on_pair_status(client, event: pair_ev):
+                qr_payload = self._extract_pair_qr(event)
+                if qr_payload:
+                    self._last_qr_payload = qr_payload
+                    self._publish_state(
+                        pair_qr=qr_payload,
+                        warning="Scan QR to pair WhatsApp account.",
+                    )
+
+            self._client.event(on_pair_status)
+
+    async def _maybe_await(self, value):
+        if asyncio.iscoroutine(value):
+            return await value
+        return value
+
+    async def _resolve_account_identifier(self):
+        if self._client is None:
+            return ""
+        if not hasattr(self._client, "get_me"):
+            return self.client_name
+        try:
+            me = await self._maybe_await(self._client.get_me())
+        except Exception:
+            return self.client_name
+        # Support both dict-like and object-like payloads.
+        for path in (
+            ("JID", "User"),
+            ("jid",),
+            ("user",),
+            ("ID",),
+        ):
+            value = self._pluck(me, *path)
+            if value:
+                return str(value)
+        return self.client_name
+
+    def _pluck(self, obj, *path):
+        current = obj
+        for key in path:
+            if current is None:
+                return None
+            if isinstance(current, dict):
+                current = current.get(key)
+                continue
+            if hasattr(current, key):
+                current = getattr(current, key)
+                continue
+            return None
+        return current
+
+    def _normalize_timestamp(self, raw_value):
+        if raw_value is None:
+            return int(time.time() * 1000)
+        try:
+            value = int(raw_value)
+        except Exception:
+            return int(time.time() * 1000)
+        # WhatsApp libs often emit seconds. Promote to ms.
+        if value < 10**12:
+            return value * 1000
+        return value
+
+    def _normalize_identifier_candidates(self, *values):
+        out = set()
+        for value in values:
+            raw = str(value or "").strip()
+            if not raw:
+                continue
+            out.add(raw)
+            if "@" in raw:
+                out.add(raw.split("@", 1)[0])
+            digits = re.sub(r"[^0-9]", "", raw)
+            if digits:
+                out.add(digits)
+                if not digits.startswith("+"):
+                    out.add(f"+{digits}")
+        return out
+
+    def _is_media_message(self, message_obj):
+        media_fields = (
+            "imageMessage",
+            "videoMessage",
+            "audioMessage",
+            "documentMessage",
+            "stickerMessage",
+            "image_message",
+            "video_message",
+            "audio_message",
+            "document_message",
+            "sticker_message",
+        )
+        for field in media_fields:
+            value = self._pluck(message_obj, field)
+            if value:
+                return True
+        return False
+
+    async def _download_event_media(self, event):
+        if not self._client:
+            return []
+        msg_obj = self._pluck(event, "message")
+        if msg_obj is None or not self._is_media_message(msg_obj):
+            return []
+        if not hasattr(self._client, "download_any"):
+            return []
+
+        try:
+            payload = await self._maybe_await(self._client.download_any(msg_obj))
+        except Exception as exc:
+            self.log.warning("whatsapp media download failed: %s", exc)
+            return []
+
+        if isinstance(payload, memoryview):
+            payload = payload.tobytes()
+        if not isinstance(payload, (bytes, bytearray)):
+            return []
+
+        filename = (
+            self._pluck(msg_obj, "documentMessage", "fileName")
+            or self._pluck(msg_obj, "document_message", "file_name")
+            or f"wa-{int(time.time())}.bin"
+        )
+        content_type = (
+            self._pluck(msg_obj, "documentMessage", "mimetype")
+            or self._pluck(msg_obj, "document_message", "mimetype")
+            or self._pluck(msg_obj, "imageMessage", "mimetype")
+            or self._pluck(msg_obj, "image_message", "mimetype")
+            or "application/octet-stream"
+        )
+        blob_key = media_bridge.put_blob(
+            service="whatsapp",
+            content=bytes(payload),
+            filename=filename,
+            content_type=content_type,
+        )
+        if not blob_key:
+            return []
+        return [
+            {
+                "blob_key": blob_key,
+                "filename": filename,
+                "content_type": content_type,
+                "size": len(payload),
+            }
+        ]
+
+    async def _handle_message_event(self, event):
+        msg_obj = self._pluck(event, "message")
+        text = (
+            self._pluck(msg_obj, "conversation")
+            or self._pluck(msg_obj, "extendedTextMessage", "text")
+            or self._pluck(msg_obj, "extended_text_message", "text")
+            or ""
+        )
+
+        sender = (
+            self._pluck(event, "info", "message_source", "sender")
+            or self._pluck(event, "info", "messageSource", "sender")
+            or ""
+        )
+        chat = (
+            self._pluck(event, "info", "message_source", "chat")
+            or self._pluck(event, "info", "messageSource", "chat")
+            or ""
+        )
+        raw_ts = (
+            self._pluck(event, "info", "timestamp")
+            or self._pluck(event, "info", "message_timestamp")
+            or self._pluck(event, "timestamp")
+        )
+        ts = self._normalize_timestamp(raw_ts)
+
+        identifier_values = self._normalize_identifier_candidates(sender, chat)
+        if not identifier_values:
+            return
+
+        identifiers = await sync_to_async(list)(
+            PersonIdentifier.objects.filter(
+                service="whatsapp",
+                identifier__in=list(identifier_values),
+            )
+        )
+        if not identifiers:
+            return
+
+        attachments = await self._download_event_media(event)
+        xmpp_attachments = []
+        if attachments:
+            fetched = await asyncio.gather(
+                *[transport.fetch_attachment(self.service, att) for att in attachments]
+            )
+            xmpp_attachments = [row for row in fetched if row]
+
+        payload = {
+            "sender": str(sender or ""),
+            "chat": str(chat or ""),
+            "raw_event": str(type(event).__name__),
+        }
+
+        for identifier in identifiers:
+            session = await history.get_chat_session(identifier.user, identifier)
+            await history.store_message(
+                session=session,
+                sender=str(sender or chat or ""),
+                text=text,
+                ts=ts,
+                outgoing=False,
+            )
+            await self.ur.xmpp.client.send_from_external(
+                identifier.user,
+                identifier,
+                text,
+                is_outgoing_message=False,
+                attachments=xmpp_attachments,
+            )
+            await self.ur.message_received(
+                self.service,
+                identifier=identifier,
+                text=text,
+                ts=ts,
+                payload=payload,
+            )
+
+    async def _handle_receipt_event(self, event):
+        sender = (
+            self._pluck(event, "info", "message_source", "sender")
+            or self._pluck(event, "info", "messageSource", "sender")
+            or ""
+        )
+        chat = (
+            self._pluck(event, "info", "message_source", "chat")
+            or self._pluck(event, "info", "messageSource", "chat")
+            or ""
+        )
+        timestamps = []
+        raw_ids = self._pluck(event, "message_ids") or []
+        if isinstance(raw_ids, list):
+            for item in raw_ids:
+                try:
+                    value = int(item)
+                    timestamps.append(value * 1000 if value < 10**12 else value)
+                except Exception:
+                    continue
+        read_ts = self._normalize_timestamp(self._pluck(event, "timestamp") or int(time.time() * 1000))
+
+        for candidate in self._normalize_identifier_candidates(sender, chat):
+            await self.ur.message_read(
+                self.service,
+                identifier=candidate,
+                message_timestamps=timestamps,
+                read_ts=read_ts,
+                read_by=sender or chat,
+                payload={"event": "receipt", "sender": str(sender), "chat": str(chat)},
+            )
+
+    async def _handle_presence_event(self, event):
+        sender = (
+            self._pluck(event, "message_source", "sender")
+            or self._pluck(event, "info", "message_source", "sender")
+            or ""
+        )
+        chat = (
+            self._pluck(event, "message_source", "chat")
+            or self._pluck(event, "info", "message_source", "chat")
+            or ""
+        )
+        presence = str(self._pluck(event, "presence") or "").strip().lower()
+
+        for candidate in self._normalize_identifier_candidates(sender, chat):
+            if presence in {"composing", "typing", "recording"}:
+                await self.ur.started_typing(
+                    self.service,
+                    identifier=candidate,
+                    payload={"presence": presence, "sender": str(sender), "chat": str(chat)},
+                )
+            elif presence:
+                await self.ur.stopped_typing(
+                    self.service,
+                    identifier=candidate,
+                    payload={"presence": presence, "sender": str(sender), "chat": str(chat)},
+                )
+
+    def _extract_pair_qr(self, event):
+        for path in (
+            ("qr",),
+            ("qr_code",),
+            ("code",),
+            ("pair_code",),
+            ("pairCode",),
+            ("url",),
+        ):
+            value = self._pluck(event, *path)
+            if value:
+                return str(value)
+        return ""
+
+    def _to_jid(self, recipient):
+        raw = str(recipient or "").strip()
+        if not raw:
+            return ""
+        if self._build_jid is not None:
+            try:
+                return self._build_jid(raw)
+            except Exception:
+                pass
+        if "@" in raw:
+            return raw
+        digits = re.sub(r"[^0-9]", "", raw)
+        if digits:
+            return f"{digits}@s.whatsapp.net"
+        return raw
+
+    async def _fetch_attachment_payload(self, attachment):
+        blob_key = (attachment or {}).get("blob_key")
+        if blob_key:
+            row = media_bridge.get_blob(blob_key)
+            if row:
+                return row
+
+        content = (attachment or {}).get("content")
+        if isinstance(content, memoryview):
+            content = content.tobytes()
+        if isinstance(content, bytes):
+            return {
+                "content": content,
+                "filename": (attachment or {}).get("filename") or "attachment.bin",
+                "content_type": (attachment or {}).get("content_type")
+                or "application/octet-stream",
+                "size": len(content),
+            }
+
+        url = (attachment or {}).get("url")
+        if url:
+            timeout = aiohttp.ClientTimeout(total=20)
+            async with aiohttp.ClientSession(timeout=timeout) as session:
+                async with session.get(url) as response:
+                    if response.status != 200:
+                        return None
+                    payload = await response.read()
+                    return {
+                        "content": payload,
+                        "filename": (attachment or {}).get("filename")
+                        or url.rstrip("/").split("/")[-1]
+                        or "attachment.bin",
+                        "content_type": (attachment or {}).get("content_type")
+                        or response.headers.get(
+                            "Content-Type", "application/octet-stream"
+                        ),
+                        "size": len(payload),
+                    }
+        return None
+
+    async def send_message_raw(self, recipient, text=None, attachments=None):
+        if not self._client:
+            return False
+        jid = self._to_jid(recipient)
+        if not jid:
+            return False
+
+        sent_any = False
+        for attachment in attachments or []:
+            payload = await self._fetch_attachment_payload(attachment)
+            if not payload:
+                continue
+            mime = str(payload.get("content_type") or "application/octet-stream").lower()
+            data = payload.get("content") or b""
+            filename = payload.get("filename") or "attachment.bin"
+
+            try:
+                if mime.startswith("image/") and hasattr(self._client, "send_image"):
+                    await self._maybe_await(self._client.send_image(jid, data, caption=""))
+                elif mime.startswith("video/") and hasattr(self._client, "send_video"):
+                    await self._maybe_await(self._client.send_video(jid, data, caption=""))
+                elif mime.startswith("audio/") and hasattr(self._client, "send_audio"):
+                    await self._maybe_await(self._client.send_audio(jid, data))
+                elif hasattr(self._client, "send_document"):
+                    await self._maybe_await(
+                        self._client.send_document(
+                            jid,
+                            data,
+                            filename=filename,
+                            mimetype=mime,
+                            caption="",
+                        )
+                    )
+                sent_any = True
+            except Exception as exc:
+                self.log.warning("whatsapp attachment send failed: %s", exc)
+
+        if text:
+            try:
+                await self._maybe_await(self._client.send_message(jid, text))
+                sent_any = True
+            except TypeError:
+                await self._maybe_await(self._client.send_message(jid, message=text))
+                sent_any = True
+            except Exception as exc:
+                self.log.warning("whatsapp text send failed: %s", exc)
+                return False
+
+        return int(time.time() * 1000) if sent_any else False
+
+    async def start_typing(self, identifier):
+        if not self._client:
+            return False
+        jid = self._to_jid(identifier)
+        if not jid:
+            return False
+        for method_name in ("send_chat_presence", "set_chat_presence"):
+            if hasattr(self._client, method_name):
+                method = getattr(self._client, method_name)
+                try:
+                    await self._maybe_await(method(jid, "composing"))
+                    return True
+                except Exception:
+                    continue
+        return False
+
+    async def stop_typing(self, identifier):
+        if not self._client:
+            return False
+        jid = self._to_jid(identifier)
+        if not jid:
+            return False
+        for method_name in ("send_chat_presence", "set_chat_presence"):
+            if hasattr(self._client, method_name):
+                method = getattr(self._client, method_name)
+                try:
+                    await self._maybe_await(method(jid, "paused"))
+                    return True
+                except Exception:
+                    continue
+        return False
+
+    async def fetch_attachment(self, attachment_ref):
+        blob_key = (attachment_ref or {}).get("blob_key")
+        if blob_key:
+            return media_bridge.get_blob(blob_key)
+        return None
+
+    def get_link_qr_png(self, device_name):
+        _ = (device_name or "").strip()
+        if not self._last_qr_payload:
+            return None
+        try:
+            return transport._as_qr_png(self._last_qr_payload)
+        except Exception:
+            return None

core/clients/xmpp.py

@@ -10,12 +10,13 @@ from slixmpp.stanza import Message
 from slixmpp.xmlstream import register_stanza_plugin
 from slixmpp.xmlstream.stanzabase import ET
 
-from core.clients import ClientBase, signalapi
+from core.clients import ClientBase
 from core.messaging import ai, history, replies, utils
 from core.models import (
     ChatSession,
     Manipulation,
     PatternMitigationAutoSettings,
+    PatternMitigationCorrection,
     PatternMitigationGame,
     PatternMitigationPlan,
     PatternMitigationRule,
@@ -91,21 +92,13 @@ class XMPPComponent(ComponentXMPP):
         sender_bare_jid = sender_parts[0]  # Always present: user@domain
         sender_username, sender_domain = sender_bare_jid.split("@", 1)
 
-        sender_resource = (
-            sender_parts[1] if len(sender_parts) > 1 else None
-        )  # Extract resource if present
-
         # Extract recipient JID (should match component JID format)
         recipient_jid = str(msg["to"])
 
         if "@" in recipient_jid:
-            recipient_username, recipient_domain = recipient_jid.split("@", 1)
+            recipient_username = recipient_jid.split("@", 1)[0]
         else:
             recipient_username = recipient_jid
-            recipient_domain = recipient_jid
-
-        # Extract message body
-        body = msg["body"] if msg["body"] else "[No Body]"
         # Parse recipient_name and recipient_service (e.g., "mark|signal")
         if "|" in recipient_username:
             person_name, service = recipient_username.split("|")
@@ -134,9 +127,15 @@ class XMPPComponent(ComponentXMPP):
             return None
 
     def _get_workspace_conversation(self, user, person):
+        primary_identifier = (
+            PersonIdentifier.objects.filter(user=user, person=person)
+            .order_by("service")
+            .first()
+        )
+        platform_type = primary_identifier.service if primary_identifier else "signal"
         conversation, _ = WorkspaceConversation.objects.get_or_create(
             user=user,
-            platform_type="signal",
+            platform_type=platform_type,
             title=f"{person.name} Workspace",
             defaults={"platform_thread_id": str(person.id)},
         )
@@ -186,6 +185,10 @@ class XMPPComponent(ComponentXMPP):
                 ".mitigation rule-del <person>|<title> | "
                 ".mitigation game-add <person>|<title>|<instructions> | "
                 ".mitigation game-del <person>|<title> | "
+                ".mitigation correction-add <person>|<title>|<clarification> | "
+                ".mitigation correction-del <person>|<title> | "
+                ".mitigation fundamentals-set <person>|<item1;item2;...> | "
+                ".mitigation plan-set <person>|<draft|active|archived>|<auto|guided> | "
                 ".mitigation auto <person>|on|off | "
                 ".mitigation auto-status <person>"
             )
@@ -214,7 +217,9 @@ class XMPPComponent(ComponentXMPP):
         if command.startswith(".mitigation show "):
             person_name = command.replace(".mitigation show ", "", 1).strip().title()
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
@@ -231,9 +236,15 @@ class XMPPComponent(ComponentXMPP):
             if len(parts) < 3:
                 sym("Usage: .mitigation rule-add <person>|<title>|<content>")
                 return True
-            person_name, title, content = parts[0].title(), parts[1], "|".join(parts[2:])
+            person_name, title, content = (
+                parts[0].title(),
+                parts[1],
+                "|".join(parts[2:]),
+            )
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
@@ -257,7 +268,9 @@ class XMPPComponent(ComponentXMPP):
                 return True
             person_name, title = parts[0].title(), "|".join(parts[1:])
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
@@ -279,9 +292,15 @@ class XMPPComponent(ComponentXMPP):
             if len(parts) < 3:
                 sym("Usage: .mitigation game-add <person>|<title>|<instructions>")
                 return True
-            person_name, title, content = parts[0].title(), parts[1], "|".join(parts[2:])
+            person_name, title, content = (
+                parts[0].title(),
+                parts[1],
+                "|".join(parts[2:]),
+            )
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
@@ -305,7 +324,9 @@ class XMPPComponent(ComponentXMPP):
                 return True
             person_name, title = parts[0].title(), "|".join(parts[1:])
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
@@ -321,6 +342,128 @@ class XMPPComponent(ComponentXMPP):
             sym("Game deleted." if deleted else "Game not found.")
             return True
 
+        if command.startswith(".mitigation correction-add "):
+            payload = command.replace(".mitigation correction-add ", "", 1)
+            parts = parse_parts(payload)
+            if len(parts) < 3:
+                sym(
+                    "Usage: .mitigation correction-add <person>|<title>|<clarification>"
+                )
+                return True
+            person_name, title, clarification = (
+                parts[0].title(),
+                parts[1],
+                "|".join(parts[2:]),
+            )
+            person = await sync_to_async(
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
+            )()
+            if not person:
+                sym("Unknown person.")
+                return True
+            plan = await sync_to_async(self._get_or_create_plan)(sender_user, person)
+            await sync_to_async(PatternMitigationCorrection.objects.create)(
+                user=sender_user,
+                plan=plan,
+                title=title[:255],
+                clarification=clarification,
+                source_phrase="",
+                perspective="second_person",
+                share_target="both",
+                language_style="adapted",
+                enabled=True,
+            )
+            sym("Correction added.")
+            return True
+
+        if command.startswith(".mitigation correction-del "):
+            payload = command.replace(".mitigation correction-del ", "", 1)
+            parts = parse_parts(payload)
+            if len(parts) < 2:
+                sym("Usage: .mitigation correction-del <person>|<title>")
+                return True
+            person_name, title = parts[0].title(), "|".join(parts[1:])
+            person = await sync_to_async(
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
+            )()
+            if not person:
+                sym("Unknown person.")
+                return True
+            plan = await sync_to_async(self._get_or_create_plan)(sender_user, person)
+            deleted, _ = await sync_to_async(
+                lambda: PatternMitigationCorrection.objects.filter(
+                    user=sender_user,
+                    plan=plan,
+                    title__iexact=title,
+                ).delete()
+            )()
+            sym("Correction deleted." if deleted else "Correction not found.")
+            return True
+
+        if command.startswith(".mitigation fundamentals-set "):
+            payload = command.replace(".mitigation fundamentals-set ", "", 1)
+            parts = parse_parts(payload)
+            if len(parts) < 2:
+                sym("Usage: .mitigation fundamentals-set <person>|<item1;item2;...>")
+                return True
+            person_name, values = parts[0].title(), "|".join(parts[1:])
+            person = await sync_to_async(
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
+            )()
+            if not person:
+                sym("Unknown person.")
+                return True
+            plan = await sync_to_async(self._get_or_create_plan)(sender_user, person)
+            items = [item.strip() for item in values.split(";") if item.strip()]
+            plan.fundamental_items = items
+            await sync_to_async(plan.save)(
+                update_fields=["fundamental_items", "updated_at"]
+            )
+            sym(f"Fundamentals updated ({len(items)}).")
+            return True
+
+        if command.startswith(".mitigation plan-set "):
+            payload = command.replace(".mitigation plan-set ", "", 1)
+            parts = parse_parts(payload)
+            if len(parts) < 3:
+                sym(
+                    "Usage: .mitigation plan-set <person>|<draft|active|archived>|<auto|guided>"
+                )
+                return True
+            person_name, status_value, mode_value = (
+                parts[0].title(),
+                parts[1].lower(),
+                parts[2].lower(),
+            )
+            person = await sync_to_async(
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
+            )()
+            if not person:
+                sym("Unknown person.")
+                return True
+            plan = await sync_to_async(self._get_or_create_plan)(sender_user, person)
+            valid_status = {key for key, _ in PatternMitigationPlan.STATUS_CHOICES}
+            valid_modes = {
+                key for key, _ in PatternMitigationPlan.CREATION_MODE_CHOICES
+            }
+            if status_value in valid_status:
+                plan.status = status_value
+            if mode_value in valid_modes:
+                plan.creation_mode = mode_value
+            await sync_to_async(plan.save)(
+                update_fields=["status", "creation_mode", "updated_at"]
+            )
+            sym(f"Plan updated: status={plan.status}, mode={plan.creation_mode}")
+            return True
+
         if command.startswith(".mitigation auto "):
             payload = command.replace(".mitigation auto ", "", 1)
             parts = parse_parts(payload)
@@ -329,31 +472,47 @@ class XMPPComponent(ComponentXMPP):
                 return True
             person_name, state = parts[0].title(), parts[1].lower()
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
                 return True
-            conversation = await sync_to_async(self._get_workspace_conversation)(sender_user, person)
-            auto_obj, _ = await sync_to_async(PatternMitigationAutoSettings.objects.get_or_create)(
+            conversation = await sync_to_async(self._get_workspace_conversation)(
+                sender_user, person
+            )
+            auto_obj, _ = await sync_to_async(
+                PatternMitigationAutoSettings.objects.get_or_create
+            )(
                 user=sender_user,
                 conversation=conversation,
             )
             auto_obj.enabled = state in {"on", "true", "1", "yes"}
             await sync_to_async(auto_obj.save)(update_fields=["enabled", "updated_at"])
-            sym(f"Automation {'enabled' if auto_obj.enabled else 'disabled'} for {person.name}.")
+            sym(
+                f"Automation {'enabled' if auto_obj.enabled else 'disabled'} for {person.name}."
+            )
             return True
 
         if command.startswith(".mitigation auto-status "):
-            person_name = command.replace(".mitigation auto-status ", "", 1).strip().title()
+            person_name = (
+                command.replace(".mitigation auto-status ", "", 1).strip().title()
+            )
             person = await sync_to_async(
-                lambda: Person.objects.filter(user=sender_user, name__iexact=person_name).first()
+                lambda: Person.objects.filter(
+                    user=sender_user, name__iexact=person_name
+                ).first()
             )()
             if not person:
                 sym("Unknown person.")
                 return True
-            conversation = await sync_to_async(self._get_workspace_conversation)(sender_user, person)
-            auto_obj, _ = await sync_to_async(PatternMitigationAutoSettings.objects.get_or_create)(
+            conversation = await sync_to_async(self._get_workspace_conversation)(
+                sender_user, person
+            )
+            auto_obj, _ = await sync_to_async(
+                PatternMitigationAutoSettings.objects.get_or_create
+            )(
                 user=sender_user,
                 conversation=conversation,
             )
@@ -383,7 +542,7 @@ class XMPPComponent(ComponentXMPP):
         """
         self.log.info(f"Chat state: Active from {msg['from']}.")
 
-        identifier = self.get_identifier(msg)
+        self.get_identifier(msg)
 
     def on_chatstate_composing(self, msg):
         """
@@ -392,6 +551,13 @@ class XMPPComponent(ComponentXMPP):
         self.log.info(f"Chat state: Composing from {msg['from']}.")
 
         identifier = self.get_identifier(msg)
+        if identifier:
+            asyncio.create_task(
+                self.ur.started_typing(
+                    "xmpp",
+                    identifier=identifier,
+                )
+            )
 
     def on_chatstate_paused(self, msg):
         """
@@ -400,6 +566,13 @@ class XMPPComponent(ComponentXMPP):
         self.log.info(f"Chat state: Paused from {msg['from']}.")
 
         identifier = self.get_identifier(msg)
+        if identifier:
+            asyncio.create_task(
+                self.ur.stopped_typing(
+                    "xmpp",
+                    identifier=identifier,
+                )
+            )
 
     def on_chatstate_inactive(self, msg):
         """
@@ -407,7 +580,7 @@ class XMPPComponent(ComponentXMPP):
         """
         self.log.info(f"Chat state: Inactive from {msg['from']}.")
 
-        identifier = self.get_identifier(msg)
+        self.get_identifier(msg)
 
     def on_chatstate_gone(self, msg):
         """
@@ -415,7 +588,7 @@ class XMPPComponent(ComponentXMPP):
         """
         self.log.info(f"Chat state: Gone from {msg['from']}.")
 
-        identifier = self.get_identifier(msg)
+        self.get_identifier(msg)
 
     def on_presence_available(self, pres):
         """
@@ -621,7 +794,9 @@ class XMPPComponent(ComponentXMPP):
         Process incoming XMPP messages.
         """
 
-        sym = lambda x: msg.reply(f"[>] {x}").send()
+        def sym(value):
+            msg.reply(f"[>] {value}").send()
+
         # self.log.info(f"Received message: {msg}")
 
         # Extract sender JID (full format: user@domain/resource)
@@ -710,7 +885,7 @@ class XMPPComponent(ComponentXMPP):
 
                     # Construct contact list response
                     contact_names = [person.name for person in persons]
-                    response_text = f"Contacts: " + ", ".join(contact_names)
+                    response_text = "Contacts: " + ", ".join(contact_names)
                     sym(response_text)
                 elif body == ".help":
                     sym("Commands: .contacts, .whoami, .mitigation help")
@@ -785,12 +960,11 @@ class XMPPComponent(ComponentXMPP):
             )
             self.log.info(f"MANIP11 {manipulations}")
             if not manipulations:
-                tss = await signalapi.send_message_raw(
-                    identifier.identifier,
+                await identifier.send(
                     body,
                     attachments,
                 )
-                self.log.info(f"Message sent unaltered")
+                self.log.info("Message sent unaltered")
                 return
 
             manip = manipulations.first()
@@ -810,12 +984,11 @@ class XMPPComponent(ComponentXMPP):
                 text=result,
                 ts=int(now().timestamp() * 1000),
             )
-            tss = await signalapi.send_message_raw(
-                identifier.identifier,
+            await identifier.send(
                 result,
                 attachments,
             )
-            self.log.info(f"Message sent with modifications")
+            self.log.info("Message sent with modifications")
 
     async def request_upload_slots(self, recipient_jid, attachments):
         """Requests upload slots for multiple attachments concurrently."""
@@ -898,7 +1071,7 @@ class XMPPComponent(ComponentXMPP):
 
         # Step 2: Request upload slots concurrently
         valid_uploads = await self.request_upload_slots(recipient_jid, attachments)
-        self.log.info(f"Got upload slots")
+        self.log.info("Got upload slots")
         if not valid_uploads:
             self.log.warning("No valid upload slots obtained.")
             # return