add changes

agent/src/agent/mcp_server.py

@@ -64,16 +64,21 @@ async def chat_with_agent(
                 }
             )
 
-        result = await agent_graph.ainvoke(Command(
-            update={
-                "last_error": None,
-                "trino_error": None,
-                "escalated": None,
-                "escalation_reason": None,
-                "refinement_count": 0,
-            },
-            resume=resume_value
-        ), config=config)
+        is_node_interrupt = False
+        if getattr(state_snapshot, "interrupts", None):
+            is_node_interrupt = True
+        elif getattr(state_snapshot, "tasks", None):
+            if any(getattr(task, "interrupts", None) for task in state_snapshot.tasks):
+                is_node_interrupt = True
+
+        # Resume the graph execution depending on how it was interrupted
+        if is_node_interrupt:
+            # Interrupted by the `interrupt()` function inside a node
+            result = await agent_graph.ainvoke(Command(resume=resume_value), config=config)
+        else:
+            # Interrupted by an `interrupt_before` breakpoint
+            await agent_graph.aupdate_state(config, resume_value)
+            result = await agent_graph.ainvoke(None, config=config)
     else:
         if not query:
             return json.dumps({"error": "Query is required for new chat session."})
@@ -142,8 +147,16 @@ async def chat_with_agent(
     final_state = await agent_graph.aget_state(config)
 
     # Check if interrupted by `interrupt()` function
-    if final_state.interrupts:
+    interrupt_val = None
+    if getattr(final_state, "interrupts", None):
         interrupt_val = final_state.interrupts[-1].value
+    elif getattr(final_state, "tasks", None):
+        for task in final_state.tasks:
+            if getattr(task, "interrupts", None):
+                interrupt_val = task.interrupts[-1].value
+                break
+
+    if interrupt_val is not None:
         return json.dumps(
             {
                 "thread_id": thread_id,

agent/src/agent/nodes/query_builder.py

@@ -7,6 +7,8 @@
 from agent.config import settings
 from agent.langfuse_client import langfuse_client
 from langgraph.types import interrupt
+
+
 async def query_builder_node(state: AgentState, config: RunnableConfig | None = None):
     """Build SQL from plan and pause for user approval."""
     runtime_flags = state.get("runtime_flags") or {}

agent/src/agent/nodes/refiner.py

@@ -106,6 +106,7 @@ async def refiner_node(state: AgentState, config: RunnableConfig | None = None):
                 "error": trino_error,
                 "schema_context": schema_context,
                 "error_history": json.dumps(error_history),
+                "user_query": state.get("user_query", ""),
             }
         )
         new_sql = clean_sql(response.content)
@@ -140,8 +141,8 @@ async def refiner_node(state: AgentState, config: RunnableConfig | None = None):
                     langfuse_client.update_current_span(
                         level="ERROR", status_message=f"ESCA write failed: {e}"
                     )
-                else:
-                    logging.error(f"ESCA write failed: {e}")
+
+                logging.error(f"ESCA write failed: {e}")
                 raise RuntimeError(f"Failed to write query result to ESCA: {e}")
 
         return {

agent/src/agent/nodes/satisfaction_check.py

@@ -62,7 +62,8 @@ async def satisfaction_check_node(state: AgentState, config: RunnableConfig | No
             failures.append(f"[CHECK_A] Execution failed: {state['trino_error']}")
 
     # ── Check B: Row Plausibility ─────────────────────────────────────────────
-    if _f(runtime_flags, "SATISFACTION_CHECK_PLAUSIBILITY", settings.SATISFACTION_CHECK_PLAUSIBILITY):
+    # Only run plausibility if there wasn't a hard execution failure
+    if not state.get("trino_error") and _f(runtime_flags, "SATISFACTION_CHECK_PLAUSIBILITY", settings.SATISFACTION_CHECK_PLAUSIBILITY):
         n = len(rows)
         min_rows = _f(runtime_flags, "SATISFACTION_MIN_ROWS", settings.SATISFACTION_MIN_ROWS)
         max_rows = _f(runtime_flags, "SATISFACTION_MAX_ROWS", settings.SATISFACTION_MAX_ROWS)

agent/src/agent/nodes/schema_explorer.py

@@ -135,9 +135,22 @@ def _build_column_context(cp: "ColumnProfile") -> dict:
     return col
 
 
+class TableJoin(BaseModel):
+    source_table: str = Field(description="Fully qualified name of the source table (catalog.schema.name)")
+    target_table: str = Field(description="Fully qualified name of the target table (catalog.schema.name)")
+    join_condition: str = Field(description="SQL condition for the join (e.g., A.id = B.a_id)")
+
+class SchemaPlan(BaseModel):
+    tables: List[str] = Field(description="List of fully qualified table names (catalog.schema.name) required for the query")
+    columns: List[str] = Field(description="Columns to select")
+    joins: List[TableJoin] = Field(default_factory=list, description="How to join the tables")
+    filters: List[str] = Field(default_factory=list, description="Any WHERE clause filters")
+
 # Define standardized Schema Explorer Output Type
 class SchemaExplorerOutput(BaseModel):
-    schema_plan: Optional[Any] = Field(
+    reasoning: str = Field(default_factory=str, description="Reasoning for the schema plan - explain your logic.")
+
+    schema_plan: Optional[SchemaPlan] = Field(
         default=None,
         description="Detailed query plan describing tables, columns, and joins.",
     )
@@ -156,6 +169,10 @@ class SchemaExplorerOutput(BaseModel):
         default_factory=list,
         description="List of fully qualified table names (catalog.schema.name) used in the plan.",
     )
+    error: Optional[str] = Field(
+        default=None, 
+        description="Error message if any"
+    )
 
 
 def get_query_embedding(text: str) -> list[float]:
@@ -189,7 +206,7 @@ def hybrid_search_tables(
     all_tables = session.exec(stmt_all).all()
 
     allowed = allowed_tables or []
-    statuses = allowed_statuses or ["production"]
+    statuses = allowed_statuses if allowed_statuses is not None else ["production"]
     allowed_tables_set = []
     allowed_ids = set()
 
@@ -430,7 +447,7 @@ def _parse_bool_flag(value) -> bool:
 
     sem = asyncio.Semaphore(profile_fetch_concurrency)
 
-    async def fetch_profile(t_id, t_name):
+    async def fetch_profile(t_id, t_name, t_fqn):
         nonlocal cache_hit_count, cache_miss_count
         async with sem:
             try:
@@ -453,22 +470,25 @@ async def fetch_profile(t_id, t_name):
                             cache_miss_count += 1
 
                 profile_res = await get_table_profile.ainvoke({"table_id": t_id})
-                return json.loads(profile_res)
+                data = json.loads(profile_res)
+                data["fully_qualified_name"] = t_fqn
+                return data
             except Exception as e:
                 print(f"Error fetching profile for {t_name}: {e}")
                 return None
 
     fetch_tasks = []
     for i, t in enumerate(candidate_tables):
+        fqn = f"{t.catalog}.{t.schema_name}.{t.name}"
         tables_info.append(
             {
                 "id": t.id,
-                "name": f"{t.catalog}.{t.schema_name}.{t.name}",
+                "name": fqn,
                 "description": "",
             }
         )
         if i < max_profiles_to_fetch:
-            fetch_tasks.append(fetch_profile(t.id, t.name))
+            fetch_tasks.append(fetch_profile(t.id, t.name, fqn))
 
     if fetch_tasks:
         results = await asyncio.gather(*fetch_tasks, return_exceptions=True)
@@ -532,7 +552,7 @@ async def fetch_profile(t_id, t_name):
     if schema_summarization and profile_details:
         try:
             summaries = [
-                f"[{p.get('table_name', 'unknown')}] {p.get('description', '') or '(no description available)'}"
+                f"[{p.get('fully_qualified_name', p.get('table_name', 'unknown'))}] {p.get('description', '') or '(no description available)'}"
                 for p in profile_details
             ]
             profiles_json_str = "\n".join(summaries)
@@ -593,13 +613,14 @@ async def fetch_profile(t_id, t_name):
             ambiguity_detected=False,
             ambiguity_message="",
             candidate_options=[],
+            error=str(e),
         )
 
     data = await _resolve_ambiguity(data, chain, tables_info, profiles_json_str, human_message, state)
 
     plan = data.schema_plan
     if plan is not None and not isinstance(plan, str):
-        plan = json.dumps(plan)
+        plan = plan.model_dump_json() if hasattr(plan, "model_dump_json") else json.dumps(plan)
     elif plan is None:
         plan = ""
 

agent/tests/test_scoping_mode.py

@@ -0,0 +1,54 @@
+import pytest
+from unittest.mock import patch, MagicMock
+
+from agent.nodes.schema_explorer import hybrid_search_tables
+from core.models.models import Table
+
+def test_hybrid_search_tables_strict_mode():
+    session_mock = MagicMock()
+
+    table1 = Table(id="t1", name="table_1", schema_name="public", catalog="cat", status="production")
+    table2 = Table(id="t2", name="table_2", schema_name="public", catalog="cat", status="production")
+    table3 = Table(id="t3", name="table_3", schema_name="public", catalog="cat", status="deprecated")
+
+    session_mock.exec.return_value.all.return_value = [table1, table2, table3]
+
+    # Mock execute for vector search
+    session_mock.execute.return_value.fetchall.return_value = [("t1",)]
+
+    # Mock session.get for final return
+    def mock_get(cls, id):
+        return {"t1": table1, "t2": table2, "t3": table3}.get(id)
+    session_mock.get.side_effect = mock_get
+
+    # Mock enrichment version for keyword search
+    session_mock.exec.return_value.first.return_value = None
+
+    # Test strict mode allows ONLY t1
+    results = hybrid_search_tables(
+        query="table", 
+        query_embedding=[0.0], 
+        session=session_mock, 
+        allowed_tables=["t1"], 
+        allowed_statuses=["production"], 
+        scoping_mode="strict"
+    )
+
+    assert len(results) == 1
+    assert results[0].id == "t1"
+
+    # Test hybrid mode allows t1 and t2 (because status="production")
+    session_mock.execute.return_value.fetchall.return_value = [("t1",), ("t2",)]
+    results_hybrid = hybrid_search_tables(
+        query="table", 
+        query_embedding=[0.0], 
+        session=session_mock, 
+        allowed_tables=["t1"], 
+        allowed_statuses=["production"], 
+        scoping_mode="hybrid"
+    )
+
+    assert len(results_hybrid) == 2
+    ids = {r.id for r in results_hybrid}
+    assert "t1" in ids
+    assert "t2" in ids

backend/alembic/versions/70e8a34ff877_add_unique_constraint_to_tables.py

@@ -0,0 +1,39 @@
+"""Add unique constraint to tables
+
+Revision ID: 70e8a34ff877
+Revises: f9a3d1c8e205
+Create Date: 2026-07-07 11:17:43.459879
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+# revision identifiers, used by Alembic.
+revision: str = '70e8a34ff877'
+down_revision: Union[str, None] = 'f9a3d1c8e205'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint('uq_table_profiles_table_id', 'table_profiles', type_='unique')
+    op.drop_index('ix_table_profiles_table_id', table_name='table_profiles')
+    op.create_index(op.f('ix_table_profiles_table_id'), 'table_profiles', ['table_id'], unique=True)
+    op.drop_column('table_profiles', 'is_partial')
+    op.create_unique_constraint('uq_table_fqn', 'tables', ['catalog', 'schema_name', 'name'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint('uq_table_fqn', 'tables', type_='unique')
+    op.add_column('table_profiles', sa.Column('is_partial', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False))
+    op.drop_index(op.f('ix_table_profiles_table_id'), table_name='table_profiles')
+    op.create_index('ix_table_profiles_table_id', 'table_profiles', ['table_id'], unique=False)
+    op.create_unique_constraint('uq_table_profiles_table_id', 'table_profiles', ['table_id'], postgresql_nulls_not_distinct=False)
+    # ### end Alembic commands ###

backend/app/config.py

@@ -20,6 +20,7 @@ class Settings(BaseSettings):
     LANGFUSE_PUBLIC_KEY: str = ""
     LANGFUSE_SECRET_KEY: str = ""
     LANGFUSE_HOST: str = "https://cloud.langfuse.com"
+    LANGFUSE_REQUEST_TIMEOUT: float = 30.0
     CORS_ORIGINS: list[str] = [
         "http://localhost:5173",
         "http://localhost:3000",

backend/app/routers/agent.py

@@ -67,6 +67,17 @@ class ChatResponse(BaseModel):
     trace_id: str | None = None
     execution_path: list[str] | None = None
 
+class TraceSpan(BaseModel):
+    span_name: str
+    start_time: str | None = None
+    duration_ms: int = 0
+    input_tokens: int = 0
+    output_tokens: int = 0
+    model: str = "N/A"
+    status: str
+    input_preview: str = ""
+    output_preview: str = ""
+
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -187,6 +198,8 @@ async def chat(request: ChatRequest) -> ChatResponse:
         "allowed_tables": request.allowed_tables,
         "allowed_statuses": request.allowed_statuses,
         "extractors": request.extractors,
+        "active_skills": request.active_skills,
+        "execution_mode": request.execution_mode,
         "hitl_enabled": request.hitl_enabled,
     }
 
@@ -254,13 +267,13 @@ async def event_generator():
     return StreamingResponse(event_generator(), media_type="text/event-stream")
 
 
-@router.get("/traces/{trace_id}")
+@router.get("/traces/{trace_id}", response_model=list[TraceSpan])
 async def get_trace_timeline(trace_id: str):
     """Fetch trace from Langfuse and normalize observations for frontend timeline."""
     auth = (settings.LANGFUSE_PUBLIC_KEY, settings.LANGFUSE_SECRET_KEY)
     url = f"{settings.LANGFUSE_HOST}/api/public/traces/{trace_id}"
 
-    async with httpx.AsyncClient() as client:
+    async with httpx.AsyncClient(timeout=settings.LANGFUSE_REQUEST_TIMEOUT) as client:
         resp = await client.get(url, auth=auth)
         if resp.status_code != 200:
             if resp.status_code == 404:
@@ -290,7 +303,7 @@ async def get_trace_timeline(trace_id: str):
 
             timeline.append(
                 {
-                    "span_name": obs.get("name") or obs.get("type"),
+                    "span_name": obs.get("name") or obs.get("type") or "Unknown Span",
                     "start_time": start_time_str,
                     "duration_ms": duration_ms,
                     "input_tokens": obs.get("promptTokens", 0),

backend/app/routers/tables.py

@@ -18,6 +18,7 @@
     UserScope,
 )
 from fastapi import APIRouter, Depends, Header, HTTPException, Query
+from sqlalchemy.exc import IntegrityError
 from sqlmodel import Session, col, select
 
 from app.config import settings
@@ -195,6 +196,20 @@ def create_table(payload: TableCreate, session: Session = Depends(get_session)):
     text_to_embed = f"Table name: {name}\nSchema: {schema_name}\nDescription: {description}\nColumns: {', '.join([c.get('name', '') for c in om_columns])}"
     embedding = get_embedding(text_to_embed)
 
+    # Check for duplicate table
+    existing = session.exec(
+        select(Table).where(
+            Table.catalog == catalog_name,
+            Table.schema_name == schema_name,
+            Table.name == name
+        )
+    ).first()
+    if existing:
+        raise HTTPException(
+            status_code=409,
+            detail=f"Table '{catalog_name}.{schema_name}.{name}' already exists."
+        )
+
     # Create the table
     table = Table(
         name=name,
@@ -207,7 +222,14 @@ def create_table(payload: TableCreate, session: Session = Depends(get_session)):
         embedding=embedding,
     )
     session.add(table)
-    session.commit()
+    try:
+        session.commit()
+    except IntegrityError:
+        session.rollback()
+        raise HTTPException(
+            status_code=409,
+            detail=f"Table '{catalog_name}.{schema_name}.{name}' already exists."
+        )
 
     # Extract columns in the format expected by the frontend
     def parse_columns(cols):

backend/dump_schema.py

@@ -0,0 +1,4 @@
+import json
+from app.main import app
+with open("../frontend/openapi.json", "w") as f:
+    json.dump(app.openapi(), f, indent=2)