added parts

app/{memory_store → agent_memory_store}/memory_store.py

@@ -1,7 +1,7 @@
 from psycopg_pool import ConnectionPool
 from langgraph.store.postgres import PostgresStore
 from langgraph.checkpoint.postgres import PostgresSaver
-from app.memory_store.embeddings import remote_embeddings
+from app.agent_memory_store.embeddings import remote_embeddings
 
 from app.core.config import settings
 
@@ -14,5 +14,7 @@ pool = ConnectionPool(
     max_size=10,
     kwargs={"autocommit": True} 
 )
+4
+
 
 memory_store = PostgresStore(pool, index={"dims": 384, "embed": remote_embeddings})

app/agents/memory_manager_agent.py

@@ -2,6 +2,7 @@ import types
 from langchain_groq import ChatGroq
 from langmem import create_memory_store_manager
 from app.schemas.memory_agent_schema import EmailMemory
+from app.agent_memory_store import memory_store
 import os
 from app.core.config import settings
 
@@ -45,11 +46,12 @@ def patch_groq_for_extractions(model: ChatGroq):
 model=ChatGroq(model="openai/gpt-oss-20b", temperature=0.2)
 model = patch_groq_for_extractions(model)
 
+namespace = ("emails", "{user_id}", "collection")
 memory_manager_agent = create_memory_store_manager(
     model, 
     schemas=[EmailMemory],
     namespace=namespace,
-    store=p_store,
+    store=memory_store,
     instructions="Extract required info from incoming mail and its reply .",
     enable_inserts=True,
     enable_deletes=True,

app/core/config.py

@@ -12,6 +12,11 @@ class Settings(BaseSettings):
     CLOUDINARY_CLOUD_NAME: str
     CLOUDINARY_API_KEY: str
     CLOUDINARY_API_SECRET: str
+
+    
+    DB_URL_FOR_CHECKPOINTER_STORE: str
+
+    DB_URL_FOR_SQL_AL:str
  
     model_config = SettingsConfigDict(
         env_file=str(BASE_DIR / ".env"),
@@ -19,4 +24,6 @@ class Settings(BaseSettings):
         extra="ignore"
     )
  
-settings = Settings()
+settings = Settings()
+
+

app/database/connection.py

@@ -0,0 +1,14 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, Session
+from app.core.config import DB_URL_FOR_SQL_AL
+engine = create_engine(
+    DB_URL_FOR_SQL_AL, 
+    echo=False, 
+    pool_pre_ping=True, 
+    pool_recycle=1800
+)
+
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+def get_session() -> Session:
+    return SessionLocal()

app/database/models.py

@@ -0,0 +1,49 @@
+from sqlalchemy import (
+    Column, Integer, Text, Boolean, DateTime, ForeignKey, String, func
+
+)
+from sqlalchemy.orm import DeclarativeBase, relationship
+
+class Base(DeclarativeBase):
+    pass
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id         = Column(Integer, primary_key=True, autoincrement=True)
+    email      = Column(String, unique=True, nullable=False)
+    created_at = Column(DateTime, default=func.now())
+
+    received_emails = relationship("ReceivedEmail", back_populates="owner")
+    sent_emails     = relationship("SentEmail",     back_populates="sender")
+
+
+class ReceivedEmail(Base):
+    __tablename__ = "received_emails"
+
+    id           = Column(Integer, primary_key=True, autoincrement=True)
+    thread_id    = Column(String, nullable=False) 
+    owner_id     = Column(Integer, ForeignKey("users.id"), nullable=False)
+    sender_email = Column(Text, nullable=False)
+    subject      = Column(Text)
+    body         = Column(Text)
+    received_at  = Column(DateTime, default=func.now())
+    is_safe      = Column(Boolean, default=False)
+    owner = relationship("User", back_populates="received_emails")
+
+
+class SentEmail(Base):
+    __tablename__ = "sent_emails"
+
+    id              = Column(Integer, primary_key=True, autoincrement=True)
+    thread_id    = Column(String, nullable=False) 
+    sender_id       = Column(Integer, ForeignKey("users.id"), nullable=False)
+    recipient_email = Column(Text,    nullable=False)
+    subject         = Column(Text)
+    body            = Column(Text)
+    sent_at         = Column(DateTime, default=func.now())
+
+    sender = relationship("User", back_populates="sent_emails")
+
+

app/database/utils.py

@@ -0,0 +1,52 @@
+from app.database.models import User, ReceivedEmail, SentEmail
+from sqlalchemy.orm import Session
+
+def get_or_create_user(session, email: str):
+    user = session.query(User).filter_by(email=email).first()
+
+    if not user:
+        user = User(email=email)
+        session.add(user)
+        session.commit()
+        session.refresh(user)
+
+    return user
+
+
+
+
+def save_received_email(session: Session, owner_id: int,thread_id: str, state: dict) -> ReceivedEmail:
+    """
+    Persists the incoming email and the agent's initial triage results.
+    """
+    email = ReceivedEmail(
+        owner_id       = owner_id,
+        thread_id      = thread_id,
+        sender_email   = state.get("sender_email_id"),
+        subject        = state.get("sender_subject"),
+        body           = state.get("sender_email_body"),
+        is_safe        = state.get("is_safe"),
+    )
+    session.add(email)
+    # session.flush()  # Use flush to get the ID without closing the transaction
+    session.commit()   # Only commit here if this is a standalone operation
+    return email
+
+
+def save_sent_email(session: Session, sender_id: int, thread_id: str,state: dict) -> SentEmail:
+    """
+    Persists the final outbound reply sent by the agent.
+    """
+    # Ensure we don't save an empty email if the agent decided not to reply
+        
+    email = SentEmail(
+        sender_id       = sender_id,
+        thread_id    = thread_id,
+        recipient_email = state.get("sender_email_id"),
+        subject         = state.get("reply_subject") or f"Re: {state.get('sender_subject')}",
+        body            = state.get("reply_email_body"),
+    )
+    session.add(email)
+    session.commit()
+    return email
+

app/nodes/store_memory_data_node.py

@@ -4,6 +4,8 @@ from app.agents.memory_manager_agent import memory_manager_agent
 from app.prompts.memory_manager_agent_prompt import memory_agent_template
 from app.state.state import EmailAgentState
 from app.agents.memory_manager_agent import memory_manager_agent 
+from app.database.connection import get_session
+from app.database.utils import save_sent_email, save_received_email
 
 def store_memory_and_data_node(state: EmailAgentState, config: RunnableConfig):
     """

app/nodes/unsafe_email_nodes.py

@@ -1,14 +1,32 @@
 from app.state.state import EmailAgentState
 from langchain_core.runnables.config import RunnableConfig
+from app.database.connection import get_session
+from app.database.utils import save_received_email
 
-def unsafe_emails_node(state: EmailAgentState,config: RunnableConfig) -> dict:
-    print(f"[QUARANTINE] {state['sender_subject']} — reason: {state['safety_reason']}")
+def unsafe_emails_node(state: EmailAgentState, config: RunnableConfig) -> dict:
+    """
+    Handles emails flagged as unsafe by persisting them to the database 
+    and logging the security reason.
+    """
+    print(f"--- [QUARANTINE SIGNAL] {state['sender_subject']} ---")
+    print(f"Reason: {state['safety_reason']}")
 
-    session=get_session()
+    user_id = state['user_id']
+    thread_id = config.get("configurable", {}).get("thread_id")
 
-    user_id=state['user_id']
+    with get_session() as session:
+        try:
+            # 2. Persist the received email even if it's unsafe (for records/logging)
+            save_received_email(session, user_id, thread_id, state)
 
-    thread_id = config.get("configurable", {}).get("thread_id")
+            session.commit()
+            print("--- [QUARANTINE] Data persisted successfully ---")
+            
+        except Exception as e:
+            # 4. Rollback in case of an OperationalError or SSL timeout
+            session.rollback()
+            print(f"--- [QUARANTINE ERROR] Failed to persist unsafe email: {e} ---")
+         
+            raise e
 
-    save_received_email(session, user_id, thread_id,state) 
-    return {}
+    return {}