feat: inline Google Sheets data source integration

- Inline DataSourceRef with full config (url, tab, columns, format, inject_as)
- Single migration: ALTER TABLE template ADD data_sources JSONB
- Discovery routes (admin-only) for sheet tab/column/preview
- Prefetch at dispatch warms Redis (60s TTL, content-hash key)
- Layer 5 injection: var -> template_vars, message -> ds_messages
- Cache miss writes back to Redis; defensive URL parsing; session close

Author: cmd-err

app/ai/voice/agents/breeze_buddy/agent/__init__.py

@@ -159,6 +159,7 @@ def __init__(
         self.greeting_text: Optional[str] = (
             None  # Resolved greeting text for LLM context
         )
+        self.ds_messages: List[Dict] = []
         self.default_vad_params: Optional[VADParams] = None
         self.default_interruption_config: Optional[InterruptionConfig] = None
 
@@ -356,6 +357,7 @@ async def _setup_daily_transport(self, runner_args: RunnerArguments) -> None:
                 self.template,
                 self.configurations,
                 self.template_vars,
+                self.ds_messages,
             ) = await load_template_config(self.lead)
         except ValueError as e:
             logger.error(f"Failed to load template config for Daily mode: {e}")
@@ -553,6 +555,7 @@ async def _setup_telephony_transport(self) -> bool:
                 self.template,
                 self.configurations,
                 self.template_vars,
+                self.ds_messages,
             ) = await load_template_config(self.lead)
         except ValueError as e:
             error_msg = f"Template loading failed: {str(e)}"
@@ -908,7 +911,9 @@ async def _handle_client_connected(self) -> None:
             self.flow_config,
             self.end_conversation_callbacks,
             self.expected_callback_response_schema,
-        ) = build_flow_config(self.flow_builder, self.template)
+        ) = build_flow_config(
+            self.flow_builder, self.template, ds_messages=self.ds_messages
+        )
 
         lead_payload = self.lead.payload or {}
 
@@ -1012,7 +1017,6 @@ async def run(self, runner_args: Optional[RunnerArguments] = None) -> None:
             else:
                 if not await self._setup_telephony_transport():
                     if self.completion_function and self.call_sid:
-
                         lead = await get_lead_by_call_id(self.call_sid)
                         # If lead is None (not found), or it doesn't have an outcome,
                         # or the outcome is not a BLOCKED_ outcome, then it's an early hangup.

app/ai/voice/agents/breeze_buddy/agent/flow.py

@@ -27,18 +27,18 @@
 
 async def load_template_config(
     lead: LeadCallTracker,
-) -> tuple[TemplateModel, Optional[ConfigurationModel], Dict[str, str]]:
+) -> tuple[TemplateModel, Optional[ConfigurationModel], Dict[str, str], List[Dict]]:
     """Load template configuration from database.
 
     Args:
         lead: The lead instance
 
     Returns:
-        Tuple of (template, configurations, template_vars)
+        Tuple of (template, configurations, template_vars, ds_messages)
     """
     flow_loader = FlowConfigLoader()
 
-    template, template_vars = await flow_loader.load_template(
+    template, template_vars, ds_messages = await flow_loader.load_template(
         reseller_id=lead.reseller_id,
         template=lead.template,
         merchant_id=lead.merchant_id if lead else None,
@@ -58,7 +58,7 @@ async def load_template_config(
     if getattr(lead, "execution_mode", None) != ExecutionMode.DAILY_STREAM:
         validate_template_compat(template)
 
-    return template, template.configurations, template_vars
+    return template, template.configurations, template_vars, ds_messages
 
 
 def setup_flow_manager(
@@ -127,12 +127,14 @@ def setup_flow_manager(
 def build_flow_config(
     flow_builder: FlowConfigBuilder,
     template: TemplateModel,
+    ds_messages: Optional[List[Dict]] = None,
 ) -> tuple[Dict[str, Any], List, Any]:
     """Build flow configuration from template.
 
     Args:
         flow_builder: Flow config builder
         template: Template model
+        ds_messages: Optional data source system messages to inject
 
     Returns:
         Tuple of (flow_config, end_conversation_callbacks, expected_callback_response_schema)
@@ -143,6 +145,9 @@ def build_flow_config(
         "expected_callback_response_schema", None
     )
 
+    if ds_messages:
+        flow_config["_data_source_messages"] = ds_messages
+
     logger.info(
         f"Built flow config with {len(flow_config['nodes'])} nodes, "
         f"initial: {flow_config['initial_node']}, "

app/ai/voice/agents/breeze_buddy/dispatch/worker.py

@@ -52,6 +52,9 @@
     _release_number,
     _run_pre_checks_for_lead,
 )
+from app.ai.voice.agents.breeze_buddy.managers.data_source_prefetch import (
+    prefetch_data_sources,
+)
 from app.ai.voice.agents.breeze_buddy.managers.utils import (
     prepare_and_store_initial_greeting,
 )
@@ -339,6 +342,8 @@ async def _dispatch(
 
             if template:
                 template = apply_playground_overrides(locked, template)
+                if template.data_sources:
+                    asyncio.create_task(prefetch_data_sources(template=template))
                 await prepare_and_store_initial_greeting(
                     lead_id=locked.id,
                     payload=locked.payload or {},

app/ai/voice/agents/breeze_buddy/managers/data_source_prefetch.py

@@ -0,0 +1,75 @@
+"""
+Data Source Prefetch Manager
+
+Pre-warms Redis with Google Sheets content for all DataSourceRefs attached to a
+template at dispatch time. Runs concurrently with greeting TTS synthesis.
+
+Cache key is scoped to a content hash so concurrent calls sharing the same sheet
+use a single cached copy.
+
+Cache key : ``datasource:{sheet_id}:{sheet_name}:{cols_hash}``
+TTL        : 60 s
+"""
+
+import asyncio
+import hashlib
+import json
+from typing import Optional
+
+from app.ai.voice.agents.breeze_buddy.template.types import DataSourceRef, TemplateModel
+from app.core.logger import logger
+from app.services.google.sheets import fetch_formatted
+from app.services.redis import get_redis_service
+
+_CACHE_TTL = 60  # seconds — shared across leads; short to keep data fresh
+_FETCH_TIMEOUT = 5.0
+
+
+def _cache_key(ref: DataSourceRef) -> str:
+    """Derive a stable cache key from the sheet config."""
+    spreadsheet_id = ref.spreadsheet_url.split("/d/")[1].split("/")[0]
+    sheet = ref.sheet_name or "_first_"
+    cols = json.dumps(sorted(ref.columns or []))
+    cols_digest = hashlib.md5(cols.encode()).hexdigest()[:8]
+    return f"datasource:{spreadsheet_id}:{sheet}:{cols_digest}:{ref.format.value}"
+
+
+async def _prefetch_one(ref: DataSourceRef) -> None:
+    if not ref.is_active:
+        return
+    try:
+        content = await asyncio.wait_for(
+            fetch_formatted(
+                spreadsheet_id=ref.spreadsheet_url.split("/d/")[1].split("/")[0],
+                sheet_name=ref.sheet_name,
+                columns=ref.columns,
+                format=ref.format.value,
+            ),
+            timeout=_FETCH_TIMEOUT,
+        )
+        cache_key = _cache_key(ref)
+        redis = await get_redis_service()
+        await redis.setex(cache_key, content, ttl_seconds=_CACHE_TTL)
+        logger.info(
+            "Prefetched data source '%s' (%d chars, TTL=%ds)",
+            ref.name,
+            len(content),
+            _CACHE_TTL,
+        )
+    except asyncio.TimeoutError:
+        logger.warning("Prefetch timeout for data source '%s'", ref.name)
+    except Exception as exc:
+        logger.error(
+            "Prefetch error for data source '%s': %s", ref.name, exc, exc_info=True
+        )
+
+
+async def prefetch_data_sources(
+    template: Optional[TemplateModel],
+) -> None:
+    if not template or not template.data_sources:
+        return
+    await asyncio.gather(
+        *[_prefetch_one(ref) for ref in template.data_sources],
+        return_exceptions=True,
+    )

app/ai/voice/agents/breeze_buddy/template/loader.py

@@ -4,12 +4,16 @@
 This module provides functionality to load templates from the database.
 """
 
-from typing import Dict, Optional, Tuple
+import asyncio
+import hashlib
+import json
+from typing import Dict, List, Optional, Tuple
 
 from app.ai.voice.agents.breeze_buddy.template.transformation_function import (
     TEMPLATE_FUNCTION_REGISTRY,
 )
 from app.ai.voice.agents.breeze_buddy.template.types import (
+    DataSourceRef,
     FlowMode,
     TemplateModel,
 )
@@ -21,6 +25,9 @@
 from app.database.accessor.breeze_buddy.template import (
     get_template_by_id_with_fallback,
 )
+from app.services.data_sources import DATA_SOURCE_UNAVAILABLE
+from app.services.google.sheets import fetch_formatted
+from app.services.redis import get_redis_service
 
 
 class FlowConfigLoader:
@@ -94,7 +101,7 @@ async def load_template(
         merchant_id: Optional[str] = None,
         call_payload: Optional[Dict[str, str]] = None,
         template_id: Optional[str] = None,
-    ) -> Tuple[TemplateModel, Dict[str, str]]:
+    ) -> Tuple[TemplateModel, Dict[str, str], List[Dict]]:
         """
         Load template and render task messages with variables.
 
@@ -123,6 +130,7 @@ async def load_template(
             )
 
         template_vars = {}
+        _ds_messages: List[Dict] = []
 
         # 1. Load credentials from credentials table (global + merchant-specific)
         try:
@@ -205,13 +213,30 @@ async def load_template(
                         f"Injected extra payload field '{field_name}' into template_vars"
                     )
 
+        # 5. Inject data_source content.
+        if template_obj.data_sources:
+            contents = await asyncio.gather(
+                *[
+                    self._fetch_one_data_source(ref)
+                    for ref in template_obj.data_sources
+                ],
+                return_exceptions=True,
+            )
+            for ref, result in zip(template_obj.data_sources, contents):
+                if isinstance(result, Exception) or result is None:
+                    continue
+                if ref.inject_as == "message":
+                    _ds_messages.append({"role": "system", "content": result})
+                else:
+                    template_vars[ref.name] = result
+
         # Direct mode has a flat structure (system_prompt + flat function list)
         # rather than a nodes array, so render the top-level message fields and
         # skip the per-node loop entirely.
         if template_obj.flow.get("mode") == FlowMode.DIRECT.value:
             self._render_direct_mode_flow(template_obj.flow, template_vars)
             logger.info(f"Rendered direct-mode flow for template {template_obj.name}")
-            return template_obj, template_vars
+            return template_obj, template_vars, _ds_messages
 
         # Get nodes from flow structure
         nodes = template_obj.flow.get("nodes", [])
@@ -274,4 +299,57 @@ async def load_template(
             node["role_messages"] = rendered_role_dicts
 
         logger.info(f"Rendered task messages for template {template_obj.name}")
-        return template_obj, template_vars
+        return template_obj, template_vars, _ds_messages
+
+    async def _fetch_one_data_source(self, ref: "DataSourceRef") -> str:
+        if not ref.is_active:
+            return DATA_SOURCE_UNAVAILABLE
+
+        try:
+            spreadsheet_id = ref.spreadsheet_url.split("/d/")[1].split("/")[0]
+        except (IndexError, AttributeError):
+            logger.warning(
+                "Data source '%s' has malformed spreadsheet_url: %s",
+                ref.name,
+                ref.spreadsheet_url,
+            )
+            return DATA_SOURCE_UNAVAILABLE
+
+        sheet = ref.sheet_name or "_first_"
+        cols = json.dumps(sorted(ref.columns or []))
+        cols_digest = hashlib.md5(cols.encode()).hexdigest()[:8]
+        cache_key = (
+            f"datasource:{spreadsheet_id}:{sheet}:{cols_digest}:{ref.format.value}"
+        )
+
+        try:
+            redis = await get_redis_service()
+            cached = await redis.get(cache_key)
+            if cached:
+                return cached
+        except Exception:
+            pass
+
+        try:
+            content = await asyncio.wait_for(
+                fetch_formatted(
+                    spreadsheet_id=spreadsheet_id,
+                    sheet_name=ref.sheet_name,
+                    columns=ref.columns,
+                    format=ref.format.value,
+                ),
+                timeout=0.8,
+            )
+        except asyncio.TimeoutError:
+            return DATA_SOURCE_UNAVAILABLE
+        except Exception:
+            return DATA_SOURCE_UNAVAILABLE
+
+        if content != DATA_SOURCE_UNAVAILABLE:
+            try:
+                redis = await get_redis_service()
+                await redis.setex(cache_key, content, ttl_seconds=60)
+            except Exception:
+                pass
+
+        return content

app/ai/voice/agents/breeze_buddy/template/types.py

@@ -2062,6 +2062,40 @@ def _default_supported_channels() -> List[Literal["voice", "chat"]]:
     return ["voice"]
 
 
+class DataSourceFormat(str, Enum):
+    MARKDOWN_TABLE = "markdown_table"
+    CSV = "csv"
+    JSON = "json"
+
+
+class DataSourceRef(BaseModel):
+    """Inline config for a data source attached to a template.
+
+    All sheet configuration lives inline — no separate ``data_source`` table.
+    Two templates sharing the same sheet each store the URL independently.
+    """
+
+    name: str = Field(description="Variable name used as {name} placeholder in prompts")
+    inject_as: Literal["var", "message"] = Field(
+        default="var",
+        description="'var' → template_vars[{name}] = content; 'message' → prepend system message",
+    )
+    spreadsheet_url: str = Field(
+        description="Full Google Sheets URL (shared with platform SA)"
+    )
+    sheet_name: Optional[str] = Field(
+        default=None, description="Tab name. NULL = first tab"
+    )
+    columns: Optional[List[str]] = Field(
+        default=None, description="Columns to include. NULL = all columns"
+    )
+    format: DataSourceFormat = Field(
+        default=DataSourceFormat.MARKDOWN_TABLE,
+        description="Output format: 'markdown_table' | 'csv' | 'json'",
+    )
+    is_active: bool = Field(default=True, description="Skip if false at call time")
+
+
 class TemplateModel(BaseModel):
     # Read-only fields (set by server, not editable via API).
     # These are intentionally excluded from ReplaceTemplateRequest so that
@@ -2081,6 +2115,10 @@ class TemplateModel(BaseModel):
     secrets: Optional[Dict[str, Any]] = None
     outbound_number_id: Optional[str] = None
     is_active: bool = True
+    data_sources: Optional[List[DataSourceRef]] = Field(
+        default=None,
+        description="Inline data source configs attached to this template",
+    )
     # Channels this template is allowed to be served on. Defaults to
     # voice-only so existing templates are unaffected. Add "chat" to
     # opt the template into the chat (text) mode flow build path
@@ -2129,6 +2167,7 @@ class CreateTemplateRequest(BaseModel):
     expected_callback_response_schema: Optional[Dict[str, Any]] = None
     configurations: Optional[ConfigurationModel] = None
     secrets: Optional[Dict[str, Any]] = None
+    data_sources: Optional[List[DataSourceRef]] = None
     supported_channels: List[Literal["voice", "chat"]] = Field(
         default_factory=_default_supported_channels,
         min_length=1,
@@ -2177,6 +2216,7 @@ class ReplaceTemplateRequest(BaseModel):
     expected_callback_response_schema: Optional[Dict[str, Any]] = None
     configurations: Optional[ConfigurationModel] = None
     secrets: Optional[Dict[str, Any]] = None
+    data_sources: Optional[List[DataSourceRef]] = None
     supported_channels: Optional[List[Literal["voice", "chat"]]] = Field(
         default=None,
         min_length=1,