fix(memory): batch-wrap create_observation and remove observation auto-creation

src/tac/context/memory.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from typing import Any, TypeVar
 
 import httpx
@@ -350,12 +351,16 @@ async def create_observation(
         """
         Create a new observation in Conversation Memory.
 
+        The Observations endpoint is a batch-create API, so the observation is
+        wrapped in an ``observations`` array.
+
         Args:
             profile_id: Profile ID to associate observation with
-            content: Observation content (the summary text or extracted fact)
+            content: Observation content (an extracted fact or note about the profile)
             source: Source system identifier (default: "conversation-intelligence")
             conversation_ids: List of conversation IDs this observation relates to
-            occurred_at: Optional timestamp when observation occurred (ISO 8601 format)
+            occurred_at: Timestamp when observation occurred (ISO 8601 format).
+                Defaults to the current time when omitted or blank.
 
         Returns:
             Dict with created observation details
@@ -366,14 +371,17 @@ 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] = {
+        observation: dict[str, Any] = {
             "content": content,
             "source": source,
         }
         if conversation_ids:
-            payload["conversationIds"] = conversation_ids
-        if occurred_at:
-            payload["occurredAt"] = occurred_at
+            observation["conversationIds"] = conversation_ids
+        if occurred_at is None or not occurred_at.strip():
+            occurred_at = datetime.now(timezone.utc).isoformat()
+        observation["occurredAt"] = occurred_at
+
+        payload: dict[str, Any] = {"observations": [observation]}
 
         try:
             async with self._get_client() as client:

src/tac/core/config.py

@@ -24,10 +24,6 @@ class ConversationIntelligenceConfig(BaseModel):
     configuration_id: str = Field(
         description="Conversation Intelligence Configuration ID",
     )
-    observation_operator_sid: str | None = Field(
-        default=None,
-        description="Operator SID for observation extraction (e.g., LY...)",
-    )
     summary_operator_sid: str | None = Field(
         default=None,
         description="Operator SID for summary extraction (e.g., LY...)",
@@ -37,26 +33,29 @@ class ConversationIntelligenceConfig(BaseModel):
         json_schema_extra={
             "example": {
                 "configuration_id": "your_ci_configuration_id",
-                "observation_operator_sid": "LY00000000000000000000000000000001",
                 "summary_operator_sid": "LY00000000000000000000000000000002",
             }
         },
     )
 
     @classmethod
     def from_env(cls) -> "ConversationIntelligenceConfig | None":
-        """Create ConversationIntelligenceConfig from CONVERSATION_INTELLIGENCE_* env vars."""
+        """Create ConversationIntelligenceConfig from CONVERSATION_INTELLIGENCE_* env vars.
+
+        Blank or whitespace-only operator SIDs are treated as not configured.
+        """
         configuration_id = os.environ.get("CONVERSATION_INTELLIGENCE_CONFIGURATION_ID")
 
         if not configuration_id:
             return None
 
+        summary_operator_sid = os.environ.get("CONVERSATION_INTELLIGENCE_SUMMARY_OPERATOR_SID")
+        if summary_operator_sid is not None and not summary_operator_sid.strip():
+            summary_operator_sid = None
+
         return cls(
             configuration_id=configuration_id,
-            observation_operator_sid=os.environ.get(
-                "CONVERSATION_INTELLIGENCE_OBSERVATION_OPERATOR_SID"
-            ),
-            summary_operator_sid=os.environ.get("CONVERSATION_INTELLIGENCE_SUMMARY_OPERATOR_SID"),
+            summary_operator_sid=summary_operator_sid,
         )
 
 
@@ -372,7 +371,6 @@ def _normalize_voice_public_domain(cls, v: str | None) -> str | None:
                 },
                 "conversation_intelligence_config": {
                     "configuration_id": "GAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
-                    "observation_operator_sid": "LYxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
                     "summary_operator_sid": "LYyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy",
                 },
             }
@@ -427,8 +425,6 @@ def from_env(cls) -> "TACConfig":
 
         - `CONVERSATION_INTELLIGENCE_CONFIGURATION_ID`: CI Service configuration ID
           for webhook filtering
-        - `CONVERSATION_INTELLIGENCE_OBSERVATION_OPERATOR_SID`: Operator SID for
-          observation extraction
         - `CONVERSATION_INTELLIGENCE_SUMMARY_OPERATOR_SID`: Operator SID for summary
           extraction
         """

src/tac/intelligence/operator_result_processor.py

@@ -82,39 +82,6 @@ def _generate_content(operator_result: "OperatorResult") -> str | None:
     return json.dumps(result) if result else None
 
 
-def _parse_observations_content(json_content: str) -> list[str]:
-    """
-    Parse JSON content to extract individual observation contents.
-
-    Expected format: {"observations": [{"content": "..."}, {"content": "..."}]}
-    Fallback: Treat raw content as single observation
-
-    Args:
-        json_content: The JSON content string
-
-    Returns:
-        List of observation content strings
-    """
-    try:
-        payload = json.loads(json_content)
-        if isinstance(payload, dict) and "observations" in payload:
-            observations = payload["observations"]
-            if isinstance(observations, list):
-                contents = []
-                for obs in observations:
-                    if isinstance(obs, dict) and obs.get("content"):
-                        contents.append(str(obs["content"]))
-                if contents:
-                    return contents
-    except (json.JSONDecodeError, TypeError):
-        # If parsing fails or the structure is unexpected, fall back to treating
-        # the entire input as a single summary in the return statement below.
-        pass
-
-    # Fallback: treat entire content as single observation
-    return [json_content] if json_content else []
-
-
 def _parse_summaries_content(json_content: str) -> list[str]:
     """
     Parse JSON content to extract individual summary contents.
@@ -161,11 +128,11 @@ class OperatorResultProcessor:
     """Processor for Conversation Intelligence webhook events.
 
     This processor handles incoming CI webhook payloads, validates them,
-    and creates observations or summaries in Conversation Memory based on the event type.
+    and creates conversation summaries in Conversation Memory based on the event type.
 
     Events are filtered by:
     - Configuration ID matching the provided config
-    - Operator SID matching observation or summary operator SID in config
+    - Operator SID matching the summary operator SID in config
 
     Example usage:
         ```python
@@ -176,16 +143,15 @@ class OperatorResultProcessor:
         conversation_memory_client = MemoryClient(...)
         config = ConversationIntelligenceConfig(
             configuration_id="GA...",
-            observation_operator_sid="LY...",
             summary_operator_sid="LY...",
         )
         processor = OperatorResultProcessor(conversation_memory_client, config)
 
         result = await processor.process_event(webhook_payload)
-        if result.success:
-            print(f"Created {result.created_count} {result.event_type}(s)")
-        elif result.skipped:
+        if result.skipped:
             print(f"Skipped: {result.skip_reason}")
+        elif result.success:
+            print(f"Created {result.created_count} {result.event_type}(s)")
         else:
             print(f"Error: {result.error}")
         ```
@@ -200,7 +166,7 @@ def __init__(
         Initialize the CI event processor.
 
         Args:
-            conversation_memory_client: MemoryClient instance for creating observations/summaries
+            conversation_memory_client: MemoryClient instance for creating summaries
             config: ConversationIntelligenceConfig for filtering events by configuration
                 ID and operator SIDs
         """
@@ -216,8 +182,9 @@ async def process_event(self, payload: dict[str, Any]) -> OperatorProcessingResu
         1. Parses the payload into an OperatorResultEvent (Pydantic validates required fields)
         2. Applies filtering logic based on intelligence configuration ID and operator SIDs
         3. Iterates over operator_results array
-        4. For each operator result: extracts profile IDs, generates content
-        5. Creates observations or summaries in Conversation Memory
+        4. For each operator result: filters by operator SID, then generates
+           content and extracts profile IDs
+        5. Creates conversation summaries in Conversation Memory
 
         Args:
             payload: The raw webhook payload dictionary
@@ -316,54 +283,23 @@ async def _process_operator_result(
         Returns:
             OperatorProcessingResult with status and count
         """
-        # Extract profile IDs from this operator result
-        profile_ids = _extract_profile_ids(operator_result)
-        if not profile_ids:
-            error_msg = f"No profile IDs found in operator result {operator_result.id}"
-            self.logger.error(error_msg)
-            return OperatorProcessingResult(
-                success=False,
-                error=error_msg,
-            )
+        # Filter by operator SID first so unrelated operators are skipped rather
+        # than failing the whole event.
+        operator_id = operator_result.operator.id if operator_result.operator else None
 
-        # Generate content from result
-        content = _generate_content(operator_result)
-        if not content:
-            error_msg = f"Failed to generate content from operator result {operator_result.id}"
-            self.logger.error(error_msg)
+        if self.config.summary_operator_sid is None:
+            # No summary operator configured - nothing to process
+            self.logger.debug("Skipping operator - summary operator SID not configured")
             return OperatorProcessingResult(
-                success=False,
-                error=error_msg,
+                success=True,
+                skipped=True,
+                skip_reason="Summary operator SID not configured",
             )
 
-        # Determine event type by operator SID and process
-        operator_id = operator_result.operator.id if operator_result.operator else None
-
-        # Check if operator matches configured SIDs
-        if (
-            self.config.observation_operator_sid
-            and operator_id == self.config.observation_operator_sid
-        ):
-            # Process as observation (SID match)
-            return await self._process_observation_event(
-                event=event,
-                operator_result=operator_result,
-                content=content,
-                profile_ids=profile_ids,
-            )
-        elif self.config.summary_operator_sid and operator_id == self.config.summary_operator_sid:
-            # Process as summary (SID match)
-            return await self._process_summary_event(
-                event=event,
-                operator_result=operator_result,
-                content=content,
-                profile_ids=profile_ids,
-            )
-        else:
-            # SIDs are configured but don't match - skip
+        if operator_id != self.config.summary_operator_sid:
+            # Configured summary SID doesn't match this operator - skip
             self.logger.debug(
                 f"Skipping operator - SID {operator_id} doesn't match "
-                f"observation ({self.config.observation_operator_sid}) or "
                 f"summary ({self.config.summary_operator_sid})"
             )
             return OperatorProcessingResult(
@@ -372,71 +308,31 @@ async def _process_operator_result(
                 skip_reason="Operator SID mismatch",
             )
 
-    async def _process_observation_event(
-        self,
-        event: OperatorResultEvent,
-        operator_result: "OperatorResult",
-        content: str,
-        profile_ids: list[str],
-    ) -> OperatorProcessingResult:
-        """
-        Process an observation event by creating observations in Conversation Memory.
-
-        Args:
-            event: The parent webhook event (for conversation_id)
-            operator_result: The individual operator result
-            content: The generated content string
-            profile_ids: List of profile IDs to create observations for
-
-        Returns:
-            OperatorProcessingResult with status and count
-        """
-        # Parse observations from content
-        observation_contents = _parse_observations_content(content)
-
-        if not observation_contents:
-            self.logger.info(f"No observations to create from operator result {operator_result.id}")
+        # Generate content from result
+        content = _generate_content(operator_result)
+        if not content:
+            self.logger.info(f"Skipping operator result {operator_result.id} with empty content")
             return OperatorProcessingResult(
                 success=True,
-                event_type="observation",
                 skipped=True,
-                skip_reason="No observation content found",
+                skip_reason="Operator result has empty content",
             )
 
-        created_count = 0
-        errors: list[str] = []
-
-        # Create observations for each profile
-        for profile_id in profile_ids:
-            for obs_content in observation_contents:
-                try:
-                    await self.conversation_memory_client.create_observation(
-                        profile_id=profile_id,
-                        content=obs_content,
-                        source="conversation-intelligence",
-                        conversation_ids=[event.conversation_id],
-                        occurred_at=operator_result.date_created,
-                    )
-                    created_count += 1
-                except Exception as e:
-                    error_msg = f"Failed to create observation for profile {profile_id}: {e}"
-                    self.logger.error(error_msg)
-                    errors.append(error_msg)
-
-        if created_count == 0 and errors:
+        # Extract profile IDs from this operator result
+        profile_ids = _extract_profile_ids(operator_result)
+        if not profile_ids:
+            self.logger.info(f"No profile IDs found in operator result {operator_result.id}")
             return OperatorProcessingResult(
-                success=False,
-                event_type="observation",
-                error="; ".join(errors),
+                success=True,
+                skipped=True,
+                skip_reason="No profile IDs found in operator result execution details",
             )
 
-        self.logger.info(
-            f"Created {created_count} observation(s) from operator result {operator_result.id}"
-        )
-        return OperatorProcessingResult(
-            success=True,
-            event_type="observation",
-            created_count=created_count,
+        return await self._process_summary_event(
+            event=event,
+            operator_result=operator_result,
+            content=content,
+            profile_ids=profile_ids,
         )
 
     async def _process_summary_event(