120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""Routing helpers for Home Assistant user-aware OpenClaw chat sessions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
DEFAULT_CHAT_AGENT_ID = "tetra-home"
|
|
SYSTEM_HA_USER_ID = "system"
|
|
|
|
DEFAULT_HA_USER_AGENT_MAP: dict[str, str] = {}
|
|
|
|
|
|
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,
|
|
user_agent_map: Any = None,
|
|
fallback_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)
|
|
try:
|
|
routing_map = parse_user_agent_map(user_agent_map)
|
|
except ValueError:
|
|
routing_map = dict(DEFAULT_HA_USER_AGENT_MAP)
|
|
fallback = normalize_optional_text(fallback_agent_id) or DEFAULT_CHAT_AGENT_ID
|
|
return routing_map.get(normalized_user_id, fallback)
|
|
|
|
|
|
def parse_user_agent_map(value: Any = None) -> dict[str, str]:
|
|
"""Parse a HA-user-id to OpenClaw-agent-id mapping from options."""
|
|
if value is None:
|
|
return dict(DEFAULT_HA_USER_AGENT_MAP)
|
|
|
|
if isinstance(value, str):
|
|
cleaned = value.strip()
|
|
if not cleaned:
|
|
return {}
|
|
try:
|
|
value = json.loads(cleaned)
|
|
except json.JSONDecodeError as err:
|
|
raise ValueError("User agent map must be a JSON object") from err
|
|
|
|
if not isinstance(value, dict):
|
|
raise ValueError("User agent map must be a JSON object")
|
|
|
|
parsed: dict[str, str] = {}
|
|
for raw_user_id, raw_agent_id in value.items():
|
|
user_id = normalize_optional_text(raw_user_id)
|
|
agent_id = normalize_optional_text(raw_agent_id)
|
|
if user_id and agent_id:
|
|
parsed[user_id] = agent_id
|
|
return parsed
|
|
|
|
|
|
def serialize_user_agent_map(value: Any = None) -> str:
|
|
"""Serialize a user-agent routing map for config entry options."""
|
|
return json.dumps(
|
|
parse_user_agent_map(value),
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
|
|
|
|
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"
|