Fix chat runtime option matching and add direct service calls (#577)

* Fix chat runtime option matching and add direct service calls

- Store inline question options in _current_options (not just CMS-sourced),
  fixing cascading failures where age/reading answers stored as raw strings
  instead of full option dicts with typed fields like age_number
- Add internal API handler registry for direct service-layer calls, bypassing
  HTTP auth for anonymous chatbot sessions (e.g. /v1/recommend)
- Fix broken import in _find_matching_connection (app.crud.chat → chat_repo)
- Resolve school name server-side from school_wriveted_id during session start
- Add CEL functions for hue profile aggregation (merge, top_keys)
- Expand seed fixtures with book catalog, themes, and flow_file loading

* Update docs for internal service calls, option matching, and seed fixtures

- Fix repository path reference (app/crud/chat_repo.py → app/repositories/)
- Document direct service call architecture for anonymous chatbot sessions
- Document choice option matching behavior (full option objects with typed fields)
- Add top_keys CEL function to aggregation docs
- Fix user scope as read/write (writable by action nodes)
- Document flow_file key for external flow JSON in seed fixtures
- Document server-side school name resolution and theme loading in /start

* Harden seed script: guard missing flow files and fix theme lookup

- _load_flow_config returns None with warning instead of raising
  FileNotFoundError when referenced flow JSON is absent
- _ensure_theme uses (name, school_id) composite filter with .first()
  to avoid MultipleResultsFound and remove unused seed_key branch

* Fix variable resolver tests to expect preserved types

substitute_object returns raw typed values for single {{var}} refs
(int 30, bool True) rather than stringified versions.

* Add unit tests for top_keys CEL function

Covers ranking, n-parameter limiting, non-numeric value filtering,
empty/non-dict inputs, and the full merge-then-rank pipeline used
in Huey hue profile aggregation.

* Fix limit parsing, internal API URL, and stale option clearing

- Safely parse and clamp limit param in recommend handler (1-50)
- Point api service WRIVETED_INTERNAL_API at internal:8888, not itself
- Always write _current_options (even empty) to clear stale options
  from previous questions, preventing incorrect option reuse

Author: hardbyte

app/api/chat.py

@@ -97,6 +97,38 @@ async def start_conversation(
             initial_state=session_data.initial_state,
         )
 
+        # Resolve school name from school_wriveted_id if not already set
+        session_state = conversation_session.state or {}
+        ctx = session_state.get("context", {})
+        school_wriveted_id = ctx.get("school_wriveted_id")
+        if school_wriveted_id and not ctx.get("school_name"):
+            try:
+                from app.repositories.school_repository import school_repository
+
+                school_obj = await school_repository.aget_by_wriveted_id_or_404(
+                    db=session, wriveted_id=school_wriveted_id
+                )
+                ctx["school_name"] = school_obj.name
+                session_state["context"] = ctx
+                await chat_repo.update_session_state(
+                    session,
+                    session_id=conversation_session.id,
+                    state_updates=session_state,
+                    expected_revision=conversation_session.revision,
+                )
+                # Refresh session to pick up updated state
+                refreshed = await chat_repo.get_session_by_id(
+                    session, conversation_session.id
+                )
+                if refreshed:
+                    conversation_session = refreshed
+            except Exception as e:
+                logger.warning(
+                    "Could not resolve school name",
+                    school_wriveted_id=school_wriveted_id,
+                    error=str(e),
+                )
+
         # Get initial node
         initial_node = await chat_runtime.get_initial_node(
             session, session_data.flow_id, conversation_session

app/repositories/cms_repository.py

@@ -374,7 +374,7 @@ async def get_random_content(
 
                     json_filter = json.dumps({key: value})
                     conditions.append(
-                        text("info @> :json_filter::jsonb").bindparams(
+                        text("info @> cast(:json_filter as jsonb)").bindparams(
                             json_filter=json_filter
                         )
                     )

app/services/action_processor.py

@@ -9,7 +9,7 @@
 - aggregate: Aggregate values from a list using various operations
 """
 
-from datetime import datetime
+import datetime
 from typing import Any, Dict
 
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -29,6 +29,21 @@
 logger = get_logger()
 
 
+def _extract_nested(data: Dict[str, Any], key_path: str) -> Any:
+    """Extract a value from a nested dict using dot notation."""
+    keys = key_path.split(".")
+    value = data
+    try:
+        for key in keys:
+            if isinstance(value, dict):
+                value = value.get(key)
+            else:
+                return None
+        return value
+    except (KeyError, TypeError):
+        return None
+
+
 class ActionNodeProcessor(NodeProcessor):
     """Processor for ACTION nodes with support for api_call actions."""
 
@@ -145,6 +160,11 @@ async def _execute_actions_sync(
 
         current_state = session.state or {}
 
+        # Inject db session and user ID for internal API calls
+        context = {**context, "db": db}
+        if session.user_id:
+            context["session_user_id"] = str(session.user_id)
+
         for i, action in enumerate(actions):
             action_type = action.get("type")
             action_id = f"{node_id}_action_{i}"
@@ -194,7 +214,7 @@ async def _execute_actions_sync(
 
         # Update session state if variables were modified
         if variables_updated:
-            current_state.update(variables_updated)
+            self._deep_merge(current_state, variables_updated)
             await chat_repo.update_session_state(
                 db,
                 session_id=session.id,
@@ -214,7 +234,7 @@ async def _execute_actions_sync(
                 "variables_updated": list(variables_updated.keys()),
                 "success": success,
                 "errors": errors,
-                "timestamp": datetime.utcnow().isoformat(),
+                "timestamp": datetime.datetime.utcnow().isoformat(),
                 "processed_async": False,
             },
         )
@@ -233,9 +253,10 @@ async def _handle_set_variable(
         value = action.get("value")
 
         if variable and value is not None:
-            # Substitute variables in value if it's a string
-            if isinstance(value, str):
-                value = self.runtime.substitute_variables(value, state)
+            # Recursively substitute variables in the value.
+            # substitute_object handles strings, lists, and dicts, and preserves
+            # typed values when the entire string is a single {{var}} reference.
+            value = self.runtime.substitute_object(value, state)
 
             self._set_nested_value(updates, variable, value)
             logger.debug(f"Set variable {variable} = {value}")
@@ -285,6 +306,51 @@ async def _handle_api_call(
     ) -> None:
         """Handle api_call action."""
         api_config_data = action.get("config", {})
+        auth_type = api_config_data.get("auth_type", "internal")
+
+        # For internal endpoints, try direct service call (bypasses HTTP + auth)
+        if auth_type == "internal":
+            from app.services.internal_api_handlers import INTERNAL_HANDLERS
+
+            endpoint = api_config_data.get("endpoint", "")
+            db = context.get("db")
+
+            if endpoint in INTERNAL_HANDLERS and db is not None:
+                resolved_body = self.runtime.substitute_object(
+                    api_config_data.get("body", {}), state
+                )
+                resolved_params = self.runtime.substitute_object(
+                    api_config_data.get("query_params", {}), state
+                )
+
+                result_data = await INTERNAL_HANDLERS[endpoint](
+                    db, resolved_body, resolved_params
+                )
+
+                response_mapping = api_config_data.get("response_mapping", {})
+                for response_path, variable_name in response_mapping.items():
+                    value = _extract_nested(result_data, response_path)
+                    if value is not None:
+                        self._set_nested_value(updates, variable_name, value)
+
+                logger.info(
+                    "Internal API call via direct service call",
+                    endpoint=endpoint,
+                    variables_updated=list(response_mapping.values()),
+                )
+                return
+
+        # For authenticated sessions, generate a short-lived JWT
+        if auth_type == "internal" and context.get("session_user_id"):
+            from app.services.security import create_access_token
+
+            token = create_access_token(
+                subject=f"Wriveted:User-Account:{context['session_user_id']}",
+                expires_delta=datetime.timedelta(minutes=5),
+            )
+            api_config_data = {**api_config_data}
+            api_config_data["auth_type"] = "bearer"
+            api_config_data["auth_config"] = {"token": token}
 
         # Create API call configuration
         api_config = ApiCallConfig(**api_config_data)
@@ -297,8 +363,9 @@ async def _handle_api_call(
         result = await api_client.execute_api_call(api_config, state, composite_scopes)
 
         if result.success:
-            # Update variables with API response
-            updates.update(result.variables_updated)
+            # Update variables with API response using nested paths
+            for var_name, var_value in result.variables_updated.items():
+                self._set_nested_value(updates, var_name, var_value)
             logger.info(
                 "API call successful",
                 endpoint=api_config.endpoint,
@@ -307,12 +374,16 @@ async def _handle_api_call(
         else:
             # Store error information
             error_var = api_config_data.get("error_variable", "api_error")
-            updates[error_var] = {
-                "error": result.error_message,
-                "status_code": result.status_code,
-                "timestamp": datetime.utcnow().isoformat(),
-                "fallback_used": result.fallback_used,
-            }
+            self._set_nested_value(
+                updates,
+                error_var,
+                {
+                    "error": result.error_message,
+                    "status_code": result.status_code,
+                    "timestamp": datetime.datetime.utcnow().isoformat(),
+                    "fallback_used": result.fallback_used,
+                },
+            )
             logger.error(
                 "API call failed",
                 endpoint=api_config.endpoint,
@@ -450,6 +521,14 @@ def _build_cel_expression(
         else:
             raise ValueError(f"Unknown aggregate operation: {operation}")
 
+    def _deep_merge(self, base: Dict[str, Any], updates: Dict[str, Any]) -> None:
+        """Recursively merge updates into base dict, preserving existing nested keys."""
+        for key, value in updates.items():
+            if key in base and isinstance(base[key], dict) and isinstance(value, dict):
+                self._deep_merge(base[key], value)
+            else:
+                base[key] = value
+
     def _get_nested_value(self, data: Dict[str, Any], key_path: str) -> Any:
         """Get nested value from dictionary using dot notation."""
         keys = key_path.split(".")

app/services/api_client.py

@@ -5,11 +5,11 @@
 with proper authentication, error handling, and response processing.
 """
 
-import logging
 from typing import Any, Dict, Optional
 
 import httpx
 from pydantic import BaseModel
+from structlog import get_logger
 
 from app.config import get_settings
 from app.services.circuit_breaker import (
@@ -18,7 +18,7 @@
     get_circuit_breaker,
 )
 
-logger = logging.getLogger(__name__)
+logger = get_logger()
 
 
 class ApiCallConfig(BaseModel):
@@ -68,7 +68,11 @@ class InternalApiClient:
 
     def __init__(self):
         self.settings = get_settings()
-        self.base_url = self.settings.WRIVETED_INTERNAL_API
+        self.base_url = (
+            str(self.settings.WRIVETED_INTERNAL_API)
+            if self.settings.WRIVETED_INTERNAL_API
+            else None
+        )
         self.session: Optional[httpx.AsyncClient] = None
 
     async def initialize(self) -> None:

app/services/cel_evaluator.py

@@ -95,6 +95,19 @@ def _cel_collect(items: List[Any]) -> List[Any]:
     return _cel_flatten(items)
 
 
+def _cel_top_keys(d: Dict[str, Any], n: int = 5) -> List[str]:
+    """Return the top N keys from a dict sorted by value descending.
+
+    Useful for converting a hue_profile dict (hue→weight) into a ranked
+    list of hue keys for the recommendation API.
+    """
+    if not isinstance(d, dict):
+        return []
+    numeric_items = [(k, v) for k, v in d.items() if isinstance(v, (int, float))]
+    numeric_items.sort(key=lambda pair: pair[1], reverse=True)
+    return [k for k, _ in numeric_items[:n]]
+
+
 # Registry of custom functions available in CEL expressions
 CUSTOM_CEL_FUNCTIONS: Dict[str, Callable] = {
     "sum": _cel_sum,
@@ -108,6 +121,7 @@ def _cel_collect(items: List[Any]) -> List[Any]:
     "merge_last": _cel_merge_last,
     "flatten": _cel_flatten,
     "collect": _cel_collect,
+    "top_keys": _cel_top_keys,
 }
 
 
@@ -174,6 +188,7 @@ def evaluate_cel_expression(
         - min(temp.quiz_results.map(x, x.time))
         - merge(selections.map(x, x.preferences))
         - flatten(items.map(x, x.tags))
+        - top_keys(user.hue_profile, 5)
     """
     try:
         if include_aggregation_functions: