infra/__init__.py
"""Infrastructure packages for DIGEST Data Vault."""
infra/aws/__init__.py
"""AWS connector and website functions for DIG-590."""
infra/aws/claude_ws_deploy_policy.json
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "Dig590RolesReadAndCreate",
"Effect": "Allow",
"Action": ["iam:GetRole", "iam:CreateRole", "iam:PutRolePolicy", "iam:AttachRolePolicy"],
"Resource": "arn:aws:iam::523608012184:role/dig590-*"
},
{
"Sid": "Dig590RolesPassToLambdaOnly",
"Effect": "Allow",
"Action": "iam:PassRole",
"Resource": "arn:aws:iam::523608012184:role/dig590-*",
"Condition": {"StringEquals": {"iam:PassedToService": "lambda.amazonaws.com"}}
},
{
"Sid": "Dig590Functions",
"Effect": "Allow",
"Action": [
"lambda:GetFunction", "lambda:CreateFunction", "lambda:UpdateFunctionCode", "lambda:UpdateFunctionConfiguration",
"lambda:GetFunctionUrlConfig", "lambda:CreateFunctionUrlConfig", "lambda:UpdateFunctionUrlConfig",
"lambda:AddPermission", "lambda:RemovePermission", "lambda:PutFunctionConcurrency"
],
"Resource": "arn:aws:lambda:us-east-1:523608012184:function:dig590-*"
},
{
"Sid": "Dig590Logs",
"Effect": "Allow",
"Action": ["logs:CreateLogGroup", "logs:PutRetentionPolicy"],
"Resource": "arn:aws:logs:us-east-1:523608012184:log-group:/aws/lambda/dig590-*"
}
]
}
infra/aws/conversation_asgi.py
"""ASGI entry point for the Ask function's response streaming through the Lambda Web Adapter, in every environment."""
from __future__ import annotations
import json
import logging
import os
import time
from infra.aws.conversation_function import (
UnverifiedAnswerError,
answer,
validate_messages,
)
from infra.aws.web_functions import (
consume_daily_allowance,
environment,
response,
valid_access_code,
)
logger = logging.getLogger()
logger.setLevel(logging.INFO)
STARTED = time.perf_counter()
COLD = True
async def send_response(send, status: int, body: dict, content_type="application/json") -> None:
payload = response(status, body)
payload["headers"]["Content-Type"] = content_type
await send({
"type": "http.response.start", "status": status,
"headers": [(key.lower().encode(), value.encode()) for key, value in payload["headers"].items()],
})
await send({"type": "http.response.body", "body": payload["body"].encode()})
def log_start(question: str, streaming: bool) -> None:
global COLD
logger.info(json.dumps({
"event": "conversation_start", "environment": environment(), "question": question,
"streaming": streaming, "cold_start": COLD,
"since_import_ms": round((time.perf_counter() - STARTED) * 1000) if COLD else 0,
}))
COLD = False
async def app(scope, receive, send) -> None:
if scope["type"] != "http":
return
method = scope["method"]
if method == "GET":
await send_response(send, 200, {"status": "ok"})
return
if method == "OPTIONS":
await send_response(send, 204, {})
return
if method != "POST":
await send_response(send, 405, {"error": "invalid_request"})
return
body = b""
while True:
message = await receive()
body += message.get("body", b"")
if not message.get("more_body"):
break
try:
value = json.loads(body or b"{}")
if not isinstance(value, dict):
raise ValueError("invalid_request")
except (ValueError, UnicodeDecodeError):
await send_response(send, 400, {"error": "invalid_request"})
return
if not valid_access_code(value.get("access_code")):
await send_response(send, 401, {"error": "access_code"})
return
try:
messages = validate_messages(value.get("messages"))
except ValueError as error:
await send_response(send, 400, {"error": str(error)})
return
if not consume_daily_allowance("conversation", int(os.environ.get("DIGEST_DAILY_CAP", "200"))):
await send_response(send, 429, {"error": "daily_cap"})
return
question = next((item["content"] for item in reversed(messages) if item["role"] == "user"), "")
headers = dict(scope.get("headers", []))
streaming = b"text/event-stream" in headers.get(b"accept", b"")
log_start(question, streaming)
if not streaming:
try:
reply, queries, usage = await answer(messages)
except UnverifiedAnswerError:
logger.exception("Conversation answer could not be verified")
await send_response(send, 502, {"error": "unverified_answer"})
return
except Exception:
logger.exception("Conversation upstream failure")
await send_response(send, 502, {"error": "upstream"})
return
logger.info(json.dumps({"event": "conversation", "queries": queries, "usage": usage}))
await send_response(send, 200, {"reply": reply, "queries": queries})
return
await send({
"type": "http.response.start", "status": 200,
"headers": [
(b"content-type", b"text/event-stream"),
(b"cache-control", b"no-store"),
(b"x-accel-buffering", b"no"),
],
})
async def on_text(text: str) -> None:
# answer() releases text only once it can no longer trip the guard (D-13), so it streams as it comes.
if text:
payload = json.dumps({"text": text}).encode()
await send({
"type": "http.response.body",
"body": b"event: delta\ndata: " + payload + b"\n\n", "more_body": True,
})
try:
reply, queries, usage = await answer(messages, on_text=on_text)
except UnverifiedAnswerError:
logger.exception("Conversation answer could not be verified")
await send({
"type": "http.response.body",
"body": b'event: error\ndata: {"error":"unverified_answer"}\n\n', "more_body": False,
})
return
except Exception:
logger.exception("Conversation upstream failure")
await send({
"type": "http.response.body",
"body": b'event: error\ndata: {"error":"upstream"}\n\n', "more_body": False,
})
return
logger.info(json.dumps({"event": "conversation", "queries": queries, "usage": usage}))
final = json.dumps({"reply": reply, "queries": queries}).encode()
await send({
"type": "http.response.body",
"body": b"event: done\ndata: " + final + b"\n\n", "more_body": False,
})
infra/aws/conversation_function.py
"""Access-protected conversation function: website -> Claude -> DV Digest MCP."""
from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import time
import httpx2
from infra.aws.web_functions import (
consume_daily_allowance,
digest_session,
json_body,
preflight,
response,
secret,
environment,
valid_access_code,
)
SYSTEM_PROMPT = """You are DV Digest's assistant on a demonstration site prepared by Ray Serrano for AmeriLife's data architecture team. All data is invented for the DIGEST Data Vault course: an acquired agency, Sunbelt Senior, and a carrier, Northfield Life, with two deliveries on 2026-10-01 and 2026-10-02.
Answer only from DV Digest's tools. The current semantic-layer catalogue below provides every metric and its valid group_by dimensions, so no metadata discovery calls are needed. For each new user question, including a repeated question, request every query-metrics call you need together in your first response to that question; after those results arrive, answer that question directly without requesting more tools. Never state a number you did not get from a query in this conversation, and say which metric and grouping each number came from.
Speak only the user's business language. Never write a metric, dimension, table or column name, such as policies_held or carrier__carrier_name, in a reply: the catalogue's names are for your tool calls only. Name a measure by its label, such as Policies held or Coverage, current, and say what it means in plain words.
This release answers six questions: which policies each client is on, with which carrier and in what role; how much coverage each carrier provides through us; on how many policies the sources agree about the agent who sold it, and on how many they disagree; where each client's record came from and when it was first known; what was known about a policy on a date and what changed since; and how many records arrived in a form we don't accept, from which source, beside the rows that arrived without a business key.
What to know: Sunbelt Senior is the acquired agency and Northfield Life the carrier; Pinecrest Mutual is a second carrier that has sent only its own name so far. Face amounts appear only where a carrier sends them, and only Northfield Life does. One sale sent policy NF-100005 without its carrier, so the vault keeps that policy twice, once with Northfield Life and once with an unknown carrier, until a later release joins the two; counted by policy number it counts once, so the sources agree on 14 policies and disagree on 1. One sale arrived without a policy number and is shown as (unknown). When asked where the sources disagree about an agent, give both counts: where they agree and where they disagree.
If a question needs something this release does not answer, say so plainly and that it is planned for a later release; for example, the same person across client numbers and agencies.
For grouped questions, report only the requested grouped totals, not individual clients or policies. Offer the detailed listing rather than producing it unless the user explicitly asks for the individual details."""
logger = logging.getLogger()
logger.setLevel(logging.INFO)
semantic_catalogue_cache = None
semantic_catalogue_key = None
class UpstreamError(RuntimeError):
pass
def validate_messages(value) -> list[dict]:
if not isinstance(value, list) or not 1 <= len(value) <= 20:
raise ValueError("too_long")
messages = []
for message in value:
if not isinstance(message, dict) or message.get("role") not in {"user", "assistant"}:
raise ValueError("invalid_request")
content = message.get("content")
if not isinstance(content, str) or not 1 <= len(content) <= 2000:
raise ValueError("too_long")
messages.append({"role": message["role"], "content": content})
return messages
def result_text(result) -> str:
parts = [item.text for item in result.content if item.type == "text"]
if result.structured_content is not None:
return json.dumps(result.structured_content)
return "\n".join(parts)
def result_payload(result):
if result.structured_content is not None:
return result.structured_content.get("result", result.structured_content)
return json.loads(result_text(result))
NON_FIGURE_NUMBERS = re.compile(
r"\b(?:release\s+v?\d+(?:\.\d+)*|question\s+\d+|q-\d+|decision\s+\d+|d-\d+"
r"|\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{2,4}"
r"|(?:january|february|march|april|may|june|july|august|september|october|november|december)"
r"\s+\d{1,2}(?:,\s*\d{4})?)\b",
re.IGNORECASE,
)
def states_figure(reply: str) -> bool:
"""Ignore planning labels and dates; every other digit needs query evidence."""
return any(character.isdigit() for character in NON_FIGURE_NUMBERS.sub("", reply))
def is_throttled(error: BaseException) -> bool:
if isinstance(error, httpx2.HTTPStatusError):
return error.response.status_code == 429
if isinstance(error, BaseExceptionGroup):
return any(is_throttled(item) for item in error.exceptions)
return False
async def call_digest(session, name: str, arguments: dict):
for attempt in range(3):
try:
return await session.call_tool(name, arguments)
except Exception as error:
if not is_throttled(error) or attempt == 2:
raise
await asyncio.sleep(0.5 * (2**attempt))
USAGE_KEYS = (
"input_tokens", "output_tokens", "cache_creation_input_tokens",
"cache_read_input_tokens",
)
def cached_tools(tools: list[dict]) -> list[dict]:
"""Keep one stable cache breakpoint after the system prompt and tool list."""
cached = [dict(tool) for tool in tools]
if cached:
cached[-1]["cache_control"] = {"type": "ephemeral"}
return cached
async def semantic_catalogue(digest) -> list[dict]:
"""Read the current metric catalogue and valid dimensions from MetricFlow."""
listed = result_payload(await call_digest(digest, "list-metrics", {}))
dimension_results = []
for metric in listed:
dimension_results.append(await call_digest(
digest, "get-dimensions", {"metric_name": metric["name"]}
))
return [{
"name": metric["name"],
"description": metric.get("description", ""),
"group_by": result_payload(dimensions),
} for metric, dimensions in zip(listed, dimension_results)]
async def current_semantic_catalogue(digest) -> list[dict]:
"""Reuse the catalogue only while the connector's active manifest is unchanged."""
global semantic_catalogue_cache, semantic_catalogue_key
manifest_key = result_payload(await call_digest(digest, "get-manifest-version", {}))
cache_hit = semantic_catalogue_cache is not None and semantic_catalogue_key == manifest_key
if not cache_hit:
catalogue = await semantic_catalogue(digest)
semantic_catalogue_cache = catalogue
semantic_catalogue_key = manifest_key
print(json.dumps({
"event": "semantic_catalogue", "cache_hit": cache_hit,
"manifest_version": manifest_key,
}), flush=True)
return semantic_catalogue_cache
async def stream_completion(client, payload: dict, headers: dict, on_text) -> dict:
"""Collect one Anthropic SSE response while forwarding its text deltas."""
content = []
usage = {}
async with client.stream(
"POST", "https://api.anthropic.com/v1/messages", headers=headers,
json={**payload, "stream": True},
) as api_response:
if api_response.status_code >= 400:
await api_response.aread()
logger.error(json.dumps({
"event": "model_error", "status": api_response.status_code,
"body": api_response.text,
}))
api_response.raise_for_status()
async for line in api_response.aiter_lines():
if not line.startswith("data:"):
continue
event = json.loads(line[5:].strip())
kind = event.get("type")
if kind == "message_start":
usage.update(event.get("message", {}).get("usage", {}))
elif kind == "content_block_start":
block = event["content_block"]
content.append(dict(block))
if block.get("type") == "tool_use":
content[-1]["_partial_json"] = ""
elif kind == "content_block_delta":
index = event["index"]
delta = event["delta"]
if delta.get("type") == "text_delta":
text = delta.get("text", "")
content[index]["text"] = content[index].get("text", "") + text
await on_text(text)
elif delta.get("type") == "thinking_delta":
content[index]["thinking"] = (
content[index].get("thinking", "") + delta.get("thinking", "")
)
elif delta.get("type") == "signature_delta":
content[index]["signature"] = (
content[index].get("signature", "") + delta.get("signature", "")
)
elif delta.get("type") == "input_json_delta":
content[index]["_partial_json"] += delta.get("partial_json", "")
elif kind == "message_delta":
usage.update(event.get("usage", {}))
for block in content:
partial = block.pop("_partial_json", None)
if partial is not None:
block["input"] = json.loads(partial) if partial else {}
return {"content": content, "usage": usage}
class UnverifiedAnswerError(Exception):
"""The final reply states numbers but no query-metrics call ran for the question."""
async def answer(messages: list[dict], on_text=None) -> tuple[str, list[dict], dict]:
model = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-5-5")
usage = {key: 0 for key in USAGE_KEYS}
queries = []
tool_calls = 0
tools_enabled = True
model_call = 0
number_verification_retry = True
retry_streams_text = False
held, released = [], []
async def gated_text(text):
# D-13: hold streamed text until the question's first query-metrics call has run; then release
# what was held and stream the rest as it arrives. Text with no query behind it waits for the guard.
if queries:
if held:
await on_text("".join(held))
released.append("".join(held))
held.clear()
await on_text(text)
released.append(text)
else:
held.append(text)
async def release_accepted(reply):
# The guard accepted the reply: release whatever is still held, or the whole reply if nothing streamed.
if on_text is None:
return
if held:
await on_text("".join(held))
held.clear()
elif not released:
await on_text(reply)
async with digest_session() as digest:
listed = await digest.list_tools()
tools = [{
"name": tool.name,
"description": tool.description or "",
"input_schema": tool.input_schema,
} for tool in listed.tools if tool.name == "query-metrics"]
catalogue = await current_semantic_catalogue(digest)
system_prompt = (
f"{SYSTEM_PROMPT}\n\nCurrent semantic-layer catalogue:\n"
f"{json.dumps(catalogue, separators=(',', ':'))}"
)
async with httpx2.AsyncClient(timeout=60.0) as client:
while True:
model_call += 1
request_payload = {
"model": model,
"max_tokens": 1024,
"thinking": {"type": "between_tools"},
"output_config": {"effort": "low"},
"system": [{
"type": "text", "text": system_prompt,
"cache_control": {"type": "ephemeral"},
}],
"messages": messages,
}
if tools_enabled:
request_payload["tools"] = cached_tools(tools)
headers = {
"x-api-key": secret("anthropic-api-key"),
"anthropic-version": "2023-06-01",
}
started = time.perf_counter()
stream_text = None if (retry_streams_text or on_text is None) else gated_text
try:
if stream_text is None:
api_response = await client.post(
"https://api.anthropic.com/v1/messages",
headers=headers, json=request_payload,
)
if api_response.status_code >= 400:
logger.error(json.dumps({
"event": "model_error", "status": api_response.status_code,
"body": api_response.text,
}))
api_response.raise_for_status()
completion = api_response.json()
else:
completion = await stream_completion(
client, request_payload, headers, stream_text
)
except (httpx2.HTTPError, ValueError) as error:
raise UpstreamError("Claude request failed") from error
call_usage = completion.get("usage", {})
print(json.dumps({
"event": "model_call", "environment": environment(), "call": model_call,
"duration_ms": round((time.perf_counter() - started) * 1000),
"streaming": stream_text is not None, "usage": call_usage,
}), flush=True)
for key in usage:
usage[key] += call_usage.get(key, 0)
content = completion.get("content", [])
messages.append({"role": "assistant", "content": content})
uses = [block for block in content if block.get("type") == "tool_use"]
if not uses:
reply = "\n".join(
block.get("text", "") for block in content if block.get("type") == "text"
).strip()
if not reply:
raise UpstreamError("Claude returned no answer")
if not queries and states_figure(reply):
if number_verification_retry:
number_verification_retry = False
retry_streams_text = True
held.clear() # unverified text is never released
messages.append({
"role": "user",
"content": (
"Your reply states numbers, but no query-metrics call has run "
"for this question in this conversation. Every number in a "
"DV Digest answer must come from a query-metrics call. Run "
"the query-metrics calls you need, then answer the question "
"again."
),
})
continue
raise UnverifiedAnswerError(
"Answer could not be verified: it states numbers but no "
"query-metrics call ran for this question"
)
await release_accepted(reply)
return reply, queries, usage
tool_results = []
execution = []
for use in uses:
name = use.get("name")
if name not in {"list-metrics", "get-dimensions", "query-metrics"}:
raise UpstreamError("Claude requested an unknown tool")
if tool_calls >= 6:
tools_enabled = False
tool_results.append({
"type": "tool_result",
"tool_use_id": use["id"],
"content": (
"Six-tool-call limit reached. Answer from the evidence already "
"gathered, or state what is unavailable; do not request more tools."
),
"is_error": True,
})
continue
tool_calls += 1
execution.append((len(tool_results), use))
tool_results.append(None)
async def execute(use):
print(json.dumps({
"event": "tool_request", "environment": environment(), "name": use["name"],
"arguments": use.get("input") or {},
}), flush=True)
try:
return await call_digest(
digest, use["name"], use.get("input") or {}
)
except Exception as error:
raise UpstreamError("DV Digest request failed") from error
called_results = await asyncio.gather(
*(execute(use) for _, use in execution)
)
for (index, use), called in zip(execution, called_results):
arguments = use.get("input") or {}
if use["name"] == "query-metrics" and not called.is_error:
try:
payload = result_payload(called)
rows = len(payload.get("rows", []))
labels = payload.get("labels") or {}
except (ValueError, TypeError, AttributeError):
rows, labels = 0, {}
queries.append({
"metric": arguments.get("metric_name", ""),
"group_by": arguments.get("group_by") or [],
"rows": rows,
"label": labels.get("metric", ""),
"group_by_labels": labels.get("group_by") or [],
})
tool_results[index] = {
"type": "tool_result",
"tool_use_id": use["id"],
"content": result_text(called),
"is_error": called.is_error,
}
messages.append({"role": "user", "content": tool_results})
def handler(event, context):
early = preflight(event)
if early is not None:
return early
try:
body = json_body(event)
except (ValueError, UnicodeDecodeError):
return response(400, {"error": "invalid_request"})
if not valid_access_code(body.get("access_code")):
return response(401, {"error": "access_code"})
try:
messages = validate_messages(body.get("messages"))
except ValueError as error:
return response(400, {"error": str(error)})
cap = int(os.environ.get("DIGEST_DAILY_CAP", "200"))
if not consume_daily_allowance("conversation", cap):
return response(429, {"error": "daily_cap"})
question = next(
(item["content"] for item in reversed(messages) if item["role"] == "user"), ""
)
try:
reply, queries, usage = asyncio.run(answer(messages))
except UnverifiedAnswerError:
logger.exception("Conversation answer could not be verified")
return response(502, {"error": "unverified_answer"})
except Exception:
logger.exception("Conversation upstream failure")
return response(502, {"error": "upstream"})
logger.info(json.dumps({
"event": "conversation",
"environment": environment(),
"question": question,
"queries": queries,
"usage": usage,
}))
return response(200, {"reply": reply, "queries": queries})
infra/aws/dbt_cloud.py
"""Minimal dbt Cloud client for governed semantic-manifest artifacts."""
from __future__ import annotations
import json
import os
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
class LocalManifestSource:
"""Reads the semantic manifest packaged with the dev connector."""
def __init__(self, path: Path | None = None) -> None:
self.path = path or Path(__file__).resolve().with_name("dev_semantic_manifest.json")
def latest_successful_run_id(self) -> int:
return 1
def semantic_manifest(self, run_id: int) -> str:
if run_id != 1:
raise RuntimeError(f"Unknown local manifest run {run_id}")
return self.path.read_text(encoding="utf-8")
class DbtCloudManifestSource:
"""Reads the latest successful QA run and its optional semantic manifest."""
def __init__(
self,
token: str,
host: str,
account_id: int,
job_id: int,
timeout_seconds: int = 30,
) -> None:
self.token = token
self.host = host.rstrip("/")
if not self.host.startswith("https://"):
self.host = f"https://{self.host}"
self.account_id = account_id
self.job_id = job_id
self.timeout_seconds = timeout_seconds
@classmethod
def from_environment(cls) -> "DbtCloudManifestSource":
required = (
"DBT_CLOUD_TOKEN", "DBT_CLOUD_HOST", "DBT_CLOUD_ACCOUNT_ID", "DBT_CLOUD_JOB_ID"
)
missing = [name for name in required if not os.environ.get(name)]
if missing:
raise RuntimeError(f"Required dbt Cloud settings are missing: {', '.join(missing)}")
return cls(
token=os.environ["DBT_CLOUD_TOKEN"],
host=os.environ["DBT_CLOUD_HOST"],
account_id=int(os.environ["DBT_CLOUD_ACCOUNT_ID"]),
job_id=int(os.environ["DBT_CLOUD_JOB_ID"]),
)
def _get(self, url: str) -> bytes:
request = Request(
url,
headers={
"Authorization": f"Bearer {self.token}",
"Accept": "application/json",
"User-Agent": "dv-digest-mcp",
},
)
with urlopen(request, timeout=self.timeout_seconds) as response:
return response.read()
def latest_successful_run_id(self) -> int | None:
query = urlencode(
{
"job_definition_id": self.job_id,
"status": 10,
"order_by": "-finished_at",
"limit": 1,
}
)
url = f"{self.host}/api/v2/accounts/{self.account_id}/runs/?{query}"
payload = json.loads(self._get(url))
runs = payload.get("data", [])
return int(runs[0]["id"]) if runs else None
def semantic_manifest(self, run_id: int) -> str | None:
url = (
f"{self.host}/api/v2/accounts/{self.account_id}/runs/"
f"{run_id}/artifacts/semantic_manifest.json"
)
try:
return self._get(url).decode("utf-8")
except HTTPError as error:
if error.code == 404:
return None
raise
infra/aws/deploy_functions.py
"""Deploy DIG-590's conversation and GitHub invitation functions by stage.
Required 1Password references are supplied through OP_ANTHROPIC_REFERENCE and
OP_ACCESS_CODE_REFERENCE. Resolved values go only to SecureString parameters.
"""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import sys
import time
import boto3
from botocore.exceptions import ClientError
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from infra.aws.deploy_lambda import build_package, op_read, wait_for_update
ACCOUNT = "523608012184"
MCP_URLS = {
"qa": "https://tmbe7c62hzbuvr6bxys2ic2ot40awyoh.lambda-url.us-east-1.on.aws/mcp",
"prod": "https://c7h6qyma3ykmda354i27jfgfei0cjtla.lambda-url.us-east-1.on.aws/mcp",
}
ORIGINS = {
"dev": "http://localhost:8000",
"qa": ["https://qa.dvdigest.integrationals.com", "http://localhost:8000"],
"prod": "https://dvdigest.integrationals.com",
}
def conversation_mcp_url(stage: str) -> str:
if stage != "dev":
return MCP_URLS[stage]
url = os.environ.get("DIGEST_DEV_MCP_URL", "").strip()
if not url:
raise RuntimeError("DIGEST_DEV_MCP_URL is required for the dev conversation deployment")
return url
def ensure_secret(ssm, prefix: str, suffix: str, reference: str) -> None:
if not reference:
raise RuntimeError(f"An approved 1Password reference is required for {suffix}")
ssm.put_parameter(
Name=f"{prefix}/{suffix}",
Value=op_read(reference),
Type="SecureString",
Tier="Standard",
Overwrite=True,
)
def ensure_role(iam, name: str, account: str, region: str, prefix: str,
secret_suffixes: list[str], kinds: list[str], connector_arn: str | None) -> tuple[str, bool]:
trust = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}],
}
created = False
try:
role = iam.get_role(RoleName=name)["Role"]
except ClientError as error:
if error.response["Error"]["Code"] != "NoSuchEntity":
raise
role = iam.create_role(
RoleName=name,
AssumeRolePolicyDocument=json.dumps(trust),
Description="DIG-590 access-protected website function",
)["Role"]
created = True
iam.attach_role_policy(
RoleName=name,
PolicyArn="arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
)
statements = [
{
"Effect": "Allow",
"Action": "ssm:GetParameter",
"Resource": [
f"arn:aws:ssm:{region}:{account}:parameter{prefix}/{suffix}"
for suffix in secret_suffixes
],
},
{
"Effect": "Allow",
"Action": ["ssm:GetParameter", "ssm:PutParameter"],
"Resource": [
f"arn:aws:ssm:{region}:{account}:parameter{prefix}/daily-{kind}-*"
for kind in kinds
],
},
]
if connector_arn:
statements.extend([
{
"Effect": "Allow",
"Action": "lambda:InvokeFunctionUrl",
"Resource": connector_arn,
"Condition": {"StringEquals": {"lambda:FunctionUrlAuthType": "AWS_IAM"}},
},
{
"Effect": "Allow",
"Action": "lambda:InvokeFunction",
"Resource": connector_arn,
"Condition": {"Bool": {"lambda:InvokedViaFunctionUrl": "true"}},
},
])
iam.put_role_policy(
RoleName=name,
PolicyName="dig590-web-function-access",
PolicyDocument=json.dumps({"Version": "2012-10-17", "Statement": statements}),
)
return role["Arn"], created
def ensure_log_group(logs, function_name: str) -> str:
name = f"/aws/lambda/{function_name}"
try:
logs.create_log_group(logGroupName=name)
except ClientError as error:
if error.response["Error"]["Code"] != "ResourceAlreadyExistsException":
raise
logs.put_retention_policy(logGroupName=name, retentionInDays=14)
return name
def ensure_function(lambda_client, name: str, role_arn: str, role_created: bool,
handler: str, timeout: int, memory: int, environment: dict,
package: Path, log_group: str, layers: list[str] | None = None) -> None:
configuration = {
"Runtime": "python3.12",
"Role": role_arn,
"Handler": handler,
"Timeout": timeout,
"MemorySize": memory,
"Environment": {"Variables": environment},
"Description": "DIG-590 access-protected website function",
"LoggingConfig": {"LogGroup": log_group},
"Layers": layers or [],
}
try:
lambda_client.get_function(FunctionName=name)
exists = True
except ClientError as error:
if error.response["Error"]["Code"] not in {"ResourceNotFound", "ResourceNotFoundException"}:
raise
exists = False
if exists:
lambda_client.update_function_code(FunctionName=name, ZipFile=package.read_bytes())
wait_for_update(lambda_client, name)
lambda_client.update_function_configuration(FunctionName=name, **configuration)
wait_for_update(lambda_client, name)
else:
if role_created:
time.sleep(10)
lambda_client.create_function(
FunctionName=name,
Code={"ZipFile": package.read_bytes()},
Architectures=["x86_64"],
PackageType="Zip",
Publish=False,
**configuration,
)
lambda_client.get_waiter("function_active").wait(
FunctionName=name, WaiterConfig={"Delay": 2, "MaxAttempts": 60}
)
lambda_client.put_function_concurrency(FunctionName=name, ReservedConcurrentExecutions=2)
def ensure_public_url(lambda_client, name: str, origin: str | list[str], invoke_mode="BUFFERED") -> str:
cors = {
"AllowOrigins": origin if isinstance(origin, list) else [origin],
"AllowMethods": ["POST"],
"AllowHeaders": ["content-type"],
"MaxAge": 300,
}
try:
config = lambda_client.get_function_url_config(FunctionName=name)
if (config["AuthType"] != "NONE" or config.get("Cors") != cors
or config.get("InvokeMode") != invoke_mode):
config = lambda_client.update_function_url_config(
FunctionName=name, AuthType="NONE", InvokeMode=invoke_mode, Cors=cors
)
except ClientError as error:
if error.response["Error"]["Code"] not in {"ResourceNotFound", "ResourceNotFoundException"}:
raise
config = lambda_client.create_function_url_config(
FunctionName=name, AuthType="NONE", InvokeMode=invoke_mode, Cors=cors
)
permissions = [
{
"StatementId": "dig590-public-function-url",
"Action": "lambda:InvokeFunctionUrl",
"FunctionUrlAuthType": "NONE",
},
{
"StatementId": "dig590-public-function-invoke",
"Action": "lambda:InvokeFunction",
"InvokedViaFunctionUrl": True,
},
]
for permission in permissions:
try:
lambda_client.add_permission(FunctionName=name, Principal="*", **permission)
except ClientError as error:
if error.response["Error"]["Code"] != "ResourceConflictException":
raise
return config["FunctionUrl"].rstrip("/") + "/"
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--stage", choices=["dev", "qa", "prod"], required=True)
parser.add_argument("--package-only", action="store_true")
args = parser.parse_args()
package, compressed, uncompressed = build_package(args.stage)
print(f"Built Lambda package: {compressed / 1024 / 1024:.1f} MiB zip, "
f"{uncompressed / 1024 / 1024:.1f} MiB unzipped")
if args.package_only:
print(package)
return
session = boto3.Session(region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
account = session.client("sts").get_caller_identity()["Account"]
if account != ACCOUNT:
raise RuntimeError(f"Expected AWS account {ACCOUNT}, got {account}")
region = session.region_name
prefix = f"/dig590/dv-digest/{args.stage}"
ssm = session.client("ssm")
ensure_secret(ssm, prefix, "anthropic-api-key", os.environ.get("OP_ANTHROPIC_REFERENCE", ""))
ensure_secret(ssm, prefix, "access-code", os.environ.get("OP_ACCESS_CODE_REFERENCE", ""))
if args.stage != "dev":
ensure_secret(
ssm, prefix, "github-invite-token",
"op://Agentic Workstation/dv-digest-invite API Credential/credential",
)
iam = session.client("iam")
logs = session.client("logs")
lambda_client = session.client("lambda")
connector_name = f"dig590-dv-digest-{args.stage}"
connector_arn = f"arn:aws:lambda:{region}:{account}:function:{connector_name}"
base_environment = {
"SSM_PARAMETER_PREFIX": prefix,
"DIGEST_STAGE": args.stage,
}
streaming = True
adapter_layer = f"arn:aws:lambda:{region}:753240598075:layer:LambdaAdapterLayerX86:30"
functions = [
{
"name": f"dig590-conversation-{args.stage}",
"handler": "run.sh" if streaming else "infra.aws.conversation_function.handler",
"timeout": 300,
"memory": 512,
"secrets": ["anthropic-api-key", "access-code"],
"kinds": ["conversation"],
"connector_arn": connector_arn if args.stage == "prod" else None,
"layers": [adapter_layer] if streaming else [],
"invoke_mode": "RESPONSE_STREAM" if streaming else "BUFFERED",
"environment": {
**base_environment,
"DIGEST_MCP_URL": conversation_mcp_url(args.stage),
"DIGEST_DAILY_CAP": "200",
"ANTHROPIC_MODEL": os.environ.get(
"ANTHROPIC_MODEL", "claude-sonnet-5-5"
),
**({
"AWS_LAMBDA_EXEC_WRAPPER": "/opt/bootstrap",
"AWS_LWA_INVOKE_MODE": "response_stream",
"AWS_LWA_PORT": "8080",
} if streaming else {}),
},
},
*([] if args.stage == "dev" else [{
"name": f"dig590-invitation-{args.stage}",
"handler": "infra.aws.invitation_function.handler",
"timeout": 30,
"memory": 256,
"secrets": ["github-invite-token", "access-code"],
"kinds": ["invite"],
"connector_arn": None,
"layers": [],
"invoke_mode": "BUFFERED",
"environment": {**base_environment, "DIGEST_INVITE_DAILY_CAP": "5"},
}]),
]
for function in functions:
role_arn, role_created = ensure_role(
iam, function["name"], account, region, prefix,
function["secrets"], function["kinds"], function["connector_arn"],
)
log_group = ensure_log_group(logs, function["name"])
ensure_function(
lambda_client, function["name"], role_arn, role_created,
function["handler"], function["timeout"], function["memory"],
function["environment"], package, log_group, function["layers"],
)
url = ensure_public_url(
lambda_client, function["name"], ORIGINS[args.stage], function["invoke_mode"]
)
print(f"Deployed {function['name']}")
print(f"Function URL: {url}")
if __name__ == "__main__":
main()
infra/aws/deploy_lambda.py
"""Build and deploy DIG-590 DV Digest with one command.
Example: python infra/aws/deploy_lambda.py --stage qa
Secrets are read from 1Password at runtime and stored only as SecureString
parameters. The script never prints or saves resolved secret values.
"""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time
import zipfile
import boto3
from botocore.exceptions import ClientError
ROOT = Path(__file__).resolve().parents[2]
CONNECTOR = ROOT / "infra" / "aws"
BUILD_ROOT = ROOT / "build"
ZIP_PATH = BUILD_ROOT / "dv-digest-lambda.zip"
JOB_IDS = {"qa": "70403104452684", "prod": "70403104453080"}
SECRET_REFERENCES = {
"snowflake-private-key": "op://Agentic Workstation/Snowflake CLAUDE_WS key/Key normalized",
"dbt-cloud-token": "op://Agentic Workstation/Claude dbt PAT/credential",
}
def run(command: list[str]) -> None:
subprocess.run(command, cwd=ROOT, check=True)
def build_package(stage: str) -> tuple[Path, int, int]:
import tempfile
local_manifest = None
if stage == "dev":
source_manifest = ROOT / "dbt" / "target" / "semantic_manifest.json"
if not source_manifest.is_file():
raise RuntimeError(f"Dev semantic manifest is missing: {source_manifest}")
local_manifest = json.loads(source_manifest.read_text(encoding="utf-8"))
if not local_manifest.get("metrics"):
raise RuntimeError("Dev semantic manifest contains no metrics")
BUILD_ROOT.mkdir(exist_ok=True)
package_dir = Path(tempfile.mkdtemp(prefix="lambda-", dir=BUILD_ROOT))
uv = shutil.which("uv")
if uv:
run([
uv, "pip", "install",
"--target", str(package_dir),
"--python-platform", "x86_64-manylinux2014",
"--python-version", "3.12",
"--only-binary=:all:",
"--link-mode=copy",
"-r", str(CONNECTOR / "requirements-lambda.txt"),
])
elif os.name == "nt":
raise RuntimeError("uv is required to build Linux Lambda packages on Windows")
else:
run([
sys.executable, "-m", "pip", "install",
"--target", str(package_dir),
"--platform", "manylinux2014_x86_64",
"--python-version", "3.12",
"--implementation", "cp",
"--abi", "cp312", "--abi", "abi3", "--abi", "none",
"--only-binary=:all:",
"-r", str(CONNECTOR / "requirements-lambda.txt"),
])
shutil.copytree(
CONNECTOR,
package_dir / "infra" / "aws",
dirs_exist_ok=True,
ignore=shutil.ignore_patterns(
"__pycache__", "*.pyc", "test_*", "deploy_lambda.py"
),
)
shutil.copy2(
ROOT / "infra" / "__init__.py", package_dir / "infra" / "__init__.py"
)
if local_manifest is not None:
(package_dir / "infra" / "aws").mkdir(parents=True, exist_ok=True)
(package_dir / "infra" / "aws" / "dev_semantic_manifest.json").write_text(
json.dumps(local_manifest), encoding="utf-8"
)
# Every environment's Ask function streams through the Lambda web adapter, which starts run.sh; the package
# carries it for every environment, so dev, QA and production run alike (Ray, 2026-11-01, answer 1a).
startup = package_dir / "run.sh"
shutil.copy2(CONNECTOR / "run.sh", startup)
startup.chmod(0o755)
for cache in package_dir.rglob("__pycache__"):
shutil.rmtree(cache)
uncompressed = sum(path.stat().st_size for path in package_dir.rglob("*") if path.is_file())
if uncompressed > 250 * 1024 * 1024:
raise RuntimeError(f"Lambda package exceeds 250 MiB unzipped: {uncompressed} bytes")
with zipfile.ZipFile(ZIP_PATH, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
for path in sorted(package_dir.rglob("*")):
if path.is_file():
archive.write(path, path.relative_to(package_dir))
compressed = ZIP_PATH.stat().st_size
if compressed > 50 * 1024 * 1024:
raise RuntimeError(f"Lambda zip exceeds the 50 MiB direct-upload limit: {compressed} bytes")
shutil.rmtree(package_dir, ignore_errors=True)
return ZIP_PATH, compressed, uncompressed
def op_executable() -> str:
candidates = [
os.environ.get("OP_CLI"),
str(ROOT.parent / "op-cli" / "op.exe"),
shutil.which("op"),
]
for candidate in candidates:
if candidate and Path(candidate).is_file():
return candidate
raise RuntimeError("1Password CLI is unavailable; set OP_CLI or add op to PATH")
def op_read(reference: str) -> str:
value = subprocess.check_output(
[op_executable(), "read", reference], text=True, encoding="utf-8"
).strip()
if not value:
raise RuntimeError("An approved 1Password reference returned no value")
return value
def secret_references(stage: str) -> dict[str, str]:
if stage == "dev":
return {"snowflake-private-key": SECRET_REFERENCES["snowflake-private-key"]}
return SECRET_REFERENCES
def ensure_parameters(ssm, prefix: str, stage: str) -> None:
for suffix, reference in secret_references(stage).items():
ssm.put_parameter(
Name=f"{prefix}/{suffix}",
Value=op_read(reference),
Type="SecureString",
Tier="Standard",
Overwrite=True,
)
def ensure_role(iam, role_name: str, account: str, region: str, prefix: str, stage: str) -> tuple[str, bool]:
trust = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}],
}
created = False
try:
role = iam.get_role(RoleName=role_name)["Role"]
except ClientError as error:
if error.response["Error"]["Code"] != "NoSuchEntity":
raise
role = iam.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=json.dumps(trust),
Description="DIG-590 DV Digest Lambda; Parameter Store and CloudWatch only",
)["Role"]
created = True
iam.attach_role_policy(
RoleName=role_name,
PolicyArn="arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
)
parameter_arns = [
f"arn:aws:ssm:{region}:{account}:parameter{prefix}/{suffix}"
for suffix in secret_references(stage)
]
policy = {
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["ssm:GetParameter", "ssm:GetParameters"],
"Resource": parameter_arns,
}],
}
iam.put_role_policy(
RoleName=role_name,
PolicyName="dig590-read-dv-digest-secrets",
PolicyDocument=json.dumps(policy),
)
return role["Arn"], created
def ensure_log_group(logs, function_name: str) -> str:
name = f"/aws/lambda/{function_name}"
try:
logs.create_log_group(logGroupName=name)
except ClientError as error:
if error.response["Error"]["Code"] != "ResourceAlreadyExistsException":
raise
logs.put_retention_policy(logGroupName=name, retentionInDays=14)
return name
def wait_for_update(lambda_client, function_name: str) -> None:
lambda_client.get_waiter("function_updated").wait(
FunctionName=function_name,
WaiterConfig={"Delay": 2, "MaxAttempts": 60},
)
def ensure_function(lambda_client, function_name: str, role_arn: str, role_created: bool,
stage: str, prefix: str, environment: dict[str, str], package: Path,
log_group: str) -> None:
configuration = {
"Runtime": "python3.12",
"Role": role_arn,
"Handler": "infra.aws.lambda_handler.handler",
"Timeout": 60,
"MemorySize": 1024,
"Environment": {"Variables": environment},
"Description": f"DIG-590 stateless MCP DV Digest ({stage})",
"LoggingConfig": {"LogGroup": log_group},
}
try:
lambda_client.get_function(FunctionName=function_name)
exists = True
except ClientError as error:
if error.response["Error"]["Code"] not in {"ResourceNotFound", "ResourceNotFoundException"}:
raise
exists = False
if exists:
lambda_client.update_function_code(FunctionName=function_name, ZipFile=package.read_bytes())
wait_for_update(lambda_client, function_name)
lambda_client.update_function_configuration(FunctionName=function_name, **configuration)
wait_for_update(lambda_client, function_name)
else:
if role_created:
time.sleep(10)
lambda_client.create_function(
FunctionName=function_name,
Code={"ZipFile": package.read_bytes()},
Architectures=["x86_64"],
PackageType="Zip",
Publish=False,
**configuration,
)
lambda_client.get_waiter("function_active").wait(
FunctionName=function_name,
WaiterConfig={"Delay": 2, "MaxAttempts": 60},
)
lambda_client.put_function_concurrency(
FunctionName=function_name,
ReservedConcurrentExecutions=6,
)
def ensure_function_url(lambda_client, function_name: str, stage: str) -> str:
public = stage in {"dev", "qa"}
auth_type = "NONE" if public else "AWS_IAM"
try:
config = lambda_client.get_function_url_config(FunctionName=function_name)
if config["AuthType"] != auth_type:
config = lambda_client.update_function_url_config(
FunctionName=function_name,
AuthType=auth_type,
InvokeMode="BUFFERED",
)
except ClientError as error:
if error.response["Error"]["Code"] not in {"ResourceNotFound", "ResourceNotFoundException"}:
raise
config = lambda_client.create_function_url_config(
FunctionName=function_name,
AuthType=auth_type,
InvokeMode="BUFFERED",
)
# Function URLs created since October 2025 require both public grants for NONE.
permissions = [
{
"StatementId": "dig590-public-function-url",
"Action": "lambda:InvokeFunctionUrl",
"FunctionUrlAuthType": "NONE",
},
{
"StatementId": "dig590-public-function-invoke",
"Action": "lambda:InvokeFunction",
"InvokedViaFunctionUrl": True,
},
]
for permission in permissions:
try:
if public:
lambda_client.add_permission(
FunctionName=function_name,
Principal="*",
**permission,
)
else:
lambda_client.remove_permission(
FunctionName=function_name,
StatementId=permission["StatementId"],
)
except ClientError as error:
ignored = "ResourceConflictException" if public else "ResourceNotFoundException"
if error.response["Error"]["Code"] != ignored:
raise
return config["FunctionUrl"].rstrip("/") + "/mcp"
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--stage", choices=["dev", "qa", "prod"], default="qa")
parser.add_argument("--package-only", action="store_true")
args = parser.parse_args()
package, compressed, uncompressed = build_package(args.stage)
print(f"Built Lambda package: {compressed / 1024 / 1024:.1f} MiB zip, "
f"{uncompressed / 1024 / 1024:.1f} MiB unzipped")
if args.package_only:
print(str(package))
return
session = boto3.Session(region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
account = session.client("sts").get_caller_identity()["Account"]
function_name = f"dig590-dv-digest-{args.stage}"
role_name = f"dig590-dv-digest-{args.stage}"
prefix = f"/dig590/dv-digest/{args.stage}"
ssm = session.client("ssm")
ensure_parameters(ssm, prefix, args.stage)
role_arn, role_created = ensure_role(
session.client("iam"), role_name, account, session.region_name, prefix, args.stage
)
log_group = ensure_log_group(session.client("logs"), function_name)
environment = {
"LOAD_SSM_SECRETS": "true",
"SSM_PARAMETER_PREFIX": prefix,
"SNOWFLAKE_USER": "CLAUDE_WS",
"SNOWFLAKE_ACCOUNT": "mgb11914.us-east-1",
"SNOWFLAKE_ROLE": "CLAUDE_WS_ROLE",
"SNOWFLAKE_WAREHOUSE": "COMPUTE_WH",
"DIGEST_STAGE": args.stage,
}
if args.stage != "dev":
environment.update({
"DBT_CLOUD_ACCOUNT_ID": "48669",
"DBT_CLOUD_JOB_ID": JOB_IDS[args.stage],
"DBT_CLOUD_HOST": op_read("op://Agentic Workstation/Claude dbt PAT/hostname"),
"DBT_CLOUD_POLL_SECONDS": "0",
})
lambda_client = session.client("lambda")
ensure_function(
lambda_client, function_name, role_arn, role_created, args.stage,
prefix, environment, package, log_group,
)
url = ensure_function_url(lambda_client, function_name, args.stage)
print(f"Deployed {function_name}")
print(f"MCP URL: {url}")
if __name__ == "__main__":
main()
infra/aws/invitation_function.py
"""Access-protected GitHub contributor invitation function."""
from __future__ import annotations
import json
import logging
import os
import re
import httpx2
from infra.aws.web_functions import (
consume_daily_allowance,
json_body,
preflight,
release_daily_allowance,
response,
secret,
valid_access_code,
)
USERNAME = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9]|-(?=[A-Za-z0-9])){0,38}")
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def invite(username: str) -> dict:
headers = {
"Authorization": f"Bearer {secret('github-invite-token')}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
"User-Agent": "dig590-dv-digest-invite",
}
try:
with httpx2.Client(timeout=30.0, headers=headers) as client:
user = client.get(f"https://api.github.com/users/{username}")
if user.status_code == 404:
return {"status": 404, "body": {"error": "no_such_user"}}
user.raise_for_status()
collaborator_url = (
"https://api.github.com/repos/rayserrano2735/digest-datavault/"
f"collaborators/{username}"
)
membership = client.get(collaborator_url)
if membership.status_code == 204:
return {"status": 200, "body": {"already": True}}
if membership.status_code != 404:
membership.raise_for_status()
cap = int(os.environ.get("DIGEST_INVITE_DAILY_CAP", "5"))
if not consume_daily_allowance("invite", cap):
return {"status": 429, "body": {"error": "daily_cap"}}
sent = False
try:
invitation = client.put(collaborator_url, json={"permission": "push"})
if invitation.status_code in {200, 201}:
sent = True
return {"status": 200, "body": {"invited": True}}
if invitation.status_code == 204:
return {"status": 200, "body": {"already": True}}
invitation.raise_for_status()
finally:
if not sent:
release_daily_allowance("invite")
except httpx2.HTTPError as error:
raise RuntimeError("GitHub invitation failed") from error
raise RuntimeError("GitHub returned an unexpected invitation response")
def handler(event, context):
early = preflight(event)
if early is not None:
return early
try:
body = json_body(event)
except (ValueError, UnicodeDecodeError):
return response(400, {"error": "invalid_request"})
if not valid_access_code(body.get("access_code")):
return response(401, {"error": "access_code"})
username = body.get("github_username")
if not isinstance(username, str) or not USERNAME.fullmatch(username):
return response(404, {"error": "no_such_user"})
try:
result = invite(username)
except RuntimeError:
logger.exception("GitHub invitation upstream failure")
result = {"status": 502, "body": {"error": "upstream"}}
logger.info(json.dumps({
"event": "repository_invitation",
"github_username": username,
"status": result["status"],
"outcome": result["body"],
}))
return response(result["status"], result["body"])
infra/aws/lambda_config.py
"""Cold-start secret loading for the Lambda connector."""
import os
import boto3
PARAMETER_FIELDS = {
"snowflake-private-key": "SNOWFLAKE_PRIVATE_KEY",
"dbt-cloud-token": "DBT_CLOUD_TOKEN",
}
def load_secrets() -> None:
"""Load approved SecureString values into memory, never into files or logs."""
if os.environ.get("LOAD_SSM_SECRETS", "true").lower() in {"0", "false", "no"}:
return
prefix = os.environ.get("SSM_PARAMETER_PREFIX", "").rstrip("/")
if not prefix:
raise RuntimeError("SSM_PARAMETER_PREFIX is required")
fields = dict(PARAMETER_FIELDS)
if os.environ.get("DIGEST_STAGE") == "dev":
fields.pop("dbt-cloud-token")
names = {f"{prefix}/{suffix}": environment for suffix, environment in fields.items()}
response = boto3.client("ssm").get_parameters(Names=list(names), WithDecryption=True)
if response.get("InvalidParameters"):
raise RuntimeError("Required Lambda secret parameters are unavailable")
values = {parameter["Name"]: parameter["Value"] for parameter in response["Parameters"]}
for name, environment in names.items():
value = values.get(name)
if not value:
raise RuntimeError("A required Lambda secret parameter is empty")
os.environ[environment] = value
infra/aws/lambda_handler.py
"""AWS Lambda Function URL entry point for stateless MCP DV Digest."""
import logging
import os
from mangum import Mangum
from mcp.server.transport_security import TransportSecuritySettings
from infra.aws.lambda_config import load_secrets
# Lambda does not provide a reliable background-thread lifecycle. Cold start
# loads the latest successful manifest; warm invocations retain the last good one.
os.environ.setdefault("DBT_CLOUD_POLL_SECONDS", "0")
logging.getLogger("snowflake").setLevel(logging.WARNING)
logging.getLogger("boto3").setLevel(logging.WARNING)
logging.getLogger("botocore").setLevel(logging.WARNING)
load_secrets()
from infra.aws.mcp_server import mcp # noqa: E402
def handler(event, context):
# MCP session managers are single-use. A fresh stateless transport per
# invocation avoids carrying a stopped manager into the next request.
app = mcp.streamable_http_app(
streamable_http_path="/mcp",
stateless_http=True,
json_response=True,
transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
)
return Mangum(app, lifespan="auto")(event, context)
infra/aws/mcp_server.py
"""DV Digest MCP service using MetricFlow and Snowflake."""
import json
import os
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.transport_security import TransportSecuritySettings
from infra.aws.metricflow_backend import MetricFlowBackend
environment = os.environ.get("DIGEST_STAGE")
if not environment:
raise RuntimeError("DIGEST_STAGE is required")
mcp = MCPServer(f"dv-digest-{environment}", version="1.0.0")
backend = MetricFlowBackend()
@mcp.tool(name="list-metrics", structured_output=False)
def list_metrics() -> str:
# One JSON array in one text block, rather than concatenated JSON objects.
return json.dumps(backend.list_metrics())
@mcp.tool(name="get-manifest-version")
def get_manifest_version() -> str:
"""Identify the manifest behind the current metrics and dimensions."""
return str(backend.active_run_id)
@mcp.tool(name="get-dimensions")
def get_dimensions(metric_name: str) -> list[str]:
try:
return backend.get_dimensions(metric_name)
except ValueError as error:
choices = ", ".join(metric["name"] for metric in backend.list_metrics())
raise ToolError(f"unknown metric {metric_name}; the metrics are {choices}") from error
@mcp.tool(name="query-metrics")
def query_metrics(metric_name: str, group_by: list[str] | None = None) -> dict:
group_by = group_by or []
dimensions = get_dimensions(metric_name)
for dimension in group_by:
if dimension not in dimensions:
raise ToolError(
f"unknown dimension {dimension}; this metric's dimensions are {', '.join(dimensions)}"
)
print(json.dumps({
"event": "metric_query_start", "metric": metric_name, "group_by": group_by,
}), flush=True)
try:
result = backend.query(metric_name, group_by)
print(json.dumps({
"event": "metric_query_success", "metric": metric_name, "group_by": group_by,
}), flush=True)
return result
except Exception as error:
print(json.dumps({
"event": "metric_query_error", "metric": metric_name,
"group_by": group_by, "error": repr(error),
}), flush=True)
raise
if __name__ == "__main__":
mcp.run(
transport="streamable-http",
host="127.0.0.1",
port=8000,
transport_security=TransportSecuritySettings(
enable_dns_rebinding_protection=True,
allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
allowed_origins=[
"http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"
],
),
)
infra/aws/metricflow_backend.py
"""MetricFlow backend with dbt Cloud semantic-manifest loading."""
from __future__ import annotations
from datetime import datetime, timezone
import logging
import os
import threading
import snowflake.connector
from cryptography.hazmat.primitives import serialization
from metricflow.engine.metricflow_engine import MetricFlowEngine, MetricFlowQueryRequest
from metricflow.protocols.sql_client import SqlEngine
from metricflow.sql.render.snowflake import SnowflakeSqlPlanRenderer
from metricflow_semantics.model.dbt_manifest_parser import parse_manifest_from_dbt_generated_manifest
from metricflow_semantics.model.semantic_manifest_lookup import SemanticManifestLookup
from infra.aws.dbt_cloud import DbtCloudManifestSource, LocalManifestSource
logger = logging.getLogger(__name__)
class CompileOnlySnowflakeClient:
sql_engine_type = SqlEngine.SNOWFLAKE
sql_plan_renderer = SnowflakeSqlPlanRenderer()
def query(self, *args, **kwargs):
raise RuntimeError("This client compiles SQL but does not execute it")
def execute(self, *args, **kwargs):
raise RuntimeError("This client compiles SQL but does not execute it")
def dry_run(self, *args, **kwargs):
raise RuntimeError("This client compiles SQL but does not execute it")
def close(self):
return None
def render_bind_parameter_key(self, bind_parameter_key: str) -> str:
return f"%({bind_parameter_key})s"
class SnowflakeRunner:
"""Runs compiled SQL on Snowflake. Opens its session lazily and opens a fresh one when Snowflake says the old
one can no longer be renewed (error 250001, after the session's master token expires; seen 2026-10-27 after
the server had run for hours), so a long-running server never needs a restart for it."""
def __init__(self) -> None:
private_key = serialization.load_pem_private_key(
os.environ["SNOWFLAKE_PRIVATE_KEY"].encode(), password=None
)
self._connect_args = dict(
user=os.environ["SNOWFLAKE_USER"],
account=os.environ["SNOWFLAKE_ACCOUNT"],
authenticator="SNOWFLAKE_JWT",
private_key=private_key.private_bytes(
serialization.Encoding.DER,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
),
role=os.environ["SNOWFLAKE_ROLE"],
warehouse=os.environ["SNOWFLAKE_WAREHOUSE"],
client_session_keep_alive=True,
session_parameters={"TIMEZONE": "UTC"},
)
self.connection = None
self._lock = threading.Lock()
def _connection(self, fresh: bool = False):
with self._lock:
if fresh and self.connection is not None:
try:
self.connection.close()
except Exception:
pass
self.connection = None
if self.connection is None or self.connection.is_closed():
self.connection = snowflake.connector.connect(**self._connect_args)
# The session runs as its role alone: the user's secondary roles, SYSADMIN among them, would unmask the
# insured's details and give the connector far more than it reads (QA run 4, 2026-11-01; Ray, answer 1a).
self.connection.cursor().execute("USE SECONDARY ROLES NONE")
return self.connection
def query(self, sql: str) -> list[dict]:
try:
return self._run(self._connection(), sql)
except snowflake.connector.errors.Error as exc:
# ReauthenticationRequest (250001) on newer connectors; an expired session token (390114) on older ones.
if type(exc).__name__ != "ReauthenticationRequest" and getattr(exc, "errno", None) not in (250001, 390114):
raise
logger.warning("Snowflake session expired (%s); reconnecting once", getattr(exc, "errno", exc))
return self._run(self._connection(fresh=True), sql)
@staticmethod
def _serialize_value(value):
if isinstance(value, datetime):
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
else:
value = value.astimezone(timezone.utc)
return value.isoformat() if hasattr(value, "isoformat") else value
@staticmethod
def _run(connection, sql: str) -> list[dict]:
with connection.cursor() as cursor:
cursor.execute(sql)
columns = [column[0].lower() for column in cursor.description]
return [
{
column: SnowflakeRunner._serialize_value(value)
for column, value in zip(columns, row)
}
for row in cursor.fetchall()
]
class MetricFlowBackend:
def __init__(
self,
manifest_source=None,
poll_interval_seconds: float | None = None,
) -> None:
self.manifest = None
self.engine = None
self.runner = SnowflakeRunner() if os.environ.get("SNOWFLAKE_ACCOUNT") else None
self._lock = threading.RLock()
self._stop_polling = threading.Event()
self._manifest_source = manifest_source or (
LocalManifestSource()
if os.environ.get("DIGEST_STAGE") == "dev"
else DbtCloudManifestSource.from_environment()
)
self._active_run_id: int | None = None
self._checked_run_id: int | None = None
self._poll_interval_seconds = (
poll_interval_seconds
if poll_interval_seconds is not None
else float(os.environ.get("DBT_CLOUD_POLL_SECONDS", "300"))
)
self.refresh_manifest(initial=True)
if self._poll_interval_seconds > 0:
thread = threading.Thread(
target=self._poll_loop,
name="dbt-cloud-manifest-poll",
daemon=True,
)
thread.start()
@staticmethod
def _build_engine(manifest) -> MetricFlowEngine:
return MetricFlowEngine(
semantic_manifest_lookup=SemanticManifestLookup(manifest),
sql_client=CompileOnlySnowflakeClient(),
)
@property
def active_run_id(self) -> int | None:
return self._active_run_id
@property
def checked_run_id(self) -> int | None:
return self._checked_run_id
def refresh_manifest(self, initial: bool = False) -> bool:
"""Require a valid manifest at startup; retain the last good manifest during polling."""
try:
run_id = self._manifest_source.latest_successful_run_id()
if run_id is None:
if initial:
raise RuntimeError("No successful dbt Cloud run is available")
return False
if run_id == self._checked_run_id:
return False
artifact = self._manifest_source.semantic_manifest(run_id)
if artifact is None:
if initial:
raise RuntimeError(f"dbt Cloud run {run_id} has no semantic manifest")
with self._lock:
self._checked_run_id = run_id
logger.info("Run %s has no semantic manifest; retaining the current manifest", run_id)
return False
candidate = parse_manifest_from_dbt_generated_manifest(artifact)
if not candidate.metrics:
if initial:
raise RuntimeError(f"dbt Cloud run {run_id} contains no metrics")
with self._lock:
self._checked_run_id = run_id
logger.info("Run %s contains no metrics; retaining the current manifest", run_id)
return False
candidate_engine = self._build_engine(candidate)
with self._lock:
self.manifest = candidate
self.engine = candidate_engine
self._active_run_id = run_id
self._checked_run_id = run_id
logger.info("Loaded semantic manifest from dbt Cloud run %s", run_id)
return True
except Exception:
if initial:
logger.error("Unable to load the required initial semantic manifest", exc_info=True)
raise
logger.warning("Unable to refresh the semantic manifest; retaining the last good one", exc_info=True)
return False
def _poll_loop(self) -> None:
while not self._stop_polling.wait(self._poll_interval_seconds):
self.refresh_manifest()
def list_metrics(self) -> list[dict]:
with self._lock:
return [
{"name": metric.name, "description": metric.description or ""}
for metric in self.manifest.metrics
]
def get_dimensions(self, metric_name: str) -> list[str]:
with self._lock:
if metric_name not in {metric.name for metric in self.manifest.metrics}:
raise ValueError(f"Unknown governed metric: {metric_name}")
return sorted(
dimension.dunder_name
for dimension in self.engine.list_dimensions(metric_names=[metric_name])
)
def compile_query(self, metric_name: str, group_by: list[str]) -> str:
with self._lock:
self.get_dimensions(metric_name)
request = MetricFlowQueryRequest.create(
metric_names=[metric_name],
group_by_names=group_by,
)
return self.engine.explain(request).sql_statement.sql
def labels(self, metric_name: str, group_by: list[str]) -> dict:
"""The business labels of a query's metric and groupings, so the consumer is shown meaning, never the layer's
names (Data Consumption Design, k17ki72, section 11). A grouping such as policy__carrier__carrier_name or
client__first_known_date__day takes the label of its dimension, carrier_name or first_known_date."""
with self._lock:
metric_labels = {m.name: (m.label or "") for m in self.manifest.metrics}
dimension_labels = {}
for model in self.manifest.semantic_models:
for dimension in model.dimensions:
if dimension.label:
dimension_labels.setdefault(dimension.name, dimension.label)
def grouping(name: str) -> str:
parts = name.split("__")
if len(parts) > 1 and parts[-1] in ("day", "week", "month", "quarter", "year"):
parts = parts[:-1]
return dimension_labels.get(parts[-1], "")
return {"metric": metric_labels.get(metric_name, ""), "group_by": [grouping(g) for g in group_by]}
def query(self, metric_name: str, group_by: list[str]) -> dict:
sql = self.compile_query(metric_name, group_by)
if self.runner is None:
return {
"metric_name": metric_name,
"group_by": group_by,
"compiled_sql": sql,
"execution": "compile only; no Snowflake connection configured",
}
return {
"metric_name": metric_name,
"group_by": group_by,
"rows": self.runner.query(sql),
"labels": self.labels(metric_name, group_by),
"execution": "snowflake",
}
infra/aws/requirements-deploy.txt
boto3==1.43.120
infra/aws/requirements-lambda.txt
mcp==2.3.0
metricflow==0.213.0
# Lambda Python 3.12 uses Amazon Linux 2; 5.1.0 requires manylinux_2_28.
snowflake-connector-python==4.8.1
mangum==0.22.0
boto3==1.43.120
uvicorn==0.38.0
infra/aws/requirements.txt
mcp==2.3.0
metricflow==0.213.0
snowflake-connector-python==5.1.0
infra/aws/run.sh
#!/bin/sh
exec python -m uvicorn infra.aws.conversation_asgi:app --host 127.0.0.1 --port 8080
infra/aws/test_dev_environment.py
"""Focused tests for the dev-only local semantic manifest path."""
import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import Mock, patch
import zipfile
from infra.aws import deploy_functions, deploy_lambda
from infra.aws.dbt_cloud import LocalManifestSource
from infra.aws.lambda_config import load_secrets
from infra.aws.metricflow_backend import MetricFlowBackend
class DevEnvironmentTests(unittest.TestCase):
def test_local_manifest_source_reads_packaged_file(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "manifest.json"
path.write_text('{"metrics": [{"name": "policies_held"}]}', encoding="utf-8")
source = LocalManifestSource(path)
self.assertEqual(source.latest_successful_run_id(), 1)
self.assertEqual(json.loads(source.semantic_manifest(1))["metrics"][0]["name"], "policies_held")
with self.assertRaisesRegex(RuntimeError, "Unknown local manifest run"):
source.semantic_manifest(2)
def test_dev_backend_selects_local_source_without_dbt_cloud_environment(self):
with patch.dict(os.environ, {"DIGEST_STAGE": "dev"}, clear=True), patch.object(
MetricFlowBackend, "refresh_manifest"
):
backend = MetricFlowBackend(poll_interval_seconds=0)
self.assertIsInstance(backend._manifest_source, LocalManifestSource)
def test_dev_secret_loading_requests_only_snowflake(self):
client = Mock()
client.get_parameters.return_value = {
"Parameters": [{"Name": "/test/snowflake-private-key", "Value": "invented-key"}],
"InvalidParameters": [],
}
environment = {
"DIGEST_STAGE": "dev",
"LOAD_SSM_SECRETS": "true",
"SSM_PARAMETER_PREFIX": "/test",
}
with patch("infra.aws.lambda_config.boto3.client", return_value=client), patch.dict(
os.environ, environment, clear=True
):
load_secrets()
self.assertEqual(
client.get_parameters.call_args.kwargs["Names"],
["/test/snowflake-private-key"],
)
def test_dev_package_rejects_missing_and_empty_manifest(self):
with tempfile.TemporaryDirectory() as directory, patch.object(
deploy_lambda, "ROOT", Path(directory)
):
with self.assertRaisesRegex(RuntimeError, "missing"):
deploy_lambda.build_package("dev")
target = Path(directory) / "dbt" / "target"
target.mkdir(parents=True)
(target / "semantic_manifest.json").write_text('{"metrics": []}', encoding="utf-8")
with self.assertRaisesRegex(RuntimeError, "no metrics"):
deploy_lambda.build_package("dev")
def test_dev_package_contains_valid_local_manifest(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
(root / "dbt" / "target").mkdir(parents=True)
manifest = {"metrics": [{"name": "policies_held"}], "semantic_models": [{}]}
(root / "dbt" / "target" / "semantic_manifest.json").write_text(
json.dumps(manifest), encoding="utf-8"
)
package = root / "package.zip"
with patch.object(deploy_lambda, "ROOT", root), patch.object(
deploy_lambda, "BUILD_ROOT", root
), patch.object(deploy_lambda, "CONNECTOR", root), patch.object(
deploy_lambda, "ZIP_PATH", package
), patch.object(deploy_lambda.shutil, "which", return_value="/usr/bin/uv"), patch.object(
deploy_lambda, "run"
), patch.object(deploy_lambda.shutil, "copytree"), patch.object(
deploy_lambda.shutil, "copy2"
), patch.object(deploy_lambda.Path, "chmod"):
deploy_lambda.build_package("dev")
with zipfile.ZipFile(package) as archive:
packaged = json.loads(archive.read("infra/aws/dev_semantic_manifest.json"))
self.assertEqual(packaged, manifest)
def test_dev_web_function_environment_is_accepted(self):
from infra.aws import web_functions
with patch.dict(os.environ, {"DIGEST_STAGE": "dev"}, clear=True):
self.assertEqual(web_functions.environment(), "dev")
self.assertEqual(web_functions.ORIGINS["dev"], "http://localhost:8000")
def test_environment_site_origins(self):
from infra.aws import web_functions
self.assertEqual(
deploy_functions.ORIGINS["qa"],
["https://qa.dvdigest.integrationals.com", "http://localhost:8000"],
)
self.assertEqual(deploy_functions.ORIGINS["prod"], "https://dvdigest.integrationals.com")
self.assertEqual(web_functions.ORIGINS, deploy_functions.ORIGINS)
def test_qa_function_url_allows_both_site_and_localhost(self):
client = Mock()
client.get_function_url_config.return_value = {
"FunctionUrl": "https://invented.lambda-url.us-east-1.on.aws/",
"AuthType": "NONE",
"InvokeMode": "BUFFERED",
"Cors": {
"AllowOrigins": deploy_functions.ORIGINS["qa"],
"AllowMethods": ["POST"],
"AllowHeaders": ["content-type"],
"MaxAge": 300,
},
}
deploy_functions.ensure_public_url(client, "invented", deploy_functions.ORIGINS["qa"])
client.update_function_url_config.assert_not_called()
def test_function_url_updates_to_response_streaming(self):
client = Mock()
client.get_function_url_config.return_value = {
"FunctionUrl": "https://invented.lambda-url.us-east-1.on.aws/",
"AuthType": "NONE",
"InvokeMode": "BUFFERED",
"Cors": {
"AllowOrigins": ["http://localhost:8000"],
"AllowMethods": ["POST"],
"AllowHeaders": ["content-type"],
"MaxAge": 300,
},
}
client.update_function_url_config.return_value = {
"FunctionUrl": "https://invented.lambda-url.us-east-1.on.aws/"
}
deploy_functions.ensure_public_url(
client, "invented", "http://localhost:8000", "RESPONSE_STREAM"
)
self.assertEqual(
client.update_function_url_config.call_args.kwargs["InvokeMode"],
"RESPONSE_STREAM",
)
def test_dev_conversation_url_is_required_and_returned(self):
with patch.dict(os.environ, {}, clear=True):
with self.assertRaisesRegex(RuntimeError, "DIGEST_DEV_MCP_URL"):
deploy_functions.conversation_mcp_url("dev")
url = "https://invented.lambda-url.us-east-1.on.aws/mcp"
with patch.dict(os.environ, {"DIGEST_DEV_MCP_URL": url}, clear=True):
self.assertEqual(deploy_functions.conversation_mcp_url("dev"), url)
if __name__ == "__main__":
unittest.main()
infra/aws/test_fixtures/semantic_manifest.yaml
---
semantic_model:
name: infra_smoke
node_relation:
database: DEV__DV__CLAUDE_WS
schema_name: MART
alias: infra_smoke
defaults:
agg_time_dimension: day
entities:
- name: smoke
type: primary
expr: day
measures:
- name: visits
expr: visits
agg: sum
dimensions:
- name: day
type: time
type_params:
time_granularity: day
---
metric:
name: total_visits
description: Disposable infrastructure metric; not part of Mary's catalogue.
type: simple
type_params:
measure: visits
---
project_configuration:
time_spines:
- node_relation:
database: DEV__DV__CLAUDE_WS
schema_name: MART
alias: mf_time_spine
primary_column:
name: day
time_granularity: day
infra/aws/test_lambda_config.py
"""Tests for Lambda secret loading with invented values only."""
import os
import unittest
from unittest.mock import Mock, patch
from infra.aws.lambda_config import load_secrets
class LambdaConfigTests(unittest.TestCase):
def test_loads_two_secure_parameters_into_environment(self):
client = Mock()
client.get_parameters.return_value = {
"Parameters": [
{"Name": "/test/snowflake-private-key", "Value": "invented-key"},
{"Name": "/test/dbt-cloud-token", "Value": "invented-token"},
],
"InvalidParameters": [],
}
environment = {
"LOAD_SSM_SECRETS": "true",
"SSM_PARAMETER_PREFIX": "/test",
}
with patch("infra.aws.lambda_config.boto3.client", return_value=client) as boto_client:
with patch.dict(os.environ, environment, clear=False):
load_secrets()
self.assertEqual(os.environ["SNOWFLAKE_PRIVATE_KEY"], "invented-key")
self.assertEqual(os.environ["DBT_CLOUD_TOKEN"], "invented-token")
boto_client.assert_called_once_with("ssm")
self.assertTrue(client.get_parameters.call_args.kwargs["WithDecryption"])
def test_local_mode_does_not_call_aws(self):
with patch("infra.aws.lambda_config.boto3.client") as boto_client:
with patch.dict(os.environ, {"LOAD_SSM_SECRETS": "false"}, clear=False):
load_secrets()
boto_client.assert_not_called()
def test_missing_prefix_fails_closed(self):
with patch("infra.aws.lambda_config.boto3.client") as boto_client:
with patch.dict(os.environ, {"LOAD_SSM_SECRETS": "true"}, clear=True):
with self.assertRaisesRegex(RuntimeError, "SSM_PARAMETER_PREFIX"):
load_secrets()
boto_client.assert_not_called()
def test_missing_parameter_fails_closed(self):
client = Mock()
client.get_parameters.return_value = {
"Parameters": [],
"InvalidParameters": ["/test/dbt-cloud-token"],
}
with patch("infra.aws.lambda_config.boto3.client", return_value=client):
with patch.dict(os.environ, {"LOAD_SSM_SECRETS": "true", "SSM_PARAMETER_PREFIX": "/test"}):
with self.assertRaisesRegex(RuntimeError, "unavailable"):
load_secrets()
if __name__ == "__main__":
unittest.main()
infra/aws/test_qa_contract.py
"""Unit tests for the QA acceptance contract, without external services."""
import json
import os
import unittest
from unittest.mock import patch
from mcp.server.mcpserver.exceptions import ToolError
with patch.dict(os.environ, {"DIGEST_STAGE": "qa"}):
with patch("infra.aws.metricflow_backend.MetricFlowBackend") as backend_class:
backend = backend_class.return_value
backend.list_metrics.return_value = [
{"name": "policies_held", "description": "Policies held"},
{"name": "findings", "description": "Rejected records"},
]
from infra.aws import mcp_server
class QaContractTests(unittest.TestCase):
def setUp(self):
backend.reset_mock()
backend.list_metrics.return_value = [
{"name": "policies_held", "description": "Policies held"},
{"name": "findings", "description": "Rejected records"},
]
def test_list_metrics_is_one_json_array(self):
result = mcp_server.list_metrics()
self.assertIsInstance(result, str)
payload = json.loads(result)
self.assertEqual([item["name"] for item in payload], ["policies_held", "findings"])
def test_unknown_metric_names_valid_choices(self):
backend.get_dimensions.side_effect = ValueError("internal detail")
with self.assertRaises(ToolError) as raised:
mcp_server.get_dimensions("not_a_metric")
message = str(raised.exception)
self.assertIn("unknown metric not_a_metric; the metrics are", message)
self.assertIn("policies_held", message)
self.assertIn("findings", message)
def test_unknown_dimension_names_valid_choices_and_does_not_query(self):
backend.get_dimensions.side_effect = None
backend.get_dimensions.return_value = ["client_policy__role", "metric_time__day"]
with self.assertRaises(ToolError) as raised:
mcp_server.query_metrics("policies_held", ["client_policy__x"])
self.assertIn(
"unknown dimension client_policy__x; this metric's dimensions are",
str(raised.exception),
)
backend.query.assert_not_called()
def test_server_identity_includes_stage_and_version(self):
self.assertEqual(mcp_server.mcp.name, "dv-digest-qa")
self.assertEqual(mcp_server.mcp.version, "1.0.0")
def test_snowflake_session_uses_utc(self):
environment = {
"SNOWFLAKE_PRIVATE_KEY": "unused-test-key",
"SNOWFLAKE_USER": "test-user",
"SNOWFLAKE_ACCOUNT": "test-account",
"SNOWFLAKE_ROLE": "test-role",
"SNOWFLAKE_WAREHOUSE": "test-warehouse",
}
with patch.dict(os.environ, environment, clear=True), patch(
"infra.aws.metricflow_backend.serialization.load_pem_private_key"
) as load_key:
load_key.return_value.private_bytes.return_value = b"test-key"
from infra.aws.metricflow_backend import SnowflakeRunner
runner = SnowflakeRunner()
self.assertEqual(runner._connect_args["session_parameters"], {"TIMEZONE": "UTC"})
if __name__ == "__main__":
unittest.main()
infra/aws/test_qa_manifest.py
"""Tests for QA manifest fallback, polling, and relation selection."""
import json
import os
from pathlib import Path
import time
import unittest
from unittest.mock import patch
from metricflow_semantic_interfaces.parsing.dir_to_model import parse_yaml_file_paths_to_semantic_manifest
from infra.aws.metricflow_backend import MetricFlowBackend
MANIFEST_PATH = Path(__file__).parent / "test_fixtures" / "semantic_manifest.yaml"
RAW_MANIFEST = parse_yaml_file_paths_to_semantic_manifest(
[str(MANIFEST_PATH)],
apply_transformations=False,
).semantic_manifest
class FakeManifestSource:
def __init__(self):
self.run_id = 1
self.artifact = None
self.fail = False
def latest_successful_run_id(self):
if self.fail:
raise RuntimeError("invented dbt Cloud failure")
return self.run_id
def semantic_manifest(self, run_id):
if self.fail:
raise RuntimeError("invented artifact failure")
return self.artifact
def qa_artifact():
payload = json.loads(RAW_MANIFEST.json())
payload["metrics"][0]["name"] = "qa_visits"
payload["metrics"][0]["description"] = "Invented QA hot-reload test metric"
payload["semantic_models"][0]["node_relation"]["database"] = "DV_QA"
payload["semantic_models"][0]["node_relation"]["schema_name"] = "QA_MART"
payload["semantic_models"][0]["node_relation"]["relation_name"] = "DV_QA.QA_MART.infra_smoke"
payload["project_configuration"]["time_spines"][0]["node_relation"]["database"] = "DV_QA"
payload["project_configuration"]["time_spines"][0]["node_relation"]["schema_name"] = "QA_MART"
payload["project_configuration"]["time_spines"][0]["node_relation"]["relation_name"] = "DV_QA.QA_MART.mf_time_spine"
return json.dumps(payload)
class QaManifestTests(unittest.TestCase):
def setUp(self):
self.environment = patch.dict(os.environ, {"SNOWFLAKE_ACCOUNT": ""})
self.environment.start()
self.source = FakeManifestSource()
self.source.artifact = qa_artifact()
self.backend = MetricFlowBackend(
manifest_source=self.source,
poll_interval_seconds=0,
)
def tearDown(self):
self.environment.stop()
def test_initial_manifest_is_required(self):
self.assertEqual([item["name"] for item in self.backend.list_metrics()], ["qa_visits"])
self.assertEqual(self.backend.checked_run_id, 1)
self.assertEqual(self.backend.active_run_id, 1)
def test_missing_artifact_fails_startup(self):
self.source.artifact = None
with self.assertRaisesRegex(RuntimeError, "no semantic manifest"):
MetricFlowBackend(manifest_source=self.source, poll_interval_seconds=0)
def test_empty_artifact_fails_startup(self):
payload = json.loads(RAW_MANIFEST.json())
payload["metrics"] = []
payload["semantic_models"] = []
self.source.artifact = json.dumps(payload)
with self.assertRaisesRegex(RuntimeError, "contains no metrics"):
MetricFlowBackend(manifest_source=self.source, poll_interval_seconds=0)
def test_dbt_failure_fails_startup(self):
self.source.fail = True
with self.assertRaisesRegex(RuntimeError, "invented dbt Cloud failure"):
MetricFlowBackend(manifest_source=self.source, poll_interval_seconds=0)
def test_new_manifest_reloads_and_drives_relation_names(self):
self.source.run_id = 2
self.source.artifact = qa_artifact()
self.assertTrue(self.backend.refresh_manifest())
self.assertEqual([item["name"] for item in self.backend.list_metrics()], ["qa_visits"])
self.assertEqual(self.backend.active_run_id, 2)
sql = self.backend.compile_query("qa_visits", ["metric_time__day"])
self.assertIn("DV_QA", sql)
self.assertIn("QA_MART", sql)
self.assertNotIn("DEV__DV__CLAUDE_WS", sql)
def test_failure_keeps_last_good_manifest_after_startup(self):
self.source.run_id = 2
self.source.artifact = qa_artifact()
self.backend.refresh_manifest()
self.source.run_id = 3
self.source.fail = True
self.assertFalse(self.backend.refresh_manifest())
self.assertEqual([item["name"] for item in self.backend.list_metrics()], ["qa_visits"])
self.assertEqual(self.backend.active_run_id, 2)
self.assertEqual(self.backend.checked_run_id, 2)
def test_background_polling_applies_manifest_without_restart(self):
self.backend._poll_interval_seconds = 0.05
polling = __import__("threading").Thread(target=self.backend._poll_loop, daemon=True)
polling.start()
self.source.run_id = 2
self.source.artifact = qa_artifact()
deadline = time.monotonic() + 3
while time.monotonic() < deadline:
if self.backend.active_run_id == 2:
break
time.sleep(0.05)
self.backend._stop_polling.set()
polling.join(timeout=1)
self.assertEqual([item["name"] for item in self.backend.list_metrics()], ["qa_visits"])
if __name__ == "__main__":
unittest.main()
infra/aws/test_web_functions.py
"""Unit tests for the conversation and invitation website functions."""
import asyncio
import json
import os
from types import SimpleNamespace
import unittest
from unittest.mock import AsyncMock, patch
import httpx2
from infra.aws import conversation_function, invitation_function, web_functions
def event(body, method="POST"):
return {
"body": json.dumps(body),
"isBase64Encoded": False,
"requestContext": {"http": {"method": method}},
}
class FakeHttpResponse:
def __init__(self, status_code=200, payload=None):
self.status_code = status_code
self.payload = payload or {}
def raise_for_status(self):
if self.status_code >= 400:
request = httpx2.Request("POST", "https://example.test")
response = httpx2.Response(self.status_code, request=request)
raise httpx2.HTTPStatusError("HTTP failure", request=request, response=response)
def json(self):
return self.payload
class WebFunctionTests(unittest.TestCase):
def setUp(self):
self.environment = patch.dict(os.environ, {
"DIGEST_STAGE": "qa",
"SSM_PARAMETER_PREFIX": "/dig590/dv-digest/qa",
"DIGEST_MCP_URL": "https://example.test/mcp",
})
self.environment.start()
web_functions._secret_cache.clear()
conversation_function.semantic_catalogue_cache = None
conversation_function.semantic_catalogue_key = None
def tearDown(self):
self.environment.stop()
web_functions._secret_cache.clear()
conversation_function.semantic_catalogue_cache = None
conversation_function.semantic_catalogue_key = None
def test_environment_is_required(self):
with patch.dict(os.environ, {}, clear=True):
with self.assertRaisesRegex(RuntimeError, "explicitly set"):
web_functions.environment()
def test_conversation_log_field_names_environment(self):
logged = []
async def fake_answer(messages, on_text=None):
return ("Northfield Life.", [], {})
with patch.object(conversation_function, "valid_access_code", return_value=True), patch.object(
conversation_function, "consume_daily_allowance", return_value=True
), patch.object(conversation_function, "answer", new=fake_answer), patch.object(
conversation_function.logger, "info",
side_effect=lambda message, *args, **kwargs: logged.append(message),
):
result = conversation_function.handler(event({
"access_code": "code",
"messages": [{"role": "user", "content": "Which carrier?"}],
}), None)
self.assertEqual(result["statusCode"], 200)
record = next(json.loads(m) for m in logged if '"event": "conversation"' in m)
self.assertEqual(record["environment"], "qa")
self.assertNotIn("stage", record)
def test_cors_preflight_allows_json_post(self):
result = conversation_function.handler(event({}, "OPTIONS"), None)
self.assertEqual(result["statusCode"], 204)
self.assertEqual(result["headers"]["Access-Control-Allow-Methods"], "POST, OPTIONS")
self.assertNotIn("Access-Control-Allow-Origin", result["headers"])
def test_wrong_access_code_is_401(self):
with patch.object(conversation_function, "valid_access_code", return_value=False):
result = conversation_function.handler(event({"access_code": "wrong"}), None)
self.assertEqual(result["statusCode"], 401)
self.assertEqual(json.loads(result["body"]), {"error": "access_code"})
def test_message_contract_rejects_overlong_history(self):
with self.assertRaisesRegex(ValueError, "too_long"):
conversation_function.validate_messages([
{"role": "user", "content": "x"}
] * 21)
with self.assertRaisesRegex(ValueError, "too_long"):
conversation_function.validate_messages([
{"role": "user", "content": "x" * 2001}
])
def test_daily_cap_response(self):
with patch.object(conversation_function, "valid_access_code", return_value=True), patch.object(
conversation_function, "consume_daily_allowance", return_value=False
):
result = conversation_function.handler(event({
"access_code": "approved",
"messages": [{"role": "user", "content": "question"}],
}), None)
self.assertEqual(result["statusCode"], 429)
self.assertEqual(json.loads(result["body"]), {"error": "daily_cap"})
def test_invitation_grants_contributor_write_access(self):
calls = []
class Client:
def __init__(self, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
def get(self, url):
calls.append(("GET", url, None))
return FakeHttpResponse(404 if "/collaborators/" in url else 200)
def put(self, url, json):
calls.append(("PUT", url, json))
return FakeHttpResponse(201)
with patch.object(invitation_function, "valid_access_code", return_value=True), patch.object(
invitation_function, "consume_daily_allowance", return_value=True
), patch.object(invitation_function, "secret", return_value="not-logged"), patch.object(
invitation_function.httpx2, "Client", Client
):
result = invitation_function.handler(event({
"access_code": "approved", "github_username": "mary-example"
}), None)
self.assertEqual(result["statusCode"], 200)
self.assertEqual(json.loads(result["body"]), {"invited": True})
self.assertEqual(calls[-1][2], {"permission": "push"})
self.assertTrue(calls[-1][1].endswith("/collaborators/mary-example"))
def test_existing_collaborator_returns_already_without_put(self):
calls = []
class Client:
def __init__(self, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
def get(self, url):
calls.append(url)
return FakeHttpResponse(204 if "/collaborators/" in url else 200)
with patch.object(invitation_function, "secret", return_value="not-logged"), patch.object(
invitation_function.httpx2, "Client", Client
):
result = invitation_function.invite("existing-user")
self.assertEqual(result, {"status": 200, "body": {"already": True}})
self.assertEqual(len(calls), 2)
def test_unknown_and_invalid_github_users_are_404(self):
with patch.object(invitation_function, "valid_access_code", return_value=True):
invalid = invitation_function.handler(event({
"access_code": "approved", "github_username": "bad/name"
}), None)
self.assertEqual(invalid["statusCode"], 404)
class MissingClient:
def __init__(self, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
def get(self, url):
return FakeHttpResponse(404)
with patch.object(invitation_function, "valid_access_code", return_value=True), patch.object(
invitation_function, "consume_daily_allowance", return_value=True
), patch.object(invitation_function, "secret", return_value="not-logged"), patch.object(
invitation_function.httpx2, "Client", MissingClient
):
missing = invitation_function.handler(event({
"access_code": "approved", "github_username": "missing-user"
}), None)
self.assertEqual(missing["statusCode"], 404)
self.assertEqual(json.loads(missing["body"]), {"error": "no_such_user"})
def test_invitation_cap_ignores_unknown_and_existing_users(self):
class Client:
def __init__(self, **kwargs):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
def get(self, url):
if url.endswith("/users/missing-user"):
return FakeHttpResponse(404)
if url.endswith("/collaborators/existing-user"):
return FakeHttpResponse(204)
return FakeHttpResponse(200)
def put(self, url, json):
raise AssertionError("No invitation should be sent")
with patch.object(invitation_function, "valid_access_code", return_value=True), patch.object(
invitation_function, "secret", return_value="not-logged"
), patch.object(
invitation_function, "consume_daily_allowance", return_value=True
) as allowance, patch.object(invitation_function.httpx2, "Client", Client):
missing = invitation_function.handler(event({
"access_code": "approved", "github_username": "missing-user"
}), None)
existing = invitation_function.handler(event({
"access_code": "approved", "github_username": "existing-user"
}), None)
self.assertEqual(missing["statusCode"], 404)
self.assertEqual(existing["statusCode"], 200)
self.assertEqual(json.loads(existing["body"]), {"already": True})
allowance.assert_not_called()
def test_digest_429_is_retried(self):
request = httpx2.Request("POST", "https://example.test/mcp")
throttled = httpx2.HTTPStatusError(
"throttled", request=request, response=httpx2.Response(429, request=request)
)
session = SimpleNamespace(call_tool=AsyncMock(side_effect=[throttled, "success"]))
with patch.object(conversation_function.asyncio, "sleep", new=AsyncMock()) as sleep:
result = asyncio.run(conversation_function.call_digest(session, "list-metrics", {}))
self.assertEqual(result, "success")
self.assertEqual(session.call_tool.await_count, 2)
sleep.assert_awaited_once_with(0.5)
def test_catalogue_cache_reloads_when_manifest_changes(self):
version = SimpleNamespace(content=[], structured_content={"result": "run-1"})
digest = SimpleNamespace(call_tool=AsyncMock(return_value=version))
load = AsyncMock(return_value=[{"name": "policies", "group_by": []}])
with patch.object(conversation_function, "semantic_catalogue", new=load):
asyncio.run(conversation_function.current_semantic_catalogue(digest))
asyncio.run(conversation_function.current_semantic_catalogue(digest))
self.assertEqual(load.await_count, 1)
version.structured_content = {"result": "run-2"}
asyncio.run(conversation_function.current_semantic_catalogue(digest))
self.assertEqual(load.await_count, 2)
self.assertEqual(digest.call_tool.await_count, 3)
self.assertEqual(conversation_function.semantic_catalogue_key, "run-2")
def test_conversation_runs_tools_and_lists_queries(self):
tool = SimpleNamespace(name="query-metrics", description="Query", input_schema={"type": "object"})
refused = SimpleNamespace(
content=[SimpleNamespace(type="text", text="unknown dimension")],
structured_content=None,
is_error=True,
)
succeeded = SimpleNamespace(
content=[], structured_content={"result": {"rows": [{"value": 1}, {"value": 2}], "labels": {"metric": "Policies", "group_by": ["Carrier code"]}}},
is_error=False,
)
digest = SimpleNamespace(
list_tools=AsyncMock(return_value=SimpleNamespace(tools=[tool])),
call_tool=AsyncMock(side_effect=[
SimpleNamespace(content=[], structured_content={"result": "run-1"}),
refused, succeeded,
]),
)
class DigestContext:
async def __aenter__(self):
return digest
async def __aexit__(self, *args):
pass
completions = [
FakeHttpResponse(payload={
"content": [
{
"type": "tool_use", "id": "call-1", "name": "query-metrics",
"input": {"metric_name": "policies", "group_by": ["carrier_code"]},
},
{
"type": "tool_use", "id": "call-2", "name": "query-metrics",
"input": {"metric_name": "policies", "group_by": ["policy__carrier_code"]},
},
],
"usage": {"input_tokens": 10, "output_tokens": 4},
}),
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "Northfield Life."}],
"usage": {"input_tokens": 20, "output_tokens": 3},
}),
]
class AnthropicClient:
def __init__(self, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
async def post(self, *args, **kwargs):
return completions.pop(0)
with patch.object(conversation_function, "digest_session", DigestContext), patch.object(
conversation_function, "semantic_catalogue", new=AsyncMock(return_value=[])
), patch.object(
conversation_function.httpx2, "AsyncClient", AnthropicClient
), patch.object(conversation_function, "secret", return_value="not-logged"):
reply, queries, usage = asyncio.run(conversation_function.answer([
{"role": "user", "content": "Which carrier?"}
]))
self.assertEqual(reply, "Northfield Life.")
self.assertEqual(queries, [{
"metric": "policies", "group_by": ["policy__carrier_code"], "rows": 2,
"label": "Policies", "group_by_labels": ["Carrier code"],
}])
self.assertEqual(usage, {
"input_tokens": 30, "output_tokens": 7,
"cache_creation_input_tokens": 0, "cache_read_input_tokens": 0,
})
def _digest_with_query_tool(self, call_results):
tool = SimpleNamespace(
name="query-metrics", description="Query", input_schema={"type": "object"}
)
digest = SimpleNamespace(
list_tools=AsyncMock(return_value=SimpleNamespace(tools=[tool])),
call_tool=AsyncMock(side_effect=call_results),
)
class DigestContext:
async def __aenter__(self):
return digest
async def __aexit__(self, *args):
pass
return DigestContext
def _patched_answer(self, completions):
class AnthropicClient:
def __init__(self, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
async def post(self, *args, **kwargs):
return completions.pop(0)
return (
patch.object(conversation_function, "semantic_catalogue", new=AsyncMock(return_value=[])),
patch.object(conversation_function.httpx2, "AsyncClient", AnthropicClient),
patch.object(conversation_function, "secret", return_value="not-logged"),
)
def test_answer_with_digits_and_no_query_asks_again_then_errors(self):
completions = [
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "About 12 policies in force."}],
"usage": {"input_tokens": 10, "output_tokens": 4},
}),
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "Roughly 40 policies in force."}],
"usage": {"input_tokens": 20, "output_tokens": 5},
}),
]
context = self._digest_with_query_tool([
SimpleNamespace(content=[], structured_content={"result": "run-1"}),
])
catalogue, client, secret_patch = self._patched_answer(completions)
with patch.object(conversation_function, "digest_session", context), catalogue, client, secret_patch:
with self.assertRaises(conversation_function.UnverifiedAnswerError):
asyncio.run(conversation_function.answer([
{"role": "user", "content": "How many policies?"}
]))
self.assertEqual(completions, [])
def test_answer_with_digits_recovers_when_retry_runs_query(self):
completed = SimpleNamespace(
content=[], structured_content={"result": {"rows": [{"value": 12}]}},
is_error=False,
)
completions = [
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "About 12 policies in force."}],
"usage": {"input_tokens": 10, "output_tokens": 4},
}),
FakeHttpResponse(payload={
"content": [{
"type": "tool_use", "id": "call-1", "name": "query-metrics",
"input": {"metric_name": "policies", "group_by": []},
}],
"usage": {"input_tokens": 20, "output_tokens": 4},
}),
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "12 policies in force."}],
"usage": {"input_tokens": 30, "output_tokens": 3},
}),
]
context = self._digest_with_query_tool([
SimpleNamespace(content=[], structured_content={"result": "run-1"}),
completed,
])
catalogue, client, secret_patch = self._patched_answer(completions)
with patch.object(conversation_function, "digest_session", context), catalogue, client, secret_patch:
reply, queries, usage = asyncio.run(conversation_function.answer([
{"role": "user", "content": "How many policies?"}
]))
self.assertEqual(reply, "12 policies in force.")
self.assertEqual(queries, [{"metric": "policies", "group_by": [], "rows": 1, "label": "", "group_by_labels": []}])
def test_answer_without_digits_returns_without_query(self):
completions = [
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "Northfield Life."}],
"usage": {"input_tokens": 10, "output_tokens": 3},
}),
]
context = self._digest_with_query_tool([
SimpleNamespace(content=[], structured_content={"result": "run-1"}),
])
catalogue, client, secret_patch = self._patched_answer(completions)
with patch.object(conversation_function, "digest_session", context), catalogue, client, secret_patch:
reply, queries, usage = asyncio.run(conversation_function.answer([
{"role": "user", "content": "Which carrier?"}
]))
self.assertEqual(reply, "Northfield Life.")
self.assertEqual(queries, [])
self.assertEqual(completions, [])
def test_figure_rule_ignores_release_names_and_question_labels(self):
completions = [
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "Release 1 and Q-9 are planning labels."}],
"usage": {"input_tokens": 10, "output_tokens": 8},
}),
]
context = self._digest_with_query_tool([
SimpleNamespace(content=[], structured_content={"result": "run-1"}),
])
catalogue, client, secret_patch = self._patched_answer(completions)
with patch.object(conversation_function, "digest_session", context), catalogue, client, secret_patch:
reply, queries, usage = asyncio.run(conversation_function.answer([
{"role": "user", "content": "Which release and question?"}
]))
self.assertEqual(reply, "Release 1 and Q-9 are planning labels.")
self.assertEqual(queries, [])
self.assertEqual(completions, [])
def test_streamed_unverified_reply_releases_no_text(self):
from infra.aws import conversation_asgi
async def unverified_answer(messages, on_text=None):
# answer() never releases unverified text (see test_function_holds_unverified_text).
raise conversation_function.UnverifiedAnswerError("unverified figure")
scope = {
"type": "http", "method": "POST",
"headers": [(b"accept", b"text/event-stream")],
}
request = json.dumps({
"access_code": "approved",
"messages": [{"role": "user", "content": "How many policies?"}],
}).encode()
receive = AsyncMock(return_value={
"type": "http.request", "body": request, "more_body": False,
})
sent = []
async def send(message):
sent.append(message)
with patch.object(conversation_asgi, "valid_access_code", return_value=True), patch.object(
conversation_asgi, "consume_daily_allowance", return_value=True
), patch.object(conversation_asgi, "answer", new=unverified_answer):
asyncio.run(conversation_asgi.app(scope, receive, send))
body = b"".join(message.get("body", b"") for message in sent)
self.assertIn(b'event: error\ndata: {"error":"unverified_answer"}', body)
self.assertNotIn(b"event: delta", body)
def test_function_holds_unverified_text(self):
"""P-13 (D-13): streamed text with no query behind it never leaves answer(), even after the retry."""
unverified = {"content": [{"type": "text", "text": "There are 12 policies."}], "usage": {}}
async def fake_stream(client, payload, headers, on_text):
await on_text("There are 12 policies.")
return unverified
completions = [FakeHttpResponse(payload=unverified)]
context = self._digest_with_query_tool([
SimpleNamespace(content=[], structured_content={"result": "run-1"}),
])
catalogue, client, secret_patch = self._patched_answer(completions)
released = []
async def on_text(text):
released.append(text)
with patch.object(conversation_function, "digest_session", context), patch.object(
conversation_function, "stream_completion", new=fake_stream
), catalogue, client, secret_patch:
with self.assertRaises(conversation_function.UnverifiedAnswerError):
asyncio.run(conversation_function.answer(
[{"role": "user", "content": "How many policies?"}], on_text=on_text))
self.assertEqual(released, [])
def test_streamed_backed_reply_streams_after_first_query(self):
"""P-15 (D-13): text the function releases reaches the page as it arrives, in pieces."""
from infra.aws import conversation_asgi
async def backed_answer(messages, on_text=None):
await on_text("Northfield Life holds ")
await on_text("12 policies.")
return "Northfield Life holds 12 policies.", [{"tool": "query-metrics"}], {"input_tokens": 1, "output_tokens": 1}
scope = {
"type": "http", "method": "POST",
"headers": [(b"accept", b"text/event-stream")],
}
request = json.dumps({
"access_code": "approved",
"messages": [{"role": "user", "content": "How many policies?"}],
}).encode()
receive = AsyncMock(return_value={
"type": "http.request", "body": request, "more_body": False,
})
sent = []
async def send(message):
sent.append(message)
with patch.object(conversation_asgi, "valid_access_code", return_value=True), patch.object(
conversation_asgi, "consume_daily_allowance", return_value=True
), patch.object(conversation_asgi, "answer", new=backed_answer):
asyncio.run(conversation_asgi.app(scope, receive, send))
body = b"".join(message.get("body", b"") for message in sent)
self.assertEqual(body.count(b"event: delta"), 2)
self.assertIn(b'"Northfield Life holds "', body)
def test_tool_call_ceiling_finishes_without_more_tools(self):
tool = SimpleNamespace(name="query-metrics", description="Query", input_schema={"type": "object"})
succeeded = SimpleNamespace(
content=[], structured_content={"result": {"rows": [{"value": 1}]}},
is_error=False,
)
digest = SimpleNamespace(
list_tools=AsyncMock(return_value=SimpleNamespace(tools=[tool])),
call_tool=AsyncMock(return_value=succeeded),
)
class DigestContext:
async def __aenter__(self):
return digest
async def __aexit__(self, *args):
pass
uses = [{
"type": "tool_use", "id": f"call-{index}", "name": "query-metrics",
"input": {"metric_name": "policies", "group_by": []},
} for index in range(7)]
completions = [
FakeHttpResponse(payload={"content": uses, "usage": {}}),
FakeHttpResponse(payload={
"content": [{"type": "text", "text": "Answer from gathered evidence."}],
"usage": {},
}),
]
request_payloads = []
class AnthropicClient:
def __init__(self, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
async def post(self, *args, **kwargs):
request_payloads.append(kwargs["json"])
return completions.pop(0)
with patch.object(conversation_function, "digest_session", DigestContext), patch.object(
conversation_function, "semantic_catalogue", new=AsyncMock(return_value=[])
), patch.object(
conversation_function.httpx2, "AsyncClient", AnthropicClient
), patch.object(conversation_function, "secret", return_value="not-logged"):
reply, queries, usage = asyncio.run(conversation_function.answer([
{"role": "user", "content": "Which carrier?"}
]))
self.assertEqual(reply, "Answer from gathered evidence.")
self.assertEqual(digest.call_tool.await_count, 7)
self.assertEqual(len(queries), 6)
self.assertNotIn("tools", request_payloads[1])
self.assertTrue(request_payloads[1]["messages"][-2]["content"][-1]["is_error"])
self.assertEqual(request_payloads[0]["thinking"], {"type": "between_tools"})
self.assertEqual(request_payloads[0]["output_config"], {"effort": "low"})
def test_prompt_cache_breakpoints_cover_system_and_tools(self):
tools = [{"name": "query-metrics", "input_schema": {"type": "object"}}]
cached = conversation_function.cached_tools(tools)
self.assertEqual(cached[-1]["cache_control"], {"type": "ephemeral"})
self.assertNotIn("cache_control", tools[0])
def test_anthropic_stream_collects_text_tools_and_cache_usage(self):
events = [
{"type": "message_start", "message": {"usage": {
"input_tokens": 20, "cache_read_input_tokens": 900,
}}},
{"type": "content_block_start", "index": 0,
"content_block": {"type": "text", "text": ""}},
{"type": "content_block_delta", "index": 0,
"delta": {"type": "text_delta", "text": "North"}},
{"type": "content_block_delta", "index": 0,
"delta": {"type": "text_delta", "text": "field."}},
{"type": "content_block_start", "index": 1,
"content_block": {"type": "tool_use", "id": "call-1", "name": "query-metrics"}},
{"type": "content_block_delta", "index": 1,
"delta": {"type": "input_json_delta", "partial_json": '{"metric_name":"policies"}'}},
{"type": "message_delta", "usage": {"output_tokens": 8}},
]
class Stream:
status_code = 200
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
def raise_for_status(self):
pass
async def aiter_lines(self):
for event in events:
yield "data: " + json.dumps(event)
class Client:
def stream(self, *args, **kwargs):
self.payload = kwargs["json"]
return Stream()
client = Client()
on_text = AsyncMock()
completion = asyncio.run(conversation_function.stream_completion(
client, {"model": "test"}, {}, on_text
))
self.assertTrue(client.payload["stream"])
self.assertEqual(completion["content"][0]["text"], "Northfield.")
self.assertEqual(completion["content"][1]["input"], {"metric_name": "policies"})
self.assertEqual(completion["usage"]["cache_read_input_tokens"], 900)
self.assertEqual(
[item.args[0] for item in on_text.await_args_list], ["North", "field."]
)
def test_anthropic_stream_preserves_thinking_and_signature(self):
events = [
{"type": "content_block_start", "index": 0,
"content_block": {"type": "thinking", "thinking": ""}},
{"type": "content_block_delta", "index": 0,
"delta": {"type": "thinking_delta", "thinking": "Consider the evidence."}},
{"type": "content_block_delta", "index": 0,
"delta": {"type": "signature_delta", "signature": "signed"}},
]
class Stream:
status_code = 200
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
def raise_for_status(self):
pass
async def aiter_lines(self):
for event in events:
yield "data: " + json.dumps(event)
class Client:
def stream(self, *args, **kwargs):
return Stream()
completion = asyncio.run(conversation_function.stream_completion(
Client(), {}, {}, AsyncMock()
))
self.assertEqual(completion["content"][0], {
"type": "thinking", "thinking": "Consider the evidence.", "signature": "signed",
})
if __name__ == "__main__":
unittest.main()
infra/aws/web_functions.py
"""Shared, dependency-light helpers for DIG-590's public web functions."""
from __future__ import annotations
from contextlib import asynccontextmanager
from datetime import datetime, timezone
import hmac
import json
import os
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.exceptions import ClientError
import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
ORIGINS = {
"dev": "http://localhost:8000",
"qa": ["https://qa.dvdigest.integrationals.com", "http://localhost:8000"],
"prod": "https://dvdigest.integrationals.com",
}
_secret_cache: dict[str, str] = {}
def environment() -> str:
value = os.environ.get("DIGEST_STAGE")
if value not in ORIGINS:
raise RuntimeError("DIGEST_STAGE must be explicitly set to dev, qa, or prod")
return value
def response(status: int, body: dict | None = None) -> dict:
return {
"statusCode": status,
# Function URL CORS adds the environment origin exactly once to every reply.
"headers": {
"Content-Type": "application/json",
"Cache-Control": "no-store",
},
"body": json.dumps(body or {}),
}
def preflight(event: dict) -> dict | None:
method = event.get("requestContext", {}).get("http", {}).get("method")
if method != "OPTIONS":
return None
result = response(204)
result["headers"].update({
"Access-Control-Allow-Methods": "POST, OPTIONS",
"Access-Control-Allow-Headers": "Content-Type",
"Access-Control-Max-Age": "300",
})
result["body"] = ""
return result
def json_body(event: dict) -> dict:
body = event.get("body") or "{}"
if event.get("isBase64Encoded"):
import base64
body = base64.b64decode(body).decode("utf-8")
value = json.loads(body)
if not isinstance(value, dict):
raise ValueError("JSON body must be an object")
return value
def secret(suffix: str) -> str:
if suffix not in _secret_cache:
prefix = os.environ["SSM_PARAMETER_PREFIX"]
value = boto3.client("ssm", region_name=os.environ.get("AWS_REGION", "us-east-1"))
parameter = value.get_parameter(Name=f"{prefix}/{suffix}", WithDecryption=True)
_secret_cache[suffix] = parameter["Parameter"]["Value"]
return _secret_cache[suffix]
def valid_access_code(provided) -> bool:
if not isinstance(provided, str) or not provided:
return False
return hmac.compare_digest(provided.encode("utf-8"), secret("access-code").encode("utf-8"))
def consume_daily_allowance(kind: str, cap: int) -> bool:
"""Increment a small per-environment UTC-day counter in Standard Parameter Store."""
day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
prefix = os.environ["SSM_PARAMETER_PREFIX"]
name = f"{prefix}/daily-{kind}-{day}"
client = boto3.client("ssm", region_name=os.environ.get("AWS_REGION", "us-east-1"))
try:
current = int(client.get_parameter(Name=name)["Parameter"]["Value"])
except ClientError as error:
if error.response["Error"]["Code"] != "ParameterNotFound":
raise
current = 0
if current >= cap:
return False
client.put_parameter(Name=name, Value=str(current + 1), Type="String", Overwrite=True)
return True
def release_daily_allowance(kind: str) -> None:
"""Release a reserved invitation when GitHub did not send one."""
day = datetime.now(timezone.utc).strftime("%Y-%m-%d")
prefix = os.environ["SSM_PARAMETER_PREFIX"]
name = f"{prefix}/daily-{kind}-{day}"
client = boto3.client("ssm", region_name=os.environ.get("AWS_REGION", "us-east-1"))
try:
current = int(client.get_parameter(Name=name)["Parameter"]["Value"])
except ClientError as error:
if error.response["Error"]["Code"] == "ParameterNotFound":
return
raise
client.put_parameter(
Name=name, Value=str(max(0, current - 1)), Type="String", Overwrite=True
)
class LambdaSigV4Auth(httpx2.Auth):
"""Sign Function URL requests without exposing credentials to application code."""
def __init__(self):
credentials = boto3.Session(
region_name=os.environ.get("AWS_REGION", "us-east-1")
).get_credentials()
if credentials is None:
raise RuntimeError("AWS credentials are unavailable")
self.credentials = credentials.get_frozen_credentials()
def auth_flow(self, request):
aws_request = AWSRequest(
method=request.method,
url=str(request.url),
headers=dict(request.headers),
data=request.content,
)
SigV4Auth(
self.credentials,
"lambda",
os.environ.get("AWS_REGION", "us-east-1"),
).add_auth(aws_request)
request.headers.update(dict(aws_request.headers))
yield request
@asynccontextmanager
async def digest_session():
auth = LambdaSigV4Auth() if environment() == "prod" else None
async with httpx2.AsyncClient(auth=auth, timeout=60.0) as client:
async with streamable_http_client(
os.environ["DIGEST_MCP_URL"], http_client=client
) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
yield session
infra/github/branch_protection.sh
#!/bin/sh
# Protection for digest-datavault's qa and main (Ray, 2026-10-28, answer 1y), as a GitHub repository ruleset.
# Platform settings are kept as a committed script, never set by hand (Ray, 2026-10-23). Rerun it to restore them.
#
# What it sets, for qa and main:
# - changes only by pull request, with one approving review (Ray's), stale approvals dismissed on new pushes;
# - no force pushes (non-fast-forward updates) and no deletion;
# - one bypass: the repository admin (Ray), so a promotion's converge still works (qa pointed at main's new commit,
# a sideways move Git calls a force push). Classic branch protection cannot exempt admins from the force-push
# rule, which the scratch-repository test of 2026-10-28 proved; a ruleset's bypass list can.
#
# Run as the repository's admin: OP_SERVICE_ACCOUNT_TOKEN set, then sh infra/github/branch_protection.sh
# Test against a scratch repository: REPO=rayserrano2735/<scratch> sh infra/github/branch_protection.sh
set -eu
REPO="${REPO:-rayserrano2735/digest-datavault}"
NAME="qa and main: pull requests only"
TOKEN=$(op read "op://Agentic Workstation/GitHub PAT/credential")
API="https://api.github.com/repos/$REPO/rulesets"
H1="Authorization: Bearer $TOKEN"; H2="Accept: application/vnd.github+json"; H3="X-GitHub-Api-Version: 2022-11-28"
BODY='{
"name": "'"$NAME"'",
"target": "branch",
"enforcement": "active",
"conditions": {"ref_name": {"include": ["refs/heads/qa", "refs/heads/main"], "exclude": []}},
"rules": [
{"type": "pull_request", "parameters": {"required_approving_review_count": 1, "dismiss_stale_reviews_on_push": true,
"require_code_owner_review": false, "require_last_push_approval": false, "required_review_thread_resolution": false}},
{"type": "non_fast_forward"},
{"type": "deletion"}
],
"bypass_actors": [{"actor_id": 5, "actor_type": "RepositoryRole", "bypass_mode": "always"}]
}'
# Idempotent: update the ruleset by name if it exists, otherwise create it.
ID=$(curl -sS -H "$H1" -H "$H2" -H "$H3" "$API" | python3 -c "
import sys, json
print(next((str(r['id']) for r in json.load(sys.stdin) if r['name'] == '$NAME'), ''))")
if [ -n "$ID" ]; then curl -sS -X PUT -H "$H1" -H "$H2" -H "$H3" "$API/$ID" -d "$BODY" > /dev/null
else curl -sS -X POST -H "$H1" -H "$H2" -H "$H3" "$API" -d "$BODY" > /dev/null; fi
# Read back what GitHub now holds.
ID=$(curl -sS -H "$H1" -H "$H2" -H "$H3" "$API" | python3 -c "
import sys, json
print(next((str(r['id']) for r in json.load(sys.stdin) if r['name'] == '$NAME'), ''))")
curl -sS -H "$H1" -H "$H2" -H "$H3" "$API/$ID" | python3 -c "
import sys, json
r = json.load(sys.stdin)
rules = {x['type']: x.get('parameters', {}) for x in r['rules']}
print('ruleset:', r['name'], '|', r['enforcement'], '| branches:', r['conditions']['ref_name']['include'])
print(' reviews required:', rules.get('pull_request', {}).get('required_approving_review_count'),
'| force pushes blocked:', 'non_fast_forward' in rules, '| deletion blocked:', 'deletion' in rules,
'| bypass:', [(b['actor_type'], b['actor_id']) for b in r.get('bypass_actors', [])])"
infra/keys/new_developer_keypair.ps1
<#
New developer key pair for Snowflake key-pair authentication (dbt Studio, snow CLI).
Run by the developer, on the developer's own machine, in their own PowerShell session, from the
repository root: .\infra\keys\new_developer_keypair.ps1 -User RAYSERRANO
1. Generates a 2048-bit RSA private key, encrypted with a passphrase OpenSSL asks for, at
$HOME\.snowflake\<user>_dbt.p8. The private key is never printed and never enters the repository.
Store the file and its passphrase in your personal password manager, not a shared vault.
2. Writes infra\snowflake\developer_keys\<USER>.sql, which registers the PUBLIC key with Snowflake.
Public keys are not secret: commit that file, then RAYSERRANO_ADMIN runs it.
3. In dbt Studio's development credentials, choose key pair, paste the private key file's contents
(BEGIN and END lines included) and enter the passphrase.
Resumable: if the private key and public key already exist (a run that stopped before step 2), it reuses
them and only writes the registration script. It never overwrites a private key; to rotate, rename it first.
-Slot chooses which of Snowflake's two key slots to fill (1: RSA_PUBLIC_KEY, 2: RSA_PUBLIC_KEY_2). Use the
slot NOT already holding a key you use elsewhere (for example DBeaver), so registering does not break it.
#>
param(
[Parameter(Mandatory = $true)][ValidatePattern('^[A-Z0-9_]+$')][string]$User,
[ValidateSet(1, 2)][int]$Slot = 2,
[string]$OpenSsl = 'C:\Program Files\Git\usr\bin\openssl.exe'
)
$ErrorActionPreference = 'Stop'
if (-not (Test-Path -LiteralPath $OpenSsl)) { throw "OpenSSL not found at $OpenSsl (Git for Windows provides it); pass -OpenSsl <path>." }
$repoRoot = Split-Path (Split-Path $PSScriptRoot -Parent) -Parent
$keyDir = Join-Path $HOME '.snowflake'
$name = $User.ToLower() + '_dbt'
$private = Join-Path $keyDir ($name + '.p8')
$public = Join-Path $keyDir ($name + '.pub')
New-Item -ItemType Directory -Path $keyDir -Force | Out-Null
$haveKeys = (Test-Path -LiteralPath $private) -and (Test-Path -LiteralPath $public)
if ($haveKeys) {
Write-Host "Reusing existing key pair: $private"
} elseif (Test-Path -LiteralPath $private) {
throw "$private exists without its public key. Rename it to start over, or extract the public key with: & '$OpenSsl' rsa -in '$private' -pubout -out '$public'"
} else {
# 1. Encrypted PKCS#8 private key; OpenSSL prompts for the passphrase on this console.
& $OpenSsl genrsa 2048 2>$null | & $OpenSsl pkcs8 -topk8 -v2 aes-256-cbc -inform PEM -out $private
if ($LASTEXITCODE -ne 0 -or -not (Test-Path -LiteralPath $private)) { throw 'Private key generation failed.' }
# Only the current user may read the private key.
$me = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name
icacls $private /inheritance:r /grant:r "$($me):(R,W)" | Out-Null
# 2. Public key (OpenSSL asks for the passphrase again to read the private key).
& $OpenSsl rsa -in $private -pubout -out $public
if ($LASTEXITCODE -ne 0) { throw 'Public key extraction failed.' }
}
$body = (Get-Content -LiteralPath $public | Where-Object { $_ -notmatch '^-----' }) -join ''
$sqlDir = Join-Path $repoRoot 'infra\snowflake\developer_keys'
New-Item -ItemType Directory -Path $sqlDir -Force | Out-Null
$sqlPath = Join-Path $sqlDir ($User + '.sql')
$stamp = (Get-Date).ToUniversalTime().ToString('yyyy-MM-dd HH:mm') + ' UTC'
$keyParam = if ($Slot -eq 2) { 'RSA_PUBLIC_KEY_2' } else { 'RSA_PUBLIC_KEY' }
@"
/*
Register $User's public key for key-pair authentication. Generated by infra/keys/new_developer_keypair.ps1
on $stamp. Public key only; the private key stays on the developer's machine. Run as RAYSERRANO_ADMIN.
*/
USE ROLE SECURITYADMIN;
ALTER USER $User SET $keyParam = '$body';
DESC USER $User; /* ${keyParam}_FP must now show a fingerprint; the other slot is unchanged */
"@ | Set-Content -LiteralPath $sqlPath -Encoding utf8
Write-Host "Private key: $private (encrypted; store it and its passphrase in your personal password manager)"
Write-Host "Registration script: $sqlPath (commit it; RAYSERRANO_ADMIN runs it)"
infra/runbooks/dbt_cloud_prod_environment.md
# dbt Cloud production environment for digest-datavault
The production twin of `dbt_cloud_qa_environment.md`, recording what was created on 2026-10-27 for release 1,
The Acquired Book (DIG-574; QA item k35ki293, sections 10 and 11).
## As created
| Setting | Value |
|---|---|
| Environment | PROD (id 70403104023957), deployment type production, engine Latest |
| Branch | custom branch `main` |
| Credentials | key pair: user DV_DEPLOY (`infra/snowflake/developer_keys/DV_DEPLOY.sql`), role DV_ROLE, warehouse COMPUTE_WH, database DV, schema VAULT |
| Landing | DBT_LANDING_STORAGE: no PROD value; the project default, LANDING, applies |
| Job | "Prod Build" (id 70403104453080), on demand, no schedule |
## The job's commands
```
dbt run-operation capture_baseline
dbt build --exclude "test_type:unit"
dbt run-operation verify_build
```
No unit tests run in production: they test the code against fixed rows and have done their work in QA
(the QA job's `--selector qa_build` step exists only there).
## Promotion, each release
Every act below runs in dev and in QA before it runs in production (Promotion standard, k12ki22).
0. **The restore point.** Before the Prod Build, `SELECT CURRENT_TIMESTAMP();` is run and kept with the promotion's
evidence: the clock the rollback below returns to. DV's Time Travel retention is one day, so the rollback is
available for 24 hours after the promotion.
1. QA certified on qa (the branch's QA item).
2. A pull request from qa into main, squash-merged by Ray.
3. qa converged with main: qa pointed at main's new commit, no file change.
4. Prod Build triggered (by Claude through the dbt Cloud API), and DV checked against DV_QA.
**Release 1's key fix only, once:** `snow sql ... -D vault=DV < infra/snowflake/correct_zero_key_satellite_rows.sql` removes the four zero-key source rows loaded before the fix (three client, one insured); its before and after counts are kept with the promotion's evidence (Ray, 2026-10-28, answer 1a).
**From release 2, What We Knew When (DIG-594):** after the Prod Build, as SYSADMIN, each script piped in:
`infra/snowflake/drop_mart_views.sql -D vault=DV` (release 1's five MART views retire);
`infra/snowflake/connector_read_grants.sql -D vault=DV` (once: the connector reads the conformed mart through one
read-only grant).
5. The structure check: `python3 infra/scripts/vault_ddl.py --check --database DV --landing LANDING -- <snow options>` compares production's deployed structure with the committed `infra/snowflake/ddl/dv.sql`; a difference fails the promotion (Ray, 2026-10-28, answer 3y).
5b. Production's connector and Ask function, after the data (Promotion standard, section 13.6):
`python infra/aws/deploy_lambda.py --stage prod`, then `python infra/aws/deploy_functions.py --stage prod`, from
main. Until the repository's GitHub Actions deploy each environment's stack (section 13.3 and 13.4), these run from
the workstation, an exception Ray granted on 2026-11-01 (answer 2a).
6. A GitHub release tagged vYYYY.M.R on the commit production built, its name leading the description, with that commit's `dv.sql` attached as the release's structure.
## First run
Run 70403230861868 on main f48c24b, 2026-10-27: 68 of 68 passed, zero warnings; DV identical to DV_QA.
Rollback for that first release: drop DV's built schemas, there being no earlier production state.
## Rollback
Within DV's retention (one day), by `infra/snowflake/rollback_vault_to_time.sql`, as SYSADMIN:
`-D vault=DV -D at='<the restore point>' -D owner=DV_ROLE`. It clones DV as it stood at the restore point, returns
the clone to DV_ROLE with its grants, and swaps it in; the undone state stays as DV_ROLLBACK for inspection. Then, in
order: `connector_read_grants.sql -D vault=DV`; `dbt run-operation declare_vault_keys` from the restored release's
commit when that release declares keys; the structure check against that commit's dv.sql; the release's commit
reverted on main and production's functions redeployed from it; DV_ROLLBACK dropped once all of that is verified.
Rehearsed in DEV__DV__CLAUDE_WS on 2026-11-01: 45 objects and every count restored, 27 keys re-declared, the
structure matching dv.sql. Rehearsed in DV_QA from qa before this release is certified.
infra/runbooks/dbt_cloud_qa_environment.md
# dbt Cloud: QA deployment environment
Ray, 2026-10-23 (answer 2y). The dbt Cloud settings below are made in dbt Cloud's interface; this runbook is
their record. Order matters.
| # | Step | Value |
|---|---|---|
| 1 | Service user | Run infra/snowflake/service_users.sql as RAYSERRANO_ADMIN, then infra/snowflake/dv_qa_role.sql and infra/snowflake/refresh_landing_tier.sql (QA tier) as SYSADMIN |
| 2 | Its key pair | `infra\keys\new_developer_keypair.ps1 -User DV_QA_DEPLOY -Slot 1`; commit the generated developer_keys/DV_QA_DEPLOY.sql; run it as RAYSERRANO_ADMIN |
| 3 | Environment | Deployment environment named QA; type staging if offered, otherwise general; engine v1 Latest |
| 4 | Branch | Custom branch `qa` |
| 5 | Credentials | User DV_QA_DEPLOY, key pair (the .p8 from step 2 and its passphrase), role DV_QA_ROLE, warehouse COMPUTE_WH, database DV_QA. Schema: any value; dbt_project.yml sets every layer's schema |
| 6 | Environment variable | DBT_LANDING_STORAGE = LANDING_QA for the QA environment |
| 7 | Job | Named "QA build", in environment QA, with these steps in order: |
```
dbt run-operation capture_baseline
dbt build --exclude "test_type:unit"
dbt build --selector qa_build
dbt run-operation verify_build
```
The first build step excludes unit tests so a fresh QA database gets true first builds (ghost records included);
the second, the selector qa_build in dbt/selectors.yml, runs everything dbt v1 can run, unit tests included. The job's first successful run must wait until increment 1 is
merged into `qa`, because before that `qa` has no dbt/ folder.
**After every QA build, the structure check** (Ray, 2026-10-28, answer 3y), from the workstation:
```
python3 infra/scripts/vault_ddl.py --check --database DV_QA --landing LANDING_QA -- <snow options>
```
It compares DV_QA's deployed raw vault, reference and mart structure with the committed `infra/snowflake/ddl/dv.sql`, which each release's branch regenerates from the developer's vault right after a dbt build (`vault_ddl.py` without `--check`). A difference fails the QA cycle: the structure QA certifies is the one on record.
**Expected after the first run** (increments 1 and 2): DV_QA.RAW_VAULT.CLIENT_H holds 37 rows, 35 client keys, the same as the
development tier's vault_checks on 2026-10-26; the QA item (k35ki293) records the full comparison.
infra/runbooks/github_machine_account.md
# GitHub machine account for Claude: rayserrano2735-claude-ws
Ray, 2026-10-23. Claude contributes to this repository as a first-class developer under its own GitHub
machine account, so its work shows as its own, and Ray, as the other developer, can approve its pull requests
(GitHub does not let an author approve their own pull request). GitHub's terms allow one machine account per
person; Ray is the human responsible for it.
## Why (found 2026-10-23)
| Finding | Effect |
|---|---|
| 31 commits signed Claude <claude@users.noreply.github.com>, an address no account of ours owns | No contributor shown; the address format belongs to whichever GitHub user is named claude |
| Every push and pull request used Ray's token | Pull request #1 was opened by rayserrano2735, so Ray could not approve it |
## Ray's steps (GitHub interface)
| # | Step | Value |
|---|---|---|
| 1 | Create the account | Username rayserrano2735-claude-ws (claude-ws belongs to an unrelated account created 2024-06-28); email maestrorayserrano@gmail.com; two-factor authentication on; password and recovery codes in Ray's personal vault. Created 2026-10-23 13:16 UTC, id 346915053 |
| 2 | Invite it | Write role (not Admin) on rayserrano2735/digest-datavault. Sent by Claude with Ray's token on 2026-10-23 (invitation 339332332) after confirming the account's creation date; accepted by Ray as the machine account |
| 3 | Its token | A classic token with the repo scope only, created signed in as the machine account. A fine-grained token cannot target a repository owned by another personal account (the resource owner list offered only the machine account itself), so the token is contained by the account's access instead: the account can reach exactly one repository. Inviting the machine account anywhere else widens what this token reaches. Verified 2026-10-23: acts as rayserrano2735-claude-ws, scopes repo, reaches only digest-datavault with push and without admin |
| 4 | Store it | 1Password vault Agentic Workstation, item "GitHub PAT claude-ws", field credential. Ray's own token stays for administration |
## Claude's steps (after step 4)
| # | Step |
|---|---|
| 5 | Read the token from the vault; confirm it acts as claude-ws and can reach only digest-datavault |
| 6 | Set the workstation's git identity for this repository to claude-ws and its GitHub no-reply address (the numeric form, read from the API) |
| 7 | Close pull request #1 and open the same branch into qa as claude-ws, with the same description; Ray reviews and approves it there |
| 8 | Record the change in the workstation canonical (bootstrap reads the new token for digest-datavault) |
Existing commits keep their old signature; history is not rewritten. The squash merge into qa carries the
pull request author's identity.
infra/scripts/dev_scoped_rebuild.sh
#!/usr/bin/env bash
# Development rebuild of one model and everything downstream of it (design D-20; guide G-12).
# Usage, from the repository root: infra/scripts/dev_scoped_rebuild.sh <model_name>
# Refuses to run unless dbt's target database is a developer database (its name starts with DEV__):
# rebuilding is legitimate only where data can be replayed, never in QA or production.
set -euo pipefail
model="${1:?usage: dev_scoped_rebuild.sh <model_name>}"
cd "$(dirname "$0")/../../dbt"
db=$(dbt debug 2>/dev/null | grep -o '"database": *"[^"]*"' | head -1 | sed 's/.*"\([^"]*\)"$/\1/' | tr '[:lower:]' '[:upper:]')
case "$db" in
DEV__*) echo "Target database $db: rebuilding $model and everything downstream." ;;
*) echo "Refused: target database '$db' is not a developer database (DEV__...). Rebuilds never run in QA or production." >&2; exit 1 ;;
esac
dbt build --full-refresh --select "${model}+"
infra/scripts/vault_ddl.py
"""
The vault's master DDL (Ray, 2026-10-28, answers 2y and 3y).
Phase 1: dbt creates the structures, so the DDL is reverse-engineered from a built vault with Snowflake's GET_DDL,
for the schemas a modeler works with: RAW_VAULT, REFERENCE and CONFORMED_DM (MART retired, mart design D-6). Database names are written as production's
(DV, LANDING), so the one file describes every tier and can be compared with any of them.
Generate, on a release's branch, right after a dbt build in the developer's database:
python3 infra/scripts/vault_ddl.py --database DEV__DV__CLAUDE_WS --landing DEV__LANDING__CLAUDE_WS -- <snow options>
Check a tier's deployed structure against the committed file (QA after its build, production after its build):
python3 infra/scripts/vault_ddl.py --check --database DV_QA --landing LANDING_QA -- <snow options>
python3 infra/scripts/vault_ddl.py --check --database DV --landing LANDING -- <snow options>
<snow options> are the Snowflake CLI's connection options, as the workstation's other scripts pass them.
--check exits 1 and shows the differences when the structure is not the one on record.
"""
import argparse, difflib, json, os, re, subprocess, sys
SCHEMAS = ["RAW_VAULT", "REFERENCE", "CONFORMED_DM"]
OUTPUT = os.path.join(os.path.dirname(__file__), "..", "snowflake", "ddl", "dv.sql")
HEADER = """-- The vault's structure: the raw vault, reference and conformed mart schemas of the DIGEST Data Vault course.
-- Phase 1: generated from a built vault by infra/scripts/vault_ddl.py (Snowflake GET_DDL), with database names
-- written as production's (DV, LANDING). Each release commits its own version.
-- To propose a change to the structure, edit this file on a branch of your own: the diff is the proposal.
"""
def ddl(database, landing, snow):
"""The normalized DDL of the three schemas, in a fixed order."""
sql = "\n".join(f"SELECT '{s}' AS S, GET_DDL('SCHEMA', '{database}.{s}', FALSE) AS D;" for s in SCHEMAS)
run = subprocess.run(["snow", "sql", "-i", "--format", "json", "--enable-templating", "NONE"] + snow,
input=sql.encode(), capture_output=True)
out = run.stdout.decode("utf-8", "replace")
try:
results = json.loads(out)
except ValueError:
sys.exit("GET_DDL failed:\n" + out[:2000] + run.stderr.decode("utf-8", "replace")[:1000])
parts = []
for rows in results:
text = rows[0]["D"]
for name, prod in ((database, "DV"), (landing, "LANDING")):
text = re.sub(rf"\b{re.escape(name)}\b", prod, text, flags=re.IGNORECASE)
# dbt signs each view's text with a comment naming its engine, version, target and run; it is not structure, and
# it differs between Fusion and dbt-core and on every run, so it is removed (QA run 2, 2026-11-01)
text = re.sub(r'/\* \{"app": "dbt".*?\} \*/', "", text)
parts.append("\n".join(line.rstrip() for line in text.strip().splitlines()))
return HEADER + "\n" + "\n\n".join(parts) + "\n"
def main():
if "--" not in sys.argv:
sys.exit("usage: vault_ddl.py [--check] --database DB --landing DB -- <snow connection options>")
i = sys.argv.index("--")
ap = argparse.ArgumentParser()
ap.add_argument("--database", required=True)
ap.add_argument("--landing", required=True)
ap.add_argument("--check", action="store_true")
args = ap.parse_args(sys.argv[1:i])
text = ddl(args.database.upper(), args.landing.upper(), sys.argv[i + 1:])
if args.check:
with open(OUTPUT, encoding="utf-8") as f:
recorded = f.read()
if text == recorded:
print(f"{args.database}: structure matches infra/snowflake/ddl/dv.sql")
return
diff = list(difflib.unified_diff(recorded.splitlines(), text.splitlines(), "dv.sql (committed)",
f"{args.database} (deployed)", lineterm="", n=1))
print(f"{args.database}: structure DIFFERS from infra/snowflake/ddl/dv.sql ({len(diff)} diff lines)")
print("\n".join(diff[:80]))
sys.exit(1)
os.makedirs(os.path.dirname(OUTPUT), exist_ok=True)
with open(OUTPUT, "w", encoding="utf-8") as f:
f.write(text)
print(f"wrote infra/snowflake/ddl/dv.sql from {args.database}: {text.count(chr(10))} lines, "
f"{len(re.findall(r'create or replace (?:transient )?table', text, re.I))} tables, "
f"{len(re.findall(r'create or replace view', text, re.I))} views")
if __name__ == "__main__":
main()
infra/scripts/workstation_setup.sh
#!/bin/sh
# The vault project's workstation setup (Provisioning — Agentic common, k1ki331, leaves a repository's own setup to
# its records). Run from the repository root after provisioning, as the workstation's user:
# sh infra/scripts/workstation_setup.sh CLAUDE_WS
# 1. The dev dbt profile: the developer databases DEV__DV__<USER> and DEV__LANDING__<USER>, key-pair login.
# 2. The MetricFlow environment, from the connector's own requirements, so Mary's questions compile here as the
# connector compiles them (Ray, 2026-11-01); boto3 for the connector's unit tests and AWS reads.
# 3. Playwright and its browser, for the QA retest (site/dev/qa_retest.py).
# 4. dbt-core with dbt-snowflake, QA's engine, beside Fusion, so the dev rehearsal runs the QA job's steps on the engine QA
# uses before any pull request into qa (Ray, 2026-11-01: nothing is supposed to fail in QA).
set -eu
USER_NAME="${1:?the workstation's Snowflake user, such as CLAUDE_WS}"
KEY="${SNOWFLAKE_KEY:-$HOME/.op_secret/claude_ws.p8}"
umask 077; mkdir -p "$HOME/.dbt"
cat > "$HOME/.dbt/profiles.yml" <<PROFILE
digest_datavault:
target: dev
outputs:
dev:
type: snowflake
account: mgb11914.us-east-1
user: $USER_NAME
private_key_path: $KEY
role: ${USER_NAME}_ROLE
warehouse: COMPUTE_WH
database: DEV__DV__$USER_NAME
schema: RAW_VAULT
threads: 8
PROFILE
umask 022
VENV="${MF_VENV:-$HOME/.agentic/mfvenv}"
python3 -m venv "$VENV"
"$VENV/bin/pip" install -q -r infra/aws/requirements.txt boto3 playwright
"$VENV/bin/python" -m playwright install --with-deps chromium > /dev/null
CORE="${DBT_CORE_VENV:-$HOME/.agentic/dbtcore}"
python3 -m venv "$CORE"
"$CORE/bin/pip" install -q dbt-core dbt-snowflake
echo "dev profile: DEV__DV__$USER_NAME over DEV__LANDING__$USER_NAME (set DBT_LANDING_STORAGE=DEV__LANDING__$USER_NAME)"
echo "MetricFlow and Playwright: $VENV"
echo "dbt-core, QA's engine: $CORE/bin/dbt"
infra/snowflake/admin_separation.sql
/*
Separate administration from development (Ray, 2026-10-23, answer 1y).
RAYSERRANO becomes Ray's developer identity; a new RAYSERRANO_ADMIN holds the admin roles.
Run by Ray, in this order: 1, 2, 2b, 3, 3b, 4, 6, then 5 after the environment tier setup. Step 4 must not run until step 3 has succeeded:
removing ACCOUNTADMIN from the only user that holds it can lock the account.
*/
/* Step 1. See what RAYSERRANO holds today, so only the admin roles move.
Read 2026-10-23: ACCOUNTADMIN, ORGADMIN, ALBA_ROLE, CIRO_ROLE, DCS_ROLE, VERA_ROLE.
ACCOUNTADMIN and ORGADMIN move; the four project roles stay.
Note: CIRO_ROLE holds SYSADMIN (SHOW GRANTS OF ROLE SYSADMIN, 2026-10-23), so RAYSERRANO
still reaches SYSADMIN through CIRO_ROLE after step 4 unless that grant is revisited. */
USE ROLE SECURITYADMIN;
SHOW GRANTS TO USER RAYSERRANO;
/* Step 2. Create the admin user and give it the two admin roles RAYSERRANO holds (ACCOUNTADMIN, ORGADMIN).
Set authentication in Snowsight after creation: a password with MFA, or a key pair. */
USE ROLE USERADMIN;
CREATE USER IF NOT EXISTS RAYSERRANO_ADMIN
DEFAULT_ROLE = ACCOUNTADMIN
COMMENT = 'Administration only. Daily development uses RAYSERRANO.';
USE ROLE ACCOUNTADMIN;
GRANT ROLE ACCOUNTADMIN TO USER RAYSERRANO_ADMIN; /* contains SECURITYADMIN, SYSADMIN, USERADMIN */
GRANT ROLE ORGADMIN TO USER RAYSERRANO_ADMIN; /* organization-level administration */
/* Step 2b. Give the admin user a way to sign in to Snowsight: a temporary password that must be
changed at first sign-in. Fill in the placeholder only in the worksheet you run; never commit a
password, and keep it in a password manager. Snowflake requires MFA enrollment for human users
signing in with a password; enroll at the first sign-in. */
USE ROLE SECURITYADMIN;
ALTER USER RAYSERRANO_ADMIN SET
PASSWORD = '<a strong temporary password>'
MUST_CHANGE_PASSWORD = TRUE
TYPE = PERSON;
/* Step 3. Sign in to Snowsight as RAYSERRANO_ADMIN while another RAYSERRANO session stays open as the
way back (Ray, 2026-10-23: his DBeaver connection), change the password, enroll MFA, then confirm:
USE ROLE ACCOUNTADMIN; SELECT CURRENT_USER(), CURRENT_ROLE();
USE ROLE ORGADMIN; SELECT CURRENT_ROLE();
USE ROLE SECURITYADMIN; SHOW GRANTS TO USER RAYSERRANO_ADMIN;
Snowflake recommends more than one user holding ACCOUNTADMIN; consider a second,
break-glass admin user before step 4. */
/* Step 3b. A break-glass administrator (Ray, 2026-10-23), created before step 4 so the account never
depends on a single admin. Read 2026-10-23: only RAYSERRANO and RAYSERRANO_ADMIN hold ACCOUNTADMIN and
ORGADMIN. Never used day to day; its own strong password and MFA, stored apart from RAYSERRANO_ADMIN's.
Run as RAYSERRANO_ADMIN; fill the placeholder only in the worksheet, then sign in once to change the
password, enroll MFA and confirm both roles, as in step 3. */
USE ROLE USERADMIN;
CREATE USER IF NOT EXISTS RAYSERRANO_BREAKGLASS
DEFAULT_ROLE = ACCOUNTADMIN
TYPE = PERSON
COMMENT = 'Break-glass administrator. Not for daily use.';
USE ROLE ACCOUNTADMIN;
GRANT ROLE ACCOUNTADMIN TO USER RAYSERRANO_BREAKGLASS;
GRANT ROLE ORGADMIN TO USER RAYSERRANO_BREAKGLASS;
USE ROLE SECURITYADMIN;
ALTER USER RAYSERRANO_BREAKGLASS SET PASSWORD = '<a strong temporary password>' MUST_CHANGE_PASSWORD = TRUE;
/* Step 4. ONLY after step 3 succeeded, logged in as RAYSERRANO_ADMIN:
USE ROLE ACCOUNTADMIN;
REVOKE ROLE ACCOUNTADMIN FROM USER RAYSERRANO;
REVOKE ROLE ORGADMIN FROM USER RAYSERRANO;
SHOW GRANTS TO USER RAYSERRANO;
*/
/* Step 5. Make RAYSERRANO a working developer, as RAYSERRANO_ADMIN, with the per-developer role
(Ray, 2026-10-23, answer 3a). DEV_RAYSERRANO_ROLE is created by the environment
tier setup (not yet written); run this step after it.
USE ROLE SECURITYADMIN;
GRANT ROLE DEV_RAYSERRANO_ROLE TO USER RAYSERRANO;
ALTER USER RAYSERRANO SET DEFAULT_ROLE = DEV_RAYSERRANO_ROLE;
SHOW GRANTS TO USER RAYSERRANO;
*/
/* Step 6. Clean slate for RAYSERRANO (Ray, 2026-10-23): revoke every remaining role, then add grants one
at a time as real needs appear, each recorded here with its reason. Read 2026-10-23: the four project
roles carried administrator-grade power: CIRO_ROLE inherits SYSADMIN, and DCS_ROLE holds CREATE USER
and CREATE INTEGRATION. The roles themselves are not changed, so services that use them are unaffected;
anything signed in AS RAYSERRANO (DBeaver, a connector) loses that access. Run as RAYSERRANO_ADMIN. */
USE ROLE SECURITYADMIN;
REVOKE ROLE ALBA_ROLE FROM USER RAYSERRANO;
REVOKE ROLE CIRO_ROLE FROM USER RAYSERRANO;
REVOKE ROLE DCS_ROLE FROM USER RAYSERRANO;
REVOKE ROLE VERA_ROLE FROM USER RAYSERRANO;
ALTER USER RAYSERRANO SET DEFAULT_ROLE = PUBLIC;
SHOW GRANTS TO USER RAYSERRANO;
/* Grants added back, one per need (append below, newest last):
- DEV_RAYSERRANO_ROLE: Data Vault course development and labs (environment_tiers.sql, part A)
- PILOT_READ_ROLE: read the course material in Pilot, read-only, granted to DEV_RAYSERRANO_ROLE, Ray's one role
(pilot_reader.sql; Ray, 2026-10-23: one role per person)
*/
infra/snowflake/connector_read_grants.sql
/*
Mary's connector reads the conformed mart as CLAUDE_WS_ROLE alone, its session's secondary roles off (Ray, 2026-11-01,
answer 1a, after QA found the connector reading through SYSADMIN). Every build grants SELECT on the conformed mart's
views (dbt_project.yml, +grants) and USAGE on its schema (an on-run-end hook); this script grants the one thing a build
cannot, USAGE on the database, which builds never drop. Run once per environment, the same in each, by the workstation
as SYSADMIN, statements on standard input:
snow sql ... -D vault=DEV__DV__CLAUDE_WS | DV_QA | DV
Refuses any other database before anything changes. Idempotent.
*/
USE ROLE SYSADMIN;
EXECUTE IMMEDIATE $$
BEGIN
IF ('<% vault %>' NOT IN ('DV_QA', 'DV') AND NOT STARTSWITH('<% vault %>', 'DEV__DV__')) THEN
RETURN 1/0; -- stops the script: only DV, DV_QA or a developer's DEV__DV__<USER>
END IF;
RETURN 'granting database usage on <% vault %>';
END;
$$;
GRANT USAGE ON DATABASE <% vault %> TO ROLE CLAUDE_WS_ROLE;
SHOW GRANTS ON DATABASE <% vault %>;
infra/snowflake/correct_zero_key_satellite_rows.sql
/*
One-time correction (Ray, 2026-10-28, answer 1a): the source rows that release 1 loaded on the zero key into two
satellites. A row without a client number or insured id describes no one, so client_sunbelt_s held three such rows
as one false client (two contradictory versions on 2026-10-01, a "change" on 2026-10-02) and insured_northfield_s
one as a false insured (NF-100013). From this fix on, the satellites read only rows with a known key and the
missing-key inspectors raise each such row as a finding; this script removes the rows loaded before the fix.
Only source rows go: the package's own placeholder rows on the zero key (record source SYSTEM or ERROR) stay.
Production holds four rows to remove (three client, one insured); a vault rebuilt after the fix holds none, so the
script is safe to rerun and removes nothing there. The way back is Snowflake Time Travel on the two tables.
Run by the workstation as SYSADMIN, naming the vault database, once, after the production build at promotion:
snow sql ... -D vault=DV < infra/snowflake/correct_zero_key_satellite_rows.sql
Refuses any database other than DV, DV_QA or a developer's DEV__DV__<USER>.
*/
USE ROLE SYSADMIN;
USE WAREHOUSE COMPUTE_WH;
/* 0. Refuse anything but a vault database. */
EXECUTE IMMEDIATE $$
BEGIN
IF ('<% vault %>' NOT IN ('DV', 'DV_QA') AND NOT STARTSWITH('<% vault %>', 'DEV__DV__')) THEN
RETURN 1/0; -- stops the script: only a vault database may be corrected
END IF;
RETURN 'correcting <% vault %>';
END;
$$;
/* 1. Before: the source rows on the zero key in each satellite. */
SELECT 'before' AS STEP,
(SELECT COUNT(*) FROM <% vault %>.RAW_VAULT.CLIENT_SUNBELT_S
WHERE HK_CLIENT_H = '00000000000000000000000000000000' AND RSRC NOT IN ('SYSTEM', 'ERROR')) AS CLIENT_ROWS,
(SELECT COUNT(*) FROM <% vault %>.RAW_VAULT.INSURED_NORTHFIELD_S
WHERE HK_INSURED_H = '00000000000000000000000000000000' AND RSRC NOT IN ('SYSTEM', 'ERROR')) AS INSURED_ROWS;
/* 2. The correction. */
DELETE FROM <% vault %>.RAW_VAULT.CLIENT_SUNBELT_S
WHERE HK_CLIENT_H = '00000000000000000000000000000000' AND RSRC NOT IN ('SYSTEM', 'ERROR');
DELETE FROM <% vault %>.RAW_VAULT.INSURED_NORTHFIELD_S
WHERE HK_INSURED_H = '00000000000000000000000000000000' AND RSRC NOT IN ('SYSTEM', 'ERROR');
/* 3. After: none left; the placeholder rows (zero key and error key) still present. Returns 0, 0, 2, 2. */
SELECT 'after' AS STEP,
(SELECT COUNT(*) FROM <% vault %>.RAW_VAULT.CLIENT_SUNBELT_S
WHERE HK_CLIENT_H = '00000000000000000000000000000000' AND RSRC NOT IN ('SYSTEM', 'ERROR')) AS CLIENT_ROWS,
(SELECT COUNT(*) FROM <% vault %>.RAW_VAULT.INSURED_NORTHFIELD_S
WHERE HK_INSURED_H = '00000000000000000000000000000000' AND RSRC NOT IN ('SYSTEM', 'ERROR')) AS INSURED_ROWS,
(SELECT COUNT(*) FROM <% vault %>.RAW_VAULT.CLIENT_SUNBELT_S WHERE RSRC IN ('SYSTEM', 'ERROR')) AS CLIENT_PLACEHOLDERS,
(SELECT COUNT(*) FROM <% vault %>.RAW_VAULT.INSURED_NORTHFIELD_S WHERE RSRC IN ('SYSTEM', 'ERROR')) AS INSURED_PLACEHOLDERS;
infra/snowflake/ddl/dv.sql
-- The vault's structure: the raw vault, reference and conformed mart schemas of the DIGEST Data Vault course.
-- Phase 1: generated from a built vault by infra/scripts/vault_ddl.py (Snowflake GET_DDL), with database names
-- written as production's (DV, LANDING). Each release commits its own version.
-- To propose a change to the structure, edit this file on a branch of your own: the diff is the proposal.
create or replace schema RAW_VAULT;
create or replace TRANSIENT TABLE AGENT_H (
HK_AGENT_H VARCHAR(16777216),
AGENT_NPN VARCHAR(10),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_AGENT_H primary key (HK_AGENT_H)
);
create or replace TRANSIENT TABLE AGENT_POLICY_L (
HK_AGENT_POLICY_L VARCHAR(16777216),
HK_AGENT_H VARCHAR(16777216),
HK_POLICY_H VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_AGENT_POLICY_L primary key (HK_AGENT_POLICY_L),
constraint FK_AGENT_POLICY_L_HK_AGENT_H foreign key (HK_AGENT_H) references DV.RAW_VAULT.AGENT_H(HK_AGENT_H),
constraint FK_AGENT_POLICY_L_HK_POLICY_H foreign key (HK_POLICY_H) references DV.RAW_VAULT.POLICY_H(HK_POLICY_H)
);
create or replace TRANSIENT TABLE CLIENT_H (
HK_CLIENT_H VARCHAR(16777216),
CLIENT_NUMBER VARCHAR(32),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_CLIENT_H primary key (HK_CLIENT_H)
);
create or replace TRANSIENT TABLE CLIENT_POLICY_L (
HK_CLIENT_POLICY_L VARCHAR(16777216),
HK_CLIENT_H VARCHAR(16777216),
HK_POLICY_H VARCHAR(16777216),
ROLE VARCHAR(16),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_CLIENT_POLICY_L primary key (HK_CLIENT_POLICY_L),
constraint FK_CLIENT_POLICY_L_HK_CLIENT_H foreign key (HK_CLIENT_H) references DV.RAW_VAULT.CLIENT_H(HK_CLIENT_H),
constraint FK_CLIENT_POLICY_L_HK_POLICY_H foreign key (HK_POLICY_H) references DV.RAW_VAULT.POLICY_H(HK_POLICY_H)
);
create or replace TRANSIENT TABLE CLIENT_POLICY_SUNBELT_S (
HK_CLIENT_POLICY_L VARCHAR(16777216),
HD_CLIENT_POLICY_SUNBELT_S VARCHAR(16777216),
RSRC VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
SOLD_ON VARCHAR(10),
AGENT_NAME VARCHAR(64),
PLAN_NAME VARCHAR(64),
constraint PK_CLIENT_POLICY_SUNBELT_S primary key (HK_CLIENT_POLICY_L, LDTS),
constraint FK_CLIENT_POLICY_SUNBELT_S_HK_CLIENT_POLICY_L foreign key (HK_CLIENT_POLICY_L) references DV.RAW_VAULT.CLIENT_POLICY_L(HK_CLIENT_POLICY_L)
);
create or replace TRANSIENT TABLE CLIENT_SUNBELT_S (
HK_CLIENT_H VARCHAR(16777216),
HD_CLIENT_SUNBELT_S VARCHAR(16777216),
RSRC VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
FIRST_NAME VARCHAR(64),
LAST_NAME VARCHAR(64),
STATE VARCHAR(9),
constraint PK_CLIENT_SUNBELT_S primary key (HK_CLIENT_H, LDTS),
constraint FK_CLIENT_SUNBELT_S_HK_CLIENT_H foreign key (HK_CLIENT_H) references DV.RAW_VAULT.CLIENT_H(HK_CLIENT_H)
);
create or replace TRANSIENT TABLE INSURED_H (
HK_INSURED_H VARCHAR(16777216),
INSURED_NUMBER VARCHAR(16),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_INSURED_H primary key (HK_INSURED_H)
);
create or replace TRANSIENT TABLE INSURED_NORTHFIELD_S (
HK_INSURED_H VARCHAR(16777216),
HD_INSURED_NORTHFIELD_S VARCHAR(16777216),
RSRC VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
INSURED_NAME VARCHAR(64),
BIRTH_DATE VARCHAR(10),
ADDRESS VARCHAR(128),
constraint PK_INSURED_NORTHFIELD_S primary key (HK_INSURED_H, LDTS),
constraint FK_INSURED_NORTHFIELD_S_HK_INSURED_H foreign key (HK_INSURED_H) references DV.RAW_VAULT.INSURED_H(HK_INSURED_H)
);
create or replace TRANSIENT TABLE INSURED_POLICY_L (
HK_INSURED_POLICY_L VARCHAR(16777216),
HK_INSURED_H VARCHAR(16777216),
HK_POLICY_H VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_INSURED_POLICY_L primary key (HK_INSURED_POLICY_L),
constraint FK_INSURED_POLICY_L_HK_INSURED_H foreign key (HK_INSURED_H) references DV.RAW_VAULT.INSURED_H(HK_INSURED_H),
constraint FK_INSURED_POLICY_L_HK_POLICY_H foreign key (HK_POLICY_H) references DV.RAW_VAULT.POLICY_H(HK_POLICY_H)
);
create or replace TRANSIENT TABLE ORGANIZATION_H (
HK_ORGANIZATION_H VARCHAR(16777216),
CARRIER_CODE VARCHAR(16),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_ORGANIZATION_H primary key (HK_ORGANIZATION_H)
);
create or replace TRANSIENT TABLE ORGANIZATION_NORTHFIELD_S (
HK_ORGANIZATION_H VARCHAR(16777216),
HD_ORGANIZATION_NORTHFIELD_S VARCHAR(16777216),
RSRC VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
CARRIER_NAME VARCHAR(64),
constraint PK_ORGANIZATION_NORTHFIELD_S primary key (HK_ORGANIZATION_H, LDTS),
constraint FK_ORGANIZATION_NORTHFIELD_S_HK_ORGANIZATION_H foreign key (HK_ORGANIZATION_H) references DV.RAW_VAULT.ORGANIZATION_H(HK_ORGANIZATION_H)
);
create or replace TRANSIENT TABLE ORGANIZATION_PINECREST_S (
HK_ORGANIZATION_H VARCHAR(16777216),
HD_ORGANIZATION_PINECREST_S VARCHAR(16777216),
RSRC VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
CARRIER_NAME VARCHAR(64),
constraint PK_ORGANIZATION_PINECREST_S primary key (HK_ORGANIZATION_H, LDTS),
constraint FK_ORGANIZATION_PINECREST_S_HK_ORGANIZATION_H foreign key (HK_ORGANIZATION_H) references DV.RAW_VAULT.ORGANIZATION_H(HK_ORGANIZATION_H)
);
create or replace TRANSIENT TABLE POLICY_H (
HK_POLICY_H VARCHAR(16777216),
CARRIER_CODE VARCHAR(8),
POLICY_NUMBER VARCHAR(32),
LDTS TIMESTAMP_NTZ(9),
RSRC VARCHAR(16777216),
constraint PK_POLICY_H primary key (HK_POLICY_H)
);
create or replace TRANSIENT TABLE POLICY_NORTHFIELD_S (
HK_POLICY_H VARCHAR(16777216),
HD_POLICY_NORTHFIELD_S VARCHAR(16777216),
RSRC VARCHAR(16777216),
LDTS TIMESTAMP_NTZ(9),
FACE_AMOUNT NUMBER(12,2),
STATUS VARCHAR(9),
EFFECTIVE_DATE VARCHAR(10),
constraint PK_POLICY_NORTHFIELD_S primary key (HK_POLICY_H, LDTS),
constraint FK_POLICY_NORTHFIELD_S_HK_POLICY_H foreign key (HK_POLICY_H) references DV.RAW_VAULT.POLICY_H(HK_POLICY_H)
);
create or replace view CLIENT_POLICY_SUNBELT_S1(
HK_CLIENT_POLICY_L,
HD_CLIENT_POLICY_SUNBELT_S,
RSRC,
LDTS,
LEDTS,
SOLD_ON,
AGENT_NAME,
PLAN_NAME
) as (
-----------------------------------------------------------------------------------------------
-- ( ) --
-- // --
-- ( )=( o ) --
-- ##### ##### # # ####### ####### ###### ####### ####### \\ --
-- # # # # # # # # # # # # # ( ) --
-- # # # # # # # # # # # --
-- ##### # # # # ##### ##### ###### ##### ##### --
-- # # ####### # # # # # # # --
-- # # # # # # # # # # # # # --
-- ##### ##### # # ####### ####### # # # ####### ####### --
-----------------------------------------------------------------------------------------------
-- Generated by datavault4dbt by Scalefree International GmbH --
-- Source & Documentation: https://www.datavault4dbt.com --
-----------------------------------------------------------------------------------------------
WITH
end_dated_source AS (
SELECT
hk_client_policy_l,
hd_client_policy_sunbelt_s,
rsrc,
ldts,
COALESCE(LEAD(ldts - INTERVAL '1 MICROSECOND') OVER (PARTITION BY hk_client_policy_l ORDER BY ldts), CAST(TO_TIMESTAMP('8888-12-31T23:59:59', 'YYYY-MM-DDTHH24:MI:SS') AS TIMESTAMP_NTZ)) AS ledts,
"SOLD_ON",
"AGENT_NAME",
"PLAN_NAME"
FROM DV.RAW_VAULT.client_policy_sunbelt_s
)
SELECT
hk_client_policy_l,
hd_client_policy_sunbelt_s,
rsrc,
ldts,
ledts,
"SOLD_ON",
"AGENT_NAME",
"PLAN_NAME"
FROM end_dated_source
)
;
create or replace view CLIENT_SUNBELT_S1(
HK_CLIENT_H,
HD_CLIENT_SUNBELT_S,
RSRC,
LDTS,
LEDTS,
FIRST_NAME,
LAST_NAME,
STATE
) as (
-----------------------------------------------------------------------------------------------
-- ( ) --
-- // --
-- ( )=( o ) --
-- ##### ##### # # ####### ####### ###### ####### ####### \\ --
-- # # # # # # # # # # # # # ( ) --
-- # # # # # # # # # # # --
-- ##### # # # # ##### ##### ###### ##### ##### --
-- # # ####### # # # # # # # --
-- # # # # # # # # # # # # # --
-- ##### ##### # # ####### ####### # # # ####### ####### --
-----------------------------------------------------------------------------------------------
-- Generated by datavault4dbt by Scalefree International GmbH --
-- Source & Documentation: https://www.datavault4dbt.com --
-----------------------------------------------------------------------------------------------
WITH
end_dated_source AS (
SELECT
hk_client_h,
hd_client_sunbelt_s,
rsrc,
ldts,
COALESCE(LEAD(ldts - INTERVAL '1 MICROSECOND') OVER (PARTITION BY hk_client_h ORDER BY ldts), CAST(TO_TIMESTAMP('8888-12-31T23:59:59', 'YYYY-MM-DDTHH24:MI:SS') AS TIMESTAMP_NTZ)) AS ledts,
"FIRST_NAME",
"LAST_NAME",
"STATE"
FROM DV.RAW_VAULT.client_sunbelt_s
)
SELECT
hk_client_h,
hd_client_sunbelt_s,
rsrc,
ldts,
ledts,
"FIRST_NAME",
"LAST_NAME",
"STATE"
FROM end_dated_source
)
;
create or replace view INSURED_NORTHFIELD_S1(
HK_INSURED_H,
HD_INSURED_NORTHFIELD_S,
RSRC,
LDTS,
LEDTS,
INSURED_NAME,
BIRTH_DATE,
ADDRESS
) as (
-----------------------------------------------------------------------------------------------
-- ( ) --
-- // --
-- ( )=( o ) --
-- ##### ##### # # ####### ####### ###### ####### ####### \\ --
-- # # # # # # # # # # # # # ( ) --
-- # # # # # # # # # # # --
-- ##### # # # # ##### ##### ###### ##### ##### --
-- # # ####### # # # # # # # --
-- # # # # # # # # # # # # # --
-- ##### ##### # # ####### ####### # # # ####### ####### --
-----------------------------------------------------------------------------------------------
-- Generated by datavault4dbt by Scalefree International GmbH --
-- Source & Documentation: https://www.datavault4dbt.com --
-----------------------------------------------------------------------------------------------
WITH
end_dated_source AS (
SELECT
hk_insured_h,
hd_insured_northfield_s,
rsrc,
ldts,
COALESCE(LEAD(ldts - INTERVAL '1 MICROSECOND') OVER (PARTITION BY hk_insured_h ORDER BY ldts), CAST(TO_TIMESTAMP('8888-12-31T23:59:59', 'YYYY-MM-DDTHH24:MI:SS') AS TIMESTAMP_NTZ)) AS ledts,
"INSURED_NAME",
"BIRTH_DATE",
"ADDRESS"
FROM DV.RAW_VAULT.insured_northfield_s
)
SELECT
hk_insured_h,
hd_insured_northfield_s,
rsrc,
ldts,
ledts,
"INSURED_NAME",
"BIRTH_DATE",
"ADDRESS"
FROM end_dated_source
)
;
create or replace view ORGANIZATION_NORTHFIELD_S1(
HK_ORGANIZATION_H,
HD_ORGANIZATION_NORTHFIELD_S,
RSRC,
LDTS,
LEDTS,
CARRIER_NAME
) as (
-----------------------------------------------------------------------------------------------
-- ( ) --
-- // --
-- ( )=( o ) --
-- ##### ##### # # ####### ####### ###### ####### ####### \\ --
-- # # # # # # # # # # # # # ( ) --
-- # # # # # # # # # # # --
-- ##### # # # # ##### ##### ###### ##### ##### --
-- # # ####### # # # # # # # --
-- # # # # # # # # # # # # # --
-- ##### ##### # # ####### ####### # # # ####### ####### --
-----------------------------------------------------------------------------------------------
-- Generated by datavault4dbt by Scalefree International GmbH --
-- Source & Documentation: https://www.datavault4dbt.com --
-----------------------------------------------------------------------------------------------
WITH
end_dated_source AS (
SELECT
hk_organization_h,
hd_organization_northfield_s,
rsrc,
ldts,
COALESCE(LEAD(ldts - INTERVAL '1 MICROSECOND') OVER (PARTITION BY hk_organization_h ORDER BY ldts), CAST(TO_TIMESTAMP('8888-12-31T23:59:59', 'YYYY-MM-DDTHH24:MI:SS') AS TIMESTAMP_NTZ)) AS ledts,
"CARRIER_NAME"
FROM DV.RAW_VAULT.organization_northfield_s
)
SELECT
hk_organization_h,
hd_organization_northfield_s,
rsrc,
ldts,
ledts,
"CARRIER_NAME"
FROM end_dated_source
)
;
create or replace view ORGANIZATION_PINECREST_S1(
HK_ORGANIZATION_H,
HD_ORGANIZATION_PINECREST_S,
RSRC,
LDTS,
LEDTS,
CARRIER_NAME
) as (
-----------------------------------------------------------------------------------------------
-- ( ) --
-- // --
-- ( )=( o ) --
-- ##### ##### # # ####### ####### ###### ####### ####### \\ --
-- # # # # # # # # # # # # # ( ) --
-- # # # # # # # # # # # --
-- ##### # # # # ##### ##### ###### ##### ##### --
-- # # ####### # # # # # # # --
-- # # # # # # # # # # # # # --
-- ##### ##### # # ####### ####### # # # ####### ####### --
-----------------------------------------------------------------------------------------------
-- Generated by datavault4dbt by Scalefree International GmbH --
-- Source & Documentation: https://www.datavault4dbt.com --
-----------------------------------------------------------------------------------------------
WITH
end_dated_source AS (
SELECT
hk_organization_h,
hd_organization_pinecrest_s,
rsrc,
ldts,
COALESCE(LEAD(ldts - INTERVAL '1 MICROSECOND') OVER (PARTITION BY hk_organization_h ORDER BY ldts), CAST(TO_TIMESTAMP('8888-12-31T23:59:59', 'YYYY-MM-DDTHH24:MI:SS') AS TIMESTAMP_NTZ)) AS ledts,
"CARRIER_NAME"
FROM DV.RAW_VAULT.organization_pinecrest_s
)
SELECT
hk_organization_h,
hd_organization_pinecrest_s,
rsrc,
ldts,
ledts,
"CARRIER_NAME"
FROM end_dated_source
)
;
create or replace view POLICY_NORTHFIELD_S1(
HK_POLICY_H,
HD_POLICY_NORTHFIELD_S,
RSRC,
LDTS,
LEDTS,
FACE_AMOUNT,
STATUS,
EFFECTIVE_DATE
) as (
-----------------------------------------------------------------------------------------------
-- ( ) --
-- // --
-- ( )=( o ) --
-- ##### ##### # # ####### ####### ###### ####### ####### \\ --
-- # # # # # # # # # # # # # ( ) --
-- # # # # # # # # # # # --
-- ##### # # # # ##### ##### ###### ##### ##### --
-- # # ####### # # # # # # # --
-- # # # # # # # # # # # # # --
-- ##### ##### # # ####### ####### # # # ####### ####### --
-----------------------------------------------------------------------------------------------
-- Generated by datavault4dbt by Scalefree International GmbH --
-- Source & Documentation: https://www.datavault4dbt.com --
-----------------------------------------------------------------------------------------------
WITH
end_dated_source AS (
SELECT
hk_policy_h,
hd_policy_northfield_s,
rsrc,
ldts,
COALESCE(LEAD(ldts - INTERVAL '1 MICROSECOND') OVER (PARTITION BY hk_policy_h ORDER BY ldts), CAST(TO_TIMESTAMP('8888-12-31T23:59:59', 'YYYY-MM-DDTHH24:MI:SS') AS TIMESTAMP_NTZ)) AS ledts,
"FACE_AMOUNT",
"STATUS",
"EFFECTIVE_DATE"
FROM DV.RAW_VAULT.policy_northfield_s
)
SELECT
hk_policy_h,
hd_policy_northfield_s,
rsrc,
ldts,
ledts,
"FACE_AMOUNT",
"STATUS",
"EFFECTIVE_DATE"
FROM end_dated_source
)
;
create or replace schema REFERENCE;
create or replace TRANSIENT TABLE SEED_BUILD_EXPECTATIONS (
VERSION_TAG VARCHAR(16777216),
TABLE_NAME VARCHAR(16777216),
METRIC_NAME VARCHAR(16777216),
CONDITION VARCHAR(16777216),
DESCRIPTION VARCHAR(16777216)
);
create or replace TRANSIENT TABLE SEED_SOURCE_SYSTEM (
SOURCE_SYSTEM_CODE VARCHAR(16777216),
SOURCE_NAME VARCHAR(16777216),
DESCRIPTION VARCHAR(16777216),
OWNER VARCHAR(16777216),
constraint PK_SEED_SOURCE_SYSTEM primary key (SOURCE_SYSTEM_CODE)
);
create or replace schema CONFORMED_DM;
create or replace view AGENT_DIM(
AGENT_KEY,
AGENT_NPN,
FULL_NAME
) as (
SELECT person_key AS agent_key, person_number AS agent_npn, full_name
FROM DV.CONFORMED_DM.person_dim
WHERE person_role = 'agent'
OR source_key IN ('SYSTEM', 'ERROR') -- the two ghost rows, mart design D-14
)
;
create or replace view AGENT_POLICY_FACT(
AGENT_POLICY_KEY,
AGENT_KEY,
POLICY_KEY,
FIRST_NAMED_BY_SOURCE_KEY,
FIRST_KNOWN_DATE
) as (
SELECT hk_agent_policy_l AS agent_policy_key,
hk_agent_h AS agent_key, hk_policy_h AS policy_key,
rsrc AS first_named_by_source_key, -- the first source to name the pair
TO_DATE(ldts) AS first_known_date
FROM DV.RAW_VAULT.agent_policy_l
WHERE rsrc NOT IN ('SYSTEM', 'ERROR')
)
;
create or replace view CARRIER_DIM(
CARRIER_KEY,
CARRIER_CODE,
CARRIER_NAME
) as (
SELECT organization_key AS carrier_key, organization_code AS carrier_code, organization_name AS carrier_name
FROM DV.CONFORMED_DM.organization_dim
WHERE source_key IN ('NFL', 'PMC') -- organizations sent through a carrier feed
OR source_key IN ('SYSTEM', 'ERROR') -- the two ghost rows, mart design D-14
)
;
create or replace view CLIENT_DIM(
CLIENT_KEY,
CLIENT_NUMBER,
FIRST_NAME,
LAST_NAME,
STATE,
SOURCE_KEY,
FIRST_KNOWN_DATE
) as (
SELECT person_key AS client_key, person_number AS client_number, first_name, last_name, state,
source_key, first_known_date
FROM DV.CONFORMED_DM.person_dim
WHERE person_role = 'client'
OR source_key IN ('SYSTEM', 'ERROR') -- the two ghost rows, mart design D-14
)
;
create or replace view CLIENT_POLICY_ROLE_FACT(
CLIENT_POLICY_ROLE_KEY,
CLIENT_KEY,
POLICY_KEY,
ROLE,
SOLD_DATE,
FIRST_KNOWN_DATE
) as (
SELECT l.hk_client_policy_l AS client_policy_role_key,
l.hk_client_h AS client_key, l.hk_policy_h AS policy_key,
l.role, -- degenerate code, mart design D-3
TRY_TO_DATE(s.sold_on) AS sold_date, -- empty when the sale gives none
TO_DATE(l.ldts) AS first_known_date
FROM DV.RAW_VAULT.client_policy_l l
LEFT JOIN DV.RAW_VAULT.client_policy_sunbelt_s1 s
ON s.hk_client_policy_l = l.hk_client_policy_l AND s.ledts = '8888-12-31 23:59:59'
WHERE l.rsrc NOT IN ('SYSTEM', 'ERROR') -- ghost rows are dimension members, never fact rows
)
;
create or replace view DAY_DIM(
DATE_DAY
) as (
with rawdata as (
with p as (
select 0 as generated_number union all select 1
), unioned as (
select
p0.generated_number * power(2, 0)
+
p1.generated_number * power(2, 1)
+
p2.generated_number * power(2, 2)
+
p3.generated_number * power(2, 3)
+
p4.generated_number * power(2, 4)
+
p5.generated_number * power(2, 5)
+
p6.generated_number * power(2, 6)
+
p7.generated_number * power(2, 7)
+
p8.generated_number * power(2, 8)
+
p9.generated_number * power(2, 9)
+ 1
as generated_number
from
p as p0
cross join
p as p1
cross join
p as p2
cross join
p as p3
cross join
p as p4
cross join
p as p5
cross join
p as p6
cross join
p as p7
cross join
p as p8
cross join
p as p9
)
select *
from unioned
where generated_number <= 730
order by generated_number
),
all_periods as (
select (
dateadd(
day,
row_number() over (order by generated_number) - 1,
to_date('2026-01-01')
)
) as date_day
from rawdata
),
filtered as (
select *
from all_periods
where date_day <= to_date('2028-01-01')
)
select * from filtered
)
;
create or replace view FINDING_FACT(
FINDING_KEY,
SOURCE_KEY,
DETECTED_DATE,
FINDER_NAME,
FINDER_TYPE,
SOURCE_TABLE,
SOURCE_COLUMN,
COLUMN_VALUE,
STATUS,
FINDING_KIND,
FINDING_ROW
) as (
SELECT finding_pk AS finding_key,
CASE source_system WHEN 'SUNBELT' THEN 'SSI' WHEN 'NORTHFIELD' THEN 'NFL' ELSE source_system END AS source_key,
TO_DATE(CONVERT_TIMEZONE('UTC', detected_at)) AS detected_date,
finder_name, finder_type, source_table, source_column, column_value, status,
CASE WHEN finder_type = 'audit' THEN 'audit'
WHEN ENDSWITH(finder_name, '_key_missing') THEN 'missing_key'
WHEN ENDSWITH(finder_name, '_form') THEN 'form' END AS finding_kind,
source_table || ': ' || column_value AS finding_row
FROM LANDING.CONFIG.data_finding
)
;
create or replace secure view INSURED_DIM(
INSURED_KEY,
FULL_NAME,
BIRTH_DATE,
ADDRESS
) as (
SELECT person_key AS insured_key, full_name,
CASE WHEN IS_ROLE_IN_SESSION('SYSADMIN') THEN birth_date ELSE LEFT(birth_date, 4) END AS birth_date,
CASE WHEN IS_ROLE_IN_SESSION('SYSADMIN') THEN address ELSE RIGHT(TRIM(address), 2) END AS address
FROM DV.CONFORMED_DM.person_dim
WHERE person_role = 'insured'
OR source_key IN ('SYSTEM', 'ERROR') -- the two ghost rows, mart design D-14
)
;
create or replace view ORGANIZATION_DIM(
ORGANIZATION_KEY,
ORGANIZATION_CODE,
ORGANIZATION_NAME,
SOURCE_KEY
) as (
SELECT h.hk_organization_h AS organization_key, h.carrier_code AS organization_code,
COALESCE(n.carrier_name, p.carrier_name, h.carrier_code) AS organization_name, h.rsrc AS source_key
FROM DV.RAW_VAULT.organization_h h
LEFT JOIN DV.RAW_VAULT.organization_northfield_s1 n
ON n.hk_organization_h = h.hk_organization_h AND n.ledts = '8888-12-31 23:59:59' AND n.rsrc = 'NFL'
LEFT JOIN DV.RAW_VAULT.organization_pinecrest_s1 p
ON p.hk_organization_h = h.hk_organization_h AND p.ledts = '8888-12-31 23:59:59' AND p.rsrc = 'PMC'
)
;
create or replace view PERSON_DIM(
PERSON_KEY,
PERSON_ROLE,
SOURCE_KEY,
FIRST_KNOWN_DATE,
PERSON_NUMBER,
FIRST_NAME,
LAST_NAME,
FULL_NAME,
STATE,
BIRTH_DATE,
ADDRESS
) as (
WITH ghosts AS (
SELECT hk_client_h AS person_key, CAST(NULL AS VARCHAR) AS person_role, rsrc AS source_key,
CAST(NULL AS DATE) AS first_known_date, client_number AS person_number,
CAST(NULL AS VARCHAR) AS first_name, CAST(NULL AS VARCHAR) AS last_name, client_number AS full_name,
CAST(NULL AS VARCHAR) AS state, CAST(NULL AS VARCHAR) AS birth_date, CAST(NULL AS VARCHAR) AS address
FROM DV.RAW_VAULT.client_h WHERE rsrc IN ('SYSTEM', 'ERROR')
),
clients AS (
SELECT h.hk_client_h, 'client', h.rsrc, TO_DATE(h.ldts), h.client_number,
s.first_name, s.last_name, TRIM(COALESCE(s.first_name, '') || ' ' || COALESCE(s.last_name, '')),
s.state, CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR)
FROM DV.RAW_VAULT.client_h h
LEFT JOIN DV.RAW_VAULT.client_sunbelt_s1 s
ON s.hk_client_h = h.hk_client_h AND s.ledts = '8888-12-31 23:59:59'
WHERE h.rsrc NOT IN ('SYSTEM', 'ERROR')
),
insureds AS (
SELECT h.hk_insured_h, 'insured', h.rsrc, TO_DATE(h.ldts), h.insured_number,
CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), s.insured_name,
CAST(NULL AS VARCHAR), s.birth_date, s.address
FROM DV.RAW_VAULT.insured_h h
LEFT JOIN DV.RAW_VAULT.insured_northfield_s1 s
ON s.hk_insured_h = h.hk_insured_h AND s.ledts = '8888-12-31 23:59:59'
WHERE h.rsrc NOT IN ('SYSTEM', 'ERROR')
),
latest_agent_name AS (
SELECT ap.hk_agent_h, s.agent_name
FROM DV.RAW_VAULT.agent_policy_l ap
JOIN DV.RAW_VAULT.client_policy_l cp ON cp.hk_policy_h = ap.hk_policy_h
JOIN DV.RAW_VAULT.client_policy_sunbelt_s1 s ON s.hk_client_policy_l = cp.hk_client_policy_l
QUALIFY ROW_NUMBER() OVER (PARTITION BY ap.hk_agent_h ORDER BY s.ldts DESC) = 1
),
agents AS (
SELECT h.hk_agent_h, 'agent', h.rsrc, TO_DATE(h.ldts), h.agent_npn,
CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), n.agent_name,
CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR)
FROM DV.RAW_VAULT.agent_h h
LEFT JOIN latest_agent_name n ON n.hk_agent_h = h.hk_agent_h
WHERE h.rsrc NOT IN ('SYSTEM', 'ERROR')
)
SELECT * FROM ghosts
UNION ALL SELECT * FROM clients
UNION ALL SELECT * FROM insureds
UNION ALL SELECT * FROM agents
)
;
create or replace view POLICY_DIM(
POLICY_KEY,
CARRIER_KEY,
POLICY_NUMBER,
PLAN_NAME,
FIRST_NAMED_BY_SOURCE_KEY
) as (
WITH latest_plan AS (
SELECT cp.hk_policy_h, s.plan_name
FROM DV.RAW_VAULT.client_policy_l cp
JOIN DV.RAW_VAULT.client_policy_sunbelt_s1 s ON s.hk_client_policy_l = cp.hk_client_policy_l
QUALIFY ROW_NUMBER() OVER (PARTITION BY cp.hk_policy_h ORDER BY s.ldts DESC) = 1
)
SELECT p.hk_policy_h AS policy_key,
COALESCE(o.hk_organization_h, '00000000000000000000000000000000') AS carrier_key,
p.policy_number, lp.plan_name, p.rsrc AS first_named_by_source_key
FROM DV.RAW_VAULT.policy_h p
LEFT JOIN DV.RAW_VAULT.organization_h o
ON o.carrier_code = p.carrier_code AND o.rsrc NOT IN ('SYSTEM', 'ERROR')
LEFT JOIN latest_plan lp ON lp.hk_policy_h = p.hk_policy_h
)
;
create or replace view POLICY_TERMS_FACT(
POLICY_TERMS_KEY,
POLICY_KEY,
VALID_FROM_DATE,
VALID_TO_DATE,
EFFECTIVE_DATE,
FACE_AMOUNT,
STATUS
) as (
SELECT md5(cast(coalesce(cast(hk_policy_h as TEXT), '_dbt_utils_surrogate_key_null_') || '-' || coalesce(cast(ldts as TEXT), '_dbt_utils_surrogate_key_null_') as TEXT)) AS policy_terms_key, -- the policy key and the load
hk_policy_h AS policy_key,
TO_DATE(ldts) AS valid_from_date,
CASE WHEN ledts = '8888-12-31 23:59:59' THEN NULL ELSE TO_DATE(ledts) END AS valid_to_date, -- empty while current, D-4
TRY_TO_DATE(effective_date) AS effective_date,
face_amount, -- the measure, US dollars as sent
status
FROM DV.RAW_VAULT.policy_northfield_s1
WHERE rsrc NOT IN ('SYSTEM', 'ERROR') -- ghost rows are never fact rows (section 3.5)
)
;
create or replace view SOURCE_DIM(
SOURCE_KEY,
SOURCE_NAME,
OWNER
) as (
SELECT source_system_code AS source_key, source_name, owner
FROM DV.REFERENCE.seed_source_system
)
;
infra/snowflake/dev_mart_schema.sql
/*
DEV__DV__CLAUDE_WS.MART was created for Kimi's connector fixture (INFRA_SMOKE, MF_TIME_SPINE), outside the
developer's role, so the developer build could not add the first mart views (Ray, 2026-10-26). The schema goes to the
developer's builder role, grants kept; the fixture's tables stay where they are. Run by the workstation as SYSADMIN.
*/
USE ROLE SYSADMIN;
GRANT OWNERSHIP ON SCHEMA DEV__DV__CLAUDE_WS.MART TO ROLE DEV_CLAUDE_WS_ROLE COPY CURRENT GRANTS;
SHOW GRANTS ON SCHEMA DEV__DV__CLAUDE_WS.MART;
infra/snowflake/drop_mart_views.sql
/*
Drop the release 1 mart's five MART views (mart design D-6, section 4.4; guide step 78): the conformed mart in
CONFORMED_DM replaces them, and DV Digest reloads from the build. Lists the MART views before and after.
Run by the workstation as SYSADMIN, statements on standard input, naming the vault database:
QA: snow sql ... -D vault=DV_QA < infra/snowflake/drop_mart_views.sql
Development: snow sql ... -D vault=DEV__DV__CLAUDE_WS < (the same)
Production: -D vault=DV, at the promotion that carries the conformed mart.
Refuses any database but DV_QA, DV or a developer's DEV__DV__<USER>. Safe to rerun.
*/
USE ROLE SYSADMIN;
USE WAREHOUSE COMPUTE_WH;
/* 0. Refuse an unknown database before anything changes. */
EXECUTE IMMEDIATE $$
BEGIN
IF ('<% vault %>' NOT IN ('DV_QA', 'DV') AND NOT STARTSWITH('<% vault %>', 'DEV__DV__')) THEN
RETURN 1/0; -- stops the script: only DV_QA, DV or a developer's DEV__DV__<USER>
END IF;
RETURN 'dropping the MART views of <% vault %>';
END;
$$;
/* 1. The MART views before the drop. */
SELECT table_name FROM <% vault %>.INFORMATION_SCHEMA.VIEWS WHERE table_schema = 'MART' ORDER BY table_name;
/* 2. Drop the five. */
DROP VIEW IF EXISTS <% vault %>.MART.CLIENT_POLICY_FACT;
DROP VIEW IF EXISTS <% vault %>.MART.AGENT_POLICY_FACT;
DROP VIEW IF EXISTS <% vault %>.MART.POLICY_FACT;
DROP VIEW IF EXISTS <% vault %>.MART.FINDING_FACT;
DROP VIEW IF EXISTS <% vault %>.MART.TIME_SPINE_DAY;
/* 3. The MART views after: none of the five. */
SELECT table_name FROM <% vault %>.INFORMATION_SCHEMA.VIEWS WHERE table_schema = 'MART' ORDER BY table_name;
infra/snowflake/dv_qa_role.sql
/*
The workstation's part of the QA split (Ray, 2026-10-26, answer 1b), after Ray's infra/snowflake/service_users.sql:
DV_QA_ROLE takes the warehouse and the ownership of DV_QA from DV_ROLE. LANDING_QA's grants move with the next
run of infra/snowflake/refresh_landing_tier.sql (QA tier). Run by the workstation as SYSADMIN, statements on
standard input. Idempotent.
*/
USE ROLE SYSADMIN;
USE WAREHOUSE COMPUTE_WH;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE DV_QA_ROLE;
GRANT OWNERSHIP ON DATABASE DV_QA TO ROLE DV_QA_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE DV_QA TO ROLE DV_QA_ROLE COPY CURRENT GRANTS;
SHOW GRANTS ON DATABASE DV_QA;
infra/snowflake/environment_tiers.sql
/*
Environment tiers for digest-datavault (Ray, 2026-10-23, answer 1y).
Tier Landing (sources, STORAGE, CONFIG) Vault (dbt builds) Builder role
production LANDING DV DV_ROLE
QA LANDING_QA (clone of LANDING) DV_QA DV_QA_ROLE
developer DEV__LANDING__<USER> (clone) DEV__DV__<USER> DEV_<USER>_ROLE
LANDING_ROLE owns every landing database; LANDING_READ_ROLE reads them. Each developer role has the same
shape on its own databases that DV_ROLE has in production and QA, so a privilege fault shows in development
first. Idempotent where Snowflake allows; the clones are replaced on every run.
PART A, run by Ray as RAYSERRANO_ADMIN: roles and role grants only.
PART B, run by the workstation as SYSADMIN, after part A: databases, clones, ownership, object grants.
WHAT EACH ROLE RECEIVES (and where):
LANDING_ROLE ownership of LANDING, LANDING_QA, DEV__LANDING__RAYSERRANO, DEV__LANDING__CLAUDE_WS,
their schemas and tables (B2); warehouse usage (B opening); granted to SYSADMIN (A)
LANDING_READ_ROLE usage on every landing database and its schemas, select on its tables (B3);
granted to SYSADMIN (A)
(QA split, Ray 2026-10-26: DV_QA_ROLE takes DV_ROLE's place on LANDING_QA and owns DV_QA; service_users.sql)
DV_ROLE on LANDING and LANDING_QA: usage, select on all tables, create table and view and
modify on STORAGE and CONFIG, insert, update, delete on CONFIG's tables (B3);
ownership of DV_QA (B5); already owns DV and holds warehouse usage
DEV_RAYSERRANO_ROLE the same as DV_ROLE, on DEV__LANDING__RAYSERRANO (B3); ownership of DEV__DV__RAYSERRANO
(B5); warehouse usage (B opening); granted to user RAYSERRANO as default role (A)
DEV_CLAUDE_WS_ROLE the same as DV_ROLE, on DEV__LANDING__CLAUDE_WS (B3); ownership of DEV__DV__CLAUDE_WS
(B5); warehouse usage (B opening); granted to CLAUDE_WS_ROLE (A)
DEFERRED (needs MANAGE GRANTS, an administrator privilege): future grants, so readers see tables created
later in STORAGE. Not needed until the first increment that writes STORAGE history; builders own the
tables they create. Found 2026-10-23 when part B's first run failed on them.
*/
/* ===================== PART A (RAYSERRANO_ADMIN) ===================== */
USE ROLE SECURITYADMIN;
CREATE ROLE IF NOT EXISTS LANDING_ROLE COMMENT = 'Owns the landing databases (LANDING, LANDING_QA, DEV__LANDING__*)';
CREATE ROLE IF NOT EXISTS LANDING_READ_ROLE COMMENT = 'Reads every landing database';
CREATE ROLE IF NOT EXISTS DEV_RAYSERRANO_ROLE COMMENT = 'Developer role for RAYSERRANO: builder of DEV__DV__RAYSERRANO on DEV__LANDING__RAYSERRANO';
CREATE ROLE IF NOT EXISTS DEV_CLAUDE_WS_ROLE COMMENT = 'Developer role for CLAUDE_WS: builder of DEV__DV__CLAUDE_WS on DEV__LANDING__CLAUDE_WS';
GRANT ROLE LANDING_ROLE TO ROLE SYSADMIN;
GRANT ROLE LANDING_READ_ROLE TO ROLE SYSADMIN;
GRANT ROLE DEV_RAYSERRANO_ROLE TO ROLE SYSADMIN;
GRANT ROLE DEV_CLAUDE_WS_ROLE TO ROLE SYSADMIN;
GRANT ROLE DEV_RAYSERRANO_ROLE TO USER RAYSERRANO; /* admin runbook step 5 */
ALTER USER RAYSERRANO SET DEFAULT_ROLE = DEV_RAYSERRANO_ROLE;
GRANT ROLE DEV_CLAUDE_WS_ROLE TO ROLE CLAUDE_WS_ROLE;
/* ===================== PART B (workstation, SYSADMIN) ===================== */
USE ROLE SYSADMIN;
USE WAREHOUSE COMPUTE_WH;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE DEV_RAYSERRANO_ROLE;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE DEV_CLAUDE_WS_ROLE;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE LANDING_ROLE;
/* B1. Clones of production landing, taken before production's CONFIG is cleared (B4), so the workstation's
development findings and baselines move into its own clone. */
CREATE OR REPLACE DATABASE LANDING_QA CLONE LANDING;
CREATE OR REPLACE DATABASE DEV__LANDING__RAYSERRANO CLONE LANDING;
CREATE OR REPLACE DATABASE DEV__LANDING__CLAUDE_WS CLONE LANDING;
/* B2. Every landing database owned by LANDING_ROLE, grants preserved. */
GRANT OWNERSHIP ON DATABASE LANDING TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE LANDING TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL TABLES IN DATABASE LANDING TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON DATABASE LANDING_QA TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE LANDING_QA TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL TABLES IN DATABASE LANDING_QA TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON DATABASE DEV__LANDING__RAYSERRANO TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE DEV__LANDING__RAYSERRANO TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL TABLES IN DATABASE DEV__LANDING__RAYSERRANO TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON DATABASE DEV__LANDING__CLAUDE_WS TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE DEV__LANDING__CLAUDE_WS TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL TABLES IN DATABASE DEV__LANDING__CLAUDE_WS TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
/* B3. Builder and reader grants, as the owner. Database-level grants do not survive a clone,
so each database is granted explicitly. */
USE ROLE LANDING_ROLE;
EXECUTE IMMEDIATE $$
DECLARE
pairs ARRAY := ARRAY_CONSTRUCT(
ARRAY_CONSTRUCT('LANDING', 'DV_ROLE'),
ARRAY_CONSTRUCT('LANDING_QA', 'DV_QA_ROLE'),
ARRAY_CONSTRUCT('DEV__LANDING__RAYSERRANO', 'DEV_RAYSERRANO_ROLE'),
ARRAY_CONSTRUCT('DEV__LANDING__CLAUDE_WS', 'DEV_CLAUDE_WS_ROLE'));
db STRING; b STRING; i INTEGER;
BEGIN
FOR i IN 0 TO ARRAY_SIZE(pairs) - 1 DO
db := pairs[i][0]; b := pairs[i][1];
EXECUTE IMMEDIATE 'GRANT USAGE ON DATABASE ' || db || ' TO ROLE ' || b;
EXECUTE IMMEDIATE 'GRANT USAGE ON ALL SCHEMAS IN DATABASE ' || db || ' TO ROLE ' || b;
EXECUTE IMMEDIATE 'GRANT SELECT ON ALL TABLES IN DATABASE ' || db || ' TO ROLE ' || b;
EXECUTE IMMEDIATE 'GRANT CREATE TABLE, CREATE VIEW, MODIFY ON SCHEMA ' || db || '.STORAGE TO ROLE ' || b;
EXECUTE IMMEDIATE 'GRANT CREATE TABLE, CREATE VIEW, MODIFY ON SCHEMA ' || db || '.CONFIG TO ROLE ' || b;
EXECUTE IMMEDIATE 'GRANT INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA ' || db || '.CONFIG TO ROLE ' || b;
EXECUTE IMMEDIATE 'GRANT USAGE ON DATABASE ' || db || ' TO ROLE LANDING_READ_ROLE';
EXECUTE IMMEDIATE 'GRANT USAGE ON ALL SCHEMAS IN DATABASE ' || db || ' TO ROLE LANDING_READ_ROLE';
EXECUTE IMMEDIATE 'GRANT SELECT ON ALL TABLES IN DATABASE ' || db || ' TO ROLE LANDING_READ_ROLE';
END FOR;
RETURN 'landing grants applied to ' || ARRAY_SIZE(pairs) || ' databases';
END;
$$;
/* B4. Production CONFIG starts clean: its findings and baselines came from the workstation's development
builds, which B1 carried into DEV__LANDING__CLAUDE_WS. CONFIG holds no source data. */
TRUNCATE TABLE LANDING.CONFIG.DATA_FINDING;
TRUNCATE TABLE LANDING.CONFIG.BUILD_BASELINE;
TRUNCATE TABLE LANDING_QA.CONFIG.DATA_FINDING;
TRUNCATE TABLE LANDING_QA.CONFIG.BUILD_BASELINE;
TRUNCATE TABLE DEV__LANDING__RAYSERRANO.CONFIG.DATA_FINDING;
TRUNCATE TABLE DEV__LANDING__RAYSERRANO.CONFIG.BUILD_BASELINE;
/* B5. Vault databases. DV (production) exists, owned by DV_ROLE. */
USE ROLE SYSADMIN;
CREATE DATABASE IF NOT EXISTS DV_QA;
GRANT OWNERSHIP ON DATABASE DV_QA TO ROLE DV_QA_ROLE COPY CURRENT GRANTS;
CREATE DATABASE IF NOT EXISTS DEV__DV__RAYSERRANO;
GRANT OWNERSHIP ON DATABASE DEV__DV__RAYSERRANO TO ROLE DEV_RAYSERRANO_ROLE COPY CURRENT GRANTS;
CREATE DATABASE IF NOT EXISTS DEV__DV__CLAUDE_WS;
GRANT OWNERSHIP ON DATABASE DEV__DV__CLAUDE_WS TO ROLE DEV_CLAUDE_WS_ROLE COPY CURRENT GRANTS;
infra/snowflake/landing.sql
/*
LANDING: landed sources, Pragmatic Data Platform storage, and testing configuration
for the DIGEST Data Vault course (increment 1 design D-1). Owned by SYSADMIN. Access is granted directly to
CLAUDE_WS_ROLE; no per-function roles (Ray, 2026-10-23, answer 3b, a recorded
deviation from guide decision G-2). Idempotent: safe to run again.
*/
USE ROLE SYSADMIN;
CREATE DATABASE IF NOT EXISTS LANDING;
CREATE SCHEMA IF NOT EXISTS LANDING.SUNBELT;
CREATE SCHEMA IF NOT EXISTS LANDING.STORAGE;
CREATE SCHEMA IF NOT EXISTS LANDING.CONFIG;
/* Source table: one acquired agency's client file (increment 1 profile, item 267). */
CREATE TABLE IF NOT EXISTS LANDING.SUNBELT.CLIENT (
CRM_ROW_ID NUMBER NOT NULL,
CLIENT_NUMBER VARCHAR(32),
FIRST_NAME VARCHAR(64),
LAST_NAME VARCHAR(64),
STATE VARCHAR(2),
LOADED_AT VARCHAR(19) NOT NULL
);
/* Findings written by inspectors (layer 5) and auditors (layer 6). Columns
FINDER_NAME and FINDER_TYPE (Ray, 2026-10-22). */
CREATE TABLE IF NOT EXISTS LANDING.CONFIG.DATA_FINDING (
FINDING_PK VARCHAR(128) NOT NULL,
SOURCE_SYSTEM VARCHAR(32) NOT NULL,
SOURCE_TABLE VARCHAR(256),
SOURCE_COLUMN VARCHAR(128),
COLUMN_VALUE VARCHAR(512),
FINDER_NAME VARCHAR(128) NOT NULL,
FINDER_TYPE VARCHAR(32) NOT NULL,
DETAIL VARCHAR(2048),
STATUS VARCHAR(32) DEFAULT 'detected',
FINDING_ID VARCHAR(32),
NOTES VARCHAR(1024),
ROOT_CAUSE VARCHAR(2048),
DETECTED_AT TIMESTAMP_TZ DEFAULT CURRENT_TIMESTAMP(),
ACKNOWLEDGED_BY VARCHAR(64),
ACKNOWLEDGED_AT TIMESTAMP_TZ
);
/* Before and after snapshots for capture_baseline and verify_build. */
CREATE TABLE IF NOT EXISTS LANDING.CONFIG.BUILD_BASELINE (
SNAPSHOT_TYPE VARCHAR(10) NOT NULL,
TABLE_NAME VARCHAR(200) NOT NULL,
METRIC_NAME VARCHAR(200) NOT NULL,
METRIC_VALUE NUMBER(38,2),
CAPTURED_AT TIMESTAMP_TZ(9) NOT NULL DEFAULT CURRENT_TIMESTAMP(),
DATABASE_NAME VARCHAR(200) NOT NULL
);
/* Direct grants from the owner to the workstation's functional role. */
GRANT USAGE ON DATABASE LANDING TO ROLE CLAUDE_WS_ROLE;
GRANT USAGE ON ALL SCHEMAS IN DATABASE LANDING TO ROLE CLAUDE_WS_ROLE;
GRANT SELECT ON ALL TABLES IN DATABASE LANDING TO ROLE CLAUDE_WS_ROLE;
GRANT INSERT, UPDATE, DELETE, TRUNCATE ON ALL TABLES IN SCHEMA LANDING.SUNBELT TO ROLE CLAUDE_WS_ROLE;
GRANT INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA LANDING.CONFIG TO ROLE CLAUDE_WS_ROLE;
/* Increment 2 (design D-22; guide step 22): Sunbelt Senior's policy sales data and Northfield Life's policy data.
Run as SYSADMIN. Ownership and reads follow the environment tiers (infra/snowflake/environment_tiers.sql). */
CREATE TABLE IF NOT EXISTS LANDING.SUNBELT.POLICY_SALE (
SALE_ROW_ID NUMBER NOT NULL, POLICY_NUMBER VARCHAR(32), CLIENT_NUMBER VARCHAR(32), CARRIER_CODE VARCHAR(8),
ROLE VARCHAR(16), SOLD_ON VARCHAR(10), AGENT_NAME VARCHAR(64), AGENT_NPN VARCHAR(10), PLAN_NAME VARCHAR(64),
LOADED_AT VARCHAR(19) NOT NULL);
CREATE SCHEMA IF NOT EXISTS LANDING.NORTHFIELD;
CREATE TABLE IF NOT EXISTS LANDING.NORTHFIELD.POLICY (
POLICY_NUMBER VARCHAR(32), INSURED_ID VARCHAR(16), INSURED_NAME VARCHAR(64), BIRTH_DATE VARCHAR(10),
ADDRESS VARCHAR(128), FACE_AMOUNT NUMBER(12,2), STATUS VARCHAR(2), EFFECTIVE_DATE VARCHAR(10),
WRITING_AGENT_NPN VARCHAR(10), LOADED_AT VARCHAR(19) NOT NULL);
GRANT OWNERSHIP ON SCHEMA LANDING.NORTHFIELD TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON TABLE LANDING.SUNBELT.POLICY_SALE TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON TABLE LANDING.NORTHFIELD.POLICY TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT USAGE ON SCHEMA LANDING.NORTHFIELD TO ROLE DV_ROLE;
GRANT USAGE ON SCHEMA LANDING.NORTHFIELD TO ROLE LANDING_READ_ROLE;
GRANT SELECT ON TABLE LANDING.SUNBELT.POLICY_SALE TO ROLE DV_ROLE;
GRANT SELECT ON TABLE LANDING.SUNBELT.POLICY_SALE TO ROLE LANDING_READ_ROLE;
GRANT SELECT ON TABLE LANDING.NORTHFIELD.POLICY TO ROLE DV_ROLE;
GRANT SELECT ON TABLE LANDING.NORTHFIELD.POLICY TO ROLE LANDING_READ_ROLE;
/* Lab 2 (design D-37; guide step 33): Pinecrest Mutual's policy data, landed by the teacher as the stand-in source
(Ray, 2026-10-25: the vault starts at landing; populating landing is outside it). Every column text, as sent.
Run as SYSADMIN; first in a developer clone, then LANDING; QA and developer landing receive it by cloning LANDING. */
CREATE SCHEMA IF NOT EXISTS LANDING.PINECREST;
CREATE TABLE IF NOT EXISTS LANDING.PINECREST.POLICY (
POLICY_NUMBER VARCHAR(32), PLAN_NAME VARCHAR(64), FACE_AMOUNT VARCHAR(16), STATUS VARCHAR(2),
EFFECTIVE_DATE VARCHAR(10), WRITING_AGENT_NPN VARCHAR(10), LOADED_AT VARCHAR(19) NOT NULL);
GRANT OWNERSHIP ON SCHEMA LANDING.PINECREST TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON TABLE LANDING.PINECREST.POLICY TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT USAGE ON SCHEMA LANDING.PINECREST TO ROLE DV_ROLE;
GRANT USAGE ON SCHEMA LANDING.PINECREST TO ROLE LANDING_READ_ROLE;
GRANT SELECT ON TABLE LANDING.PINECREST.POLICY TO ROLE DV_ROLE;
GRANT SELECT ON TABLE LANDING.PINECREST.POLICY TO ROLE LANDING_READ_ROLE;
/* The carriers' organization records (mart design D-11; profile P-21; guide step 52): each carrier sends its own
organization record, carrier code and name, through its feed. Every column text, as sent.
Run as SYSADMIN; first in a developer clone, then LANDING; QA and developer landing receive them by cloning LANDING. */
CREATE TABLE IF NOT EXISTS LANDING.NORTHFIELD.ORGANIZATION (
CARRIER_CODE VARCHAR(8), CARRIER_NAME VARCHAR(64), LOADED_AT VARCHAR(19) NOT NULL);
CREATE TABLE IF NOT EXISTS LANDING.PINECREST.ORGANIZATION (
CARRIER_CODE VARCHAR(8), CARRIER_NAME VARCHAR(64), LOADED_AT VARCHAR(19) NOT NULL);
GRANT OWNERSHIP ON TABLE LANDING.NORTHFIELD.ORGANIZATION TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON TABLE LANDING.PINECREST.ORGANIZATION TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT SELECT ON TABLE LANDING.NORTHFIELD.ORGANIZATION TO ROLE DV_ROLE;
GRANT SELECT ON TABLE LANDING.NORTHFIELD.ORGANIZATION TO ROLE LANDING_READ_ROLE;
GRANT SELECT ON TABLE LANDING.PINECREST.ORGANIZATION TO ROLE DV_ROLE;
GRANT SELECT ON TABLE LANDING.PINECREST.ORGANIZATION TO ROLE LANDING_READ_ROLE;
infra/snowflake/landing_increment2_dev_claude_ws.sql
/*
Increment 2's landing tables in the workstation's developer clone only (Ray, 2026-10-25, answer 1b):
the same tables as landing.sql's increment 2 block, created in DEV__LANDING__CLAUDE_WS instead of
refreshing both developer clones with environment_tiers.sql part B, so DEV__LANDING__RAYSERRANO is untouched.
A deviation from guide step 22, recorded in the post-implementation guide. Run as SYSADMIN. Grants mirror
what part B gives the clone's existing tables.
*/
CREATE TABLE IF NOT EXISTS DEV__LANDING__CLAUDE_WS.SUNBELT.POLICY_SALE (
SALE_ROW_ID NUMBER NOT NULL, POLICY_NUMBER VARCHAR(32), CLIENT_NUMBER VARCHAR(32), CARRIER_CODE VARCHAR(8),
ROLE VARCHAR(16), SOLD_ON VARCHAR(10), AGENT_NAME VARCHAR(64), AGENT_NPN VARCHAR(10), PLAN_NAME VARCHAR(64),
LOADED_AT VARCHAR(19) NOT NULL);
CREATE SCHEMA IF NOT EXISTS DEV__LANDING__CLAUDE_WS.NORTHFIELD;
CREATE TABLE IF NOT EXISTS DEV__LANDING__CLAUDE_WS.NORTHFIELD.POLICY (
POLICY_NUMBER VARCHAR(32), INSURED_ID VARCHAR(16), INSURED_NAME VARCHAR(64), BIRTH_DATE VARCHAR(10),
ADDRESS VARCHAR(128), FACE_AMOUNT NUMBER(12,2), STATUS VARCHAR(2), EFFECTIVE_DATE VARCHAR(10),
WRITING_AGENT_NPN VARCHAR(10), LOADED_AT VARCHAR(19) NOT NULL);
GRANT OWNERSHIP ON SCHEMA DEV__LANDING__CLAUDE_WS.NORTHFIELD TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON TABLE DEV__LANDING__CLAUDE_WS.SUNBELT.POLICY_SALE TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON TABLE DEV__LANDING__CLAUDE_WS.NORTHFIELD.POLICY TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT USAGE ON SCHEMA DEV__LANDING__CLAUDE_WS.NORTHFIELD TO ROLE CLAUDE_WS_ROLE;
GRANT USAGE ON SCHEMA DEV__LANDING__CLAUDE_WS.NORTHFIELD TO ROLE DEV_CLAUDE_WS_ROLE;
GRANT USAGE ON SCHEMA DEV__LANDING__CLAUDE_WS.NORTHFIELD TO ROLE DV_ROLE;
GRANT USAGE ON SCHEMA DEV__LANDING__CLAUDE_WS.NORTHFIELD TO ROLE LANDING_READ_ROLE;
GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON TABLE DEV__LANDING__CLAUDE_WS.SUNBELT.POLICY_SALE TO ROLE CLAUDE_WS_ROLE;
GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON TABLE DEV__LANDING__CLAUDE_WS.NORTHFIELD.POLICY TO ROLE CLAUDE_WS_ROLE;
GRANT SELECT ON TABLE DEV__LANDING__CLAUDE_WS.SUNBELT.POLICY_SALE TO ROLE DEV_CLAUDE_WS_ROLE;
GRANT SELECT ON TABLE DEV__LANDING__CLAUDE_WS.NORTHFIELD.POLICY TO ROLE DEV_CLAUDE_WS_ROLE;
GRANT SELECT ON TABLE DEV__LANDING__CLAUDE_WS.SUNBELT.POLICY_SALE TO ROLE DV_ROLE;
GRANT SELECT ON TABLE DEV__LANDING__CLAUDE_WS.NORTHFIELD.POLICY TO ROLE DV_ROLE;
GRANT SELECT ON TABLE DEV__LANDING__CLAUDE_WS.SUNBELT.POLICY_SALE TO ROLE LANDING_READ_ROLE;
GRANT SELECT ON TABLE DEV__LANDING__CLAUDE_WS.NORTHFIELD.POLICY TO ROLE LANDING_READ_ROLE;
infra/snowflake/pilot_reader.sql
/*
Read-only access to Pilot (AGENTIC.DATASKILLS) for the learner (Ray, 2026-10-23, answer 1y).
The course's lessons, records, delivery documents, rubric and acceptance log live in Pilot; the learner
reads them, and only the procedures and the teacher write them. First grant added back to RAYSERRANO
after the clean slate (admin_separation.sql, step 6).
Ownership read 2026-10-23: SYSADMIN owns AGENTIC, DATASKILLS, its 17 tables and 3 views.
PART A, run by Ray as RAYSERRANO_ADMIN: the role, its grant to Ray's role, and future grants (which need MANAGE GRANTS).
PART B, run by the workstation as SYSADMIN (the owner), after part A: grants on existing objects.
*/
/* ===================== PART A (RAYSERRANO_ADMIN) ===================== */
USE ROLE SECURITYADMIN;
CREATE ROLE IF NOT EXISTS PILOT_READ_ROLE COMMENT = 'Reads Pilot (AGENTIC.DATASKILLS); never writes';
GRANT ROLE PILOT_READ_ROLE TO ROLE SYSADMIN;
/* Granted to Ray's own role, his developer role (Ray, 2026-10-23: one role per person; no separate
functional role). Read-only, so it adds nothing a dbt build could misuse. */
GRANT ROLE PILOT_READ_ROLE TO ROLE DEV_RAYSERRANO_ROLE;
GRANT SELECT ON FUTURE TABLES IN SCHEMA AGENTIC.DATASKILLS TO ROLE PILOT_READ_ROLE;
GRANT SELECT ON FUTURE VIEWS IN SCHEMA AGENTIC.DATASKILLS TO ROLE PILOT_READ_ROLE;
/* ===================== PART B (workstation, SYSADMIN) ===================== */
USE ROLE SYSADMIN;
GRANT USAGE ON WAREHOUSE COMPUTE_WH TO ROLE PILOT_READ_ROLE;
GRANT USAGE ON DATABASE AGENTIC TO ROLE PILOT_READ_ROLE;
GRANT USAGE ON SCHEMA AGENTIC.DATASKILLS TO ROLE PILOT_READ_ROLE;
GRANT SELECT ON ALL TABLES IN SCHEMA AGENTIC.DATASKILLS TO ROLE PILOT_READ_ROLE;
GRANT SELECT ON ALL VIEWS IN SCHEMA AGENTIC.DATASKILLS TO ROLE PILOT_READ_ROLE;
infra/snowflake/refresh_landing_tier.sql
/*
Refresh one non-production landing tier from production landing (Ray, 2026-10-26, answer 1y; generalized to every
tier 2026-10-28). The tiers are environment_tiers.sql's: LANDING_QA built by DV_QA_ROLE, and each developer's
DEV__LANDING__<USER> built by DEV_<USER>_ROLE. The tier is replaced by a fresh clone of LANDING, owned by
LANDING_ROLE, production's builder (DV_ROLE) removed, the tier's own builder and the reader granted, and its CONFIG
emptied: production's builds write their findings and baselines into LANDING.CONFIG and the clone copies them
(found 2026-10-28: 3 findings and 34 baselines), so every tier starts with no findings and no baselines of its own.
Run by the workstation as SYSADMIN, statements on standard input, naming the tier:
QA: snow sql ... -D landing=LANDING_QA -D builder=DV_QA_ROLE < infra/snowflake/refresh_landing_tier.sql
Development: snow sql ... -D landing=DEV__LANDING__CLAUDE_WS -D builder=DEV_CLAUDE_WS_ROLE < (the same)
Safe to rerun: the clone is replaced each time. Refuses production: the target must not be LANDING.
*/
USE ROLE SYSADMIN;
USE WAREHOUSE COMPUTE_WH;
/* 0. Refuse production and unknown tiers before anything is replaced. */
EXECUTE IMMEDIATE $$
BEGIN
IF ('<% landing %>' NOT IN ('LANDING_QA') AND NOT STARTSWITH('<% landing %>', 'DEV__LANDING__')) THEN
RETURN 1/0; -- stops the script: only LANDING_QA or a developer's DEV__LANDING__<USER> may be refreshed
END IF;
RETURN 'refreshing <% landing %> for <% builder %>';
END;
$$;
/* 1. A fresh clone of production landing. */
CREATE OR REPLACE DATABASE <% landing %> CLONE LANDING;
/* 2. Owned by LANDING_ROLE, grants preserved. */
GRANT OWNERSHIP ON DATABASE <% landing %> TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE <% landing %> TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL TABLES IN DATABASE <% landing %> TO ROLE LANDING_ROLE COPY CURRENT GRANTS;
/* 3. Builder and reader grants, as the owner. Database-level grants do not survive a clone, but the clone's tables
keep production's object grants, so production's builder, DV_ROLE, is removed first: each tier is built by its
own builder alone (Ray, 2026-10-26, answer 1b). */
USE ROLE LANDING_ROLE;
REVOKE ALL PRIVILEGES ON ALL TABLES IN DATABASE <% landing %> FROM ROLE DV_ROLE;
REVOKE ALL PRIVILEGES ON ALL SCHEMAS IN DATABASE <% landing %> FROM ROLE DV_ROLE;
REVOKE ALL PRIVILEGES ON DATABASE <% landing %> FROM ROLE DV_ROLE;
GRANT USAGE ON DATABASE <% landing %> TO ROLE <% builder %>;
GRANT USAGE ON ALL SCHEMAS IN DATABASE <% landing %> TO ROLE <% builder %>;
GRANT SELECT ON ALL TABLES IN DATABASE <% landing %> TO ROLE <% builder %>;
GRANT CREATE TABLE, CREATE VIEW, MODIFY ON SCHEMA <% landing %>.STORAGE TO ROLE <% builder %>;
GRANT CREATE TABLE, CREATE VIEW, MODIFY ON SCHEMA <% landing %>.CONFIG TO ROLE <% builder %>;
GRANT INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA <% landing %>.CONFIG TO ROLE <% builder %>;
GRANT USAGE ON DATABASE <% landing %> TO ROLE LANDING_READ_ROLE;
GRANT USAGE ON ALL SCHEMAS IN DATABASE <% landing %> TO ROLE LANDING_READ_ROLE;
GRANT SELECT ON ALL TABLES IN DATABASE <% landing %> TO ROLE LANDING_READ_ROLE;
/* 4. The tier's own findings and baselines start empty: the clone copied production's. As the owner. */
TRUNCATE TABLE <% landing %>.CONFIG.DATA_FINDING;
TRUNCATE TABLE <% landing %>.CONFIG.BUILD_BASELINE;
USE ROLE SYSADMIN;
/* 5. Check: every LANDING source table has its twin in the tier with the same row count; CONFIG is checked in step
6 instead, since the tier's is emptied on purpose. Returns zero rows when equal. */
SELECT p.TABLE_SCHEMA, p.TABLE_NAME, p.ROW_COUNT AS LANDING_ROWS, q.ROW_COUNT AS TIER_ROWS
FROM LANDING.INFORMATION_SCHEMA.TABLES p
FULL JOIN <% landing %>.INFORMATION_SCHEMA.TABLES q
ON q.TABLE_SCHEMA = p.TABLE_SCHEMA AND q.TABLE_NAME = p.TABLE_NAME
WHERE COALESCE(p.TABLE_SCHEMA, q.TABLE_SCHEMA) NOT IN ('INFORMATION_SCHEMA', 'CONFIG')
AND (p.TABLE_NAME IS NULL OR q.TABLE_NAME IS NULL OR p.ROW_COUNT <> q.ROW_COUNT);
/* 6. Check: the tier's CONFIG is empty. Returns 0 and 0. */
SELECT (SELECT COUNT(*) FROM <% landing %>.CONFIG.DATA_FINDING) AS TIER_FINDINGS,
(SELECT COUNT(*) FROM <% landing %>.CONFIG.BUILD_BASELINE) AS TIER_BASELINES;
infra/snowflake/reset_vault_tier.sql
/*
Clear a vault tier before a build that must be a true first build (Ray, 2026-10-26, answer 1y: D-41's QA rerun;
generalized 2026-11-01 for the dev rehearsal, which runs everything QA will run before any pull request into qa).
Drops the tier's built schemas; the next build recreates them, ghost records included. The MART views of release 1
are dropped by infra/snowflake/drop_mart_views.sql after the build. Run with infra/snowflake/refresh_landing_tier.sql
for the same tier, so landing and vault start clean together.
Run by the workstation as SYSADMIN, statements on standard input, naming the vault:
QA: snow sql ... -D vault=DV_QA
Development: snow sql ... -D vault=DEV__DV__CLAUDE_WS
Refuses any database but DV_QA or a developer's DEV__DV__<USER>, before anything changes. Idempotent.
*/
USE ROLE SYSADMIN;
EXECUTE IMMEDIATE $$
BEGIN
IF ('<% vault %>' <> 'DV_QA' AND NOT STARTSWITH('<% vault %>', 'DEV__DV__')) THEN
RETURN 1/0; -- stops the script: only DV_QA or a developer's DEV__DV__<USER>, never production
END IF;
RETURN 'clearing the built schemas of <% vault %>';
END;
$$;
DROP SCHEMA IF EXISTS <% vault %>.STAGING;
DROP SCHEMA IF EXISTS <% vault %>.RAW_VAULT;
DROP SCHEMA IF EXISTS <% vault %>.REFERENCE;
DROP SCHEMA IF EXISTS <% vault %>.CONFORMED_DM;
SELECT SCHEMA_NAME FROM <% vault %>.INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME NOT IN ('INFORMATION_SCHEMA', 'PUBLIC');
infra/snowflake/rollback_vault_to_time.sql
/*
Roll a vault back to a recorded moment by Snowflake Time Travel (Ray, 2026-11-01: "we have time travel"; answer 1y).
Before a promotion's production build, the promotion records the clock (SELECT CURRENT_TIMESTAMP()) with its
evidence; if the release must be undone, this script clones the vault as it stood at that clock, returns the clone
and everything in it to the vault's owner with their grants, and swaps it in. The undone state stays behind as
<vault>_ROLLBACK for inspection, and is dropped afterwards (DROP DATABASE <vault>_ROLLBACK, as SYSADMIN, once the rollback is verified). Time Travel reaches back as far as
the vault's retention, one day on 2026-11-01. After the swap, in order: infra/snowflake/connector_read_grants.sql for the
vault, since a clone does not carry grants on the database itself; dbt run-operation declare_vault_keys from the
restored release's commit, since a clone at a past moment does not carry declared keys (found in the dev rehearsal,
2026-11-01: 27 keys re-declared); the structure check against that commit's dv.sql; the code side is the revert of the release's
commit on main and the functions redeployed from it.
Run by the workstation as SYSADMIN, statements on standard input, naming the vault, the clock and the owner:
snow sql ... -D vault=DV -D at='2026-11-01 16:30:00 +0000' -D owner=DV_ROLE
Refuses any database but DV, DV_QA or a developer's DEV__DV__<USER>, before anything changes. Rehearsed in a
developer vault and in DV_QA before any production use, as every act is.
*/
USE ROLE SYSADMIN;
EXECUTE IMMEDIATE $$
BEGIN
IF ('<% vault %>' NOT IN ('DV_QA', 'DV') AND NOT STARTSWITH('<% vault %>', 'DEV__DV__')) THEN
RETURN 1/0; -- stops the script: only DV, DV_QA or a developer's DEV__DV__<USER>
END IF;
RETURN 'rolling <% vault %> back to <% at %>';
END;
$$;
CREATE OR REPLACE DATABASE <% vault %>_ROLLBACK CLONE <% vault %> AT (TIMESTAMP => '<% at %>'::TIMESTAMP_TZ);
GRANT OWNERSHIP ON DATABASE <% vault %>_ROLLBACK TO ROLE <% owner %> COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL SCHEMAS IN DATABASE <% vault %>_ROLLBACK TO ROLE <% owner %> COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL TABLES IN DATABASE <% vault %>_ROLLBACK TO ROLE <% owner %> COPY CURRENT GRANTS;
GRANT OWNERSHIP ON ALL VIEWS IN DATABASE <% vault %>_ROLLBACK TO ROLE <% owner %> COPY CURRENT GRANTS;
ALTER DATABASE <% vault %> SWAP WITH <% vault %>_ROLLBACK;
/* The vault as restored: its schemas, tables and views. */
SELECT TABLE_SCHEMA, TABLE_TYPE, COUNT(*) AS N FROM <% vault %>.INFORMATION_SCHEMA.TABLES
WHERE TABLE_SCHEMA <> 'INFORMATION_SCHEMA' GROUP BY 1, 2 ORDER BY 1, 2;
infra/snowflake/service_users.sql
/*
Service users: the identities dbt Cloud deployment jobs run as (Ray, 2026-10-23, answer 2y; split and renamed
2026-10-26, answers 1b and 2y). A service user is named for the data product it serves, DV, never for the tool
that runs it. QA and production are split, so a QA job pointed at production by mistake fails on privileges
instead of writing there:
DV_QA_DEPLOY role DV_QA_ROLE owns DV_QA; reads and writes LANDING_QA only (environment_tiers.sql, B3 and B5)
DV_DEPLOY role DV_ROLE owns DV; reads and writes LANDING
Both sign in only by key pair. Run by Ray as RAYSERRANO_ADMIN, before the workstation's part
(infra/snowflake/dv_qa_role.sql). Then register each public key with
infra/keys/new_developer_keypair.ps1 -User DV_QA_DEPLOY -Slot 1
infra/keys/new_developer_keypair.ps1 -User DV_DEPLOY -Slot 1
and put each private key only in its dbt Cloud environment's credentials.
DBT_DEPLOY, the earlier name, is dropped if it was ever created; it held no key in use.
*/
USE ROLE USERADMIN;
CREATE ROLE IF NOT EXISTS DV_QA_ROLE
COMMENT = 'Builder of DV_QA, the QA tier of the DV data product. Reads and writes LANDING_QA only.';
CREATE USER IF NOT EXISTS DV_QA_DEPLOY
TYPE = SERVICE
DEFAULT_ROLE = DV_QA_ROLE
DEFAULT_WAREHOUSE = COMPUTE_WH
COMMENT = 'dbt Cloud QA environment of the DV data product. Key pair only.';
CREATE USER IF NOT EXISTS DV_DEPLOY
TYPE = SERVICE
DEFAULT_ROLE = DV_ROLE
DEFAULT_WAREHOUSE = COMPUTE_WH
COMMENT = 'dbt Cloud production environment of the DV data product. Key pair only.';
DROP USER IF EXISTS DBT_DEPLOY;
USE ROLE SECURITYADMIN;
GRANT ROLE DV_QA_ROLE TO ROLE SYSADMIN;
GRANT ROLE DV_QA_ROLE TO USER DV_QA_DEPLOY;
GRANT ROLE DV_ROLE TO USER DV_DEPLOY;
SHOW GRANTS TO USER DV_QA_DEPLOY;
SHOW GRANTS TO USER DV_DEPLOY;