""" Pinecone client for Self Vision POC. Uses the existing `live-assistant-index-v2` index. Stores conversations as metadata on dummy zero-vectors (no embeddings, no semantic search). All operations are wrapped in try/except — failures are logged, never crash the app. """ import os import time import logging from pinecone import Pinecone logger = logging.getLogger("self_vision.pinecone") INDEX_NAME = "live-assistant-index-v2" # Keys used across POCs KEY_ONBOARDING = "onboarding_conversation" KEY_SELF_VISION = "self_vision_conversation" KEY_ROADMAP = "roadmap_output" # Namespace for roadmap_id → user_id reverse index NAMESPACE_ROADMAP_INDEX = "_roadmap_index" class PineconeClient: """Thin wrapper around Pinecone for key-value conversation storage.""" def __init__(self): api_key = os.getenv("PINECONE_API_KEY", "") if not api_key: logger.warning("PINECONE_API_KEY not set — Pinecone operations will be no-ops") self._index = None self._dimension = 3072 return try: pc = Pinecone(api_key=api_key) self._index = pc.Index(INDEX_NAME) # Discover index dimension dynamically stats = self._index.describe_index_stats() self._dimension = stats.get("dimension", 1536) logger.info( "Pinecone connected: index=%s dimension=%d", INDEX_NAME, self._dimension, ) except Exception as exc: logger.error("Pinecone init failed: %s", exc) self._index = None self._dimension = 3072 # ------------------------------------------------------------------ # READ # ------------------------------------------------------------------ def fetch_conversation(self, user_id: str, key: str) -> dict: """ Fetch a conversation blob from namespace=user_id, id=key. Returns: dict with: - 'text': the conversation text (empty string if missing) - 'status': 'success' | 'empty' | 'error' - 'elapsed_ms': fetch duration in milliseconds """ start = time.time() if self._index is None: elapsed = (time.time() - start) * 1000 logger.warning("FETCH_ERROR: Pinecone not initialised") return {"text": "", "status": "error", "elapsed_ms": elapsed} try: vector_id = f"{user_id}_{key}" result = self._index.fetch(ids=[vector_id], namespace=user_id) elapsed = (time.time() - start) * 1000 vectors = result.get("vectors", {}) if vector_id in vectors: metadata = vectors[vector_id].get("metadata", {}) text = metadata.get("text", "") if text: logger.info( "Fetched key=%s from namespace=%s (%d chars, %.0fms)", vector_id, user_id, len(text), elapsed, ) return {"text": text, "status": "success", "elapsed_ms": elapsed} logger.info( "No record for key=%s in namespace=%s (%.0fms)", vector_id, user_id, elapsed, ) return {"text": "", "status": "empty", "elapsed_ms": elapsed} except Exception as exc: elapsed = (time.time() - start) * 1000 logger.error("FETCH_ERROR: %s (%.0fms)", exc, elapsed) return {"text": "", "status": "error", "elapsed_ms": elapsed} # ------------------------------------------------------------------ # WRITE # ------------------------------------------------------------------ def upsert_conversation(self, user_id: str, key: str, text: str) -> dict: """ Upsert a single conversation blob. namespace = user_id id = key metadata = {"text": } vector = dummy zero-vector (no embeddings used) Returns: dict with: - 'status': 'success' | 'error' - 'elapsed_ms': upsert duration in milliseconds """ start = time.time() if self._index is None: elapsed = (time.time() - start) * 1000 logger.warning("UPSERT_ERROR: Pinecone not initialised") return {"status": "error", "elapsed_ms": elapsed} try: dummy_vector = [0.1] * self._dimension vector_id = f"{user_id}_{key}" self._index.upsert( vectors=[ { "id": vector_id, "values": dummy_vector, "metadata": {"text": text}, } ], namespace=user_id, ) elapsed = (time.time() - start) * 1000 logger.info( "Upserted key=%s in namespace=%s (%d chars, %.0fms)", vector_id, user_id, len(text), elapsed, ) # Auto-store roadmap_id → user_id mapping when upserting a roadmap if key == KEY_ROADMAP and text: try: import json roadmap_data = json.loads(text) roadmap_id = roadmap_data.get("roadmap_id") or roadmap_data.get("ai_roadmap_id", "") if roadmap_id: self.store_roadmap_mapping(roadmap_id, user_id) except Exception as map_exc: logger.warning("Could not store roadmap mapping: %s", map_exc) return {"status": "success", "elapsed_ms": elapsed} except Exception as exc: elapsed = (time.time() - start) * 1000 logger.error("UPSERT_ERROR: %s (%.0fms)", exc, elapsed) return {"status": "error", "elapsed_ms": elapsed} # ------------------------------------------------------------------ # ROADMAP INDEX — reverse lookup: roadmap_id → user_id # ------------------------------------------------------------------ def store_roadmap_mapping(self, roadmap_id: str, user_id: str) -> dict: """ Store a roadmap_id → user_id mapping in the reverse index. namespace = _roadmap_index id = roadmap_id metadata = {"user_id": user_id} """ start = time.time() if self._index is None: elapsed = (time.time() - start) * 1000 return {"status": "error", "elapsed_ms": elapsed} try: dummy_vector = [0.1] * self._dimension self._index.upsert( vectors=[ { "id": roadmap_id, "values": dummy_vector, "metadata": {"user_id": user_id}, } ], namespace=NAMESPACE_ROADMAP_INDEX, ) elapsed = (time.time() - start) * 1000 logger.info( "Stored roadmap mapping: %s → %s (%.0fms)", roadmap_id, user_id, elapsed, ) return {"status": "success", "elapsed_ms": elapsed} except Exception as exc: elapsed = (time.time() - start) * 1000 logger.error("ROADMAP_INDEX_ERROR: %s (%.0fms)", exc, elapsed) return {"status": "error", "elapsed_ms": elapsed} def resolve_user_from_roadmap(self, roadmap_id: str) -> dict: """ Look up user_id from a roadmap_id via the reverse index. Returns: dict with: - 'user_id': the resolved user_id (empty string if not found) - 'status': 'success' | 'empty' | 'error' - 'elapsed_ms': lookup duration in milliseconds """ start = time.time() if self._index is None: elapsed = (time.time() - start) * 1000 return {"user_id": "", "status": "error", "elapsed_ms": elapsed} try: result = self._index.fetch( ids=[roadmap_id], namespace=NAMESPACE_ROADMAP_INDEX, ) elapsed = (time.time() - start) * 1000 vectors = result.get("vectors", {}) if roadmap_id in vectors: metadata = vectors[roadmap_id].get("metadata", {}) user_id = metadata.get("user_id", "") if user_id: logger.info( "Resolved roadmap %s → user %s (%.0fms)", roadmap_id, user_id, elapsed, ) return {"user_id": user_id, "status": "success", "elapsed_ms": elapsed} logger.info( "No mapping for roadmap_id=%s (%.0fms)", roadmap_id, elapsed, ) return {"user_id": "", "status": "empty", "elapsed_ms": elapsed} except Exception as exc: elapsed = (time.time() - start) * 1000 logger.error("ROADMAP_RESOLVE_ERROR: %s (%.0fms)", exc, elapsed) return {"user_id": "", "status": "error", "elapsed_ms": elapsed}