Overview

What You'll Need

What You'll Learn

What You'll Build

Setup

Create Snowflake Objects

USE ROLE SYSADMIN;

CREATE DATABASE IF NOT EXISTS PIPELINE_AUDITOR_DB;
CREATE SCHEMA IF NOT EXISTS PIPELINE_AUDITOR_DB.AUDITOR;

Scaffold the App

git clone https://github.com/sfc-gh-JCHAEARLEY/sfguide-pipeline-auditor-with-cortex-agent-sdk.git
cd sfguide-pipeline-auditor-with-cortex-agent-sdk

npm install

Project Structure

pipeline-auditor/
  app/
    api/
      audit/          # POST to start audit, GET to poll progress
      suggest-fix/    # POST to request fix, chat/ for follow-up, dismiss/
      databases/      # GET available databases
      schemas/        # GET schemas in a database
      chat/           # POST for follow-up chat on the audit thread
    page.tsx          # Root page -- providers + AuditDashboard
  components/
    AuditDashboard.tsx   # Main layout orchestrator
    AuditHeader.tsx      # Database/schema dropdowns + scope picker
    AuditSidebar.tsx     # Real-time tool call progress
    ChatThread.tsx       # Message thread with inline report accordion
    EmptyState.tsx       # Landing page
  hooks/
    useAudit.ts          # Poll-based audit state machine
    useSuggestFix.ts     # Fix request + multi-turn follow-up
  lib/
    agent-client.ts      # CortexAgentClient factory
    audit-engine.ts      # Composable system prompt + JSON schema
    audit-store.ts       # In-memory job store for poll-based streaming
    types.ts             # Shared TypeScript types
  app.yml                # SAR build/run manifest

Environment Variables (Local Dev Only)

SNOWFLAKE_ACCOUNT=your_account_identifier
SNOWFLAKE_PAT=your_personal_access_token

Build the Audit Engine

Agent Client Factory

import { CortexAgentClient, OAuthAuth } from "@snowflake/cortex-agent-sdk";
import fs from "fs";

const TOKEN_PATH = "/snowflake/session/token";

export function createAgentClient(): CortexAgentClient {
  const account = process.env.SNOWFLAKE_ACCOUNT || "";

  // In SAR (SPCS), the runtime mounts an OAuth token at a known path
  if (fs.existsSync(TOKEN_PATH)) {
    const token = fs.readFileSync(TOKEN_PATH, "utf-8").trim();
    return new CortexAgentClient({
      account,
      auth: new OAuthAuth(token),
    });
  }

  // Fall back to PAT for local development
  const pat = process.env.SNOWFLAKE_PAT || "";
  return new CortexAgentClient({ account, auth: pat });
}

Composable System Prompt

export const PROMPT_SECTIONS: Record<string, string> = {
  tables_freshness: `
TABLES & FRESHNESS:
   - Tables and row counts:
     SELECT table_schema, table_name, table_type, row_count
     FROM <DATABASE>.INFORMATION_SCHEMA.TABLES
     WHERE table_schema NOT IN ('INFORMATION_SCHEMA')
     ORDER BY table_schema, table_name;
   - Check freshness -- identify stale tables:
     SELECT table_schema, table_name, table_type, row_count, last_altered,
            DATEDIFF('hour', last_altered, CURRENT_TIMESTAMP()) AS hours_since_update,
            CASE
              WHEN DATEDIFF('hour', last_altered, CURRENT_TIMESTAMP()) > 168 THEN 'critical'
              WHEN DATEDIFF('hour', last_altered, CURRENT_TIMESTAMP()) > 24 THEN 'warning'
              ELSE 'fresh'
            END AS freshness_status
     FROM <DATABASE>.INFORMATION_SCHEMA.TABLES
     WHERE table_schema NOT IN ('INFORMATION_SCHEMA') AND table_type = 'BASE TABLE'
     ORDER BY hours_since_update DESC;
   - Check for row count anomalies (empty tables that shouldn't be)`,

  dynamic_tables: `
DYNAMIC TABLES:
   - Discover: SHOW DYNAMIC TABLES IN DATABASE <DATABASE>;
   - Check scheduling_state (should be ACTIVE)
   - Check refresh history for failures ...`,

  tasks: `
TASKS:
   - Discover: SHOW TASKS IN DATABASE <DATABASE>;
   - Check execution history for failures via TASK_HISTORY table function
   - Flag suspended tasks and tasks with recent failures`,

  streams: `
STREAMS:
   - Discover: SHOW STREAMS IN DATABASE <DATABASE>;
   - Check each stream's stale status (stale = true is CRITICAL)
   - Check stale_after timestamp -- if approaching, flag as warning`,

  pipes: `
PIPES:
   - Discover: SHOW PIPES IN DATABASE <DATABASE>;
   - Check pipe status (RUNNING, STOPPED_CLONING, PAUSED, STALLED)
   - Check recent copy history for errors ...`,

  procedures: `
STORED PROCEDURES:
   - Discover: SHOW PROCEDURES IN DATABASE <DATABASE>;
   - Identify ETL-related procedures
   - Flag procedures that do data movement but have no associated task`,
};

export function buildSystemPrompt(
  database: string,
  scope: string[],
  schema: string,
): string {
  const allScopes = Object.keys(PROMPT_SECTIONS);
  const activeScopes = scope.length > 0 ? scope : allScopes;

  // Use the single-schema preamble when a schema is specified
  let prompt = schema
    ? PROMPT_PREAMBLE_SINGLE_SCHEMA
    : PROMPT_PREAMBLE_ALL_SCHEMAS;

  for (const key of allScopes) {
    if (activeScopes.includes(key)) {
      prompt += PROMPT_SECTIONS[key];
    }
  }
  prompt += PROMPT_FOOTER;

  // Replace <DATABASE> and <SCHEMA> placeholders throughout
  prompt = prompt.replace(/<DATABASE>/g, database);
  if (schema) {
    prompt = prompt.replace(/<SCHEMA>/g, schema);
    // Rewrite SHOW ... IN DATABASE to SHOW ... IN SCHEMA
    // Rewrite INFORMATION_SCHEMA filters to target the single schema
  }

  return prompt;
}

In-Memory Job Store

export interface AuditJob {
  events: Record<string, unknown>[];
  done: boolean;
  error?: string;
}

export const jobStore = new Map<string, AuditJob>();

export function pushJobEvent(
  jobId: string,
  event: Record<string, unknown>,
): void {
  const job = jobStore.get(jobId);
  if (job) job.events.push(event);
}

The Audit API Route

async function runAuditJob(jobId, database, scope, schema) {
  const client = createAgentClient();
  const systemPrompt = buildSystemPrompt(database, scope, schema);
  const userPrompt = buildAuditUserPrompt(database, scope, schema);

  const messages = [
    {
      role: "user",
      content: [{ type: "text", text: systemPrompt + "\n\n" + userPrompt }],
    },
  ];

  const stream = client.codingAgent.stream({
    permissionPolicy: "always_allow",
    messages,
  });

  let accumulatedText = "";

  // Use the SDK's callback API to process events
  stream.on("text", (delta, accumulated) => {
    accumulatedText = accumulated;
    // Try to parse as JSON report on each chunk
    try {
      const parsed = JSON.parse(accumulated);
      if (parsed.findings) {
        pushJobEvent(jobId, { type: "report", report: parsed });
      }
    } catch { /* not complete JSON yet */ }
  });

  stream.on("toolUse", (ev) => {
    // Push progress events so the sidebar shows tool execution
    pushJobEvent(jobId, { type: "tool_progress", toolName: ev.name });
  });

  stream.on("thinking", (delta) => {
    pushJobEvent(jobId, { type: "thinking", text: delta });
  });

  await stream.done();
  // Job is complete — mark done in the job store
}

Key concepts:

• Coding agent sessions give the AI access to sql_execute so it can query your Snowflake account directly
• Composable prompts let you scope audits to specific object types for faster, targeted checks
• Poll-based streaming avoids long-lived HTTP connections while still giving the user real-time progress
• Structured JSON output means the report is machine-parseable, not just freeform text

Build the Dashboard UI

Page Root and Providers

"use client";

import { AppThemeProvider } from "@/components/ThemeContext";
import { AuditDashboard } from "@/components/AuditDashboard";

export default function Page() {
  return (
    <AppThemeProvider>
      <AuditDashboard />
    </AppThemeProvider>
  );
}

AuditDashboard -- The Orchestrator

export function AuditDashboard() {
  const {
    isAuditing, isLoading, report, messages,
    toolProgress, error, stats,
    startAudit, cancelAudit, resetView,
  } = useAudit();

  const { fixState, requestFix, sendFollowUp, dismissFix } = useSuggestFix();

  return (
    <Box sx={{ height: "100vh", display: "flex", flexDirection: "column" }}>
      <AuditHeader
        onStartAudit={(db, scope, schema) => startAudit(db, scope, schema)}
        onCancel={cancelAudit}
        onReset={resetView}
        isLoading={isLoading}
        isAuditing={isAuditing}
      />

      <Box sx={{ flex: 1, overflow: "hidden", display: "flex" }}>
        {!hasAuditData ? (
          <EmptyState />
        ) : (
          <>
            {/* Chat thread with inline report accordion */}
            <Box sx={{ flex: 1, overflow: "auto" }}>
              <ChatThread
                messages={messages}
                report={report}
                toolProgress={toolProgress}
                isLoading={isLoading}
                onSuggestFix={handleSuggestFix}
                fixState={fixState}
              />
            </Box>

            {/* Real-time stats sidebar */}
            <AuditSidebar
              toolProgress={toolProgress}
              report={report}
              stats={stats}
              isLoading={isLoading}
            />
          </>
        )}
      </Box>
    </Box>
  );
}

The useAudit Hook -- Poll-Based State Machine

const POLL_INTERVAL_MS = 2000;

export async function pollJob(
  jobId: string,
  onEvent: (event: StreamEvent) => void,
  signal: AbortSignal,
): Promise<void> {
  let cursor = 0;
  while (!signal.aborted) {
    const resp = await fetch(
      `/api/audit/progress?jobId=${encodeURIComponent(jobId)}&after=${cursor}`,
      { credentials: "include", signal },
    );
    const data = await resp.json();
    for (const event of data.events) {
      onEvent(event);
    }
    cursor = data.cursor;
    if (data.done) return;
    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
  }
}

Key concepts:

• Cursor-based polling ensures no events are missed even if a poll is delayed
• AbortController lets the user cancel a running audit cleanly
• useEffect + fetch handles the database/schema dropdowns with loading states

Add Suggest Fix

The Suggest Fix API

async function runSuggestFixJob(jobId, finding, database) {
  const systemPrompt = `You are a Snowflake pipeline remediation expert. 
When given a finding, respond with a clear, actionable fix. 
Use SQL code blocks for any commands. 
If the user asks you to run SQL, you may use sql_execute.
Target database: ${database}.`;

  const userPrompt = `Suggest a concrete fix for this pipeline audit finding:

Finding:
- Severity: ${finding.severity}
- Category: ${finding.category}
- Object: ${finding.object}
- Message: ${finding.message}
${finding.details ? `- Details: ${JSON.stringify(finding.details)}` : ""}

Rules:
- Do NOT run any tools or queries. Respond with your answer directly.
- Provide exact SQL commands in fenced code blocks.
- Explain briefly why the fix works.
- Keep the response under 500 words.`;

  const messages = [{
    role: "user",
    content: [{ type: "text", text: systemPrompt + "\n\n" + userPrompt }],
  }];

  const client = createAgentClient();
  const stream = client.codingAgent.stream({
    permissionPolicy: "always_allow",
    messages,
  });

  // Use callbacks to stream text to the job store
  stream.on("text", (delta) => {
    pushJobEvent(jobId, { type: "text", text: delta });
  });

  await stream.done();

  // Save the conversation thread for follow-up chat
  setFixSession([...messages, {
    role: "assistant",
    content: [{ type: "text", text: stream.currentText }],
  }]);
}

Multi-Turn Follow-Up Chat

async function runFixChatJob(jobId: string, message: string) {
  const fixThread = getFixSession();
  const client = createAgentClient();

  // Append user message to the thread
  fixThread.push({
    role: "user",
    content: [{ type: "text", text: message }],
  });

  // Stream with full conversation history for context
  const stream = client.codingAgent.stream({
    permissionPolicy: "always_allow",
    messages: fixThread,
  });

  // ... accumulate text, push events, update thread
}

The useSuggestFix Hook

export function useSuggestFix() {
  const [state, setState] = useState<SuggestFixState>(IDLE);

  const requestFix = useCallback(async (index, finding, database) => {
    // POST to /api/suggest-fix, poll for text events
    // Update state with streaming assistant messages
  }, []);

  const sendFollowUp = useCallback(async (message) => {
    // POST to /api/suggest-fix/chat, poll for response
    // Append user message + streaming assistant response
  }, []);

  const dismissFix = useCallback(() => {
    // Abort, reset state, POST to /api/suggest-fix/dismiss
  }, []);

  return { fixState: state, requestFix, sendFollowUp, dismissFix };
}

Copy and Run

Key concepts:

• Separate agent sessions for audit vs. fix keep concerns isolated and prompts focused
• Multi-turn chat with full message history gives the agent context for iterative refinement
• Per-job thread state (stored in the job store) keeps conversations isolated between requests
• The coding agent uses permissionPolicy: "always_allow" because it needs to run SQL queries autonomously. Deploy the app with a read-only role to limit what the agent can do.

Deploy and Run

Generate the Deployment Manifest

install:
  commands:
    - ["npm", "ci", "--include=dev"]

build:
  commands:
    - ["npm", "run", "build"]
    - ["cp", "-r", ".next/static", ".next/standalone/.next/static"]
    - ["cp", "-r", "public", ".next/standalone/public"]
    - ["rm", "-rf", "node_modules"]

run:
  command: ["node", ".next/standalone/server.js"]

Set Up and Deploy

snow app setup

snow app deploy

Open the App

snow app status

Run Your First Audit

Local Development

npm run dev

Conclusion And Resources

What You Learned

Extending It

Related Resources

Cleanup

DROP DATABASE IF EXISTS PIPELINE_AUDITOR_DB;
-- Also drop the Application Service if deployed:
DROP APPLICATION SERVICE IF EXISTS pipeline_auditor;