Snowflake Pipeline Guardian

Snowflake Pipeline Guardian

Overview

Pipeline symptoms cross object boundaries, but dependency order does not prove causality. This skill classifies a bounded evidence bundle, walks supplied dependency paths, and keeps observed facts, derived findings, hypotheses, and unknowns separate.

Use scripts/analyze_pipeline_state.py for deterministic classification. Read references/current-state.md before collecting live evidence and references/recovery-matrix.md before writing a recovery plan.

Hard boundaries

• Read-only diagnosis only. Never execute or emit runnable DDL, DML, task execution, refresh, resume, stream replacement, pipe refresh, or replay SQL.
• Never call an empty, privilege-hidden, stale, capped, context-mismatched, or untrusted result healthy or complete.
• Never treat current control-plane state as historical run proof, or lagged Account Usage history as current state.
• Never infer causality from graph order or a current/history disagreement.
• Never request or retain raw SQL text, object names, query IDs, file or stage paths, notification endpoints, integration names, free-text errors, or raw SYSTEM$PIPE_STATUS JSON. Use the reviewed hashed projections only.
• Never use SYSTEM$STREAM_HAS_DATA for diagnosis; calling it can affect a stream's staleness behavior.
• Treat hashes as pseudonymous operational data, not anonymization.

Trusted evidence contract

The live collector has six reviewed schema-2 surface types:

1. pipeline: explicit half-open UTC history for task runs, dynamic-table refreshes, and copy loads.
2. pipeline-task-current: current role-visible task inventory.
3. pipeline-stream-current: current role-visible stream inventory.
4. pipeline-dynamic-table-current: current role-visible dynamic-table inventory.
5. pipeline-pipe-current: current role-visible pipe inventory.
6. pipeline-pipe-status: one selector-bound, privacy-projected status receipt for each pipe in the current pipe inventory; zero status receipts are correct only when that inventory is empty.

Each receipt must be live CLI output with exactly one same-statement execution_context, the reviewed SQL/template/result hashes, exact datasets, finite documented state/status domains, the exact fixed non_claims, the declared cap, no collector error, and a valid self-hash. Invalid receipts never reach finding classification. All receipts must agree on organization/account, collector user, primary role and role type, secondary roles, and UTC timezone. Their observations may span at most 15 minutes and may be at most 15 minutes old.

Receipt self-hashes are integrity checks, not trust anchors. Put the complete receipts in one collector_receipts array, calculate the canonical bundle digest at a separate trusted local boundary, record it separately, and then supply it to the analyzer:

python3 "${CLAUDE_SKILL_DIR}/scripts/analyze_pipeline_state.py" \
  --input ./pipeline-evidence-bundle.json --print-input-sha256

python3 "${CLAUDE_SKILL_DIR}/scripts/analyze_pipeline_state.py" \
  --input ./pipeline-evidence-bundle.json \
  --trusted-input-sha256 "sha256:<separately-recorded-digest>" \
  --evaluated-at "<explicit-UTC-evaluation-time>"

A matching digest proves only byte identity with the operator-recorded bundle. It is not a signature, collector identity, or proof of Snowflake origin. Offline-normalized pipeline receipts are diagnostic-only and cannot support positive completeness claims.

Prerequisites

Choose an explicit UTC history window no longer than seven days. Both bounds are required and the end is exclusive:

python3 "${CLAUDE_SKILL_DIR}/scripts/collect_snowflake_evidence.py" \
  --surface pipeline --connection <approved-readonly-profile> \
  --window-start <window-start-UTC> \
  --window-end <exclusive-window-end-UTC> \
  --output ./pipeline-history.json

Collect each current inventory surface once under the same authorization context:

python3 "${CLAUDE_SKILL_DIR}/scripts/collect_snowflake_evidence.py" \
  --surface pipeline-task-current --connection <approved-readonly-profile> \
  --output ./pipeline-task-current.json

Repeat with pipeline-stream-current, pipeline-dynamic-table-current, and pipeline-pipe-current. For every hash in current_pipes, collect exactly one status receipt using its corresponding validated three-part unquoted identifier inside the trusted operator environment:

python3 "${CLAUDE_SKILL_DIR}/scripts/collect_snowflake_evidence.py" \
  --surface pipeline-pipe-status --connection <approved-readonly-profile> \
  --pipe DATABASE.SCHEMA.PIPE --output ./pipeline-pipe-status.json

The raw pipe selector is not receipt evidence. A successful receipt binds its fingerprint to the account-scoped object hash and hashes a receipt-only SQL rendering where that scoped hash replaces the selector. The analyzer recomputes both bindings. An error receipt retains only the reviewed template digest and a null selector fingerprint, so it cannot expose a dictionary-testable pipe name. Do not place the raw identifier in the bundle. History-window timestamps are safe selector values and are retained so the analyzer can recompute the rendered history-query digest.

History is capped independently at 5,000 task, refresh, and copy rows. Current inventories are capped at 10,000 objects each; reaching a cap is incomplete. Task completions are considered settled only through observation minus 45 minutes, completed dynamic-table refreshes through observation minus 3 hours, and completed copy loads through observation minus 48 hours. Executing refreshes are excluded. The receipt records each exact settled_through_utc. Evidence after a settlement cutoff is unknown, not absent.

See references/observability-queries.md for the exact field and nonclaim map, and references/privilege-and-boundaries.md before requesting access.