Route OpenClaw chat by HA user

custom_components/openclaw/routing.py

@@ -0,0 +1,71 @@
+"""Routing helpers for Home Assistant user-aware OpenClaw chat sessions."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+DEFAULT_CHAT_AGENT_ID = "tetra-home"
+SYSTEM_HA_USER_ID = "system"
+
+HA_USER_AGENT_MAP: dict[str, str] = {
+    "bd3f666f43f941e886ef007a794eefe8": "main",
+    "1d2be9a622584cf0abb1c243d12a9959": "family-krzysztof",
+    "eca5887898544556840878ab37a35fbc": "family-ewa",
+    "2f5bf9f09478400082380bf7f26966d9": "family-tosia",
+    "f613d90efe8b4afea63c4a5a7ef7b559": "family-domi",
+}
+
+
+def normalize_optional_text(value: Any) -> str | None:
+    """Return a stripped string, or None for blank/non-string values."""
+    if not isinstance(value, str):
+        return None
+    cleaned = value.strip()
+    return cleaned or None
+
+
+def normalize_ha_user_id(value: Any) -> str:
+    """Return a stable HA user id for routing/history scope."""
+    return normalize_optional_text(value) or SYSTEM_HA_USER_ID
+
+
+def resolve_agent_id(
+    ha_user_id: str | None,
+    explicit_agent_id: str | None = None,
+) -> str:
+    """Resolve the OpenClaw agent for a Home Assistant user."""
+    if explicit_agent_id:
+        return explicit_agent_id
+    normalized_user_id = normalize_ha_user_id(ha_user_id)
+    return HA_USER_AGENT_MAP.get(normalized_user_id, DEFAULT_CHAT_AGENT_ID)
+
+
+def resolve_model_override(value: Any) -> str | None:
+    """Return a provider model override, excluding OpenClaw pseudo-model ids."""
+    model = normalize_optional_text(value)
+    if model and model.startswith("openclaw/"):
+        return None
+    return model
+
+
+def build_scoped_session_id(
+    raw_session_id: str | None,
+    ha_user_id: str | None,
+    agent_id: str | None,
+) -> str:
+    """Build the backend/gateway session id scoped by user, agent, and raw session."""
+    raw_session = normalize_optional_text(raw_session_id) or "default"
+    user = normalize_ha_user_id(ha_user_id)
+    agent = normalize_optional_text(agent_id) or DEFAULT_CHAT_AGENT_ID
+    return (
+        f"ha_user_{_sanitize_session_part(user)}"
+        f"__agent_{_sanitize_session_part(agent)}"
+        f"__session_{_sanitize_session_part(raw_session)}"
+    )
+
+
+def _sanitize_session_part(value: str) -> str:
+    """Keep scoped session ids header-safe and readable."""
+    cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip())
+    return cleaned.strip("_") or "default"