{
  "nbformat": 4,
  "nbformat_minor": 5,
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "version": "3.13.0"
    },
    "blog_metadata": {
      "topic": "Fabric Data Agents Choosing Query Language by Context: The Governance Implications",
      "slug": "fabric-data-agents-choosing-query-language-by-context-the-go",
      "generated_by": "LinkedIn Post Generator + Azure OpenAI",
      "generated_at": "2026-07-10T13:48:11.322Z"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Fabric Data Agents Choosing Query Language by Context: The Governance Implications\n",
        "\n",
        "This notebook turns the blog post into a hands-on governance validation exercise. It demonstrates how query-path selection can be treated as a control-plane policy decision, with reproducible Python examples for routing, exception handling, provenance, and observability.\n",
        "\n",
        "The core idea is simple: in a multi-path Fabric environment, the important question is not just what an agent can do, but which execution path is approved."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "%pip install pandas matplotlib"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass, asdict\n",
        "from typing import List, Dict, Optional\n",
        "from datetime import datetime, timedelta\n",
        "import json\n",
        "import random\n",
        "import pandas as pd\n",
        "import matplotlib.pyplot as plt"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Policy flow as executable routing logic\n",
        "\n",
        "The blog post presents a governance pattern where a user question is evaluated by policy before any execution route is chosen. In practice, this means the control plane decides whether the request should go to a governed semantic model, a direct engine path by exception, or be denied for external access.\n",
        "\n",
        "The code below converts that routing idea into a simple Python policy function and tests it with several sample questions."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "\n",
        "@dataclass\n",
        "class PolicyRequest:\n",
        "    user: str\n",
        "    question: str\n",
        "    has_exception: bool = False\n",
        "    needs_external: bool = False\n",
        "\n",
        "\n",
        "def approved_path(req: PolicyRequest) -> str:\n",
        "    q = req.question.lower()\n",
        "    if req.needs_external or \"external api\" in q or \"internet\" in q or \"external\" in q:\n",
        "        return \"deny_external_access\"\n",
        "    if any(term in q for term in [\"raw sql\", \"kql\", \"spark sql\", \"warehouse table\", \"warehouse tables\"]):\n",
        "        return \"direct_engine_by_exception\" if req.has_exception else \"semantic_model_first\"\n",
        "    return \"semantic_model_first\"\n",
        "\n",
        "\n",
        "samples = [\n",
        "    PolicyRequest(\"analyst@contoso.com\", \"Compare gross margin by region from the semantic model\"),\n",
        "    PolicyRequest(\"engineer@contoso.com\", \"Run raw SQL against warehouse tables\", has_exception=False),\n",
        "    PolicyRequest(\"engineer@contoso.com\", \"Run raw SQL against warehouse tables\", has_exception=True),\n",
        "    PolicyRequest(\"research@contoso.com\", \"Use external API to enrich customer records\", needs_external=True),\n",
        "]\n",
        "\n",
        "results = [{\"user\": s.user, \"question\": s.question, \"approved_path\": approved_path(s)} for s in samples]\n",
        "pd.DataFrame(results)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Lightweight classifier from the blog post\n",
        "\n",
        "This is the blog's first policy prototype. It classifies a request into an approved execution path and shows that terms like \"raw SQL\" or \"KQL\" do not automatically win; they only route to a direct engine path when an approved exception exists."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "# Policy prototype: classify a question into an approved execution path\n",
        "from dataclasses import dataclass\n",
        "\n",
        "@dataclass\n",
        "class RequestContext:\n",
        "    user: str\n",
        "    question: str\n",
        "    has_exception: bool = False\n",
        "    needs_external: bool = False\n",
        "\n",
        "def classify_request(ctx: RequestContext) -> str:\n",
        "    q = ctx.question.lower()\n",
        "    if ctx.needs_external or \"external api\" in q or \"internet\" in q:\n",
        "        return \"deny_external_access\"\n",
        "    if any(term in q for term in [\"raw sql\", \"kql\", \"spark sql\", \"warehouse table\"]):\n",
        "        return \"direct_engine_by_exception\" if ctx.has_exception else \"semantic_model_first\"\n",
        "    return \"semantic_model_first\"\n",
        "\n",
        "ctx = RequestContext(user=\"analyst@contoso.com\", question=\"Compare sales by region from the semantic model\")\n",
        "print(classify_request(ctx))\n",
        "\n",
        "validation_cases = [\n",
        "    RequestContext(\"analyst@contoso.com\", \"Show gross margin by region\"),\n",
        "    RequestContext(\"engineer@contoso.com\", \"Run raw SQL on warehouse table sales_fact\", has_exception=False),\n",
        "    RequestContext(\"engineer@contoso.com\", \"Run raw SQL on warehouse table sales_fact\", has_exception=True),\n",
        "    RequestContext(\"research@contoso.com\", \"Use external API for enrichment\", needs_external=True),\n",
        "    RequestContext(\"ops@contoso.com\", \"Query KQL for telemetry\", has_exception=True),\n",
        "]\n",
        "\n",
        "pd.DataFrame([\n",
        "    {\n",
        "        \"user\": c.user,\n",
        "        \"question\": c.question,\n",
        "        \"has_exception\": c.has_exception,\n",
        "        \"needs_external\": c.needs_external,\n",
        "        \"path\": classify_request(c),\n",
        "    }\n",
        "    for c in validation_cases\n",
        "])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Attach governance reasons and controls to the route\n",
        "\n",
        "Classification alone is not enough for governance. A useful policy engine should also explain why a path was chosen and which controls apply, such as certified metrics, lineage, approval IDs, or outbound access review."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "# Policy prototype: attach governance reasons and controls to the chosen path\n",
        "from dataclasses import dataclass\n",
        "\n",
        "@dataclass\n",
        "class Decision:\n",
        "    path: str\n",
        "    reason: str\n",
        "    controls: list[str]\n",
        "\n",
        "def evaluate(path: str) -> Decision:\n",
        "    if path == \"semantic_model_first\":\n",
        "        return Decision(path, \"Default governed route\", [\"RLS/OLS\", \"Certified metrics\", \"Lineage\"])\n",
        "    if path == \"direct_engine_by_exception\":\n",
        "        return Decision(path, \"Approved exception for engine-native query\", [\"Approval ID\", \"Scoped dataset\", \"Audit logging\"])\n",
        "    return Decision(\"deny_external_access\", \"Outbound access not approved\", [\"Block execution\", \"Review workspace settings\"])\n",
        "\n",
        "decision = evaluate(\"direct_engine_by_exception\")\n",
        "print(decision)\n",
        "\n",
        "paths = [\"semantic_model_first\", \"direct_engine_by_exception\", \"deny_external_access\"]\n",
        "decision_df = pd.DataFrame([\n",
        "    {\n",
        "        \"path\": evaluate(p).path,\n",
        "        \"reason\": evaluate(p).reason,\n",
        "        \"controls\": \", \".join(evaluate(p).controls),\n",
        "    }\n",
        "    for p in paths\n",
        "])\n",
        "decision_df"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## End-to-end evaluation for multiple user questions\n",
        "\n",
        "The next example runs a small batch of requests through a routing function. This is useful for validating whether your default hierarchy really keeps business questions on the semantic path and only allows lower-level engine access when policy explicitly permits it."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "# Policy prototype: end-to-end evaluation for multiple user questions\n",
        "from dataclasses import dataclass\n",
        "\n",
        "@dataclass\n",
        "class Request:\n",
        "    question: str\n",
        "    has_exception: bool\n",
        "    needs_external: bool\n",
        "\n",
        "def route(req: Request) -> str:\n",
        "    q = req.question.lower()\n",
        "    if req.needs_external or \"call external\" in q:\n",
        "        return \"deny_external_access\"\n",
        "    if \"sql\" in q or \"kql\" in q:\n",
        "        return \"direct_engine_by_exception\" if req.has_exception else \"semantic_model_first\"\n",
        "    return \"semantic_model_first\"\n",
        "\n",
        "samples = [\n",
        "    Request(\"Show margin by product\", False, False),\n",
        "    Request(\"Run raw SQL against warehouse tables\", True, False),\n",
        "    Request(\"Call external enrichment API for customer data\", False, True),\n",
        "]\n",
        "\n",
        "for s in samples:\n",
        "    print({\"question\": s.question, \"path\": route(s)})\n",
        "\n",
        "batch_df = pd.DataFrame([\n",
        "    {\"question\": s.question, \"has_exception\": s.has_exception, \"needs_external\": s.needs_external, \"path\": route(s)}\n",
        "    for s in samples\n",
        "])\n",
        "batch_df"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Simulate the sequence of user, agent, policy engine, and governance log\n",
        "\n",
        "The blog also describes a sequence where the user asks a question, the agent asks the policy engine for an approved route, the selected engine executes, and the decision is written to a governance log. The code below simulates that sequence in Python and produces answer provenance for each request."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from dataclasses import dataclass\n",
        "from datetime import datetime\n",
        "\n",
        "@dataclass\n",
        "class ProvenanceRecord:\n",
        "    user: str\n",
        "    question: str\n",
        "    policy_decision: str\n",
        "    engine_used: str\n",
        "    source_used: str\n",
        "    semantic_layer: bool\n",
        "    timestamp: str\n",
        "    exception_id: Optional[str] = None\n",
        "    fallback: Optional[str] = None\n",
        "\n",
        "\n",
        "def execute_request(user: str, question: str, has_exception: bool = False, needs_external: bool = False) -> ProvenanceRecord:\n",
        "    ctx = RequestContext(user=user, question=question, has_exception=has_exception, needs_external=needs_external)\n",
        "    path = classify_request(ctx)\n",
        "\n",
        "    if path == \"semantic_model_first\":\n",
        "        return ProvenanceRecord(\n",
        "            user=user,\n",
        "            question=question,\n",
        "            policy_decision=path,\n",
        "            engine_used=\"semantic_query_engine\",\n",
        "            source_used=\"certified_semantic_model\",\n",
        "            semantic_layer=True,\n",
        "            timestamp=datetime.utcnow().isoformat(),\n",
        "        )\n",
        "    if path == \"direct_engine_by_exception\":\n",
        "        return ProvenanceRecord(\n",
        "            user=user,\n",
        "            question=question,\n",
        "            policy_decision=path,\n",
        "            engine_used=\"warehouse_sql_engine\",\n",
        "            source_used=\"warehouse.sales_fact\",\n",
        "            semantic_layer=False,\n",
        "            timestamp=datetime.utcnow().isoformat(),\n",
        "            exception_id=\"EXC-2026-001\",\n",
        "        )\n",
        "    return ProvenanceRecord(\n",
        "        user=user,\n",
        "        question=question,\n",
        "        policy_decision=\"deny_external_access\",\n",
        "        engine_used=\"none\",\n",
        "        source_used=\"none\",\n",
        "        semantic_layer=False,\n",
        "        timestamp=datetime.utcnow().isoformat(),\n",
        "    )\n",
        "\n",
        "requests = [\n",
        "    (\"analyst@contoso.com\", \"Gross margin by region\", False, False),\n",
        "    (\"engineer@contoso.com\", \"Run raw SQL against warehouse table sales_fact\", True, False),\n",
        "    (\"research@contoso.com\", \"Use external API for competitor pricing\", False, True),\n",
        "]\n",
        "\n",
        "records = [asdict(execute_request(*r)) for r in requests]\n",
        "pd.DataFrame(records)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Governance evidence: workspace outbound access protection settings\n",
        "\n",
        "The original post included PowerShell examples for documenting outbound access protection and approved destinations. Since this notebook uses Python, the next cells recreate the same governance evidence as Python objects and tables."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "workspace = {\n",
        "    \"Name\": \"Finance-Analytics\",\n",
        "    \"Id\": \"ws-001\",\n",
        "    \"OutboundAccessProtection\": \"Enabled\",\n",
        "    \"AllowedDestinations\": [\"contoso-sql.database.windows.net\", \"api.internal.contoso.com\"],\n",
        "}\n",
        "\n",
        "evidence = {\n",
        "    \"WorkspaceName\": workspace[\"Name\"],\n",
        "    \"WorkspaceId\": workspace[\"Id\"],\n",
        "    \"OutboundAccessProtection\": workspace[\"OutboundAccessProtection\"],\n",
        "    \"AllowedDestinations\": \"; \".join(workspace[\"AllowedDestinations\"]),\n",
        "    \"ReviewedOn\": datetime.utcnow().isoformat(timespec=\"seconds\"),\n",
        "}\n",
        "\n",
        "pd.DataFrame([evidence])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Governance evidence: review approved and denied data connection rules\n",
        "\n",
        "This example mirrors the blog's rule review pattern. It gives you a lightweight way to inspect which destinations are approved, which are denied, and whether your external access posture matches policy."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "rules = [\n",
        "    {\"RuleName\": \"AllowInternalSQL\", \"Type\": \"SQL\", \"Destination\": \"contoso-sql.database.windows.net\", \"Status\": \"Approved\"},\n",
        "    {\"RuleName\": \"AllowInternalAPI\", \"Type\": \"HTTPS\", \"Destination\": \"api.internal.contoso.com\", \"Status\": \"Approved\"},\n",
        "    {\"RuleName\": \"BlockPublicStorage\", \"Type\": \"HTTPS\", \"Destination\": \"*.blob.core.windows.net\", \"Status\": \"Denied\"},\n",
        "]\n",
        "\n",
        "rules_df = pd.DataFrame(rules).sort_values([\"Status\", \"RuleName\"]).reset_index(drop=True)\n",
        "rules_df"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Governance evidence: export a workspace governance snapshot to JSON\n",
        "\n",
        "A practical governance program needs durable evidence. This cell exports a small workspace governance snapshot to a JSON file so you can validate how review metadata, outbound protection, and connection rules might be captured for audit."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "workspace_review = {\n",
        "    \"Workspace\": \"Finance-Analytics\",\n",
        "    \"ReviewedBy\": \"admin@contoso.com\",\n",
        "    \"ReviewedOn\": datetime.utcnow().strftime(\"%Y-%m-%d\"),\n",
        "    \"OutboundAccessProtection\": \"Enabled\",\n",
        "    \"DataConnectionRules\": [\n",
        "        {\"Name\": \"AllowInternalSQL\", \"Status\": \"Approved\"},\n",
        "        {\"Name\": \"BlockPublicStorage\", \"Status\": \"Denied\"},\n",
        "    ],\n",
        "}\n",
        "\n",
        "path = \"workspace-governance-review.json\"\n",
        "with open(path, \"w\", encoding=\"utf-8\") as f:\n",
        "    json.dump(workspace_review, f, indent=2)\n",
        "\n",
        "import os\n",
        "pd.DataFrame([\n",
        "    {\n",
        "        \"FullName\": os.path.abspath(path),\n",
        "        \"Length\": os.path.getsize(path),\n",
        "        \"LastWriteTime\": datetime.fromtimestamp(os.path.getmtime(path)).isoformat(timespec=\"seconds\"),\n",
        "    }\n",
        "])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Reproduce the routing problem described in the finance review\n",
        "\n",
        "The blog describes a case where the same gross margin question produced two different answers because one path used a certified semantic model and another used direct warehouse tables with a different returns filter. The following example recreates that mismatch with simple sample data."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "sales = pd.DataFrame([\n",
        "    {\"region\": \"North\", \"revenue\": 1000, \"cost\": 600, \"returns\": 50},\n",
        "    {\"region\": \"South\", \"revenue\": 1200, \"cost\": 700, \"returns\": 120},\n",
        "    {\"region\": \"West\", \"revenue\": 900, \"cost\": 500, \"returns\": 30},\n",
        "])\n",
        "\n",
        "# Governed semantic definition: gross margin uses net revenue after returns.\n",
        "semantic_answer = sales.assign(net_revenue=sales[\"revenue\"] - sales[\"returns\"])\n",
        "semantic_answer[\"gross_margin\"] = semantic_answer[\"net_revenue\"] - semantic_answer[\"cost\"]\n",
        "semantic_result = semantic_answer[[\"region\", \"gross_margin\"]].copy()\n",
        "semantic_result[\"path\"] = \"semantic_model_first\"\n",
        "\n",
        "# Direct warehouse path with a different filter/definition: returns ignored.\n",
        "warehouse_answer = sales.copy()\n",
        "warehouse_answer[\"gross_margin\"] = warehouse_answer[\"revenue\"] - warehouse_answer[\"cost\"]\n",
        "warehouse_result = warehouse_answer[[\"region\", \"gross_margin\"]].copy()\n",
        "warehouse_result[\"path\"] = \"direct_engine_by_exception\"\n",
        "\n",
        "comparison = pd.concat([semantic_result, warehouse_result], ignore_index=True)\n",
        "comparison.sort_values([\"region\", \"path\"]).reset_index(drop=True)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Quantify the semantic-versus-direct variance\n",
        "\n",
        "Once different routes can produce different answers, governance needs a way to measure the variance. This cell compares the semantic and direct-engine answers side by side and calculates the difference by region."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "semantic_only = semantic_result[[\"region\", \"gross_margin\"]].rename(columns={\"gross_margin\": \"semantic_gross_margin\"})\n",
        "warehouse_only = warehouse_result[[\"region\", \"gross_margin\"]].rename(columns={\"gross_margin\": \"warehouse_gross_margin\"})\n",
        "variance = semantic_only.merge(warehouse_only, on=\"region\")\n",
        "variance[\"difference\"] = variance[\"warehouse_gross_margin\"] - variance[\"semantic_gross_margin\"]\n",
        "variance"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Model a governance-first routing hierarchy\n",
        "\n",
        "The blog recommends a clear hierarchy: semantic model first, ontology-backed meaning where available, direct engine only by approved exception, notebook execution only for bounded workflows, and external access denied by default. The next function encodes that hierarchy more explicitly."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "@dataclass\n",
        "class GovernanceContext:\n",
        "    user: str\n",
        "    question: str\n",
        "    metric_type: str = \"business_metric\"\n",
        "    ontology_available: bool = True\n",
        "    has_exception: bool = False\n",
        "    notebook_allowed: bool = False\n",
        "    needs_external: bool = False\n",
        "\n",
        "\n",
        "def route_with_hierarchy(ctx: GovernanceContext) -> Dict[str, str]:\n",
        "    q = ctx.question.lower()\n",
        "\n",
        "    if ctx.needs_external or \"external\" in q or \"internet\" in q:\n",
        "        return {\"path\": \"deny_external_access\", \"reason\": \"External access denied by default\"}\n",
        "\n",
        "    if ctx.metric_type == \"business_metric\":\n",
        "        if ctx.ontology_available:\n",
        "            return {\"path\": \"semantic_model_first\", \"reason\": \"Business metric with ontology-backed meaning\"}\n",
        "        return {\"path\": \"semantic_model_first\", \"reason\": \"Business metric defaults to semantic model\"}\n",
        "\n",
        "    if any(term in q for term in [\"notebook\", \"python workflow\", \"feature engineering\"]):\n",
        "        if ctx.notebook_allowed:\n",
        "            return {\"path\": \"notebook_bounded_workflow\", \"reason\": \"Approved bounded analytical workflow\"}\n",
        "        return {\"path\": \"semantic_model_first\", \"reason\": \"Notebook path not approved; fallback to governed route\"}\n",
        "\n",
        "    if any(term in q for term in [\"sql\", \"kql\", \"spark sql\", \"warehouse\"]):\n",
        "        if ctx.has_exception:\n",
        "            return {\"path\": \"direct_engine_by_exception\", \"reason\": \"Approved exception for engine-native access\"}\n",
        "        return {\"path\": \"semantic_model_first\", \"reason\": \"No exception; use governed semantic path\"}\n",
        "\n",
        "    return {\"path\": \"semantic_model_first\", \"reason\": \"Default governed route\"}\n",
        "\n",
        "hierarchy_cases = [\n",
        "    GovernanceContext(\"analyst@contoso.com\", \"Gross margin by region\", metric_type=\"business_metric\", ontology_available=True),\n",
        "    GovernanceContext(\"scientist@contoso.com\", \"Run notebook for feature engineering\", metric_type=\"advanced_analysis\", notebook_allowed=True),\n",
        "    GovernanceContext(\"engineer@contoso.com\", \"Run SQL on warehouse\", metric_type=\"advanced_analysis\", has_exception=True),\n",
        "    GovernanceContext(\"research@contoso.com\", \"Use external internet source\", metric_type=\"advanced_analysis\", needs_external=True),\n",
        "]\n",
        "\n",
        "pd.DataFrame([\n",
        "    {\"user\": c.user, \"question\": c.question, **route_with_hierarchy(c)}\n",
        "    for c in hierarchy_cases\n",
        "])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Make fallback behavior explicit\n",
        "\n",
        "A key governance point in the blog is that fallback behavior should never be left to model improvisation. This example shows a deterministic fallback policy: retry the semantic path once, then either return a governed message or route to a lower-level engine only if an approved exception exists."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "def execute_with_fallback(question: str, semantic_available: bool, has_exception: bool) -> Dict[str, str]:\n",
        "    if semantic_available:\n",
        "        return {\n",
        "            \"final_path\": \"semantic_model_first\",\n",
        "            \"fallback\": \"none\",\n",
        "            \"message\": \"Answered using certified semantic model\",\n",
        "        }\n",
        "\n",
        "    if has_exception:\n",
        "        return {\n",
        "            \"final_path\": \"direct_engine_by_exception\",\n",
        "            \"fallback\": \"semantic_failed_then_exception_route\",\n",
        "            \"message\": \"Semantic path unavailable; used approved direct engine fallback\",\n",
        "        }\n",
        "\n",
        "    return {\n",
        "        \"final_path\": \"cannot_answer_governed\",\n",
        "        \"fallback\": \"semantic_failed_no_exception\",\n",
        "        \"message\": \"Cannot answer under current governance policy\",\n",
        "    }\n",
        "\n",
        "fallback_cases = [\n",
        "    {\"question\": \"Gross margin by region\", \"semantic_available\": True, \"has_exception\": False},\n",
        "    {\"question\": \"Gross margin by region\", \"semantic_available\": False, \"has_exception\": True},\n",
        "    {\"question\": \"Gross margin by region\", \"semantic_available\": False, \"has_exception\": False},\n",
        "]\n",
        "\n",
        "pd.DataFrame([{**c, **execute_with_fallback(**c)} for c in fallback_cases])"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Bind exceptions to scope and time\n",
        "\n",
        "The operating model in the blog recommends that exceptions be tied to a named use case, owner, source scope, review date, and logging requirements. The code below creates a simple exception register and validates whether an exception is still active."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "exception_register = pd.DataFrame([\n",
        "    {\n",
        "        \"exception_id\": \"EXC-2026-001\",\n",
        "        \"use_case\": \"Finance reconciliation\",\n",
        "        \"owner\": \"finance.data@contoso.com\",\n",
        "        \"source_scope\": \"warehouse.sales_fact\",\n",
        "        \"review_date\": (datetime.utcnow() + timedelta(days=30)).date().isoformat(),\n",
        "        \"logging_required\": True,\n",
        "    },\n",
        "    {\n",
        "        \"exception_id\": \"EXC-2025-099\",\n",
        "        \"use_case\": \"Legacy telemetry investigation\",\n",
        "        \"owner\": \"ops.data@contoso.com\",\n",
        "        \"source_scope\": \"kql.telemetry_db\",\n",
        "        \"review_date\": (datetime.utcnow() - timedelta(days=10)).date().isoformat(),\n",
        "        \"logging_required\": True,\n",
        "    },\n",
        "])\n",
        "\n",
        "exception_register[\"active\"] = pd.to_datetime(exception_register[\"review_date\"]) >= pd.Timestamp.utcnow().tz_localize(None)\n",
        "exception_register"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Require answer provenance for every response\n",
        "\n",
        "The blog argues that every answer should surface source used, engine used, semantic layer involvement, policy decision, timestamp, and exception ID if applicable. This cell generates a small provenance table that can be attached to responses or stored in a governance log."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "provenance_log = pd.DataFrame([\n",
        "    {\n",
        "        \"question\": \"Gross margin by region\",\n",
        "        \"source_used\": \"certified_semantic_model\",\n",
        "        \"engine_used\": \"semantic_query_engine\",\n",
        "        \"semantic_layer\": True,\n",
        "        \"policy_decision\": \"semantic_model_first\",\n",
        "        \"timestamp\": datetime.utcnow().isoformat(timespec=\"seconds\"),\n",
        "        \"exception_id\": None,\n",
        "    },\n",
        "    {\n",
        "        \"question\": \"Run raw SQL against warehouse tables\",\n",
        "        \"source_used\": \"warehouse.sales_fact\",\n",
        "        \"engine_used\": \"warehouse_sql_engine\",\n",
        "        \"semantic_layer\": False,\n",
        "        \"policy_decision\": \"direct_engine_by_exception\",\n",
        "        \"timestamp\": datetime.utcnow().isoformat(timespec=\"seconds\"),\n",
        "        \"exception_id\": \"EXC-2026-001\",\n",
        "    },\n",
        "    {\n",
        "        \"question\": \"Use external API for enrichment\",\n",
        "        \"source_used\": \"none\",\n",
        "        \"engine_used\": \"none\",\n",
        "        \"semantic_layer\": False,\n",
        "        \"policy_decision\": \"deny_external_access\",\n",
        "        \"timestamp\": datetime.utcnow().isoformat(timespec=\"seconds\"),\n",
        "        \"exception_id\": None,\n",
        "    },\n",
        "])\n",
        "\n",
        "provenance_log"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Review route telemetry monthly, not prompts\n",
        "\n",
        "The post recommends monitoring route telemetry rather than focusing only on prompts. This example creates synthetic telemetry for route usage, latency, and cost so you can inspect whether direct-engine usage is rising, fallbacks are increasing, or expensive paths are becoming common."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "random.seed(42)\n",
        "route_types = [\"semantic_model_first\", \"direct_engine_by_exception\", \"notebook_bounded_workflow\", \"deny_external_access\"]\n",
        "telemetry = []\n",
        "for day in range(1, 31):\n",
        "    for _ in range(20):\n",
        "        route = random.choices(route_types, weights=[0.7, 0.15, 0.1, 0.05])[0]\n",
        "        telemetry.append({\n",
        "            \"day\": day,\n",
        "            \"route\": route,\n",
        "            \"latency_seconds\": round(random.uniform(1, 3) if route == \"semantic_model_first\" else random.uniform(3, 12), 2),\n",
        "            \"cost_units\": round(random.uniform(1, 2) if route == \"semantic_model_first\" else random.uniform(3, 8), 2),\n",
        "            \"fallback\": random.choice([\"none\", \"semantic_failed_then_exception_route\"]) if route == \"direct_engine_by_exception\" else \"none\",\n",
        "        })\n",
        "\n",
        "telemetry_df = pd.DataFrame(telemetry)\n",
        "telemetry_df.head()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Analyze route usage, latency, and cost\n",
        "\n",
        "This cell summarizes the telemetry so you can validate the governance concerns from the blog: rising direct-engine usage, slowest paths, most frequent fallbacks, and highest-cost workloads."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "summary = telemetry_df.groupby(\"route\").agg(\n",
        "    requests=(\"route\", \"size\"),\n",
        "    avg_latency_seconds=(\"latency_seconds\", \"mean\"),\n",
        "    avg_cost_units=(\"cost_units\", \"mean\"),\n",
        ").reset_index().sort_values(\"requests\", ascending=False)\n",
        "\n",
        "fallback_summary = telemetry_df[telemetry_df[\"fallback\"] != \"none\"].groupby(\"fallback\").size().reset_index(name=\"count\")\n",
        "\n",
        "print(\"Route summary:\")\n",
        "display(summary)\n",
        "print(\"Fallback summary:\")\n",
        "display(fallback_summary)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Visualize route distribution and average cost\n",
        "\n",
        "A quick chart makes it easier to spot whether your environment is drifting away from semantic-first behavior. The first chart shows route frequency, and the second shows average cost by route."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "fig, axes = plt.subplots(1, 2, figsize=(12, 4))\n",
        "\n",
        "summary.plot(kind=\"bar\", x=\"route\", y=\"requests\", ax=axes[0], legend=False, title=\"Route Usage\")\n",
        "summary.plot(kind=\"bar\", x=\"route\", y=\"avg_cost_units\", ax=axes[1], legend=False, title=\"Average Cost by Route\", color=\"orange\")\n",
        "\n",
        "axes[0].set_ylabel(\"Requests\")\n",
        "axes[1].set_ylabel(\"Cost Units\")\n",
        "plt.tight_layout()\n",
        "plt.show()"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Simple governance scorecard from 1 to 5\n",
        "\n",
        "The blog closes by asking teams to rate their query-routing governance and identify where it breaks: semantic authority, exception control, or observability. This cell creates a simple scorecard and computes an overall maturity rating."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "scorecard = {\n",
        "    \"semantic_authority\": 4,\n",
        "    \"exception_control\": 3,\n",
        "    \"observability\": 3,\n",
        "    \"fallback_governance\": 4,\n",
        "    \"least_privilege_defaults\": 5,\n",
        "}\n",
        "\n",
        "score_df = pd.DataFrame(list(scorecard.items()), columns=[\"dimension\", \"score\"])\n",
        "overall_score = round(score_df[\"score\"].mean(), 2)\n",
        "weakest_area = score_df.sort_values(\"score\").iloc[0][\"dimension\"]\n",
        "\n",
        "print(f\"Overall governance score: {overall_score} / 5\")\n",
        "print(f\"Weakest area: {weakest_area}\")\n",
        "score_df.sort_values(\"score\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summary\n",
        "\n",
        "This notebook validated the blog's central claim: query-language selection is not just a developer convenience, it is a governance decision. By modeling routing as a control-plane policy, you can keep semantic models authoritative, constrain direct engine access to approved exceptions, deny unapproved external access, and capture provenance for every answer.\n",
        "\n",
        "## Next Steps\n",
        "\n",
        "1. Replace the toy classifiers with your real routing policy rules.\n",
        "2. Connect provenance logging to your operational telemetry store.\n",
        "3. Add exception expiry checks and owner notifications.\n",
        "4. Measure route drift monthly to ensure semantic-first remains the default.\n",
        "5. Review your current maturity score and identify whether the main gap is semantic authority, exception control, or observability."
      ]
    }
  ]
}