app.py

#!/usr/bin/env python3
"""A dependency-free, production-shaped Universal SQL prototype."""

import json
import os
import re
import threading
import time
import uuid
from collections import defaultdict
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlencode, urlparse
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from services.connector_client import RemoteConnectorClient, RemoteConnectorError
from services.control_plane_client import EntitlementHTTPClient, RateLimitHTTPClient
from services.entitlement_service import EntitlementService, EntitlementError
from services.rate_limit_service import InMemoryTokenBucketRateLimiter, RateLimitRequest
from services.telemetry import Telemetry

ROOT = Path(__file__).parent


class QueryError(Exception):
    def __init__(self, code, message, status=400, **extra):
        super().__init__(message)
        self.code, self.message, self.status, self.extra = code, message, status, extra


class AuthService:
    """Local OIDC-shaped authenticator; production replaces this with JWT validation."""

    def authenticate(self, request):
        user = request.get("user")
        authorization = request.get("authorization", "")
        if not user and authorization.lower().startswith("bearer demo-"):
            user = authorization[12:]
        if not user:
            raise QueryError("ENTITLEMENT_DENIED", "A user identity or OIDC bearer token is required.", 401)
        return user


class TenantRegistry:
    def __init__(self, policy_data):
        self.tenants = {tenant_id: {"status": "active"} for tenant_id in policy_data.get("tenants", {})}

    def ensure_active(self, tenant_id):
        if self.tenants.get(tenant_id, {}).get("status") != "active":
            raise QueryError("TENANT_DISABLED", f"Tenant {tenant_id} is disabled.", 403)


class MetadataCatalog:
    def __init__(self):
        self.connectors = {}

    def register(self, connector):
        self.connectors[connector.name] = {"table": connector.table, "capabilities": connector.capabilities}

    def describe(self, source):
        return self.connectors.get(source)


class SecretKeyStore:
    """Tenant-scoped key references; plaintext secrets never enter the prototype."""

    def tenant_key_ref(self, tenant_id):
        return f"kms://tenant/{tenant_id}/active"


class AuditLog:
    def __init__(self):
        self.events = []

    def record(self, **event):
        self.events.append({"at": time.time(), **event})


class Observability:
    def __init__(self, service_name="universal-sql"):
        self.telemetry = Telemetry(service_name=service_name)
        self.metrics = self.telemetry.counters

    def increment(self, name, labels=None):
        self.telemetry.increment(name, labels=labels)

    def start_span(self, name, attributes=None, traceparent=None):
        return self.telemetry.start_span(name, attributes, traceparent=traceparent)

    def observe(self, name, value, labels=None):
        self.telemetry.observe(name, value, labels=labels)

    def metrics_text(self):
        return self.telemetry.metrics_text()


class AsyncJobRunner:
    def __init__(self):
        self.jobs = {}

    def submit(self, request, reason):
        job_id = f"job_{uuid.uuid4().hex[:12]}"
        self.jobs[job_id] = {"id": job_id, "status": "queued", "reason": reason, "request": request}
        return self.jobs[job_id]

    def get(self, job_id):
        return self.jobs.get(job_id)


class TokenBucket:
    def __init__(self, capacity, refill_per_sec):
        self.capacity = capacity
        self.refill_per_sec = refill_per_sec
        self.tokens = float(capacity)
        self.updated_at = time.time()

    def take(self):
        now = time.time()
        self.tokens = min(self.capacity, self.tokens + (now - self.updated_at) * self.refill_per_sec)
        self.updated_at = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True, 0
        retry_ms = int(((1 - self.tokens) / self.refill_per_sec) * 1000) if self.refill_per_sec else 0
        return False, retry_ms


class TenantFairScheduler:
    """Weighted deficit round-robin with a per-tenant in-flight cap."""

    def __init__(self, weights=None, max_inflight=32):
        self.weights = weights or {}
        self.max_inflight = max_inflight
        self.queues = defaultdict(list)
        self.deficits = defaultdict(int)
        self.inflight = defaultdict(int)
        self.order = list(self.weights)

    def enqueue(self, tenant_id, request_id, cost=1):
        if tenant_id not in self.weights:
            self.weights[tenant_id] = 1
            self.order.append(tenant_id)
        self.queues[tenant_id].append({"tenant_id": tenant_id, "request_id": request_id, "cost": max(1, cost)})

    def next(self):
        if not self.order or not any(self.queues.values()):
            return None
        for tenant_id in self.order:
            self.deficits[tenant_id] += self.weights[tenant_id]
        candidates = [tenant_id for tenant_id in self.order if self.queues[tenant_id] and self.inflight[tenant_id] < self.max_inflight]
        if not candidates:
            return None
        tenant_id = max(candidates, key=lambda item: (self.deficits[item], -self.order.index(item)))
        request = self.queues[tenant_id][0]
        if self.deficits[tenant_id] < request["cost"]:
            return None
        self.queues[tenant_id].pop(0)
        self.deficits[tenant_id] -= request["cost"]
        self.inflight[tenant_id] += 1
        return request

    def complete(self, tenant_id):
        self.inflight[tenant_id] = max(0, self.inflight[tenant_id] - 1)


class RateLimitService:
    def __init__(self, capacity, refill_per_sec, external_client=None):
        self.capacity, self.refill_per_sec = capacity, refill_per_sec
        self.external_client = external_client
        self.external_remaining = {}
        self.buckets = defaultdict(lambda: TokenBucket(capacity, refill_per_sec))
        self.scheduler = TenantFairScheduler(max_inflight=32)
        # The contract-backed adapter is enabled for positive refill rates.
        # Zero-refill mode remains on the legacy bucket because it is used by
        # deterministic tests and is not representable by the contract's
        # positive refill invariant.
        self.contract = (
            InMemoryTokenBucketRateLimiter(capacity, refill_per_sec)
            if refill_per_sec > 0 else None
        )

    def admit(self, tenant_id, user_id, connector, request_id, cost=1):
        self.scheduler.enqueue(tenant_id, request_id, cost=cost)
        scheduled = self.scheduler.next()
        if not scheduled:
            return False, 0
        scope = f"{tenant_id}:{user_id}"
        if self.external_client is not None:
            status, response = self.external_client.admit({
                "connector": connector,
                "tenant": tenant_id,
                "user": user_id,
                "tokens": cost,
            })
            if status != 200 or not isinstance(response, dict) or not isinstance(response.get("allowed"), bool):
                self.scheduler.complete(tenant_id)
                retry_ms = response.get("error", {}).get("retry_after_ms", 1000) if isinstance(response, dict) else 1000
                return False, int(retry_ms or 1000)
            self.external_remaining[scope] = int(response.get("remaining_tokens", 0))
            allowed, retry_ms = bool(response["allowed"]), int(response.get("retry_after_ms", 0))
        elif self.contract is None:
            allowed, retry_ms = self.take(scope)
        else:
            # Keep the compatibility-visible legacy bucket synchronized so
            # callers and existing operational tests can still inspect it.
            legacy_bucket = self.buckets[scope]
            now = time.time()
            elapsed = max(0.0, now - legacy_bucket.updated_at)
            legacy_bucket.tokens = min(
                self.capacity,
                legacy_bucket.tokens + elapsed * self.refill_per_sec,
            )
            legacy_bucket.updated_at = now
            contract_key = (connector, tenant_id, user_id)
            self.contract._buckets[contract_key] = (
                max(0.0, legacy_bucket.tokens), time.monotonic()
            )
            decision = self.contract.admit(
                RateLimitRequest(connector=connector, tenant=tenant_id, user=user_id, tokens=cost)
            )
            allowed, retry_ms = decision.allowed, decision.retry_after_ms
            legacy_bucket.tokens = float(decision.remaining_tokens)
            legacy_bucket.updated_at = now
        self.scheduler.complete(tenant_id)
        return allowed, retry_ms

    def take(self, scope):
        return self.buckets[scope].take()

    def remaining(self, scope=None):
        if scope is not None:
            if self.external_client is not None:
                return self.external_remaining.get(scope, 0)
            return int(self.buckets[scope].tokens)
        return sum(int(bucket.tokens) for bucket in self.buckets.values())


class FreshnessCache:
    def __init__(self):
        self.entries = {}

    def get(self, key, max_staleness_ms):
        entry = self.entries.get(key)
        if not entry:
            return None
        age_ms = int((time.time() - entry["fetched_at"]) * 1000)
        if age_ms <= max_staleness_ms:
            return entry["rows"], age_ms
        return None

    def put(self, key, rows):
        self.entries[key] = {"rows": rows, "fetched_at": time.time()}


class FetchStore:
    """Raw tenant-scoped source snapshots; authorization is never stored in the rows."""

    def __init__(self):
        self.entries = {}

    def refresh(self, source, rows):
        self.entries[source] = {"rows": [dict(row) for row in rows], "refreshed_at": time.time()}

    def get(self, source, max_staleness_ms):
        entry = self.entries.get(source)
        if not entry:
            raise QueryError("STALE_DATA", f"Fetch store has no snapshot for {source}; refresh the fetch store first.")
        age_ms = int((time.time() - entry["refreshed_at"]) * 1000)
        if age_ms > max_staleness_ms:
            raise QueryError("STALE_DATA", f"Fetch store snapshot for {source} is {age_ms} ms old; refresh or increase max_staleness_ms.", 409, source=source, freshness_ms=age_ms)
        return entry["rows"], age_ms


class MaterializationStore:
    """Short-lived query materializations; fetch store is the durable-ish captured-data path."""

    def __init__(self, ttl_ms=300_000):
        self.ttl_ms = ttl_ms
        self.entries = {}

    def put(self, key, rows):
        self.entries[key] = {"rows": rows, "created_at": time.time()}

    def get(self, key):
        entry = self.entries.get(key)
        if not entry or (time.time() - entry["created_at"]) * 1000 > self.ttl_ms:
            return None
        return entry["rows"]


class FreshnessLayer:
    def __init__(self, cache, fetch_store):
        self.cache, self.fetch_store = cache, fetch_store


class EntitlementSnapshotStore:
    def __init__(self):
        self.entries = {}

    def get(self, key, max_age_ms):
        entry = self.entries.get(key)
        if not entry:
            return None
        age_ms = int((time.time() - entry["checked_at"]) * 1000)
        if age_ms <= max_age_ms:
            return dict(entry["entitlements"]), age_ms
        return None

    def put(self, key, entitlements):
        self.entries[key] = {"entitlements": dict(entitlements), "checked_at": time.time()}


class EntitlementPolicy:
    def __init__(self, path=ROOT / "policy.json"):
        self.data = json.loads(Path(path).read_text())

    def principal(self, username):
        principal = self.data["users"].get(username)
        if not principal:
            raise QueryError("ENTITLEMENT_DENIED", "Unknown user or tenant policy; access is denied.", 403)
        return {**principal, "user_id": username}

    def check_sources(self, principal, sources):
        for source in sources:
            scope = f"{source}:read"
            if scope not in principal["scopes"]:
                raise QueryError("ENTITLEMENT_DENIED", f"Missing required source scope: {scope}.", 403)

    def live_source_entitlements(self, principal, source):
        """Represents a live permission/membership check against the source."""
        if f"{source}:read" not in principal["scopes"]:
            raise QueryError("ENTITLEMENT_DENIED", f"Live check denied source scope: {source}:read.", 403)
        if source == "github":
            return {"allowed_repos": list(principal.get("allowed_repos", [])), "scopes": list(principal["scopes"])}
        if source == "jira":
            return {"allowed_projects": list(principal.get("allowed_projects", [])), "scopes": list(principal["scopes"])}
        raise QueryError("ENTITLEMENT_DENIED", f"No entitlement contract exists for {source}.", 403)

    def row_allowed(self, principal, source, row):
        if source == "github":
            return row.get("repo") in principal.get("allowed_repos", [])
        if source == "jira":
            return row.get("project") in principal.get("allowed_projects", [])
        return False

    def mask(self, principal, qualified_column, value):
        if qualified_column == "jira.description" and qualified_column not in principal.get("columns", []):
            return "[REDACTED]"
        return value


class ConnectorSDK:
    def __init__(self, name, table, rows, delay_ms=0, mode="fixture", version="v1"):
        self.name, self.table, self.rows, self.delay_ms, self.mode, self.version = name, table, rows, delay_ms, mode, version
        self.capabilities = {
            "tables": [table],
            "operations": ["select", "where", "paginate"],
            "filter_operators": ["=", "LIKE", "IN"],
            "projection": True,
            "semi_join": True,
            "max_in_values": 50,
            "page_size": 100,
            "max_concurrency": 8,
        }
        self.token_version = 1

    def refresh_token(self):
        self.token_version += 1

    def fetch_page(self, page=0, page_size=None, predicate=None):
        page_size = page_size or self.capabilities["page_size"]
        rows = [row for row in self.rows if not predicate or predicate(row)]
        start = page * page_size
        return rows[start:start + page_size], start + page_size < len(rows)

    def fetch(self, predicate=None):
        started = time.perf_counter()
        if self.delay_ms:
            time.sleep(self.delay_ms / 1000)
        rows = [row for row in self.rows if not predicate or predicate(row)]
        return rows, int((time.perf_counter() - started) * 1000)


class ConnectorWorker:
    """Worker boundary for retries, concurrency, auth refresh, and connector SDK calls."""

    def __init__(self, sdk, remote_url=None):
        self.sdk = sdk
        self.remote = RemoteConnectorClient(remote_url) if remote_url else None

    @property
    def name(self): return self.sdk.name

    @property
    def table(self): return self.sdk.table

    @property
    def rows(self): return self.sdk.rows

    @property
    def delay_ms(self): return self.sdk.delay_ms

    @property
    def mode(self): return self.sdk.mode

    @property
    def capabilities(self): return self.sdk.capabilities

    def fetch(self, predicate=None, filters=None, columns=None, traceparent=None):
        if self.remote:
            try:
                return self.remote.fetch(self.name, self.table, filters or [], columns=columns, traceparent=traceparent)
            except RemoteConnectorError as error:
                raise QueryError(error.code, error.message, error.status, source=self.name)
        rows, duration_ms = self.sdk.fetch(predicate)
        if columns is not None:
            rows = [{column: row.get(column) for column in columns} for row in rows]
        return rows, duration_ms


class LiveConnectorMixin:
    def _json_request(self, method, url, headers=None, body=None):
        request = Request(url, headers=headers or {}, data=body, method=method)
        try:
            with urlopen(request, timeout=15) as response:
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as error:
            if error.code == 401:
                raise QueryError("SOURCE_AUTH_FAILED", f"{self.name} authentication failed.", 502, source=self.name)
            if error.code == 429:
                retry_after = error.headers.get("Retry-After", "1")
                raise QueryError("RATE_LIMIT_EXHAUSTED", f"{self.name} API rate limit exhausted; retry after {retry_after} seconds.", 429, source=self.name, retry_after_ms=int(float(retry_after) * 1000))
            raise QueryError("SOURCE_TIMEOUT", f"{self.name} API returned HTTP {error.code}.", 502, source=self.name)
        except URLError as error:
            raise QueryError("SOURCE_TIMEOUT", f"{self.name} API request failed: {error.reason}.", 504, source=self.name)


class GitHubConnector(LiveConnectorMixin, ConnectorSDK):
    def __init__(self, token=None, repos=None, fixture_rows=None, api_base_url=None, version="github.v1"):
        super().__init__("github", "pull_requests", fixture_rows or [], mode="live" if token else "fixture", version=version)
        self.token, self.repos, self.api_base_url = token, repos or [], (api_base_url or os.getenv("GITHUB_API_URL", "https://api.github.com")).rstrip("/")
        self.capabilities.update({"operations": ["select", "where", "paginate"], "api": "GitHub REST pull requests"})
        self.jira_key_pattern = re.compile(os.getenv("GITHUB_JIRA_KEY_REGEX", r"\b([A-Z][A-Z0-9]+-\d+)\b"))

    def fetch(self, predicate=None):
        if self.mode != "live":
            return super().fetch(predicate)
        started = time.perf_counter()
        rows = []
        headers = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", "Authorization": f"Bearer {self.token}"}
        for repo in self.repos:
            page = 1
            while True:
                url = f"{self.api_base_url}/repos/{repo}/pulls?state=all&per_page=100&page={page}"
                payload = self._json_request("GET", url, headers=headers)
                if not payload:
                    break
                rows.extend(self._normalize(item, repo) for item in payload)
                if len(payload) < 100:
                    break
                page += 1
        if predicate:
            rows = [row for row in rows if predicate(row)]
        return rows, int((time.perf_counter() - started) * 1000)

    def _normalize(self, item, repo):
        text = f"{item.get('title', '')} {item.get('body') or ''}"
        key_match = self.jira_key_pattern.search(text)
        return {"id": item.get("number"), "repo": repo, "title": item.get("title", ""), "issue_key": key_match.group(1) if key_match else None, "author": (item.get("user") or {}).get("login"), "state": "merged" if item.get("merged_at") else item.get("state"), "updated_at": item.get("updated_at"), "url": item.get("html_url")}


class JiraConnector(LiveConnectorMixin, ConnectorSDK):
    def __init__(self, base_url=None, email=None, api_token=None, projects=None, fixture_rows=None, bearer_token=None, version="jira.v1"):
        super().__init__("jira", "issues", fixture_rows or [], mode="live" if bearer_token or (base_url and email and api_token) else "fixture", version=version)
        self.base_url, self.email, self.api_token, self.projects, self.bearer_token = (base_url or "").rstrip("/"), email, api_token, projects or [], bearer_token
        self.capabilities.update({"operations": ["select", "where", "paginate"], "api": "Jira Cloud REST v3 issue search"})

    def fetch(self, predicate=None):
        if self.mode != "live":
            return super().fetch(predicate)
        started = time.perf_counter()
        auth = base64.b64encode(f"{self.email}:{self.api_token}".encode()).decode()
        headers = {"Accept": "application/json", "Content-Type": "application/json", "Authorization": f"Bearer {self.bearer_token}" if self.bearer_token else f"Basic {auth}"}
        jql = f"project in ({','.join(self.projects)}) ORDER BY updated DESC" if self.projects else "ORDER BY updated DESC"
        rows, next_page_token = [], None
        for _ in range(100):
            body = json.dumps({"jql": jql, "fields": ["summary", "description", "status", "project", "updated"], "maxResults": 100, **({"nextPageToken": next_page_token} if next_page_token else {})}).encode()
            payload = self._json_request("POST", f"{self.base_url}/rest/api/3/search/jql", headers=headers, body=body)
            rows.extend(self._normalize(item) for item in payload.get("issues", []))
            next_page_token = payload.get("nextPageToken")
            if not next_page_token:
                break
        if predicate:
            rows = [row for row in rows if predicate(row)]
        return rows, int((time.perf_counter() - started) * 1000)

    def _normalize(self, item):
        fields = item.get("fields") or {}
        return {"id": item.get("key"), "project": (fields.get("project") or {}).get("key"), "summary": fields.get("summary", ""), "description": self._text(fields.get("description")), "status": (fields.get("status") or {}).get("name"), "updated_at": fields.get("updated")}

    @staticmethod
    def _text(value):
        if isinstance(value, str):
            return value
        if isinstance(value, dict):
            if "text" in value:
                return str(value["text"])
            return " ".join(JiraConnector._text(child) for child in value.get("content", []))
        if isinstance(value, list):
            return " ".join(JiraConnector._text(child) for child in value)
        return ""


def _connector_registry_path():
    return Path(os.getenv("CONNECTOR_CONFIG_PATH", ROOT / "fixtures/connectors.json"))


def build_connectors(registry_path=None):
    registry_path = Path(registry_path or _connector_registry_path())
    registry = json.loads(registry_path.read_text())
    connectors = {}
    for entry in registry.get("connectors", []):
        if not entry.get("enabled", True):
            continue
        name = entry["name"].lower()
        fixture_rows = json.loads((registry_path.parent / entry["fixture"]).read_text())
        if name == "github":
            sdk = GitHubConnector(fixture_rows=fixture_rows, version=entry["version"])
        elif name == "jira":
            sdk = JiraConnector(fixture_rows=fixture_rows, version=entry["version"])
        else:
            raise QueryError("INVALID_QUERY", f"Unsupported fixture connector: {name}.", 500)
        if sdk.table != entry["table"]:
            raise QueryError("INVALID_QUERY", f"Connector registry table mismatch for {name}.", 500)
        connectors[name] = sdk
    return connectors


class QueryPlanner:
    QUERY = re.compile(
        r"^\s*SELECT\s+(?P<select>.+?)\s+FROM\s+(?P<table>\w+)\.(?P<table_name>\w+)(?:\s+(?P<alias>\w+))?"
        r"(?:\s+JOIN\s+(?P<join_source>\w+)\.(?P<join_table>\w+)(?:\s+(?P<join_alias>\w+))?\s+ON\s+(?P<on_left>\w+\.\w+)\s*=\s*(?P<on_right>\w+\.\w+))?"
        r"(?:\s+WHERE\s+(?P<where>.+?))?(?:\s+ORDER\s+BY\s+(?P<order>\w+\.\w+)(?:\s+(?P<direction>ASC|DESC))?)?"
        r"(?:\s+LIMIT\s+(?P<limit>\d+))?(?:\s+OFFSET\s+(?P<offset>\d+))?\s*$",
        re.IGNORECASE,
    )

    def parse(self, sql):
        match = self.QUERY.match(sql or "")
        if not match:
            raise QueryError("INVALID_QUERY", "Use SELECT ... FROM source.table [JOIN ... ON ...] [WHERE ...] [ORDER BY ...] [LIMIT n OFFSET n].")
        data = match.groupdict()
        columns = [item.strip().lower() for item in data["select"].split(",")]
        if not columns or any(not re.fullmatch(r"\w+\.\w+|\*", col) for col in columns):
            raise QueryError("INVALID_QUERY", "Projection must use qualified columns such as github.title.")
        source_alias = data["alias"] or data["table"]
        join_alias = data["join_alias"] or data["join_source"] if data["join_source"] else None
        where = self._conditions(data["where"])
        return {
            "columns": columns,
            "sources": [(data["table"].lower(), source_alias.lower(), data["table_name"].lower())]
            + ([(data["join_source"].lower(), join_alias.lower(), data["join_table"].lower())] if data["join_source"] else []),
            "join": (data["on_left"].lower(), data["on_right"].lower()) if data["on_left"] else None,
            "where": where,
            "order": (data["order"].lower(), (data["direction"] or "ASC").upper()) if data["order"] else None,
            "limit": int(data["limit"] or 100),
            "offset": int(data["offset"] or 0),
        }

    def _conditions(self, raw):
        if not raw:
            return []
        conditions = []
        for term in re.split(r"\s+AND\s+", raw, flags=re.IGNORECASE):
            match = re.fullmatch(r"(\w+\.\w+)\s*(=|LIKE)\s*'([^']*)'", term.strip(), re.IGNORECASE)
            if not match:
                raise QueryError("INVALID_QUERY", "WHERE supports qualified equality or LIKE predicates joined by AND.")
            conditions.append((match.group(1).lower(), match.group(2).upper(), match.group(3)))
        return conditions


class QueryGateway:
    def __init__(self, rate_capacity=20, rate_refill_per_sec=2, entitlement_ttl_ms=None, force_live_entitlements=None):
        policy = EntitlementPolicy()
        self.policy = policy
        self.auth = AuthService()
        self.tenant_registry = TenantRegistry(policy.data)
        self.catalog = MetadataCatalog()
        self.secret_keys = SecretKeyStore()
        self.planner = QueryPlanner()
        remote_url = os.getenv("CONNECTOR_SERVICE_URL")
        self.connectors = {source: ConnectorWorker(sdk, remote_url=remote_url) for source, sdk in build_connectors().items()}
        for connector in self.connectors.values():
            self.catalog.register(connector)
        self.rate_capacity, self.rate_refill_per_sec = rate_capacity, rate_refill_per_sec
        try:
            control_plane_timeout = float(os.getenv("CONTROL_PLANE_TIMEOUT_SECONDS", "0.5"))
        except ValueError:
            control_plane_timeout = 0.5
        entitlement_url = os.getenv("ENTITLEMENT_SERVICE_URL")
        rate_limit_url = os.getenv("RATE_LIMIT_SERVICE_URL")
        self.entitlement_client = EntitlementHTTPClient(entitlement_url, control_plane_timeout) if entitlement_url else None
        rate_limit_client = RateLimitHTTPClient(rate_limit_url, control_plane_timeout) if rate_limit_url else None
        self.rate_limits = RateLimitService(rate_capacity, rate_refill_per_sec, external_client=rate_limit_client)
        self.buckets = self.rate_limits.buckets
        self.cache = FreshnessCache()
        self.fetch_store = FetchStore()
        self.freshness = FreshnessLayer(self.cache, self.fetch_store)
        self.materializations = MaterializationStore()
        self.entitlement_snapshots = EntitlementSnapshotStore()
        self.entitlement_service = EntitlementService(policy_data=policy.data)
        self.entitlement_ttl_ms = entitlement_ttl_ms or {source: 30_000 for source in self.connectors}
        self.force_live_entitlements = (os.getenv("FORCE_LIVE_ENTITLEMENTS", "false").lower() == "true") if force_live_entitlements is None else force_live_entitlements
        self.observability = Observability()
        self.active_spans = {}
        self.audit = AuditLog()
        self.async_jobs = AsyncJobRunner()
        self.components = {
            "control_plane": {"tenant_registry", "metadata_catalog", "policy_store", "secrets_keys"},
            "capture_plane": {"connector_workers", "connector_sdk", "freshness_layer", "fetch_store"},
            "query_plane": {"query_gateway", "query_planner", "entitlement_service", "rate_limit_service", "freshness_layer", "fetch_store", "materialization", "async_jobs"},
            "data_plane": {"connector_workers", "query_gateway", "query_planner", "freshness_layer", "fetch_store", "materialization", "async_jobs"},
            "authn": self.auth,
            "tenant_registry": self.tenant_registry,
            "metadata_catalog": self.catalog,
            "policy_store": self.policy,
            "secrets_keys": self.secret_keys,
            "query_gateway": self,
            "query_planner": self.planner,
            "connector_sdk": {source: worker.sdk for source, worker in self.connectors.items()},
            "connector_workers": self.connectors,
            "entitlement_service": self.entitlement_service,
            "rate_limit_service": self.rate_limits,
            "fair_scheduler": self.rate_limits.scheduler,
            "freshness_layer": self.freshness,
            "fetch_store": self.fetch_store,
            "materialization": self.materializations,
            "async_jobs": self.async_jobs,
            "audit": self.audit,
            "observability": self.observability,
        }
        self.lock = threading.Lock()

    def refresh_fetch_store(self, sources=None):
        sources = sources or list(self.connectors)
        refreshed = {}
        for source in sources:
            if source not in self.connectors:
                raise QueryError("INVALID_QUERY", f"Unknown connector: {source}.")
            rows, duration_ms = self.connectors[source].fetch()
            self.fetch_store.refresh(source, rows)
            refreshed[source] = {"rows": len(rows), "duration_ms": duration_ms}
        self.observability.increment("universal_sql_fetch_store_refresh_total")
        return refreshed

    def register_fixture_connector(self, config):
        name = str(config.get("name", "")).lower().strip()
        version = str(config.get("version", "")).strip()
        table = str(config.get("table", "")).lower().strip()
        fixture = str(config.get("fixture", "")).strip()
        expected_tables = {"github": "pull_requests", "jira": "issues"}
        if name not in expected_tables:
            raise QueryError("INVALID_QUERY", "Only github and jira fixture connectors are supported by this prototype.")
        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", version):
            raise QueryError("INVALID_QUERY", "Connector version must contain only letters, numbers, dots, underscores, or hyphens.")
        if table != expected_tables[name]:
            raise QueryError("INVALID_QUERY", f"Connector table must be {expected_tables[name]} for {name}.")
        fixture_path_value = Path(fixture)
        if fixture_path_value.is_absolute() or fixture_path_value.name != fixture:
            raise QueryError("INVALID_QUERY", "Fixture must be a relative JSON filename.")
        if fixture_path_value.suffix.lower() != ".json":
            raise QueryError("INVALID_QUERY", "Fixture must be a JSON file.")
        registry_path = _connector_registry_path()
        fixture_path = (registry_path.parent / fixture_path_value).resolve()
        registry_root = registry_path.parent.resolve()
        try:
            fixture_path.relative_to(registry_root)
        except ValueError:
            raise QueryError("INVALID_QUERY", "Fixture must remain inside the connector registry directory.")
        try:
            fixture_rows = json.loads(fixture_path.read_text())
        except (OSError, ValueError) as error:
            raise QueryError("INVALID_QUERY", f"Fixture could not be loaded: {error}.")
        if not isinstance(fixture_rows, list):
            raise QueryError("INVALID_QUERY", "Fixture must contain a JSON array of rows.")
        try:
            registry = json.loads(registry_path.read_text())
        except (OSError, ValueError) as error:
            raise QueryError("INVALID_QUERY", f"Connector registry could not be loaded: {error}.")
        entries = registry.setdefault("connectors", [])
        entry = {"name": name, "version": version, "table": table, "fixture": fixture, "enabled": bool(config.get("enabled", True))}
        for index, current in enumerate(entries):
            if current.get("name", "").lower() == name:
                entries[index] = entry
                break
        else:
            entries.append(entry)
        registry_path.parent.mkdir(parents=True, exist_ok=True)
        temporary_path = registry_path.with_name(f".{registry_path.name}.tmp")
        temporary_path.write_text(json.dumps(registry, indent=2) + "\n")
        os.replace(temporary_path, registry_path)
        self.connectors = {source: ConnectorWorker(sdk) for source, sdk in build_connectors(registry_path).items()}
        self.catalog = MetadataCatalog()
        for connector in self.connectors.values():
            self.catalog.register(connector)
        self.fetch_store.entries = {}
        self.entitlement_snapshots.entries = {}
        self.entitlement_ttl_ms = {source: self.entitlement_ttl_ms.get(source, 30_000) for source in self.connectors}
        self.components["connector_sdk"] = {source: worker.sdk for source, worker in self.connectors.items()}
        self.components["connector_workers"] = self.connectors
        self.audit.record(decision="ADMIN_CONNECTOR_REGISTERED", connector=name, version=version)
        return self.admin_status()

    def query(self, request):
        trace_id = uuid.uuid4().hex
        started = time.perf_counter()
        span = self.observability.start_span("query.execute", {"query.trace_id": trace_id}, traceparent=request.get("_traceparent"))
        span.__enter__()
        trace_id = span.trace_id
        span.set_attribute("query.trace_id", trace_id)
        self.active_spans[trace_id] = span
        try:
            principal = self.policy.principal(self.auth.authenticate(request))
            self.tenant_registry.ensure_active(principal["tenant_id"])
            plan = self.planner.parse(request.get("sql", ""))
            self._validate_plan(plan)
            sources = [source for source, _, _ in plan["sources"]]
            self.policy.check_sources(principal, sources)
            self._authorize_with_contract(principal, sources, plan)
            source_principals, entitlement_metadata = self._entitlements(principal, sources, request)
            mode = request.get("mode", "realtime")
            if mode not in {"realtime", "fetch_store"}:
                raise QueryError("INVALID_QUERY", "mode must be realtime or fetch_store.")
            if mode == "fetch_store":
                if request.get("refresh_fetch_store"):
                    self.refresh_fetch_store(sources)
                rows, connector_times, freshness_metadata = self._execute_fetch_store(plan, principal, source_principals, int(request.get("max_staleness_ms", 0)))
                if request.get("materialize"):
                    materialization_id = f"mat_{uuid.uuid4().hex[:12]}"
                    self.materializations.put(materialization_id, rows)
                    freshness_metadata["materialization_id"] = materialization_id
                self.observability.increment(
                    "universal_sql_fetch_store_queries_total",
                    labels={"tenant_id": principal["tenant_id"]},
                )
                return self._success(plan, rows, freshness_metadata["freshness_ms"], True, trace_id, started, connector_times, mode, entitlement_metadata, freshness_metadata, principal, self._local_pushdown_plan(plan, strategy="fetch_store"))
            max_staleness = int(request.get("max_staleness_ms", 0))
            cache_key = json.dumps([principal["tenant_id"], principal["user_id"], plan], sort_keys=True)
            cached = self.cache.get(cache_key, max_staleness)
            if cached:
                rows, freshness_ms = cached
                self.observability.increment(
                    "universal_sql_cache_hits_total",
                    labels={"tenant_id": principal["tenant_id"]},
                )
                return self._success(plan, rows, freshness_ms, True, trace_id, started, {}, mode, entitlement_metadata, {"freshness_ms": freshness_ms, "fetch_store_hit": False}, principal, self._local_pushdown_plan(plan, strategy="cache"))
            allowed, retry_ms = self.rate_limits.admit(principal["tenant_id"], principal["user_id"], sources[0], trace_id)
            if not allowed:
                self.observability.increment(
                    "universal_sql_rate_limit_exhausted_total",
                    labels={"tenant_id": principal["tenant_id"]},
                )
                if request.get("async"):
                    job = self.async_jobs.submit(request, "source rate limit exhausted")
                    raise QueryError("RATE_LIMIT_EXHAUSTED", "Source refresh budget is exhausted; the query was queued for asynchronous execution.", 202, retry_after_ms=retry_ms, job_id=job["id"])
                raise QueryError("RATE_LIMIT_EXHAUSTED", "Source refresh budget is exhausted; retry after the indicated delay or increase max_staleness_ms to use cached data.", 429, retry_after_ms=retry_ms)
            rows, connector_times, pushdown = self._execute(plan, principal, source_principals, traceparent=span.traceparent)
            self.cache.put(cache_key, rows)
            self.observability.increment(
                "universal_sql_queries_total",
                labels={"tenant_id": principal["tenant_id"]},
            )
            return self._success(plan, rows, 0, False, trace_id, started, connector_times, mode, entitlement_metadata, {"freshness_ms": 0, "fetch_store_hit": False, "source_refresh_admitted": True}, principal, pushdown)
        except QueryError as error:
            labels = {"code": error.code}
            if "principal" in locals():
                labels["tenant_id"] = principal["tenant_id"]
            self.observability.increment("universal_sql_errors_total", labels=labels)
            self.audit.record(trace_id=trace_id, decision=error.code, user=request.get("user"), error=error.code)
            duration_labels = {"tenant_id": labels["tenant_id"]} if "tenant_id" in labels else None
            self._finish_span(trace_id, {"query.outcome": "error", "error.code": error.code}, labels=duration_labels)
            return {"error": {"code": error.code, "message": error.message, "trace_id": trace_id, **error.extra}, "status": error.status}

    def _finish_span(self, trace_id, attributes, labels=None):
        span = self.active_spans.pop(trace_id, None)
        if not span:
            return
        for key, value in attributes.items():
            span.set_attribute(key, value)
        span.end()
        self.observability.observe("universal_sql_query_duration_ms", span.duration_ms, labels=labels)

    def _validate_plan(self, plan):
        for source, alias, table in plan["sources"]:
            if source not in self.connectors or self.connectors[source].table != table:
                raise QueryError("INVALID_QUERY", f"Unknown source table: {source}.{table}.")
        allowed = {f"{alias}.{key}" for source, alias, _ in plan["sources"] for key in self.connectors[source].rows[0]}
        references = plan["columns"] + [left for left, _, _ in plan["where"]]
        if plan["order"]:
            references.append(plan["order"][0])
        if plan["join"]:
            references.extend(plan["join"])
        if "*" not in references and any(ref not in allowed for ref in references):
            raise QueryError("INVALID_QUERY", "Query references a column not present in the connector schema.")

    def _entitlements(self, principal, sources, request):
        requested_ttls = request.get("max_entitlement_age_ms", {})
        force_live = self.force_live_entitlements or bool(request.get("force_live_entitlements", False))
        source_principals, metadata = {}, {}
        for source in sources:
            max_age = int(requested_ttls.get(source, self.entitlement_ttl_ms.get(source, 30_000)))
            key = f"{principal['tenant_id']}:{principal['user_id']}:{source}"
            cached = None if force_live else self.entitlement_snapshots.get(key, max_age)
            if cached:
                source_principals[source], age_ms = cached
                metadata[source] = {"source": "snapshot", "age_ms": age_ms}
            else:
                try:
                    source_principals[source] = self._authorize_with_contract(
                        principal, [source], {"columns": []}
                    )["sources"][source]
                except (EntitlementError, KeyError):
                    # Preserve the original local policy behavior as an
                    # explicit compatibility fallback for malformed or older
                    # policy adapters.
                    source_principals[source] = self.policy.live_source_entitlements(principal, source)
                self.entitlement_snapshots.put(key, source_principals[source])
                metadata[source] = {"source": "live", "age_ms": 0}
                self.observability.increment(
                    "universal_sql_live_entitlement_checks_total",
                    labels={"tenant_id": principal["tenant_id"]},
                )
        return source_principals, metadata

    def _authorize_with_contract(self, principal, sources, plan):
        """Run the transport-neutral entitlement contract at the gateway edge."""
        columns = plan.get("columns", []) if isinstance(plan, dict) else []
        request = {
            "user": principal["user_id"],
            "tenant_id": principal["tenant_id"],
            "sources": list(sources),
            "columns": columns,
        }
        if self.entitlement_client is not None:
            status, response = self.entitlement_client.authorize(request)
        else:
            status, response = self.entitlement_service.handle(request)
        if status != 200:
            error = response.get("error", {})
            raise QueryError(
                error.get("code", "ENTITLEMENT_DENIED"),
                error.get("message", "Entitlement denied."),
                status,
            )
        return response

    def _source_predicates(self, plan, source, alias):
        return [(field.split(".", 1)[1], op, value) for field, op, value in plan["where"] if field.startswith(alias + ".")]

    @staticmethod
    def _filter_text(column, operator, value):
        if operator == "IN":
            return f"{column} IN ({', '.join(repr(item) for item in value)})"
        return f"{column} {operator} {value!r}"

    def _pushdown_for_source(self, plan, source, alias, extra_filters=None):
        connector = self.connectors[source]
        capabilities = connector.capabilities
        source_predicates = self._source_predicates(plan, source, alias)
        candidates = list(source_predicates) + list(extra_filters or [])
        supported = set(capabilities.get("filter_operators", []))
        max_in_values = int(capabilities.get("max_in_values", 0) or 0)
        pushed, local = [], []
        for column, operator, value in candidates:
            can_push = operator in supported
            if operator == "IN":
                can_push = can_push and bool(value) and (not max_in_values or len(value) <= max_in_values)
            if can_push:
                pushed.append({"column": column, "operator": operator, "value": value})
            else:
                local.append((column, operator, value))

        schema = list(connector.rows[0]) if connector.rows else []
        required = set(schema if plan["columns"] == ["*"] else [])
        required.update(column.split(".", 1)[1] for column in plan["columns"] if column != "*" and column.startswith(alias + "."))
        required.update(column for column, _, _ in source_predicates)
        if plan["join"]:
            required.update(column.split(".", 1)[1] for column in plan["join"] if column.startswith(alias + "."))
        if plan["order"] and plan["order"][0].startswith(alias + "."):
            required.add(plan["order"][0].split(".", 1)[1])
        required.add("repo" if source == "github" else "project")
        projected = sorted(required) if capabilities.get("projection", False) else schema
        return {
            "filters": pushed,
            "local_filters": local,
            "columns": projected,
            "pushed_text": [self._filter_text(item["column"], item["operator"], item["value"]) for item in pushed],
            "local_text": [self._filter_text(column, operator, value) for column, operator, value in local],
        }

    def _local_pushdown_plan(self, plan, strategy):
        pushed_filters, local_filters, projected_columns = {}, {}, {}
        for source, alias, _ in plan["sources"]:
            details = self._pushdown_for_source(plan, source, alias)
            pushed_filters[source] = details["pushed_text"] if strategy not in {"cache", "fetch_store"} else []
            local_filters[source] = details["local_text"] if strategy not in {"cache", "fetch_store"} else [self._filter_text(column, operator, value) for column, operator, value in self._source_predicates(plan, source, alias)]
            projected_columns[source] = details["columns"]
        return {"strategy": strategy, "pushed_filters": pushed_filters, "local_filters": local_filters, "projected_columns": projected_columns}

    def _semi_join_candidate(self, plan):
        if len(plan["sources"]) != 2 or not plan["join"]:
            return None
        source_by_alias = {alias: source for source, alias, _ in plan["sources"]}
        left, right = plan["join"]
        left_alias, left_key = left.split(".", 1)
        right_alias, right_key = right.split(".", 1)
        candidates = [
            (source_by_alias[right_alias], right_alias, right_key, source_by_alias[left_alias], left_alias, left_key),
            (source_by_alias[left_alias], left_alias, left_key, source_by_alias[right_alias], right_alias, right_key),
        ]
        for driver_source, driver_alias, driver_key, target_source, target_alias, target_key in candidates:
            target_capabilities = self.connectors[target_source].capabilities
            if not target_capabilities.get("semi_join") or "IN" not in set(target_capabilities.get("filter_operators", [])):
                continue
            if len(self.connectors[driver_source].rows) <= len(self.connectors[target_source].rows):
                return {
                    "driver_source": driver_source,
                    "driver_alias": driver_alias,
                    "driver_key": driver_key,
                    "target_source": target_source,
                    "target_alias": target_alias,
                    "target_key": target_key,
                }
        return None

    def _execute(self, plan, principal, source_principals, traceparent=None):
        source_rows, connector_times = {}, {}
        pushdown = {"strategy": "source_filter" if len(plan["sources"]) == 1 else "local_join", "pushed_filters": {}, "local_filters": {}, "projected_columns": {}}
        semi_join = self._semi_join_candidate(plan)
        extra_filters = {}

        def fetch_source(source, alias):
            details = self._pushdown_for_source(plan, source, alias, extra_filters.get(source))
            connector = self.connectors[source]
            all_predicates = self._source_predicates(plan, source, alias) + list(extra_filters.get(source, []))

            def predicate(row, predicates=all_predicates, source=source):
                return self._row_allowed(source_principals[source], source, row) and all(self._match(row.get(field), op, value) for field, op, value in predicates)

            fetched_rows, connector_ms = connector.fetch(predicate, filters=details["filters"], columns=details["columns"], traceparent=traceparent)
            source_rows[alias] = [{f"{alias}.{key}": value for key, value in row.items()} for row in fetched_rows if self._row_allowed(source_principals[source], source, row)]
            connector_times[source] = connector_ms
            pushdown["pushed_filters"][source] = details["pushed_text"]
            pushdown["local_filters"][source] = details["local_text"]
            pushdown["projected_columns"][source] = details["columns"]

        if semi_join:
            fetch_source(semi_join["driver_source"], semi_join["driver_alias"])
            driver_keys = sorted({row.get(f"{semi_join['driver_alias']}.{semi_join['driver_key']}") for row in source_rows[semi_join["driver_alias"]] if row.get(f"{semi_join['driver_alias']}.{semi_join['driver_key']}") is not None})
            max_values = int(self.connectors[semi_join["target_source"]].capabilities.get("max_in_values", 0) or 0)
            if driver_keys and (not max_values or len(driver_keys) <= max_values):
                extra_filters[semi_join["target_source"]] = [(semi_join["target_key"], "IN", driver_keys)]
                pushdown.update({"strategy": "semi_join", "driver": semi_join["driver_source"]})

        for source, alias, _ in plan["sources"]:
            if alias not in source_rows:
                fetch_source(source, alias)
        if len(plan["sources"]) == 1:
            joined = source_rows[plan["sources"][0][1]]
        else:
            left_alias, right_alias = plan["sources"][0][1], plan["sources"][1][1]
            left_key, right_key = plan["join"]
            left_key, right_key = left_key.split(".", 1)[1], right_key.split(".", 1)[1]
            joined = [{**left, **right} for left in source_rows[left_alias] for right in source_rows[right_alias] if left.get(f"{left_alias}.{left_key}") == right.get(f"{right_alias}.{right_key}")]
        if plan["where"]:
            joined = [row for row in joined if all(self._match(row.get(field), op, value) for field, op, value in plan["where"])]
        if plan["order"]:
            field, direction = plan["order"]
            joined.sort(key=lambda row: (row.get(field) is None, row.get(field)), reverse=direction == "DESC")
        selected = plan["columns"]
        if selected == ["*"]:
            selected = list(joined[0]) if joined else []
        joined = joined[plan["offset"]:plan["offset"] + plan["limit"]]
        output = [{column: self.policy.mask(principal, column, row.get(column)) for column in selected} for row in joined]
        return output, connector_times, pushdown

    def _execute_fetch_store(self, plan, principal, source_principals, max_staleness_ms):
        source_rows, freshness, connector_times = {}, {}, {}
        for source, alias, _ in plan["sources"]:
            raw_rows, age_ms = self.fetch_store.get(source, max_staleness_ms)
            predicates = [(field.split(".", 1)[1], op, value) for field, op, value in plan["where"] if field.startswith(alias + ".")]
            filtered = [row for row in raw_rows if self._row_allowed(source_principals[source], source, row) and all(self._match(row.get(field), op, value) for field, op, value in predicates)]
            source_rows[alias] = [{f"{alias}.{key}": value for key, value in row.items()} for row in filtered]
            freshness[source] = age_ms
            connector_times[source] = 0
        if len(plan["sources"]) == 1:
            joined = source_rows[plan["sources"][0][1]]
        else:
            left_alias, right_alias = plan["sources"][0][1], plan["sources"][1][1]
            left_key, right_key = (key.split(".", 1)[1] for key in plan["join"])
            joined = [{**left, **right} for left in source_rows[left_alias] for right in source_rows[right_alias] if left.get(f"{left_alias}.{left_key}") == right.get(f"{right_alias}.{right_key}")]
        if plan["where"]:
            joined = [row for row in joined if all(self._match(row.get(field), op, value) for field, op, value in plan["where"])]
        if plan["order"]:
            field, direction = plan["order"]
            joined.sort(key=lambda row: (row.get(field) is None, row.get(field)), reverse=direction == "DESC")
        selected = list(joined[0]) if plan["columns"] == ["*"] and joined else plan["columns"]
        joined = joined[plan["offset"]:plan["offset"] + plan["limit"]]
        output = [{column: self.policy.mask(principal, column, row.get(column)) for column in selected} for row in joined]
        return output, connector_times, {"freshness_ms": max(freshness.values(), default=0), "connector_freshness_ms": freshness, "fetch_store_hit": True}

    @staticmethod
    def _row_allowed(source_principal, source, row):
        if source == "github":
            return row.get("repo") in source_principal.get("allowed_repos", [])
        if source == "jira":
            return row.get("project") in source_principal.get("allowed_projects", [])
        return False

    @staticmethod
    def _match(actual, operator, expected):
        if operator == "=":
            return str(actual).lower() == expected.lower()
        if operator == "IN":
            return str(actual).lower() in {str(value).lower() for value in expected}
        pattern = re.escape(expected).replace("%", ".*")
        return re.fullmatch(pattern, str(actual), flags=re.IGNORECASE) is not None

    def _success(self, plan, rows, freshness_ms, cache_hit, trace_id, started, connector_times, mode, entitlement_metadata, freshness_metadata, principal, pushdown):
        scope = f"{principal['tenant_id']}:{principal['user_id']}"
        metadata = {"freshness_ms": freshness_ms, "cache_hit": cache_hit, "source_mode": mode, "fetch_store_hit": freshness_metadata.get("fetch_store_hit", False), "materialization_id": freshness_metadata.get("materialization_id"), "connector_freshness_ms": freshness_metadata.get("connector_freshness_ms", {}), "entitlement_snapshot_age_ms": {source: item["age_ms"] for source, item in entitlement_metadata.items()}, "entitlement_source": {source: item["source"] for source, item in entitlement_metadata.items()}, "connector_capabilities": {source: self.catalog.describe(source)["capabilities"] for source in entitlement_metadata}, "connector_versions": {source: self.connectors[source].sdk.version for source in entitlement_metadata}, "pushdown": pushdown, "key_scope": "tenant-scoped KMS reference", "rate_limit_status": {"scope": "tenant:user", "remaining": self.rate_limits.remaining(scope), "source_refresh_admitted": freshness_metadata.get("source_refresh_admitted", False)}, "trace_id": trace_id, "duration_ms": int((time.perf_counter() - started) * 1000), "connector_ms": connector_times}
        self.audit.record(trace_id=trace_id, decision="ALLOW", result_count=len(rows), source_mode=mode, entitlement_source=metadata["entitlement_source"])
        span_attributes = {"query.outcome": "ok", "query.rows": len(rows), "query.cache_hit": cache_hit}
        span_attributes.update({f"connector.{source}.duration_ms": duration for source, duration in connector_times.items()})
        labels = {"tenant_id": principal["tenant_id"]}
        for duration in connector_times.values():
            self.observability.observe("universal_sql_connector_duration_ms", duration, labels=labels)
        self._finish_span(trace_id, span_attributes, labels=labels)
        return {"columns": plan["columns"] if plan["columns"] != ["*"] else list(rows[0]) if rows else [], "rows": rows, "metadata": metadata}

    def _remaining(self, principal):
        return self.rate_limits.remaining(f"{principal['tenant_id']}:{principal['user_id']}")

    def admin_status(self):
        connectors = {}
        for source, worker in self.connectors.items():
            sdk = worker.sdk
            store_entry = self.fetch_store.entries.get(source)
            connectors[source] = {
                "mode": sdk.mode,
                "table": sdk.table,
                "version": sdk.version,
                "configured": sdk.mode == "live",
                "capabilities": sdk.capabilities,
                "fetch_store_age_ms": int((time.time() - store_entry["refreshed_at"]) * 1000) if store_entry else None,
                "scope_count": len(getattr(sdk, "repos", []) or getattr(sdk, "projects", [])),
            }
        return {"connector_mode": "fixture", "connector_registry": os.getenv("CONNECTOR_CONFIG_PATH", str(ROOT / "fixtures/connectors.json")), "force_live_entitlements": self.force_live_entitlements, "connectors": connectors, "tenant_count": len(self.tenant_registry.tenants), "audit_events": len(self.audit.events)}

    def admin_policy(self):
        return {user: {key: value for key, value in data.items() if key in {"tenant_id", "roles", "scopes", "allowed_repos", "allowed_projects", "columns"}} for user, data in self.policy.data.get("users", {}).items()}

    def admin_update_policy(self, request):
        user = request.get("user")
        if user not in self.policy.data.get("users", {}):
            raise QueryError("ENTITLEMENT_DENIED", "Policy user does not exist.", 404)
        for key in ("allowed_repos", "allowed_projects", "columns"):
            if key in request:
                if not isinstance(request[key], list) or not all(isinstance(value, str) for value in request[key]):
                    raise QueryError("INVALID_QUERY", f"Policy field {key} must be a list of strings.")
                self.policy.data["users"][user][key] = request[key]
        self.entitlement_snapshots.entries = {key: value for key, value in self.entitlement_snapshots.entries.items() if f":{user}:" not in key}
        self.audit.record(decision="ADMIN_POLICY_UPDATE", user=user)
        return self.admin_policy()[user]


ADMIN_HTML = r"""<!doctype html>
<html><head><meta charset="utf-8"><title>Universal SQL Admin</title>
<style>body{font:15px system-ui;margin:32px;background:#f8fafc;color:#0f172a}main{max-width:1000px;margin:auto}section{background:white;border:1px solid #cbd5e1;border-radius:12px;padding:20px;margin:16px 0}button{background:#2563eb;color:white;border:0;border-radius:6px;padding:9px 13px;cursor:pointer}input,textarea{padding:8px;border:1px solid #94a3b8;border-radius:6px;margin:4px;width:96%}pre{background:#0f172a;color:#e2e8f0;padding:14px;border-radius:8px;overflow:auto}.ok{color:#15803d}.warn{color:#b45309}</style></head>
<body><main><h1>Universal SQL Admin</h1><p>This reviewer console uses deterministic mocked GitHub and Jira data. No provider credentials or external calls are required.</p>
<section><h2>Admin token</h2><input id="token" type="password" value="local-admin-token" placeholder="X-Admin-Token (override if ADMIN_TOKEN is set)"><button onclick="loadAll()">Load status</button></section>
<section><h2>Mock connectors</h2><pre id="status">Not loaded</pre><p>Connector versions and fixture configuration are shown in status. Register or replace a local fixture connector below; the registry reloads immediately.</p><h3>Register mock connector</h3><input id="connectorName" placeholder="Connector: github or jira"><input id="connectorVersion" placeholder="Version: github.v2"><input id="connectorTable" placeholder="Table: pull_requests or issues"><input id="connectorFixture" placeholder="Fixture filename: github.json"><button onclick="registerConnector()">Register connector</button><button onclick="refreshStore()">Refresh fetch store</button><p>Refresh captures the local GitHub pull-request and Jira issue fixtures into the fetch store for query testing.</p></section>
<section><h2>Policy</h2><pre id="policy">Not loaded</pre><input id="user" placeholder="policy user, e.g. alice"><textarea id="repos" placeholder="allowed repos, comma separated"></textarea><textarea id="projects" placeholder="allowed projects, comma separated"></textarea><button onclick="savePolicy()">Save row policy</button></section>
<script>
const headers=()=>({'X-Admin-Token':document.getElementById('token').value,'Content-Type':'application/json'});
async function loadAll(){const s=await fetch('/admin/api/status',{headers:headers()});document.getElementById('status').textContent=JSON.stringify(await s.json(),null,2);const p=await fetch('/admin/api/policy',{headers:headers()});document.getElementById('policy').textContent=JSON.stringify(await p.json(),null,2)}
async function refreshStore(){const r=await fetch('/admin/api/refresh',{method:'POST',headers:headers(),body:'{}'});document.getElementById('status').textContent=JSON.stringify(await r.json(),null,2)}
async function registerConnector(){const body={name:document.getElementById('connectorName').value,version:document.getElementById('connectorVersion').value,table:document.getElementById('connectorTable').value,fixture:document.getElementById('connectorFixture').value,enabled:true};const r=await fetch('/admin/api/connector-registry',{method:'POST',headers:headers(),body:JSON.stringify(body)});document.getElementById('status').textContent=JSON.stringify(await r.json(),null,2)}
async function savePolicy(){const r=await fetch('/admin/api/policy',{method:'POST',headers:headers(),body:JSON.stringify({user:document.getElementById('user').value,allowed_repos:document.getElementById('repos').value.split(',').map(x=>x.trim()).filter(Boolean),allowed_projects:document.getElementById('projects').value.split(',').map(x=>x.trim()).filter(Boolean)})});document.getElementById('policy').textContent=JSON.stringify(await r.json(),null,2)}
</script></main></body></html>"""


class Handler(BaseHTTPRequestHandler):
    gateway = QueryGateway()

    def _send(self, status, payload, content_type="application/json"):
        body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        if isinstance(payload, dict) and payload.get("error", {}).get("retry_after_ms") is not None:
            self.send_header("Retry-After", str(max(1, payload["error"]["retry_after_ms"] // 1000)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = urlparse(self.path).path
        if path == "/admin":
            return self._send(200, ADMIN_HTML.encode(), "text/html; charset=utf-8")
        if path.startswith("/admin/api/"):
            if self.headers.get("X-Admin-Token") != os.getenv("ADMIN_TOKEN", "local-admin-token"):
                return self._send(403, {"error": {"code": "ENTITLEMENT_DENIED", "message": "Admin token is invalid. Use local-admin-token unless ADMIN_TOKEN was configured explicitly."}})
            if path == "/admin/api/status":
                return self._send(200, self.gateway.admin_status())
            if path == "/admin/api/policy":
                return self._send(200, self.gateway.admin_policy())
        if path == "/healthz":
            return self._send(200, {"status": "ok"})
        if path == "/metrics":
            return self._send(200, self.gateway.observability.metrics_text().encode(), "text/plain; version=0.0.4")
        if path == "/v1/components":
            return self._send(200, {"components": sorted(self.gateway.components)})
        if path.startswith("/v1/jobs/"):
            job = self.gateway.async_jobs.get(path.rsplit("/", 1)[-1])
            return self._send(200 if job else 404, job or {"error": {"code": "NOT_FOUND", "message": "Job not found."}})
        return self._send(404, {"error": {"code": "NOT_FOUND", "message": "Route not found."}})

    def do_POST(self):
        path = urlparse(self.path).path
        if path.startswith("/admin/api/"):
            if self.headers.get("X-Admin-Token") != os.getenv("ADMIN_TOKEN", "local-admin-token"):
                return self._send(403, {"error": {"code": "ENTITLEMENT_DENIED", "message": "Admin token is invalid. Use local-admin-token unless ADMIN_TOKEN was configured explicitly."}})
            try:
                body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
                request = json.loads(body or b"{}")
                if path == "/admin/api/refresh":
                    return self._send(200, {"status": "refreshed", "sources": self.gateway.refresh_fetch_store(request.get("sources"))})
                if path == "/admin/api/connector-registry":
                    return self._send(200, {"status": "registered", **self.gateway.register_fixture_connector(request)})
                if path == "/admin/api/policy":
                    return self._send(200, {"status": "updated", "policy": self.gateway.admin_update_policy(request)})
            except QueryError as error:
                return self._send(error.status, {"error": {"code": error.code, "message": error.message, **error.extra}})
            except (ValueError, json.JSONDecodeError):
                return self._send(400, {"error": {"code": "INVALID_QUERY", "message": "Admin body must be valid JSON."}})
        if path == "/v1/fetch-store/refresh":
            expected = os.getenv("FETCH_STORE_REFRESH_TOKEN", "local-demo-token")
            if self.headers.get("X-Refresh-Token") != expected:
                return self._send(403, {"error": {"code": "ENTITLEMENT_DENIED", "message": "Fetch-store refresh requires the configured internal refresh token."}})
            try:
                body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
                request = json.loads(body or b"{}")
                result = self.gateway.refresh_fetch_store(request.get("sources"))
                return self._send(200, {"status": "refreshed", "sources": result})
            except QueryError as error:
                return self._send(error.status, {"error": {"code": error.code, "message": error.message, **error.extra}})
            except (ValueError, json.JSONDecodeError):
                return self._send(400, {"error": {"code": "INVALID_QUERY", "message": "Refresh body must be valid JSON."}})
        if path != "/v1/query":
            return self._send(404, {"error": {"code": "NOT_FOUND", "message": "Route not found."}})
        try:
            body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
            request = json.loads(body)
            request["_traceparent"] = self.headers.get("traceparent")
            result = self.gateway.query(request)
        except (ValueError, json.JSONDecodeError):
            return self._send(400, {"error": {"code": "INVALID_QUERY", "message": "Request body must be valid JSON."}})
        return self._send(result.pop("status", 200), result)

    def log_message(self, *_):
        return


def serve(host="0.0.0.0", port=8080):
    server = ThreadingHTTPServer((host, port), Handler)
    print(f"universal-sql listening on http://{host}:{port}", flush=True)
    server.serve_forever()


if __name__ == "__main__":
    serve()