rex/assistant.py
Rama master · réplica de lectura del repositorio original.
"""Async assistant orchestration."""
from __future__ import annotations
import asyncio
import logging
import re
import threading
import time
from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, cast
from .actions.dispatcher import _UNDO_PATTERN, ActionResult
from .assistant_errors import ConfigurationError, IdentityRequiredError
from .calendar_service import get_calendar_service
from .config import Settings, settings
from .context.revisions import ContextCacheRequest
from .followup_engine import FollowupEngine
from .ha_bridge import HABridge
from .history_store import HistoryStore
from .identity import validate_user_id
from .latency import LatencyTrace
from .llm_client import LanguageModel
from .memory import trim_history
from .model_router import ModelRouteDecision, ModelRouter
from .plugins import PluginSpec
from .provider_reliability import classify_provider_failure
from .runtime.events import EventKind, EventObserver, TurnEvent, TurnEventStream
from .runtime.invocation import current_turn_invocation
from .runtime.turn import (
AuthorizationSnapshotRef,
ResponseMode,
TurnContext,
TurnScope,
)
from .runtime.turn_engine import TurnEngine
from .runtime_paths import household_data_path
logger = logging.getLogger(__name__)
# Deterministic fail-closed error for private operations without an identity
# (issue #303). Must never include paths, credentials, or other user IDs.
_IDENTITY_REQUIRED_MESSAGE = (
"No user identity is bound for this operation. "
"Construct Assistant(user_id=...) or pass active_user_id with the request."
)
__all__ = [
"Assistant",
"ConversationTurn",
"FollowupEngine",
"_UNDO_PATTERN",
"get_calendar_service",
]
_UNVERIFIED_ACTION_CLAIM_PATTERNS = (
re.compile(
r"\bi(?:'ve| have| just)?\s+"
r"(?:added|put|saved|sent|scheduled|set|changed|updated|deleted|removed)\b",
re.IGNORECASE,
),
re.compile(
r"\b(?:done|okay|sure)[,.\s]+(?:i(?:'ve| have| just)\s+)?"
r"(?:added|put|saved|sent|scheduled|set|changed|updated|deleted|removed)\b",
re.IGNORECASE,
),
re.compile(
r"\b(?:it|that|this)\s+(?:has\s+been|was)\s+"
r"(?:added|put|saved|sent|scheduled|set|changed|updated|deleted|removed)\b",
re.IGNORECASE,
),
re.compile(
r"\b(?:i(?:'ve| have| just)|done|okay|sure)[^.!?\n]{0,80}"
r"\bcreated\s+(?:an?\s+)?(?:event|task|reminder|calendar|file|note)\b",
re.IGNORECASE,
),
)
_TOOL_REQUEST_PREFIX = "TOOL_REQUEST:"
_ACTION_CLAIM_STREAM_PREFIXES = (
"i",
"i'",
"i have",
"i've",
"i just",
"i added",
"done",
"okay",
"sure",
"it",
"that",
"this",
)
@dataclass
class ConversationTurn:
speaker: str
text: str
class Assistant:
"""Coordinate wake-word, transcription, language model, and plugins."""
def __init__(
self,
*,
history_limit: int | None = None,
plugins: Iterable[PluginSpec] | None = None,
settings_obj: Settings | None = None,
transcripts_dir: str | Path | None = None,
user_id: str | None = None,
) -> None:
self._settings = settings_obj or settings
self._llm = LanguageModel(config=self._settings)
# Identity binding (issue #303): an omitted user_id leaves the
# instance explicitly unbound. A missing identity never becomes
# "default" and never inherits settings.user_id — private requests
# on an unbound instance must supply a validated active_user_id.
# Explicit user_id="default" remains a valid, deliberate selection
# of the profile named "default".
self._user_id: str | None = validate_user_id(user_id) if user_id is not None else None
# Per-user in-memory history windows keyed by user id (issue #303).
# ``self._history`` is a property that resolves to the current
# ``self._user_id``'s list, so identified-speaker requests never see
# another user's turns.
self._histories: dict[str, list[ConversationTurn]] = {}
self._history_limit = history_limit or self._settings.max_memory_items
self._plugins = list(plugins or [])
self._transcripts_dir = Path(transcripts_dir or self._settings.transcripts_dir)
# Conversation history persistence. Creating the store handle is a
# neutral operation; per-user rows are only read once an identity is
# bound.
self._history_store: HistoryStore | None = None
self._prune_timer: threading.Timer | None = None
if getattr(self._settings, "persist_history", True):
try:
db_path = getattr(self._settings, "history_db_path", None)
if db_path is None:
db_path = household_data_path("history.db")
self._history_store = HistoryStore(db_path=db_path)
if self._user_id is not None:
# Preload the last 50 turns into in-memory history
stored = self._history_store.load_history(self._user_id, limit=50)
self._history = [
ConversationTurn(speaker=row["role"], text=row["content"]) for row in stored
]
# Run an initial prune at startup and schedule daily repeats
self._schedule_daily_prune()
except Exception as exc:
logger.warning("Failed to initialize HistoryStore: %s", exc)
self._history_store = None
# Follow-up engine for natural conversation cues (US-018: init extracted
# to followup_engine). Follow-up sessions are user-private state, so an
# unbound assistant defers initialization until a request arrives with a
# validated identity (see _ensure_followup_session).
self._followup_lock = asyncio.Lock()
self._followup_engine: Any = None
self._pending_followups: dict[str, str | None] = {}
self._followup_sessions: set[str] = set()
self._followup_bootstrap_pending = self._user_id is None
if self._user_id is not None:
from .followup_engine import init_followup_engine as _init_fe
self._followup_engine, _pending = _init_fe(self._settings, self._user_id)
self._pending_followups[self._user_id] = _pending
self._followup_sessions.add(self._user_id)
# Model router for task-category-based model selection.
# Pass routing config so the router can probe Ollama availability at init.
_routing_cfg = getattr(self._settings, "model_routing", None)
_ollama_url: str = getattr(
self._settings, "ollama_base_url", ModelRouter._DEFAULT_OLLAMA_URL
)
self._router = ModelRouter(
ollama_base_url=_ollama_url,
routing_config=_routing_cfg,
)
# Canonical turn runtime. Surfaces remain adapters; generate_reply owns
# the shared non-streaming orchestration path from US-095 onward.
self._turn_engine = TurnEngine()
self._turn_event_observer: EventObserver | None = None
# Route all tool calls through OpenClaw ToolBridge (US-P7-008)
from .openclaw.tool_bridge import ToolBridge
self._tool_router_fn = ToolBridge().route_if_tool_request
# Skill trainer, registry, and router (US-SK-003, US-SK-004)
from .skills.registry import SkillRegistry
from .skills.router import SkillRouter
from .skills.trainer import SkillTrainer
_skills_path = getattr(self._settings, "skills_path", None)
self._skill_registry = SkillRegistry(skills_path=_skills_path)
self._skill_trainer = SkillTrainer()
self._skill_router = SkillRouter(self._skill_registry)
# Auto tool dispatcher (US-TD-002). Dynamic OpenClaw discovery updates
# the same canonical capability registry before the dispatcher is built.
from .openclaw.capability_sync import initialize_openclaw_capability_sync
from .tools.dispatcher import ToolDispatcher
from .tools.registry import get_default_registry
tool_registry = get_default_registry()
initialize_openclaw_capability_sync(
self._settings,
registry=tool_registry.capability_registry,
tool_registry=tool_registry,
)
self._tool_dispatcher = ToolDispatcher(tool_registry, config=self._settings)
# First-class timers and alarms (US-120) use one process-wide,
# deadline-driven runtime. Starting it with Assistant startup restores
# persisted deadlines and reconciles overdue items after a restart.
from .timekeeping.runtime import ensure_timekeeping_runtime
self._timekeeping_runtime = ensure_timekeeping_runtime()
# Shopping list voice handler (US-SL-002)
from .shopping_list import ShoppingList
from .shopping_list_handler import ShoppingListHandler
_sl_path = getattr(self._settings, "shopping_list_path", None)
_shopping_list = ShoppingList(path=_sl_path) if _sl_path else ShoppingList()
self._shopping_list_handler: ShoppingListHandler | None = ShoppingListHandler(
_shopping_list
)
# Music Assistant provider client remains available only behind the
# canonical media adapter. Conversational routing never calls it directly.
from .integrations.music_assistant import MusicAssistantClient
_ma_url = getattr(self._settings, "music_assistant_url", None)
_ma_token = getattr(self._settings, "music_assistant_token", None)
_ma_client = MusicAssistantClient(base_url=_ma_url, token=_ma_token)
self._music_assistant_client = _ma_client
self._music_handler = None
# Device state query handler (US-028)
from .ha.state_handler import DeviceStateHandler
_ha_url = getattr(self._settings, "ha_base_url", None)
_ha_token = getattr(self._settings, "ha_token", None)
self._device_state_handler: DeviceStateHandler | None = DeviceStateHandler(
base_url=_ha_url,
token=_ha_token,
)
# Response cache for repeated factual queries (US-LAT-004)
from .response_cache import ResponseCache
_cache_ttl = float(getattr(self._settings, "response_cache_ttl", 300.0))
self._response_cache: ResponseCache | None = (
ResponseCache(ttl=_cache_ttl) if _cache_ttl > 0 else None
)
# Only create HABridge if HA is configured
self._ha_bridge: HABridge | None = None
if self._settings.ha_base_url and self._settings.ha_token:
try:
self._ha_bridge = HABridge()
logger.info("Home Assistant bridge initialized")
except Exception as exc:
logger.warning("Failed to initialize Home Assistant bridge: %s", exc)
self._ha_bridge = None
# Canonical media runtime (US-121). This initial snapshot is deliberately
# conservative; Task 6 owns dynamic refresh and group-management surfaces.
self._configure_media_service(music_assistant_client=_ma_client)
# Tool result post-processing handler (US-013)
from .tools.result_handler import ToolResultHandler
self._result_handler = ToolResultHandler(
tool_router_fn=self._tool_router_fn,
ha_bridge=self._ha_bridge,
)
# Context builder: assembles system prompt, messages, and user facts (US-014)
# while US-123 source policy supplies content-free cache invalidation.
from .context.active import get_active_context_store
from .context.builder import ContextBuilder
from .context.source_policy import ContextSourcePolicyStore
self._context_source_policy = ContextSourcePolicyStore()
self._active_context_store = get_active_context_store()
self._context_builder = ContextBuilder(
settings=self._settings,
history=[],
user_id=self._user_id,
followup_engine=self._followup_engine,
# Live per-user view: resolves to the request user's history at
# build time instead of a construction-time snapshot (#303). The
# optional argument keeps request identities out of shared state.
history_provider=lambda user_id=None: self._history_for(
user_id if user_id is not None else self._require_user_id()
),
source_policy_store=self._context_source_policy,
active_context_store=self._active_context_store,
)
# Intent router: handles direct-reply shortcuts without LLM (US-015)
from .intent.router import IntentRouter
self._intent_router = IntentRouter(tool_context_fn=self._build_tool_context)
# Proactive suggestion engine (US-036)
from .suggestions.engine import SuggestionEngine
from .suggestions.pattern_detector import PatternEntry
_dismissed_path = getattr(self._settings, "dismissed_suggestions_path", None)
_automations_path = getattr(self._settings, "automations_path", None)
self._suggestion_engine: SuggestionEngine | None = SuggestionEngine(
dismissed_path=_dismissed_path,
automations_path=_automations_path,
)
# In-memory command log for pattern detection (wall-clock timestamps),
# keyed by user_id so one user's commands never seed another user's
# suggestions (issue #303).
self._pattern_entries: dict[str, list[PatternEntry]] = {}
# Response builder: cache, TTS cleaning, suggestions, followups (US-017)
from .response.builder import ResponseBuilder
self._response_builder = ResponseBuilder(
settings=self._settings,
response_cache=self._response_cache,
suggestion_engine=self._suggestion_engine,
followup_engine=self._followup_engine,
)
# Action dispatcher: skill invocation, HA routing, tool dispatch, LLM, post-process (US-016)
from .actions.dispatcher import ActionDispatcher
self._action_dispatcher = ActionDispatcher(
context_builder=self._context_builder,
llm=self._llm,
result_handler=self._result_handler,
ha_bridge=self._ha_bridge,
tool_dispatcher=self._tool_dispatcher,
skill_trainer=self._skill_trainer,
skill_registry=self._skill_registry,
skill_router=self._skill_router,
shopping_list_handler=self._shopping_list_handler,
music_handler=None,
device_state_handler=self._device_state_handler,
suggestion_engine=self._suggestion_engine,
pattern_entries=self._pattern_entries,
active_context_store=self._active_context_store,
build_tool_context_fn=self._build_tool_context,
model_call_fn_builder=self._build_tool_model_call,
run_plugins_fn=self._run_plugins,
)
def _configure_media_service(self, *, music_assistant_client: Any | None = None) -> None:
"""Configure canonical media tools with command-time discovery refresh."""
from .audio.speaker_discovery import get_speaker_discovery
from .identity import list_known_users
from .media.adapters import (
HomeAssistantMediaAdapter,
MusicAssistantAdapter,
SmartSpeakerAdapter,
)
from .media.groups import SpeakerGroupStore
from .media.models import AudioTarget, TargetKind, TargetProviderAdapter
from .media.registry import AudioTargetRegistry
from .media.service import MediaService
from .media.tools import set_media_service
from .permissions import get_permissions
adapters: dict[str, TargetProviderAdapter] = {}
class CachedSpeakerDiscovery:
def discover_now(self):
return get_speaker_discovery().get_cached_speakers()
smart_adapter = SmartSpeakerAdapter(CachedSpeakerDiscovery())
adapters[smart_adapter.provider] = smart_adapter
ha_bridge = getattr(self, "_ha_bridge", None)
if ha_bridge is not None:
ha_adapter = HomeAssistantMediaAdapter(ha_bridge)
adapters[ha_adapter.provider] = ha_adapter
if music_assistant_client is not None:
ma_adapter = MusicAssistantAdapter(music_assistant_client)
adapters[ma_adapter.provider] = ma_adapter
def normalize(value: str) -> str:
return " ".join(value.casefold().replace("_", " ").split())
def build_registry() -> AudioTargetRegistry:
targets: list[AudioTarget] = []
for adapter in adapters.values():
try:
targets.extend(adapter.discover_targets())
except Exception as exc:
logger.warning("Failed to refresh %s media targets: %s", adapter.provider, exc)
base_by_id = {target.id: target for target in targets}
group_store = SpeakerGroupStore(
target_exists=base_by_id.__contains__,
target_capabilities=lambda target_id: base_by_id[target_id].capabilities,
)
try:
groups = group_store.list()
except (KeyError, ValueError) as exc:
logger.warning("Failed to load canonical speaker groups: %s", exc)
groups = ()
for group in groups:
members = tuple(base_by_id[member_id] for member_id in group.member_ids)
online = bool(members) and all(member.online for member in members)
targets.append(
AudioTarget(
id=group.id,
native_id=group.id,
provider="group",
kind=TargetKind.GROUP,
display_name=group.name,
aliases=(),
room=None,
capabilities=frozenset(),
online=online,
health="configured" if online else "member_unavailable",
)
)
known_users: set[str] = set()
bound_user = getattr(self, "_user_id", None)
if isinstance(bound_user, str):
known_users.add(bound_user)
try:
known_users.update(
str(user["id"])
for user in list_known_users()
if isinstance(user, dict) and isinstance(user.get("id"), str)
)
except Exception as exc:
logger.warning("Failed to enumerate users for media authorization: %s", exc)
base_ids = frozenset(base_by_id)
group_members = {group.id: frozenset(group.member_ids) for group in groups}
authorized: dict[str, frozenset[str]] = {}
for user_id in known_users:
try:
permissions = set(get_permissions(user_id))
except Exception as exc:
logger.warning("Failed to resolve media permissions for %s: %s", user_id, exc)
permissions = set()
allowed_base = (
base_ids if permissions.intersection({"ha_control", "admin"}) else frozenset()
)
allowed_groups = {
group_id
for group_id, member_ids in group_members.items()
if member_ids and member_ids.issubset(allowed_base)
}
authorized[user_id] = frozenset((*allowed_base, *allowed_groups))
origin_targets: dict[str, str] = {}
room_map = getattr(self._settings, "device_room_map", {}) or {}
if isinstance(room_map, dict):
for device_id, room_name in room_map.items():
if not isinstance(device_id, str) or not device_id.strip():
continue
if not isinstance(room_name, str) or not room_name.strip():
continue
wanted = normalize(room_name)
matches = []
for target in base_by_id.values():
labels = {
normalize(target.display_name),
normalize(target.native_id.rsplit(".", 1)[-1]),
}
if target.room:
labels.add(normalize(target.room))
if wanted in labels:
matches.append(target)
if len(matches) == 1:
origin_targets[device_id] = matches[0].id
registry = AudioTargetRegistry(
targets,
authorized_target_ids=authorized,
origin_device_targets=origin_targets,
)
self._media_registry = registry
self._speaker_group_store = group_store
return registry
registry = build_registry()
service = MediaService(
registry=registry,
adapters=adapters,
registry_refresher=build_registry,
)
self._media_service = service
set_media_service(service)
def _schedule_daily_prune(self) -> None:
"""Prune old history turns and schedule the next prune in 24 hours.
Runs once immediately at startup, then repeats daily via a daemon thread.
Safe to call if ``_history_store`` is None (no-op).
"""
if self._history_store is None or self._user_id is None:
return
retention_days = int(getattr(self._settings, "history_retention_days", 30))
try:
deleted = self._history_store.prune(self._user_id, keep_days=retention_days)
if deleted:
logger.debug(
"Pruned %d old history turns for user %s (retention=%d days)",
deleted,
self._user_id,
retention_days,
)
except Exception as exc:
logger.warning("History prune failed: %s", exc)
# Schedule next prune in 24 hours (daemon so it doesn't block process exit)
timer = threading.Timer(86400, self._schedule_daily_prune)
timer.daemon = True
timer.start()
self._prune_timer = timer
@property
def user_id(self) -> str | None:
"""Get the bound user ID for this assistant session (None when unbound)."""
return self._user_id
# ------------------------------------------------------------------
# Identity resolution (issue #303)
# ------------------------------------------------------------------
def _require_user_id(self) -> str:
"""Return the bound user ID, failing closed when the instance is unbound."""
user_id: str | None = getattr(self, "_user_id", None)
if user_id is None:
raise IdentityRequiredError(_IDENTITY_REQUIRED_MESSAGE)
return user_id
def _resolve_request_user_id(self, active_user_id: str | None) -> str:
"""Resolve the validated identity a request operates as.
The effective identity comes only from an explicit validated request
identity or the explicit validated constructor identity. A missing
identity fails closed before any private state is touched; it never
becomes ``"default"``.
Raises:
ValueError: If *active_user_id* fails canonical validation.
IdentityRequiredError: If no identity is available at all.
"""
if active_user_id is not None:
return validate_user_id(active_user_id)
return self._require_user_id()
# ------------------------------------------------------------------
# Per-user in-memory history (issue #303)
# ------------------------------------------------------------------
def _load_persisted_history(self, user_id: str) -> list[ConversationTurn]:
"""Load the most recent persisted turns for *user_id* (empty when no store)."""
store = getattr(self, "_history_store", None)
if store is None:
return []
try:
stored = store.load_history(user_id, limit=50)
return [ConversationTurn(speaker=row["role"], text=row["content"]) for row in stored]
except Exception as exc:
logger.warning("Failed to load history for user %s: %s", user_id, exc)
return []
def _history_for(self, user_id: str) -> list[ConversationTurn]:
"""Return the in-memory history window for *user_id*, loading it lazily."""
histories: dict[str, list[ConversationTurn]] = self.__dict__.setdefault("_histories", {})
hist = histories.get(user_id)
if hist is None:
hist = self._load_persisted_history(user_id)
histories[user_id] = hist
return hist
@property
def _history(self) -> list[ConversationTurn]:
"""History window for the bound ``self._user_id`` (fails closed unbound)."""
return self._history_for(self._require_user_id())
@_history.setter
def _history(self, value: list[ConversationTurn]) -> None:
self.__dict__.setdefault("_histories", {})[self._require_user_id()] = list(value)
# ------------------------------------------------------------------
# Per-user pending follow-ups (issue #303)
# ------------------------------------------------------------------
def _pending_followups_map(self) -> dict[str, str | None]:
"""Return the per-user pending-followup map (``__new__``-fixture safe)."""
pending: dict[str, str | None] = self.__dict__.setdefault("_pending_followups", {})
return pending
@property
def _pending_followup(self) -> str | None:
"""Pending follow-up cue for the bound user (None when unbound)."""
user_id = getattr(self, "_user_id", None)
if user_id is None:
return None
return self._pending_followups_map().get(user_id)
@_pending_followup.setter
def _pending_followup(self, value: str | None) -> None:
user_id = getattr(self, "_user_id", None)
if user_id is None:
if value is None:
return
raise IdentityRequiredError(_IDENTITY_REQUIRED_MESSAGE)
self._pending_followups_map()[user_id] = value
def _ensure_followup_session(self, user_id: str) -> None:
"""Initialize or attach *user_id*'s follow-up session exactly once.
For a bound constructor identity this happened in ``__init__``. For
additional request identities (and the first identity on an unbound
instance) the user-private cue state is only touched here, after the
identity has been validated.
"""
sessions: set[str] = self.__dict__.setdefault("_followup_sessions", set())
if user_id in sessions:
return
sessions.add(user_id)
engine = getattr(self, "_followup_engine", None)
if engine is None:
# Only genuinely unbound real constructions defer engine creation;
# test shells built via __new__ never bootstrap implicitly.
if not getattr(self, "_followup_bootstrap_pending", False):
return
from .followup_engine import init_followup_engine as _init_fe
self._followup_bootstrap_pending = False
engine, pending = _init_fe(self._settings, user_id)
self._followup_engine = engine
self._pending_followups_map()[user_id] = pending
return
try:
if hasattr(engine, "start_session"):
engine.start_session(user_id)
pending = None
if hasattr(engine, "get_followup_prompt"):
pending = engine.get_followup_prompt(user_id)
self._pending_followups_map()[user_id] = pending
except Exception as exc:
logger.debug("Follow-up session init failed for user: %s", exc)
@property
def has_pending_followup(self) -> bool:
"""Check if there's a pending follow-up for this session."""
return self._pending_followup is not None
@property
def pending_followup_prompt(self) -> str | None:
"""Get the pending follow-up prompt if any."""
return self._pending_followup
def _get_followup_lock(self) -> asyncio.Lock:
"""Return the follow-up lock, creating it lazily when needed."""
lock = getattr(self, "_followup_lock", None)
if lock is None:
lock = asyncio.Lock()
self._followup_lock = lock
return lock
async def _prepare_model_input(
self,
transcript: str,
*,
voice_mode: bool = False,
active_user_id: str | None = None,
tool_context: str | None = None,
) -> tuple[str, list[dict[str, str]]]:
if not transcript.strip():
raise ValueError("Transcript must not be empty")
# Fail closed before any per-user context or cue state is touched.
user_id = self._resolve_request_user_id(active_user_id)
ctx = self._context_builder.build(
transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
tool_context=tool_context,
)
prompt = ctx.prompt
messages = ctx.messages
async with self._get_followup_lock():
pending = self._pending_followups_map().get(user_id)
if pending:
followup_text = (
f'You may want to ask the user: "{pending}" '
"as a natural conversation starter."
)
followup_hint = f"\n[Note: {followup_text}]"
prompt = prompt + followup_hint
messages.insert(-1, {"role": "system", "content": followup_text})
self._pending_followups_map()[user_id] = None
engine = self._followup_engine
if engine and hasattr(engine, "mark_current_cue_asked"):
try:
engine.mark_current_cue_asked(user_id)
except Exception as exc:
logger.debug("mark_current_cue_asked failed: %s", exc)
return prompt, messages
async def _prepare_prompt(
self,
transcript: str,
*,
voice_mode: bool = False,
active_user_id: str | None = None,
tool_context: str | None = None,
) -> str:
prompt, _messages = await self._prepare_model_input(
transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
tool_context=tool_context,
)
return prompt
def _latency_provider_model(self) -> tuple[str, str]:
"""Return latency-safe provider/model identifiers without deprecated config access."""
settings_obj = getattr(self, "_settings", None)
llm_config = getattr(settings_obj, "llm", None)
provider = getattr(llm_config, "llm_provider", None)
if provider is None:
provider = getattr(settings_obj, "llm_provider", "unknown")
llm_obj = getattr(self, "_llm", None)
model = getattr(llm_obj, "model_name", None) or getattr(llm_config, "model_name", None)
if model is None:
model = getattr(settings_obj, "llm_model", "unknown")
return str(provider or "unknown"), str(model or "unknown")
def _provider_attempt_latency_ms(self, started_ns: int) -> float:
return max(0.0, (time.perf_counter_ns() - started_ns) / 1_000_000)
def _record_provider_success(self, started_ns: int) -> None:
router = getattr(self, "_router", None)
if not isinstance(router, ModelRouter):
return
provider, _model = self._latency_provider_model()
router.record_provider_success(
provider, latency_ms=self._provider_attempt_latency_ms(started_ns)
)
def _record_provider_failure(self, exc: BaseException, started_ns: int) -> None:
router = getattr(self, "_router", None)
if not isinstance(router, ModelRouter):
return
provider, _model = self._latency_provider_model()
router.record_provider_failure(
provider,
classify_provider_failure(exc),
latency_ms=self._provider_attempt_latency_ms(started_ns),
)
def _generate_model_reply(self, prompt: str, messages: list[dict[str, str]]) -> str:
started_ns = time.perf_counter_ns()
try:
try:
reply = self._llm.generate(messages=messages)
except TypeError:
reply = self._llm.generate(prompt)
except Exception as exc:
self._record_provider_failure(exc, started_ns)
raise
self._record_provider_success(started_ns)
return reply
def _track_provider_stream(self, stream: Iterable[str], started_ns: int) -> Iterable[str]:
try:
yield from stream
except Exception as exc:
self._record_provider_failure(exc, started_ns)
raise
self._record_provider_success(started_ns)
def _stream_model_reply(self, prompt: str, messages: list[dict[str, str]]) -> Iterable[str]:
started_ns = time.perf_counter_ns()
try:
try:
stream = self._llm.stream(messages=messages)
except TypeError:
stream = self._llm.stream(prompt)
except Exception as exc:
self._record_provider_failure(exc, started_ns)
raise
return self._track_provider_stream(stream, started_ns)
async def _post_process_completion(
self, transcript: str, completion: str, *, user_id: str | None = None
) -> str:
from rex.mobile_api.action_context import ( # noqa: PLC0415
mobile_action_context_active,
)
plugin_enrichments = (
[] if mobile_action_context_active() else await self._run_plugins(transcript)
)
return await self._result_handler.process(
transcript,
completion,
tool_context=self._build_tool_context(user_id),
model_call_fn=self._build_tool_model_call(transcript, user_id=user_id),
plugin_enrichments=plugin_enrichments,
)
def _record_completion(
self, transcript: str, completion: str, *, user_id: str | None = None
) -> None:
"""Persist and remember a completed turn under its owning user.
*user_id* overrides ``self._user_id`` so early-return paths (intent
shortcuts, cache hits) and post-``_end_request`` recording attribute
the turn to the identified speaker rather than the session default
(issue #303).
"""
uid = user_id if user_id is not None else getattr(self, "_user_id", None)
if uid is None:
raise IdentityRequiredError(_IDENTITY_REQUIRED_MESSAGE)
# Never write history, transcripts, or in-memory windows under an
# unvalidated key (issue #303).
uid = validate_user_id(uid)
now = datetime.now(UTC)
history_store = getattr(self, "_history_store", None)
if history_store is not None:
try:
history_store.save_turn(uid, "user", transcript, now)
history_store.save_turn(uid, "assistant", completion, now)
except Exception as exc:
logger.warning("Failed to persist conversation turn: %s", exc)
hist = self._history_for(uid)
hist.append(ConversationTurn("user", transcript))
hist.append(ConversationTurn("assistant", completion))
self._histories[uid] = [
ConversationTurn(**item) if isinstance(item, dict) else item
for item in trim_history(hist, limit=self._history_limit) # type: ignore[arg-type]
]
self._log_turn(transcript, completion, user_id=uid)
def _looks_like_unverified_action_claim(self, completion: str) -> bool:
return any(pattern.search(completion) for pattern in _UNVERIFIED_ACTION_CLAIM_PATTERNS)
def _stream_tool_prefix_state(self, text: str) -> str:
stripped = text.lstrip().upper()
if not stripped:
return "pending"
prefix = _TOOL_REQUEST_PREFIX
if prefix.startswith(stripped):
return "pending"
if stripped.startswith(prefix):
return "pending"
lowered = text.lstrip().lower()
if self._looks_like_unverified_action_claim(text):
return "pending"
if len(lowered) < 80 and any(
prefix.startswith(lowered) or lowered.startswith(prefix)
for prefix in _ACTION_CLAIM_STREAM_PREFIXES
):
return "pending"
return "text"
async def _stream_home_assistant_completion(
self,
transcript: str,
*,
loop: asyncio.AbstractEventLoop,
latency_trace: LatencyTrace,
) -> str | None:
from rex.mobile_api.action_context import ( # noqa: PLC0415
mobile_action_context_active,
mobile_scope_granted,
run_in_executor_with_mobile_context,
)
if not (
self._ha_bridge
and self._ha_bridge.enabled
and not mobile_action_context_active()
and mobile_scope_granted("home.control")
):
return None
latency_trace.start("tool")
try:
return cast(
str | None,
await run_in_executor_with_mobile_context(
loop, self._ha_bridge.process_transcript, transcript
),
)
finally:
latency_trace.end("tool")
def _validate_model_response(
self,
result: ActionResult,
*,
turn_events: TurnEventStream,
route: str,
) -> str | None:
"""Return a failure reason when a model-backed response is clearly invalid."""
if not result.model_generated:
return None
from rex.response.validation import ( # noqa: PLC0415
MODEL_FAILURE_MESSAGE,
validate_model_output,
)
response = result.response
validation = validate_model_output(response)
if validation.valid:
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{"stage": "output_validation", "status": "passed"},
)
return None
provider, model = self._latency_provider_model()
reason = validation.reason or "invalid_output"
logger.warning(
"model_output_validation_failed provider=%s model=%s route=%s output_length=%d reason=%s",
provider,
model,
route,
len(response),
reason,
)
result.response = MODEL_FAILURE_MESSAGE
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{
"stage": "output_validation",
"status": "model_failure",
"reason": reason,
},
)
return reason
async def _deliver_safe_response(
self,
text: str,
*,
turn_events: TurnEventStream,
response_sink: Callable[[str], Awaitable[None]] | None,
latency_trace: LatencyTrace,
stream_started_ns: int | None,
) -> None:
"""Deliver only post-validation text as ordered sentence chunks."""
if response_sink is None:
return
from rex.runtime.cancellation import ( # noqa: PLC0415
await_with_cancellation,
current_turn_cancellation,
)
from rex.voice.transcripts import _split_into_sentences # noqa: PLC0415
cancellation = current_turn_cancellation()
chunks = _split_into_sentences(text)
if not chunks and text.strip():
chunks = [text.strip()]
for index, chunk in enumerate(chunks):
if cancellation is not None:
cancellation.raise_if_cancelled()
if index == 0 and stream_started_ns is not None:
latency_trace.add_duration_ms(
"first_token",
(time.perf_counter_ns() - stream_started_ns) / 1_000_000,
)
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{"stage": "delta", "index": index, "kind": "sentence"},
)
await await_with_cancellation(response_sink(chunk))
def _compose_event_observer(self, explicit: EventObserver | None) -> EventObserver | None:
"""Combine per-call and legacy observers without shared request state."""
legacy = getattr(self, "_turn_event_observer", None)
if explicit is None:
return legacy
if legacy is None or explicit == legacy:
return explicit
def observe(event) -> None: # noqa: ANN001
legacy(event)
explicit(event)
return observe
async def stream_reply(
self,
transcript: str,
*,
voice_mode: bool = False,
active_user_id: str | None = None,
event_observer: EventObserver | None = None,
) -> AsyncIterator[str]:
"""Stream verified response sentences from the canonical TurnEngine path."""
effective_user_id = self._resolve_request_user_id(active_user_id)
turn_context = self._build_turn_context(effective_user_id, voice_mode=voice_mode)
latency_provider, latency_model = self._latency_provider_model()
latency_trace = LatencyTrace(
channel="voice" if voice_mode else "chat",
provider=latency_provider,
model=latency_model,
settings_id="voice_stream" if voice_mode else "text_stream",
)
stream_started_ns = time.perf_counter_ns()
queue: asyncio.Queue[object] = asyncio.Queue()
sentinel = object()
legacy_observer = getattr(self, "_turn_event_observer", None)
queued_observer = (
event_observer
if event_observer is not None and event_observer != legacy_observer
else None
)
def stream_observer(event: TurnEvent) -> None:
if legacy_observer is not None:
legacy_observer(event)
if queued_observer is not None:
queue.put_nowait(("event", event))
observer = (
stream_observer if legacy_observer is not None or queued_observer is not None else None
)
async def response_sink(chunk: str) -> None:
await queue.put(("chunk", chunk))
async def run_turn() -> None:
try:
await self._get_or_create_turn_engine().execute_async(
turn_context,
lambda turn_events: self._run_reply_turn(
turn_events,
turn_context=turn_context,
transcript=transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
effective_user_id=effective_user_id,
latency_trace=latency_trace,
latency_event="chat_stream_latency",
response_sink=response_sink,
stream_started_ns=stream_started_ns,
),
on_event=observer,
)
except BaseException as exc:
await queue.put(("error", exc))
finally:
await queue.put(sentinel)
task = asyncio.create_task(run_turn())
try:
while True:
item = await queue.get()
if item is sentinel:
break
kind, payload = cast(tuple[str, object], item)
if kind == "event":
if queued_observer is not None:
queued_observer(cast(TurnEvent, payload))
continue
if kind == "error":
raise cast(BaseException, payload)
yield str(payload)
finally:
await task
def _get_or_create_turn_engine(self) -> TurnEngine:
"""Return the canonical turn engine, creating it for legacy test shells."""
engine = getattr(self, "_turn_engine", None)
if engine is None:
engine = TurnEngine()
self._turn_engine = engine
return engine
def _build_turn_context(self, effective_user_id: str, *, voice_mode: bool) -> TurnContext:
"""Create a turn context without widening existing runtime authority."""
invocation = current_turn_invocation()
return TurnContext.create(
user_id=effective_user_id,
scope=TurnScope.USER,
source=invocation.source,
device_id=invocation.device_id,
response_mode=ResponseMode.VOICE if voice_mode else ResponseMode.SCREEN,
authorization=AuthorizationSnapshotRef(
policy_ref="rex-policy:existing-runtime",
permission_ref=f"rex-permissions:validated-user:{effective_user_id}",
),
)
def _prepare_model_route(
self,
transcript: str,
*,
latency_trace: LatencyTrace,
turn_events: TurnEventStream,
) -> tuple[Callable[[], None] | None, ModelRouteDecision | None]:
"""Apply and publish the request-scoped model route."""
route_cleanup, route_decision = self._begin_request(transcript)
if route_decision is not None and route_decision.model:
latency_trace.model = route_decision.model
else:
latency_trace.model = str(getattr(self._llm, "model_name", None) or latency_trace.model)
latency_trace.end("routing")
route_details: dict[str, object] = {
"stage": "model_router",
"model": latency_trace.model,
}
if route_decision is not None:
route_details.update(route_decision.to_metadata())
turn_events.emit(EventKind.ROUTE_PROGRESS, route_details)
return route_cleanup, route_decision
def _context_cache_request(self, turn_context: TurnContext) -> ContextCacheRequest:
"""Build cache identity from the active request-scoped provider/model."""
provider = getattr(self._llm, "provider", None)
if not isinstance(provider, str) or not provider.strip():
provider = str(getattr(self._settings, "llm_provider", None) or "unknown")
active_model = getattr(self._llm, "active_model_name", None)
model_name = active_model() if callable(active_model) else None
if not isinstance(model_name, str) or not model_name.strip():
model_name = getattr(self._llm, "model_name", None)
if not isinstance(model_name, str) or not model_name.strip():
model_name = str(getattr(self._settings, "llm_model", None) or "unknown")
return ContextCacheRequest(
user_id=turn_context.user_id if turn_context.scope is TurnScope.USER else None,
scope=turn_context.scope,
authorization=turn_context.authorization,
model_provider=provider,
model_name=model_name,
)
async def _run_reply_turn(
self,
turn_events: TurnEventStream,
*,
turn_context: TurnContext,
transcript: str,
voice_mode: bool,
active_user_id: str | None,
effective_user_id: str,
latency_trace: LatencyTrace,
latency_event: str,
response_sink: Callable[[str], Awaitable[None]] | None = None,
stream_started_ns: int | None = None,
) -> str:
"""Run the shared verified reply pipeline for text and streaming delivery."""
loop = asyncio.get_running_loop()
check_cancelled = turn_context.cancellation.raise_if_cancelled
check_cancelled()
self._ensure_followup_session(effective_user_id)
latency_trace.start("routing")
intent = self._get_or_create_intent_router().route(
transcript,
settings=self._settings,
suggestion_engine=getattr(self, "_suggestion_engine", None),
user_id=effective_user_id,
)
check_cancelled()
turn_events.emit(
EventKind.ROUTE_PROGRESS,
{
"stage": "intent",
"handled": bool(intent.handled),
"intent_type": intent.intent_type or "unknown",
},
)
if intent.handled and not (intent.intent_type == "greeting" and active_user_id is not None):
latency_trace.end("routing")
latency_trace.start("completion")
completion = cast(str, intent.response)
check_cancelled()
self._record_completion(transcript, completion, user_id=effective_user_id)
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{"stage": "completed", "source": "intent", "history_recorded": True},
)
await self._deliver_safe_response(
completion,
turn_events=turn_events,
response_sink=response_sink,
latency_trace=latency_trace,
stream_started_ns=stream_started_ns,
)
latency_trace.end("completion")
latency_trace.finish()
latency_trace.log_summary(logger, event=latency_event)
return completion
check_cancelled()
cached = None
if self._should_check_response_cache(intent):
cached = self._get_or_create_response_builder().check_cache(
transcript, user_id=effective_user_id
)
check_cancelled()
turn_events.emit(
EventKind.ROUTE_PROGRESS,
{"stage": "cache", "cache_hit": cached is not None},
)
if cached is not None:
latency_trace.end("routing")
latency_trace.start("completion")
completion = cast(str, cached)
check_cancelled()
self._record_completion(transcript, completion, user_id=effective_user_id)
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{"stage": "completed", "source": "cache", "history_recorded": True},
)
await self._deliver_safe_response(
completion,
turn_events=turn_events,
response_sink=response_sink,
latency_trace=latency_trace,
stream_started_ns=stream_started_ns,
)
latency_trace.end("completion")
latency_trace.finish()
latency_trace.log_summary(logger, event=latency_event)
return completion
check_cancelled()
route_cleanup, _route_decision = self._prepare_model_route(
transcript,
latency_trace=latency_trace,
turn_events=turn_events,
)
model_failure_reason: str | None = None
try:
check_cancelled()
cache_request = self._context_cache_request(turn_context)
context = self._get_or_create_context_builder().build(
transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
cache_request=cache_request,
)
check_cancelled()
turn_events.emit(
EventKind.CONTEXT_PROGRESS,
{"stage": "built", "scope": turn_context.scope.value},
)
result = await self._get_or_create_action_dispatcher().dispatch(
intent,
context,
transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
user_id=effective_user_id,
loop=loop,
latency_trace=latency_trace,
turn_events=turn_events,
)
check_cancelled()
model_failure_reason = self._validate_model_response(
result,
turn_events=turn_events,
route=intent.intent_type or "action_dispatch",
)
latency_trace.start("completion")
if model_failure_reason is None:
self._prepare_contextual_suggestion(
effective_user_id,
response_text=str(result.response),
)
final = self._get_or_create_response_builder().build(
result, context, transcript=transcript, user_id=effective_user_id
)
check_cancelled()
completion = final.text
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{"stage": "response_builder", "status": "completed"},
)
else:
completion = str(result.response)
except Exception:
latency_trace.finish()
latency_trace.log_summary(logger, event=latency_event)
raise
finally:
self._end_request(route_cleanup)
check_cancelled()
self._record_completion(transcript, completion, user_id=effective_user_id)
check_cancelled()
turn_events.emit(
EventKind.RESPONSE_PROGRESS,
{"stage": "history", "history_recorded": True},
)
await self._deliver_safe_response(
completion,
turn_events=turn_events,
response_sink=response_sink,
latency_trace=latency_trace,
stream_started_ns=stream_started_ns,
)
if model_failure_reason is not None:
turn_events.finish(
EventKind.FAILED,
{
"failure_kind": "model_failure",
"reason": model_failure_reason,
},
)
latency_trace.end("completion")
latency_trace.finish()
latency_trace.log_summary(logger, event=latency_event)
return completion
@staticmethod
def _should_check_response_cache(intent: Any) -> bool:
"""Current-info replies must be freshly verified instead of served from cache."""
return getattr(intent, "intent_type", None) != "current_info"
async def generate_reply(
self,
transcript: str,
*,
voice_mode: bool = False,
active_user_id: str | None = None,
event_observer: EventObserver | None = None,
) -> str:
"""Generate one verified reply through the canonical TurnEngine pipeline."""
effective_user_id = self._resolve_request_user_id(active_user_id)
turn_context = self._build_turn_context(effective_user_id, voice_mode=voice_mode)
observer = self._compose_event_observer(event_observer)
latency_provider, latency_model = self._latency_provider_model()
latency_trace = LatencyTrace(
channel="voice" if voice_mode else "chat",
provider=latency_provider,
model=latency_model,
settings_id="voice" if voice_mode else "text",
)
return await self._get_or_create_turn_engine().execute_async(
turn_context,
lambda turn_events: self._run_reply_turn(
turn_events,
turn_context=turn_context,
transcript=transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
effective_user_id=effective_user_id,
latency_trace=latency_trace,
latency_event="chat_latency",
),
on_event=observer,
)
def _begin_request(
self, transcript: str
) -> tuple[Callable[[], None] | None, ModelRouteDecision | None]:
"""Apply one request-scoped model route and return its cleanup callback."""
router = getattr(self, "_router", None)
base_model: str | None = getattr(self._llm, "model_name", None)
decision: ModelRouteDecision | None = None
cleanup: Callable[[], None] | None = None
if router is None:
return cleanup, decision
routing_cfg = getattr(self._settings, "model_routing", None)
if isinstance(router, ModelRouter):
decision = router.decide(
transcript,
routing_config=routing_cfg,
current_model=base_model or "",
routing_mode=str(
getattr(self._settings, "llm_routing_mode", "local_preferred")
or "local_preferred"
),
provider=str(
getattr(self._settings, "llm_provider", "")
or getattr(self._llm, "provider", None)
or ""
),
)
if decision.fallback_reason == "local_only_provider_conflict":
raise ConfigurationError(
"llm_routing_mode='local_only' requires a local LLM provider."
)
if decision.fallback_reason == "cloud_provider_unconfigured":
raise ConfigurationError(
"llm_routing_mode='cloud_only' requires an explicitly configured cloud provider."
)
resolved = decision.model
category = decision.category
else:
category = router.classify(transcript)
resolved = router.resolve_model(category, routing_cfg)
if resolved and resolved != base_model:
logger.debug("model_router: classified as %s, using %s", category, resolved)
supports_scoped_model = callable(
getattr(type(self._llm), "set_request_model", None)
) and callable(getattr(type(self._llm), "reset_request_model", None))
if supports_scoped_model:
set_request_model = self._llm.set_request_model
reset_request_model = self._llm.reset_request_model
token = set_request_model(resolved)
def _reset_scoped_model() -> None:
reset_request_model(token)
cleanup = _reset_scoped_model
elif hasattr(self._llm, "model_name"):
self._llm.model_name = resolved
def _restore_legacy_model() -> None:
self._llm.model_name = base_model
cleanup = _restore_legacy_model
else:
logger.debug(
"model_router: classified as %s, using %s",
category,
base_model or "default",
)
return cleanup, decision
@staticmethod
def _end_request(cleanup: Callable[[], None] | None) -> None:
"""Restore request-local model state without mutating other turns."""
if cleanup is not None:
cleanup()
async def _run_plugins(self, transcript: str) -> list[str]:
from rex.runtime.cancellation import ( # noqa: PLC0415
TurnCancelledError,
await_with_cancellation,
)
loop = asyncio.get_running_loop()
results: list[str] = []
for spec in self._plugins:
try:
result = await await_with_cancellation(
loop.run_in_executor(None, spec.plugin.process, transcript)
)
except TurnCancelledError:
raise
except Exception as exc: # pragma: no cover - defensive guard
logger.warning("Plugin %s failed: %s", spec.name, exc)
continue
if result:
results.append(str(result))
return results
# Context building constants and methods have been extracted to
# rex.context.builder.ContextBuilder (US-014). The thin delegates below
# preserve backward compatibility for tests that call these methods directly.
@property
def _VOICE_CONCISE_INSTRUCTION(self) -> str: # noqa: N802
from .context.builder import _VOICE_CONCISE_INSTRUCTION
return _VOICE_CONCISE_INSTRUCTION
def _get_or_create_situational_assembler(self):
"""Return the privacy-bound situational assembler for proactive evaluation."""
assembler = getattr(self, "_situational_assembler", None)
if assembler is not None:
return assembler
from .context.situational import SituationalAssembler
from .knowledge_base import get_knowledge_base
from .user_facts import recall_all
calendar = getattr(self, "_calendar_service", None)
if calendar is None:
try:
calendar = get_calendar_service()
except Exception:
calendar = None
assembler = SituationalAssembler(
source_policy_store=self._get_or_create_context_source_policy(),
calendar_service=calendar,
knowledge_base=get_knowledge_base(),
active_context_store=self._get_or_create_active_context_store(),
memory_reader=recall_all,
current_info_readers=getattr(self, "_situational_current_info_readers", None),
)
self._situational_assembler = assembler
return assembler
def _get_or_create_proactive_evaluator(self):
evaluator = getattr(self, "_proactive_evaluator", None)
if evaluator is None:
from .proactivity.evaluator import ProactiveOpportunityEvaluator
evaluator = ProactiveOpportunityEvaluator()
self._proactive_evaluator = evaluator
return evaluator
def _get_or_create_context_privacy_service(self):
service = getattr(self, "_context_privacy_service", None)
if service is None:
from .context.privacy import get_context_privacy_service
service = get_context_privacy_service()
self._context_privacy_service = service
return service
def _prepare_contextual_suggestion(self, user_id: str, *, response_text: str) -> None:
"""Queue one high-signal contextual suggestion without widening authority."""
engine = getattr(self, "_suggestion_engine", None)
if engine is None or "?" in response_text:
return
try:
privacy = self._get_or_create_context_privacy_service()
preferences = privacy.preference_store.get(user_id)
if not preferences.proactive_assistance:
return
assembler = self._get_or_create_situational_assembler()
evaluator = self._get_or_create_proactive_evaluator()
snapshot = assembler.build(user_id=user_id)
required = evaluator.required_current_info(snapshot)
if required:
snapshot = assembler.enrich_current_info(snapshot, required=required)
candidates = evaluator.evaluate(snapshot)
if candidates:
engine.get_contextual_suggestion(candidates, user_id=user_id)
except Exception as exc:
logger.debug("Contextual suggestion evaluation skipped: %s", type(exc).__name__)
def _get_or_create_response_builder(self):
"""Return self._response_builder, creating one lazily for __new__-based tests."""
rb = getattr(self, "_response_builder", None)
if rb is None:
from .response.builder import ResponseBuilder
rb = ResponseBuilder(
settings=getattr(self, "_settings", None),
response_cache=getattr(self, "_response_cache", None),
suggestion_engine=getattr(self, "_suggestion_engine", None),
followup_engine=getattr(self, "_followup_engine", None),
)
self._response_builder = rb
return rb
def _get_or_create_action_dispatcher(self):
"""Return self._action_dispatcher, creating one lazily for __new__-based tests."""
ad = getattr(self, "_action_dispatcher", None)
if ad is None:
from .actions.dispatcher import ActionDispatcher
ad = ActionDispatcher(
context_builder=self._get_or_create_context_builder(),
llm=getattr(self, "_llm", None),
result_handler=getattr(self, "_result_handler", None),
ha_bridge=getattr(self, "_ha_bridge", None),
tool_dispatcher=getattr(self, "_tool_dispatcher", None),
skill_trainer=getattr(self, "_skill_trainer", None),
skill_registry=getattr(self, "_skill_registry", None),
skill_router=getattr(self, "_skill_router", None),
shopping_list_handler=getattr(self, "_shopping_list_handler", None),
music_handler=None,
device_state_handler=getattr(self, "_device_state_handler", None),
suggestion_engine=getattr(self, "_suggestion_engine", None),
pattern_entries=getattr(self, "_pattern_entries", None),
active_context_store=self._get_or_create_active_context_store(),
build_tool_context_fn=self._build_tool_context,
model_call_fn_builder=self._build_tool_model_call,
run_plugins_fn=self._run_plugins,
)
self._action_dispatcher = ad
return ad
def _get_or_create_intent_router(self):
"""Return self._intent_router, creating one lazily for __new__-based tests."""
ir = getattr(self, "_intent_router", None)
if ir is None:
from .intent.router import IntentRouter
ir = IntentRouter(tool_context_fn=self._build_tool_context)
self._intent_router = ir
return ir
def _get_or_create_active_context_store(self):
"""Return the canonical short-lived active-reference store."""
store = getattr(self, "_active_context_store", None)
if store is None:
from .context.active import get_active_context_store
store = get_active_context_store()
self._active_context_store = store
return store
def _get_or_create_context_source_policy(self):
"""Return the canonical source-policy store, creating it lazily."""
store = getattr(self, "_context_source_policy", None)
if store is None:
from .context.source_policy import ContextSourcePolicyStore
store = ContextSourcePolicyStore()
self._context_source_policy = store
return store
def _get_or_create_context_builder(self):
"""Return self._context_builder, creating one lazily for __new__-based tests."""
cb = getattr(self, "_context_builder", None)
if cb is None:
from .context.builder import ContextBuilder
cb = ContextBuilder(
settings=getattr(self, "_settings", None),
history=[],
user_id=getattr(self, "_user_id", None),
followup_engine=getattr(self, "_followup_engine", None),
history_provider=lambda user_id=None: self._history_for(
user_id if user_id is not None else self._require_user_id()
),
source_policy_store=self._get_or_create_context_source_policy(),
active_context_store=self._get_or_create_active_context_store(),
)
self._context_builder = cb
return cb
def _build_system_context(self) -> str:
"""Delegate to ContextBuilder.build_system_context() (US-014)."""
return self._get_or_create_context_builder().build_system_context() # type: ignore[no-any-return]
def _build_prompt(
self,
transcript: str,
*,
voice_mode: bool = False,
active_user_id: str | None = None,
tool_context: str | None = None,
) -> str:
"""Delegate to ContextBuilder.build() (US-014)."""
return ( # type: ignore[no-any-return]
self._get_or_create_context_builder()
.build(
transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
tool_context=tool_context,
)
.prompt
)
def _build_messages(
self,
transcript: str,
*,
voice_mode: bool = False,
active_user_id: str | None = None,
tool_context: str | None = None,
) -> list[dict[str, str]]:
"""Delegate to ContextBuilder.build() (US-014)."""
return ( # type: ignore[no-any-return]
self._get_or_create_context_builder()
.build(
transcript,
voice_mode=voice_mode,
active_user_id=active_user_id,
tool_context=tool_context,
)
.messages
)
def _get_or_create_location_context_service(self):
"""Return the canonical per-user location service, creating it lazily."""
service = getattr(self, "_location_context_service", None)
if service is None:
from .context.location_policy import LocationContextService, LocationGrantStore
grants = LocationGrantStore(
source_policy_store=self._get_or_create_context_source_policy()
)
service = LocationContextService(grant_store=grants)
self._location_context_service = service
return service
def _build_tool_context(self, user_id: str | None = None) -> dict[str, str]:
"""Return static defaults plus authorized user-specific location context."""
ctx: dict[str, str] = {}
settings = getattr(self, "_settings", None)
location: str | None = getattr(settings, "default_location", None)
timezone: str | None = getattr(settings, "default_timezone", None)
if location:
ctx["location"] = location
if timezone:
ctx["timezone"] = timezone
if user_id is not None:
from .context.location_policy import LocationUsePurpose
service = self._get_or_create_location_context_service()
personal = service.get_for_assistance(user_id, LocationUsePurpose.TOOL_CONTEXT)
if personal is not None:
if personal.city:
ctx["location"] = personal.city
if personal.timezone:
ctx["timezone"] = personal.timezone
return ctx
def _build_tool_model_call(self, transcript: str, *, user_id: str | None = None):
# Tool re-prompts read the same user's history as the rest of the
# request; missing identity fails closed (issue #303).
history = self._history_for(user_id) if user_id is not None else self._history
base_messages = [
{"role": "system", "content": self._context_builder.build_system_context()},
{
"role": "system",
"content": (
"A tool has already been executed for the user's request. "
"Use the tool result to answer the user directly in plain language. "
"Do not emit TOOL_REQUEST. "
"Do not output JSON. "
"Do not request another tool unless the tool result is missing or invalid."
),
},
*[{"role": turn.speaker, "content": turn.text} for turn in history[-4:]],
{"role": "user", "content": transcript},
]
def model_call(tool_message: dict[str, str]) -> str:
messages = base_messages + [
{
"role": "system",
"content": (
"The next message is the tool result. "
"Answer the user naturally using that result. "
"Do not emit TOOL_REQUEST. "
"Do not output JSON."
),
},
tool_message,
]
reply = self._llm.generate(messages=messages)
if isinstance(reply, str) and reply.lstrip().startswith("TOOL_REQUEST:"):
messages.extend(
[
{"role": "assistant", "content": reply},
{
"role": "system",
"content": (
"That was incorrect. "
"You already have the tool result. "
"Answer the user directly in one short natural sentence. "
"Do not emit TOOL_REQUEST. "
"Do not output JSON."
),
},
]
)
reply = self._llm.generate(messages=messages)
return reply
return model_call
def history(self) -> list[ConversationTurn]:
return list(self._history)
def _log_turn(self, transcript: str, reply: str, *, user_id: str | None = None) -> None:
# Resolve and validate the owner before any filesystem path is built:
# transcripts are written under a per-user directory and must never
# land under an invented or unvalidated identity (issue #303).
uid = user_id if user_id is not None else getattr(self, "_user_id", None)
if uid is None:
raise IdentityRequiredError(_IDENTITY_REQUIRED_MESSAGE)
uid = validate_user_id(uid)
try:
self._transcripts_dir.mkdir(parents=True, exist_ok=True)
user_dir = self._transcripts_dir / uid
user_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(UTC)
file_path = user_dir / f"{timestamp:%Y-%m-%d}.txt"
with file_path.open("a", encoding="utf-8") as handle:
handle.write(f"{timestamp:%H:%M:%S} user: {transcript.strip()}\n")
handle.write(f"{timestamp:%H:%M:%S} assistant: {reply.strip()}\n\n")
except Exception: # pragma: no cover - logging must not break replies
logger.exception("Failed to persist transcript entry")
__all__ = ["Assistant", "ConversationTurn", "PluginSpec"]