feat: implement feature flags & execution modes builder

agent/src/agent/config.py

@@ -27,6 +27,10 @@ class AgentSettings(BaseSettings):
     JEEN_API_KEY: str = ""       # If empty, agent gracefully skips fetching
     SKILLS_HOT_RELOAD: bool = False  # If true, bypass Redis cache for skills
 
+    # ── G4: Feature Flags & Execution Modes ──────────────────────────────────
+    BACKEND_URL: str = ""  # Studio backend URL for flag reads (e.g. http://backend:8000)
+                           # If empty, FlagBridge falls back to env-var defaults
+
 
     # Langfuse prompt names
     LANGFUSE_PROMPT_EXTRACTOR: str = "text2sql/extractor"

agent/src/agent/graph.py

@@ -20,6 +20,7 @@
 from langchain_core.prompts import ChatPromptTemplate
 from agent.state import AgentState
 from agent.nodes.extractor import extractor_node
+from agent.nodes.init_flags import init_flags_node
 from agent.nodes.init_skills import init_skills_node
 from agent.nodes.schema_explorer import schema_explorer_node, MAX_SCHEMA_RETRIES
 from agent.nodes.query_builder import query_builder_node
@@ -193,6 +194,7 @@ def route_rejection(state: AgentState) -> str:
 workflow = StateGraph(AgentState)
 
 workflow.add_node("validate_config", validate_config_node)
+workflow.add_node("init_flags", init_flags_node)
 workflow.add_node("init_skills", init_skills_node)
 workflow.add_node("extractor", extractor_node)
 workflow.add_node("schema_explorer", schema_explorer_node)
@@ -203,9 +205,10 @@ def route_rejection(state: AgentState) -> str:
 workflow.add_node("hitl_escalation", hitl_escalation_node)
 workflow.add_node("finalizer", finalizer_node)
 
-# Entry: validate config before anything else (G2-01 fail-fast)
+# Entry: validate config → resolve flags → load skills → start reasoning
 workflow.add_edge(START, "validate_config")
-workflow.add_edge("validate_config", "init_skills")
+workflow.add_edge("validate_config", "init_flags")
+workflow.add_edge("init_flags", "init_skills")
 workflow.add_edge("init_skills", "extractor")
 workflow.add_edge("extractor", "schema_explorer")
 

agent/src/agent/llm.py

@@ -1,16 +1,67 @@
 import logging
+from typing import Optional
+
 from langchain_openai import ChatOpenAI
+
 from agent.config import settings
-from typing import Optional
 
+logger = logging.getLogger(__name__)
+
+# Flag name → LLM_MODEL env-var fallback for each node
+_NODE_MODEL_FLAGS: dict[str, str] = {
+    "extractor":          "EXTRACTOR_MODEL",
+    "schema_explorer":    "SCHEMA_SUMMARY_MODEL",
+    "query_builder":      "QUERY_BUILDER_MODEL",
+    "refiner":            "REFINER_MODEL",
+    "satisfaction_check": "SATISFACTION_JUDGE_MODEL",
+    "routing":            "QUERY_BUILDER_MODEL",  # rejection router reuses QB model
+    "default":            "QUERY_BUILDER_MODEL",
+}
+
+_NODE_TEMP_FLAGS: dict[str, str] = {
+    "extractor":     "EXTRACTOR_TEMPERATURE",
+    "query_builder": "QUERY_BUILDER_TEMPERATURE",
+}
+
+
+def get_llm(
+    node: str = "default",
+    temperature: Optional[float] = None,
+    runtime_flags: Optional[dict] = None,
+) -> ChatOpenAI:
+    """
+    Factory for per-node LLM instances.
+
+    Priority for model/temperature selection:
+      1. runtime_flags (resolved by init_flags_node from DB + execution mode)
+      2. AgentSettings env-var defaults
+
+    Args:
+        node:          Name of the calling graph node (used to pick the right flag).
+        temperature:   Optional hard override — bypasses flag resolution.
+        runtime_flags: The state["runtime_flags"] dict from the current invocation.
+                       Pass None when initialising at module level (will use env defaults).
+    """
+    flags = runtime_flags or {}
+
+    # Resolve model
+    model_flag = _NODE_MODEL_FLAGS.get(node, "QUERY_BUILDER_MODEL")
+    model = flags.get(model_flag) or settings.LLM_MODEL
+
+    # Resolve temperature
+    if temperature is None:
+        temp_flag = _NODE_TEMP_FLAGS.get(node)
+        temperature = float(flags.get(temp_flag, 0.0)) if temp_flag else 0.0
+
+    logger.debug(
+        "Instantiating LLM for node='%s': model='%s' temperature=%.2f",
+        node,
+        model,
+        temperature,
+    )
 
-# We could dynamically configure settings based on the node name.
-# For now, it delegates to agent.config settings.
-def get_llm(node: str = "default", temperature: Optional[float] = 0.0) -> ChatOpenAI:
-    """Factory function for instantiating the unified LLM."""
-    logging.debug(f"Instantiating LLM for node: {node}")
     return ChatOpenAI(
-        model=settings.LLM_MODEL,
+        model=model,
         base_url=settings.LLM_BASE_URL,
         api_key=settings.LLM_API_KEY,
         temperature=temperature,

agent/src/agent/mcp_server.py

@@ -20,9 +20,24 @@ async def chat_with_agent(
     allowed_statuses: list[str] | None = None,
     extractors: list[str] | None = None,
     active_skills: list[str] | None = None,
+    execution_mode: str | None = None,
     hitl_enabled: bool = True,
 ) -> str:
-    """Run the Text2SQL agent to answer database queries."""
+    """Run the Text2SQL agent to answer database queries.
+
+    Args:
+        query:          The natural language question to answer.
+        thread_id:      Optional thread ID for session continuity.
+        resume_value:   HITL resume payload (pass after receiving an interrupt).
+        allowed_tables: Restrict the agent to specific tables.
+        allowed_statuses: Filter tables by status.
+        extractors:     List of extractor names/IDs to use.
+        active_skills:  List of Jeen skill UUIDs to inject.
+        execution_mode: Named configuration preset (e.g. 'cost_saving',
+                        'high_quality', 'benchmark'). Overrides flag defaults
+                        for this invocation only.
+        hitl_enabled:   If False, skip all human-in-the-loop interrupts.
+    """
     thread_id = thread_id or str(uuid.uuid4())
     config = {
         "configurable": {"thread_id": thread_id},
@@ -91,6 +106,7 @@ async def chat_with_agent(
                 "allowed_statuses": allowed_statuses,
                 "active_extractors": active_extractors,
                 "active_skills": active_skills,
+                "execution_mode": execution_mode,
                 "non_interactive": not hitl_enabled,
             },
             config=config,

agent/src/agent/nodes/extractor.py

@@ -45,8 +45,8 @@ def extract(self, query: str) -> List[ContextEntry]:
 
 
 class LLMExtractor(BaseExtractor):
-    def __init__(self):
-        self.llm = get_llm("extractor")
+    def __init__(self, runtime_flags: dict | None = None):
+        self.llm = get_llm("extractor", runtime_flags=runtime_flags)
 
         langfuse_prompt = langfuse_client.get_prompt(settings.LANGFUSE_PROMPT_EXTRACTOR)
         self.prompt = ChatPromptTemplate.from_messages(
@@ -102,10 +102,11 @@ def extractor_node(state: AgentState):
     """Enrich the user query with additional context to help downstream phases."""
     user_query = state["user_query"]
     active_extractors = state.get("active_extractors") or []
+    runtime_flags = state.get("runtime_flags") or {}
 
     import concurrent.futures
 
-    extractors: List[BaseExtractor] = [TimeExtractor(), LLMExtractor()]
+    extractors: List[BaseExtractor] = [TimeExtractor(), LLMExtractor(runtime_flags=runtime_flags)]
 
     for ext_info in active_extractors:
         extractors.append(HTTPExtractor(ext_info["url"], ext_info["name"]))

agent/src/agent/nodes/init_flags.py

@@ -0,0 +1,66 @@
+"""
+init_flags_node (G4)
+====================
+Runs immediately after validate_config, before init_skills.
+
+Responsibilities:
+  1. Call FlagBridge.resolve_flags(execution_mode) to merge:
+       mode overrides → DB flags → env-var defaults
+  2. Write the resolved dict to state["runtime_flags"]
+  3. Log runtime_flags to Langfuse trace metadata for full observability
+"""
+
+import logging
+
+from agent.langfuse_client import langfuse_client
+from agent.state import AgentState
+from agent.utils.flag_bridge import FlagBridge
+
+logger = logging.getLogger(__name__)
+
+_flag_bridge = FlagBridge()
+
+
+async def init_flags_node(state: AgentState) -> dict:
+    """
+    Resolve all runtime configuration flags for this invocation.
+
+    The resolved dict is stored in state["runtime_flags"] and read by every
+    downstream node instead of directly accessing AgentSettings env vars.
+    This guarantees that:
+      - DS team changes in the Studio UI take effect within the cache TTL (30s).
+      - Execution mode overrides are applied consistently to all nodes.
+      - Every Langfuse trace carries the exact config used for that query.
+    """
+    execution_mode: str | None = state.get("execution_mode")
+
+    try:
+        runtime_flags = await _flag_bridge.resolve_flags(execution_mode)
+    except Exception as exc:
+        logger.warning("init_flags_node: FlagBridge failed (%s), using env defaults", exc)
+        # FlagBridge already handles its own fallback internally, so this is a
+        # safety net for any unexpected error in the bridge itself.
+        from agent.utils.flag_bridge import _ENV_DEFAULTS
+        runtime_flags = dict(_ENV_DEFAULTS)
+
+    # Emit to Langfuse for observability
+    try:
+        trace_id = langfuse_client.get_current_trace_id()
+        if trace_id:
+            langfuse_client.trace(
+                id=trace_id,
+                metadata={
+                    "runtime_flags": runtime_flags,
+                    "execution_mode": execution_mode or "default",
+                },
+            )
+    except Exception as exc:
+        logger.warning("init_flags_node: Langfuse trace failed: %s", exc)
+
+    logger.info(
+        "init_flags_node: resolved %d flags (mode=%s)",
+        len(runtime_flags),
+        execution_mode or "default",
+    )
+
+    return {"runtime_flags": runtime_flags}

agent/src/agent/nodes/init_skills.py

@@ -14,13 +14,23 @@ async def init_skills_node(state: AgentState) -> dict:
     keeping reasoning nodes pure and state reproducible.
     """
     active_skills = state.get("active_skills")
+    runtime_flags = state.get("runtime_flags") or {}
 
-    if not active_skills:
+    from agent.config import settings
+    skills_enabled = bool(runtime_flags.get("SKILLS_ENABLED", settings.SKILLS_ENABLED))
+    hot_reload = bool(runtime_flags.get("SKILLS_HOT_RELOAD", settings.SKILLS_HOT_RELOAD))
+    cache_ttl = int(runtime_flags.get("SKILLS_CACHE_TTL", settings.SKILLS_CACHE_TTL))
+
+    if not skills_enabled or not active_skills:
         return {"loaded_skills": None}
 
     try:
         _skill_registry.redis = get_redis_client()
-        loaded_skills = await _skill_registry.get_skills(active_skills)
+        loaded_skills = await _skill_registry.get_skills(
+            active_skills,
+            hot_reload=hot_reload,
+            cache_ttl=cache_ttl,
+        )
 
         if loaded_skills:
             try:

agent/src/agent/nodes/query_builder.py

@@ -4,12 +4,9 @@
 from agent.config import settings
 from agent.langfuse_client import langfuse_client
 from langgraph.types import interrupt
-
-llm = get_llm("query_builder")
-
-
 def query_builder_node(state: AgentState):
     """Build SQL from plan and pause for user approval."""
+    runtime_flags = state.get("runtime_flags") or {}
     feedback = state.get("feedback")
     feedback_str = f"\nUser Feedback to apply: {feedback}" if feedback else ""
 
@@ -23,7 +20,8 @@ def query_builder_node(state: AgentState):
 
     langfuse_prompt = langfuse_client.get_prompt(settings.LANGFUSE_PROMPT_QUERY_BUILDER)
     prompt = ChatPromptTemplate.from_messages(langfuse_prompt.get_langchain_prompt())
-    chain = prompt | llm
+    _llm = get_llm("query_builder", runtime_flags=runtime_flags)
+    chain = prompt | _llm
     response = chain.invoke(
         {
             "schema_plan": state.get("schema_plan"),

agent/src/agent/nodes/refiner.py

@@ -6,7 +6,6 @@
 from core import execute_query_sync
 from agent.config import settings
 from agent.langfuse_client import langfuse_client
-from agent.langfuse_client import langfuse_client
 from langchain_core.prompts import ChatPromptTemplate
 from agent.llm import get_llm
 from agent.utils.sql import clean_sql
@@ -33,6 +32,10 @@ async def refiner_node(state: AgentState):
     sql = state.get("sql_query")
     count = state.get("refinement_count", 0)
     error_history = state.get("error_history") or []
+    runtime_flags = state.get("runtime_flags") or {}
+
+    # Resolve per-invocation limit (DS-tunable via flags)
+    max_iterations = int(runtime_flags.get("MAX_REFINER_ITERATIONS", MAX_REFINER_ITERATIONS))
 
     # Execute against Trino
     try:
@@ -49,14 +52,14 @@ async def refiner_node(state: AgentState):
 
     if not success:
         # If we reached the refinement limit, just stop and don't prompt LLM
-        if count >= MAX_REFINER_ITERATIONS:
+        if count >= max_iterations:
             return {
                 "trino_error": trino_error,
                 "last_error": trino_error,
                 "refinement_count": count + 1,
                 "error_history": error_history,
                 "escalation_reason": (
-                    f"Refiner exhausted {MAX_REFINER_ITERATIONS} iterations. "
+                    f"Refiner exhausted {max_iterations} iterations. "
                     f"Last Trino error: {trino_error}"
                 ),
             }
@@ -65,7 +68,8 @@ async def refiner_node(state: AgentState):
         prompt = ChatPromptTemplate.from_messages(
             langfuse_prompt.get_langchain_prompt()
         )
-        chain = prompt | llm
+        _llm = get_llm("refiner", runtime_flags=runtime_flags)
+        chain = prompt | _llm
 
         schema_context = build_refiner_schema_context(state)