Saved conversations follow the agent's answer settings in the API

.github/workflows/standalone-product.yml

@@ -54,6 +54,32 @@ jobs:
         run: python -m pip install --disable-pip-version-check fastapi==0.150.0 pydantic==2.14.1
       - name: Verify session continuation stays on its own agent
         run: python -m unittest discover -s tests/backend -p test_session_scope.py -v
+  session-storage:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    services:
+      postgres:
+        image: postgres:16.6-alpine
+        env:
+          POSTGRES_USER: session_storage_admin
+          POSTGRES_PASSWORD: synthetic-ci-only
+          POSTGRES_DB: session_storage
+        ports:
+          - 127.0.0.1:15441:5432
+        options: >-
+          --health-cmd "pg_isready -U session_storage_admin -d session_storage"
+          --health-interval 2s --health-timeout 2s --health-retries 20
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+      - name: Install only the pinned API-test, model and PostgreSQL driver dependencies
+        run: python -m pip install --disable-pip-version-check fastapi==0.150.0 pydantic==2.14.1 sqlalchemy==2.0.36 sqlmodel==0.0.22 asyncpg==0.32.0
+      - name: Session isolation on the product's queries against real PostgreSQL
+        env:
+          SESSION_STORAGE_TEST_URL: postgresql+asyncpg://session_storage_admin:synthetic-ci-only@127.0.0.1:15441/session_storage
+        run: python -m unittest discover -s tests/backend -p test_session_storage.py -v
   evaluation-reliability:
     runs-on: ubuntu-latest
     timeout-minutes: 5

backend/api/v1/routers/chat.py

@@ -350,12 +350,20 @@ async def export_session_csv(
         build_export_s3_key, generate_export_presigned_url,
         check_s3_object_exists, CSV_EXPORT_URL_EXPIRY,
     )
-    chat_log = await get_chat_log_for_export(session_id, chat_id)
+    # This path names no agent, so an agent key is scoped by an `agent_id` query parameter the
+    # caller chooses: only that agent's turns can be found with it.
+    key_agent = auth.get("agent_id") if auth.get("auth_type") == "agent_api_key" else None
+    chat_log = await get_chat_log_for_export(session_id, chat_id, agent_id=key_agent)
     if not chat_log:
         detail = f"Chat turn {chat_id} not found" if chat_id else "No SQL result found for this session"
         raise HTTPException(status_code=404, detail=detail)
     if not chat_log.sql_query:
         raise HTTPException(status_code=404, detail="This chat turn has no SQL result to export")
+    # The CSV is the turn's rows: an agent that hides result rows does not hand them out here
+    # either. Checked against the agent that produced the turn.
+    _, _, show_db_result, _ = await get_agent_with_visibility(chat_log.agent_id)
+    if not show_db_result:
+        raise HTTPException(status_code=403, detail="Result rows are hidden for this agent")
     s3_key = build_export_s3_key(chat_log.agent_id, session_id, chat_log.chat_id)
     if not await check_s3_object_exists(s3_key):
         raise HTTPException(status_code=404, detail="Export file not available. The query may have been made before export was enabled.")

backend/api/v1/routers/sessions.py

@@ -28,6 +28,7 @@
     UpdateSessionRequest,
 )
 from agent.utils.authorization import require_agent_permission
+from api.v1.utils.chat_helper import get_agent_with_visibility
 
 router = APIRouter(prefix="/v1")
 
@@ -144,8 +145,10 @@ async def get_session(
                 detail=f"Session {session_id} not found for agent {agent_id}"
             )
 
-        # Get conversation data
-        conversation_data = await get_session_data(session_id)
+        # Get conversation data, under the agent's answer visibility settings
+        _, show_sql_query, show_db_result, _ = await get_agent_with_visibility(agent_id)
+        conversation_data = await get_session_data(
+            session_id, agent_id=agent_id, show_sql_query=show_sql_query, show_db_result=show_db_result)
 
         return {
             "session_id": session_details['session_id'],

backend/api/v1/utils/chat_helper.py

@@ -17,14 +17,21 @@
 from sqlmodel import select
 
 
-async def get_past_conversation_safe(session_id):
-    """Get past conversation for a session"""
+async def get_past_conversation_safe(session_id, agent_id):
+    """Get past conversation for a session — this agent's turns only.
+
+    A session id can be shared by more than one agent (a chat service may send its own thread
+    id to every agent it calls). Reading by session id alone replayed one agent's questions and
+    answers into another agent's model context.
+    """
     try:
         start_time = time.time()
         messages = []
         async for session in get_async_session():
             res = await session.execute(
-                select(ChatLogs).where(ChatLogs.session_id == session_id).order_by(ChatLogs.created_at)
+                select(ChatLogs).where(
+                    ChatLogs.session_id == session_id, ChatLogs.agent_id == agent_id,
+                ).order_by(ChatLogs.created_at)
             )
             rows = res.scalars().all()
             for chat_log in rows:
@@ -45,24 +52,27 @@ async def get_past_messages_safe(session_id, user_id, agent_id, user_query):
             session_info = await agent_log_new_session(user_id, agent_id, user_query)
             return {"past_messages": [], "session_id": session_info['session_id']}
         else:
-            past_messages = await get_past_conversation_safe(session_id)
+            past_messages = await get_past_conversation_safe(session_id, agent_id)
             return {"past_messages": past_messages, "session_id": session_id}
     except Exception as e:
         logger.error(f"Error in get_past_messages_safe: {str(e)}")
         return {"past_messages": [], "session_id": session_id}
 
 
 async def ensure_session_belongs_to_agent(session_id: Optional[str], agent_id: str) -> None:
-    """Refuse to continue a session that was not created on this agent.
+    """Refuse a session_id that this API created for a different agent.
 
-    History is read by session_id alone (get_past_conversation_safe), so an unchecked id would
-    replay another agent's questions and answers into this agent's model context and append the
-    new turn to that foreign session. A new conversation (no id) is always allowed.
+    Runs before any history read or model call, so continuing another agent's conversation is
+    refused outright rather than only stripped of its history (which
+    `get_past_conversation_safe` does in any case). A session id this API never created (a chat
+    service's own thread id) continues with this agent's turns only. A new conversation (no id)
+    is always allowed.
     """
     if not session_id or session_id in ["", "string"]:
         return
-    from session.db_utils.session_db_utils import get_session_details
-    if not await get_session_details(session_id, agent_id):
+    from session.db_utils.session_db_utils import get_session_agent_ids
+    owners = await get_session_agent_ids(session_id)
+    if owners and agent_id not in owners:
         raise HTTPException(status_code=404, detail=f"Session {session_id} not found for agent {agent_id}")
 
 
@@ -328,41 +338,40 @@ async def get_agent_runtime_db(agent_id: str) -> Tuple[dict, float]:
     return db_response.get("db_details"), elapsed
 
 
-async def get_chat_log_for_export(session_id: str, chat_id: Optional[str]):
-    """Fetch the ChatLogs row for a CSV export request."""
+async def get_chat_log_for_export(session_id: str, chat_id: Optional[str], agent_id: Optional[str] = None):
+    """Fetch the ChatLogs row for a CSV export request; only `agent_id`'s turns when given."""
+    scope = [ChatLogs.agent_id == agent_id] if agent_id else []
     async for db_session in get_async_session():
         if chat_id:
             res = await db_session.execute(
                 select(ChatLogs).where(
                     ChatLogs.chat_id == chat_id,
                     ChatLogs.session_id == session_id,
+                    *scope,
                 ).limit(1)
             )
         else:
             res = await db_session.execute(
                 select(ChatLogs).where(
                     ChatLogs.session_id == session_id,
                     ChatLogs.sql_query.isnot(None),
+                    *scope,
                 ).order_by(ChatLogs.created_at.desc()).limit(1)
             )
         return res.scalar_one_or_none()
 
 
 async def get_session_messages_data(session_id: str, agent_id: str) -> dict:
-    """Verify agent exists then return session conversation data."""
-    async for session in get_async_session():
-        result = await session.execute(
-            select(ChatbotDetails).where(ChatbotDetails.agent_id == agent_id)
-        )
-        agent = result.scalar_one_or_none()
-        if not agent:
-            raise HTTPException(status_code=404, detail=f"Agent {agent_id} not found")
+    """Verify agent exists then return session conversation data, under the agent's answer
+    visibility settings (the same ones live chat applies)."""
+    _, show_sql_query, show_db_result, _ = await get_agent_with_visibility(agent_id)
 
     from session.db_utils.session_db_utils import get_session_data, get_session_details
     session_detail = await get_session_details(session_id, agent_id)
     if not session_detail:
         return {"message": "No session data found"}
-    session_data = await get_session_data(session_id)
+    session_data = await get_session_data(
+        session_id, agent_id=agent_id, show_sql_query=show_sql_query, show_db_result=show_db_result)
     if session_data:
         return {
             "session_id": session_detail['session_id'],

backend/api/v1/utils/feedback_helper.py

@@ -49,7 +49,8 @@ async def process_feedback(agent_id: str, body) -> dict:
         session_id=body.session_id,
         chat_id=body.chat_id,
         feedback=normalized_label,
-        corrected_sql=sql_to_store
+        corrected_sql=sql_to_store,
+        agent_id=agent_id,
     )
 
     if normalized_label == "rejected" and body.corrected_query:

backend/session/db_utils/session_db_utils.py

@@ -59,6 +59,14 @@ async def get_past_conversation(session_id):
 		logger.error(f"Error getting past conversation: {e}")
 		raise
 
+async def get_session_agent_ids(session_id: str) -> set:
+	"""The agents a session created by this API belongs to; empty when it never created one."""
+	async with get_async_session() as session:
+		res = await session.execute(
+			select(ChatSessions.chatbot_id, ChatSessions.agent_id).where(ChatSessions.session_id == session_id)
+		)
+		return {agent for row in res.all() for agent in row if agent}
+
 async def get_session_details(session_id: str, agent_id: str):
 	"""
 	Return dict with session_id, created_time, session_title or None.
@@ -382,13 +390,24 @@ def _chart_for_replay(row) -> Optional[dict]:
 	return None
 
 
-async def get_session_data(session_id: str):
+async def get_session_data(session_id: str, *, agent_id: str, show_sql_query: bool, show_db_result: bool):
+	"""A saved conversation — this agent's turns — under the agent's answer visibility settings.
+
+	All three are required (no default) so every caller decides. Only `agent_id`'s turns are
+	served: a session id can carry another agent's turns too (a chat service may send its own
+	thread id to every agent it calls). Live chat leaves the SQL out when `show_sql_query` is
+	off and the rows out when `show_db_result` is off, and a replay must not hand back what the
+	live answer withheld. Pass the flags from `get_agent_with_visibility`, the same resolution
+	live chat uses.
+	"""
 	try:
 		logger.info(f"Getting session data for session_id: {session_id}")
 		session_data = []
 		async with get_async_session() as session:
 			res = await session.execute(
-				select(ChatLogs).where(ChatLogs.session_id == session_id).order_by(ChatLogs.created_at)
+				select(ChatLogs).where(
+					ChatLogs.session_id == session_id, ChatLogs.agent_id == agent_id,
+				).order_by(ChatLogs.created_at)
 			)
 			rows = res.all()
 			logger.info(f"Found {len(rows) if rows else 0} messages for session {session_id}")
@@ -417,23 +436,28 @@ async def get_session_data(session_id: str):
 					}
 					# Build debug info (no dedicated debug_info column in DB)
 					if any([row_obj.user_query, row_obj.refined_query, row_obj.sql_query, row_obj.db_output]):
-						assistant_message["debug_info"] = {
+						debug_info = {
 							"query": row_obj.user_query,
 							"refine_query": row_obj.refined_query,
-							"sql_query": row_obj.sql_query,
-							"db_output": row_obj.db_output
 						}
+						# Same omission as the live answer: the key is left out, not nulled.
+						if show_sql_query:
+							debug_info["sql_query"] = row_obj.sql_query
+						if show_db_result:
+							debug_info["db_output"] = row_obj.db_output
+						assistant_message["debug_info"] = debug_info
 					else:
 						assistant_message["debug_info"] = None
 					# The ONLY change to this payload: the chart is sent instead of a
 					# hardcoded None. Every other key, debug_info included, is untouched.
 					assistant_message["graph_output"] = _chart_for_replay(row_obj)
 					# Additive: the column-labelled result block stored for this turn
 					# ({columns, rows, row_count, truncated}); `db_output` above keeps bare rows
-					# without column names. None for turns stored before the block existed.
+					# without column names. None for turns stored before the block existed, and
+					# when the agent hides result rows.
 					stored_debug = getattr(row_obj, "debug_info", None)
 					assistant_message["result"] = (
-						stored_debug.get("db_result") if isinstance(stored_debug, dict) else None
+						stored_debug.get("db_result") if show_db_result and isinstance(stored_debug, dict) else None
 					)
 					session_data.append(assistant_message)
 		logger.info(f"Returning {len(session_data)} total messages for session {session_id}")
@@ -453,10 +477,13 @@ async def get_session_chat_counter_from_db(session_id: str):
 		logger.error(f"Error getting session chat counter: {e}")
 		return None
 
-async def add_user_feedback(session_id: str, chat_id: str, feedback: str, corrected_sql: str = None):
+async def add_user_feedback(session_id: str, chat_id: str, feedback: str, corrected_sql: str = None, *, agent_id: str):
+	"""Record feedback on one of `agent_id`'s turns. A turn of another agent is not found, so a
+	caller allowed to use one agent cannot rewrite another agent's stored verdict or SQL."""
 	try:
 		async with get_async_session() as session:
-			res = await session.execute(select(ChatLogs).where(ChatLogs.session_id == session_id, ChatLogs.chat_id == chat_id))
+			res = await session.execute(select(ChatLogs).where(
+				ChatLogs.session_id == session_id, ChatLogs.chat_id == chat_id, ChatLogs.agent_id == agent_id))
 			row = res.first()
 			if not row:
 				return

docs/api.md

@@ -24,6 +24,19 @@ uses the signed HTTP-only session cookie; no hosted Auth service is required.
 | `GET /v1/agents/{id}/sessions` | Saved conversations on the agent |
 | `GET /v1/agents/{id}/sessions/{session_id}` | A saved conversation, each answer with its SQL, `result` rows and chart |
 
+An agent only ever sees its own turns of a conversation: chat history and replays
+are read for that agent alone. A `session_id` this API created for another agent
+returns HTTP404 before any history is read or a model is called; a `session_id`
+the API did not create (for example a chat service's own thread id) continues
+with this agent's turns only.
+
+`show_sql_query` and `show_db_result` apply to saved conversations exactly as to
+live answers: a replay leaves out the SQL and the rows (`debug_info.db_output`
+and `result`) the agent hides, and `GET /v1/sessions/{session_id}/export` returns
+HTTP403 for an agent that hides result rows; with an agent key it finds only that
+agent's turns. Both settings are read when the
+conversation is served, so turning one back on shows the stored SQL or rows again.
+
 Use the generated OpenAPI contract for complete request/response fields and
 streaming options. `tests/evaluation/run.py` is a small working client example:
 it loads configuration privately, creates the sample agent/source/table, issues

docs/mcp.md

@@ -127,23 +127,25 @@ and `chart_output`. Send a follow-up to
 - The agent's source credentials should be a dedicated read-only database
   account. The adapter cannot widen what that account or the agent's table
   visibility allows; it only forwards questions.
-- A conversation can be continued only on the agent that created it: a
-  `session_id` from another agent returns HTTP 404 before any history is read
-  or any model is called. Deployments running an API version older than this
-  change do not perform that check.
-- Saved conversations replayed through the API include SQL and rows regardless
-  of the agent's visibility settings. The adapter applies the agent's
-  `show_sql_query` and `show_db_result` to them, matching live answers, but a
-  caller using the API directly with the same key can read them.
+- An agent only ever sees its own turns of a conversation: history and replays
+  are read for that agent alone, and a `session_id` created on another agent
+  returns HTTP 404 before any history is read or any model is called.
+  Deployments running an API version older than this change do not perform
+  that check.
+- Saved conversations follow the agent's `show_sql_query` and `show_db_result`
+  in the API itself, exactly as live answers do: a replay leaves out the SQL and
+  rows the agent hides, and CSV export returns HTTP 403 when rows are hidden.
+  Deployments running an API version older than this change return them; the
+  adapter applies the same settings either way.
 - An agent key can list and read every conversation on its own agent, not only
   the ones it created.
 
 ## Verification
 
 `tests/mcp` checks the adapter through real MCP client sessions, in memory and
 over stdio, against a synthetic API with the product's credential rules. CI runs
-it with `tests/backend/test_session_scope.py`, which covers the session check
-and saved-result replay in the API itself.
+it with `tests/backend/test_session_scope.py`, which covers the session check,
+saved-result replay and saved-conversation visibility in the API itself.
 
 The integration was also run end to end on a disposable installation of this
 branch (Compose stack, default model profile, the synthetic `sample_sales`