feat(memory): use List Observations API for memory_mode="once"

src/tac/channels/base.py

@@ -283,9 +283,13 @@ async def _retrieve_memory_if_enabled(
                     )
                     memory_response = session.cached_memory
                 else:
-                    # First retrieval - use empty query and cache result
+                    # First retrieval - use List Observations API to fetch all observations
+                    # once and cache for the call. Avoids per-turn semantic search overhead
+                    # (~750ms) and enables prefix caching since the memory block is static.
+                    # Control how many observations are fetched via TWILIO_MEMORY_OBSERVATIONS_LIMIT
+                    # (default 20, max 500). Example: TWILIO_MEMORY_OBSERVATIONS_LIMIT=500
                     try:
-                        memory_response = await self.tac.retrieve_memory(session, query=None)
+                        memory_response = await self.tac.list_observations(session)
                         session.cached_memory = memory_response
                         self.logger.debug(
                             "Memory retrieved and cached",

src/tac/context/memory.py

@@ -283,6 +283,55 @@ async def create_profile(
             raise ValueError(f"CreateProfile response missing 'id' field: {data!r}")
         return profile_id
 
+    async def list_observations(
+        self,
+        profile_id: str,
+        limit: int = 500,
+    ) -> MemoryRetrievalResponse:
+        """
+        Fetch all observations for a profile via the List Observations API.
+
+        Unlike the Recall API, this does no semantic search — it returns all observations
+        up to `limit`. Ideal for voice calls where you want to fetch the full profile
+        once at call start and cache it for the duration.
+
+        Args:
+            profile_id: Profile ID (TTID format)
+            limit: Max observations to return (default 500)
+
+        Returns:
+            MemoryRetrievalResponse populated with observations.
+            Returns empty MemoryRetrievalResponse() on error.
+        """
+        endpoint = f"/v1/Stores/{self.store_id}/Profiles/{profile_id}/Observations"
+        url = f"{self.base_url}{endpoint}"
+
+        try:
+            async with self._get_client() as client:
+                response = await client.get(url, params={"limit": limit})
+                response.raise_for_status()
+                data = response.json()
+                observations = data.get("observations", [])
+                return MemoryRetrievalResponse(observations=observations)
+
+        except httpx.HTTPError as e:
+            response_text = (
+                getattr(e.response, "text", "No response body")
+                if hasattr(e, "response")
+                else "No response"
+            )
+            self.logger.error(
+                f"Failed to list observations from Conversation Memory: {e}\n"
+                f"URL: {url}\n"
+                f"Profile ID: {profile_id}\n"
+                f"Response: {response_text}"
+            )
+            return MemoryRetrievalResponse()
+
+        except Exception as e:
+            self.logger.error(f"Failed to parse list observations response: {e}")
+            return MemoryRetrievalResponse()
+
     async def create_observation(
         self,
         profile_id: str,
@@ -310,14 +359,13 @@ async def create_observation(
         endpoint = f"/v1/Stores/{self.store_id}/Profiles/{profile_id}/Observations"
         url = f"{self.base_url}{endpoint}"
 
-        payload: dict[str, Any] = {
-            "content": content,
-            "source": source,
-        }
+        observation: dict[str, Any] = {"content": content, "source": source}
         if conversation_ids:
-            payload["conversationIds"] = conversation_ids
+            observation["conversationIds"] = conversation_ids
         if occurred_at:
-            payload["occurredAt"] = occurred_at
+            observation["occurredAt"] = occurred_at
+
+        payload: dict[str, Any] = {"observations": [observation]}
 
         try:
             async with self._get_client() as client:

src/tac/core/tac.py

@@ -229,6 +229,62 @@ async def retrieve_memory(
             )
             return TACMemoryResponse(communications)
 
+    async def list_observations(
+        self,
+        conversation_context: ConversationSession,
+    ) -> TACMemoryResponse:
+        """Fetch all observations for a profile via the List Observations API.
+
+        Unlike retrieve_memory (Recall API), this fetches the full observation set without
+        semantic search. Used by memory_mode="once" — fetch everything once at call start,
+        cache for the duration. Enables prefix caching since the memory block never changes
+        between turns.
+
+        Falls back to retrieve_memory if conversation_memory_client is not configured.
+
+        Args:
+            conversation_context: Session containing profile information.
+
+        Returns:
+            Memory response containing all observations.
+        """
+        if self.conversation_memory_client is None:
+            return await self.retrieve_memory(conversation_context)
+
+        try:
+            if not conversation_context.profile_id:
+                self.logger.debug(
+                    "profile_id not found, attempting to lookup profile using address"
+                )
+                if conversation_context.author_info and conversation_context.author_info.address:
+                    address = conversation_context.author_info.address
+                    id_type = "email" if "@" in address else "phone"
+                    lookup_response: ProfileLookupResponse = (
+                        await self.conversation_memory_client.lookup_profile(
+                            id_type=id_type,
+                            value=address,
+                        )
+                    )
+                    if lookup_response.profiles:
+                        conversation_context.profile_id = lookup_response.profiles[0]
+                    else:
+                        self.logger.debug("No profile found, returning empty memory.")
+                        return TACMemoryResponse([])
+                else:
+                    self.logger.debug("No profile_id or address available, returning empty memory.")
+                    return TACMemoryResponse([])
+
+            cfg = self.config.memory_config
+            memory_response = await self.conversation_memory_client.list_observations(
+                profile_id=conversation_context.profile_id,
+                limit=cfg.observations_limit or 500,
+            )
+            return TACMemoryResponse(memory_response)
+
+        except Exception as e:
+            self.logger.warning(f"list_observations failed: {e}. Returning empty memory.")
+            return TACMemoryResponse([])
+
     async def process_cintel_event(
         self,
         payload: dict[str, Any],