83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""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] = {
|
|
# TODO: Move this user-to-agent map into config entry options.
|
|
"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 an OpenClaw agent-scoped session key for HA user chat sessions."""
|
|
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"agent:{_sanitize_session_part(agent)}"
|
|
f":ha:user:{_sanitize_session_part(user)}"
|
|
f":session:{_sanitize_session_part(raw_session)}"
|
|
)
|
|
|
|
|
|
def build_scoped_session_prefix(
|
|
ha_user_id: str | None,
|
|
agent_id: str | None,
|
|
) -> str:
|
|
"""Build the common OpenClaw session-key prefix for a HA user and agent."""
|
|
user = normalize_ha_user_id(ha_user_id)
|
|
agent = normalize_optional_text(agent_id) or DEFAULT_CHAT_AGENT_ID
|
|
return f"agent:{_sanitize_session_part(agent)}:ha:user:{_sanitize_session_part(user)}: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().lower())
|
|
return cleaned.strip("._-") or "default"
|