WIP experimentation

.claude/skills/data-workspace/SKILL.md

@@ -0,0 +1,314 @@
+---
+name: data-workspace
+description: Unified workflow for data analysis and pipeline building in workspaces
+allowed-tools:
+  - Read
+  - Write
+  - Edit
+  - Bash
+  - Glob
+  - Grep
+  - AskUserQuestion
+---
+
+# Data Workspace
+
+Autonomous workflow for data analysis and pipeline building. When the user asks a question or requests a transform, execute the full loop without prompting:
+
+1. **Discover tables** - Fetch metadata for relevant tables, save to `tables/*.yaml`
+2. **Write the file** - Create SQL/Python with proper headers
+3. **Sync to workspace API** - POST/PUT the transform
+4. **Run the transform** - Execute and check results
+5. **Present results** - Show sample data to user
+
+## Project Organization
+
+**One workspace = one analysis project.** Each project directory is a self-contained analysis:
+
+```
+cohort-retention-analysis/      # Created by /new-workspace
+├── .env                        # API key for this workspace
+├── workspace.yaml              # Workspace ID, database, URL
+├── tables/                     # Table schemas for THIS analysis
+│   ├── orders.yaml
+│   └── people.yaml
+├── questions/                  # Ad-hoc analysis queries
+│   └── q01_cohort_sizes.sql
+└── transforms/                 # Pipeline transforms
+    └── retention_cohort.sql
+```
+
+**For a new analysis topic, create a new workspace:**
+```bash
+cd ~/projects
+claude
+# /new-workspace → creates "customer-churn-analysis/"
+```
+
+This keeps each analysis isolated with its own:
+- Workspace (isolated schema in the database)
+- API credentials
+- Table metadata
+- Transforms
+
+**Do NOT mix unrelated analyses in one project.** If you're switching topics, create a new workspace.
+
+## Reading Configuration
+
+Always read config at the start:
+
+```bash
+# Read .env properly (handles values containing =)
+API_KEY=$(grep METABASE_API_KEY .env | cut -d= -f2-)
+WS_ID=$(grep '^id:' workspace.yaml | awk '{print $2}')
+DB_ID=$(grep '^database_id:' workspace.yaml | awk '{print $2}')
+URL=$(grep '^metabase_url:' workspace.yaml | awk '{print $2}')
+```
+
+## Step 1: Discover Tables
+
+Before writing any queries, fetch metadata for tables the user mentions or that are relevant to the question.
+
+### Using fetch_table.py (Recommended)
+
+Use the fetch_table.py script for fetching table metadata:
+
+```bash
+# List all available tables
+python3 ~/.claude/skills/workspace-admin/fetch_table.py --list
+
+# Fetch a specific table by ID
+python3 ~/.claude/skills/workspace-admin/fetch_table.py 1042
+
+# Fetch a table by name
+python3 ~/.claude/skills/workspace-admin/fetch_table.py --name orders
+
+# Fetch all tables at once
+python3 ~/.claude/skills/workspace-admin/fetch_table.py --all
+```
+
+This creates clean YAML files in `tables/`:
+
+```yaml
+name: orders
+schema: public
+database_id: 19
+table_id: 1042
+columns:
+  - name: id
+    type: Integer
+    semantic_type: PK
+  - name: customer_id
+    type: Integer
+    semantic_type: FK
+  - name: total
+    type: Decimal
+  - name: created_at
+    type: DateTime
+```
+
+### Check Existing Table Metadata
+
+Before fetching, check what we already have:
+
+```bash
+ls tables/
+```
+
+Read existing YAML files to understand available columns.
+
+## File Format Examples
+
+### SQL Question (questions/*.sql)
+
+```sql
+-- name: Revenue by Customer Segment
+-- description: Calculates total revenue grouped by customer segment
+-- target_schema: public
+-- target_table: revenue_by_segment
+
+SELECT
+    c.segment,
+    COUNT(DISTINCT o.customer_id) AS customer_count,
+    SUM(o.total) AS total_revenue,
+    AVG(o.total) AS avg_order_value
+FROM orders o
+JOIN customers c ON o.customer_id = c.id
+WHERE o.status = 'completed'
+GROUP BY c.segment
+ORDER BY total_revenue DESC
+```
+
+### SQL Transform (transforms/*.sql)
+
+```sql
+-- name: Enriched Orders
+-- description: Orders joined with customer and product details
+-- target_schema: public
+-- target_table: enriched_orders
+
+SELECT
+    o.id AS order_id,
+    o.created_at AS order_date,
+    o.total AS order_total,
+    c.id AS customer_id,
+    c.name AS customer_name,
+    c.segment AS customer_segment,
+    p.id AS product_id,
+    p.name AS product_name,
+    p.category AS product_category
+FROM orders o
+JOIN customers c ON o.customer_id = c.id
+JOIN order_items oi ON o.id = oi.order_id
+JOIN products p ON oi.product_id = p.id
+```
+
+### Python Transform (transforms/*.py)
+
+```python
+# name: Customer Cohort Analysis
+# description: Assigns customers to monthly cohorts based on first purchase
+# source_tables: {"orders": 1042, "customers": 1015}
+# target_schema: public
+# target_table: customer_cohorts
+
+import pandas as pd
+
+# First purchase date per customer
+first_purchase = orders_df.groupby('customer_id')['created_at'].min().reset_index()
+first_purchase.columns = ['customer_id', 'first_purchase_date']
+first_purchase['cohort_month'] = pd.to_datetime(first_purchase['first_purchase_date']).dt.to_period('M')
+
+# Join with customer data
+result = customers_df.merge(first_purchase, left_on='id', right_on='customer_id')
+result = result[['id', 'name', 'email', 'segment', 'cohort_month']]
+```
+
+## Autonomous Workflow
+
+When user asks a question or requests a transform:
+
+### Step 1: Discover and Fetch Table Metadata
+
+First, identify which tables are needed for this request. Then:
+
+1. Check if `tables/*.yaml` already exists for those tables
+2. If not, fetch from API and create the YAML files
+3. Read the YAML files to understand available columns
+
+```bash
+# Check existing
+ls tables/
+
+# If needed tables are missing, fetch them (see Step 1: Discover Tables above)
+```
+
+### Step 2: Understand the Request
+
+Determine:
+- Analysis mode (question) → `questions/` folder
+- Pipeline mode (transform) → `transforms/` folder
+- SQL or Python
+
+### Step 3: Write the File
+
+Create the SQL or Python file with proper header comments. Use descriptive filenames:
+- Questions: `q01_revenue_by_segment.sql`
+- Transforms: `enriched_orders.sql`, `customer_cohorts.py`
+
+### Step 4: Sync and Run
+
+Use sync_transform.py to sync files to the workspace and run them:
+
+```bash
+# Sync a single file
+python3 ~/.claude/skills/workspace-admin/sync_transform.py questions/q01_analysis.sql
+
+# Sync and run
+python3 ~/.claude/skills/workspace-admin/sync_transform.py --run questions/q01_analysis.sql
+
+# Sync all files
+python3 ~/.claude/skills/workspace-admin/sync_transform.py --all
+
+# Sync all and run
+python3 ~/.claude/skills/workspace-admin/sync_transform.py --all --run
+```
+
+The script:
+- Parses file headers automatically
+- Creates new transforms (POST) or updates existing (PUT)
+- Writes ref_id back to the file after creation
+- Optionally runs the transform with `--run`
+
+### Step 5: Preview Results with Dry-Run
+
+Use dry-run to preview transform output WITHOUT persisting to the isolated table:
+
+```bash
+# After syncing, use the ref_id to dry-run
+REF_ID="lucid-ferret-a852"  # From sync response or file header
+
+curl -s -X POST \
+  -H "x-api-key: $API_KEY" \
+  "$URL/api/ee/workspace/$WS_ID/transform/$REF_ID/dry-run" | jq '.rows[:10]'
+```
+
+Dry-run returns up to 2000 rows of preview data without writing to the database.
+
+### Step 6: Run and Verify
+
+Once satisfied with dry-run results, run the transform to persist:
+
+```bash
+curl -s -X POST \
+  -H "x-api-key: $API_KEY" \
+  "$URL/api/ee/workspace/$WS_ID/transform/$REF_ID/run"
+```
+
+The run response includes row count and status.
+
+### Step 7: Present Results
+
+Show the user:
+- What was created/updated
+- Row count from run response
+- Sample data from dry-run (first 10-20 rows as a table)
+- Any errors or warnings
+
+## Scratch Queries
+
+For quick exploratory queries, create a temporary transform and dry-run it:
+
+1. Create a scratch transform file
+2. Sync it to get a ref_id
+3. Dry-run to see results
+4. Delete if not needed
+
+Or use the sync_transform.py script which handles this:
+
+```bash
+# Create a quick scratch query
+echo '-- name: scratch
+-- target_schema: public
+-- target_table: _scratch
+
+SELECT COUNT(*) as total FROM orders' > /tmp/scratch.sql
+
+# Sync it
+python3 ~/.claude/skills/workspace-admin/sync_transform.py /tmp/scratch.sql
+```
+
+**Note:** The dry-run endpoint is per-transform: `POST /api/ee/workspace/:ws/transform/:ref/dry-run`
+
+## Error Handling
+
+If transform fails:
+1. Show the error message
+2. Identify the issue (SQL syntax, missing table, etc.)
+3. Fix the file
+4. Re-sync and re-run
+
+Common errors:
+- `relation "X" does not exist` → Check table name in schema
+- `column "X" does not exist` → Check column names in tables/*.yaml
+- `syntax error` → Fix SQL syntax

.claude/skills/workspace-admin/SKILL.md

@@ -0,0 +1,388 @@
+---
+name: workspace-admin
+description: Initialize analysis projects and merge to production (superuser operations)
+allowed-tools:
+  - Read
+  - Write
+  - Edit
+  - Bash
+  - Glob
+  - Grep
+  - AskUserQuestion
+---
+
+# Workspace Admin
+
+This skill handles superuser operations for data workspace projects:
+- **`/new-workspace`** - Create a new analysis project with workspace
+- **`/merge-workspace`** - Merge workspace to production and archive
+
+## Bootstrap Setup
+
+This skill must be installed globally to bootstrap new projects:
+
+```bash
+mkdir -p ~/.claude/skills/workspace-admin
+# Copy this SKILL.md to ~/.claude/skills/workspace-admin/SKILL.md
+```
+
+After that, you can run `/new-workspace` from any directory to create a new analysis project.
+
+## Prerequisites
+
+You need a **superuser API key** to use this skill. This is different from the workspace service user key that gets generated for day-to-day operations.
+
+## Commands
+
+### `/new-workspace` - Initialize a New Analysis Project
+
+Creates a new analysis project directory with:
+- Workspace created via Metabase API
+- Local directory structure for files
+- `.env` with the workspace's service user API key
+- Skills written into the project (self-contained)
+
+**Workflow:**
+
+1. Ask user for:
+   - Project name (will be used for directory and workspace name)
+   - Metabase URL (default: http://localhost:3000)
+   - Superuser API key
+   - Database ID (optional - can be set when first transform is created)
+
+2. Create workspace via API:
+   ```bash
+   curl -s -X POST \
+     -H "x-api-key: $SUPERUSER_KEY" \
+     -H "Content-Type: application/json" \
+     -d '{"name": "PROJECT_NAME"}' \
+     "$METABASE_URL/api/ee/workspace/"
+   ```
+
+3. Extract the `api_key` from the response - this is the service user key for the workspace (only returned at creation time).
+
+4. Create directory structure and write all files (see templates below)
+
+5. Output success message with next steps.
+
+**Example:**
+```
+User: /new-workspace
+Assistant: I'll help you create a new analysis project.
+
+What would you like to name this project?
+User: customer-churn-analysis
+
+What's your Metabase URL? (default: http://localhost:3000)
+User: http://localhost:3000
+
+Please provide your superuser API key:
+User: mb_superuser_abc123...
+
+Creating workspace...
+Created workspace ID 2856 with service user API key.
+
+Creating project directory structure...
+Done!
+
+Your project is ready at ./customer-churn-analysis
+
+Next steps:
+  cd customer-churn-analysis
+  claude
+
+Then ask questions or build transforms!
+```
+
+### `/merge-workspace` - Merge to Production
+
+Merges all transforms from the workspace to production and archives the workspace.
+
+**Workflow:**
+
+1. Read workspace.yaml to get workspace ID and Metabase URL
+2. Prompt for superuser API key
+3. Call merge endpoint:
+   ```bash
+   curl -s -X POST \
+     -H "x-api-key: $SUPERUSER_KEY" \
+     -H "Content-Type: application/json" \
+     -d '{"commit-message": "Merged from analysis project"}' \
+     "$METABASE_URL/api/ee/workspace/$WS_ID/merge"
+   ```
+4. Report results and update workspace.yaml status
+
+---
+
+## File Templates for /new-workspace
+
+When running `/new-workspace`, create these files in the project directory:
+
+### Directory Structure
+```
+PROJECT_NAME/
+├── .claude/
+│   └── skills/
+│       ├── workspace-admin/
+│       │   ├── SKILL.md
+│       │   └── fetch_table.py    # Table metadata fetcher
+│       ├── workspace-files/SKILL.md
+│       ├── workspace-sync/SKILL.md
+│       └── data-workspace/SKILL.md
+├── .env
+├── .gitignore
+├── workspace.yaml
+├── tables/
+├── questions/
+└── transforms/
+```
+
+### .gitignore
+```
+.env
+```
+
+### .env
+```
+METABASE_API_KEY=<api_key from workspace creation response>
+```
+
+### workspace.yaml
+```yaml
+id: <workspace_id>
+name: "<project_name>"
+database_id: <database_id or null>
+status: uninitialized
+metabase_url: <url>
+```
+
+### .claude/skills/workspace-files/SKILL.md
+
+```markdown
+---
+name: workspace-files
+description: Manage local file representations of transforms and table schemas
+allowed-tools: [Read, Write, Edit, Bash, Glob, Grep]
+---
+
+# Workspace Files
+
+Manages local files in a data workspace project.
+
+## SQL File Format (questions/*.sql or transforms/*.sql)
+
+```sql
+-- name: Revenue by Customer Segment
+-- description: Calculates total revenue grouped by customer segment
+-- target_schema: public
+-- target_table: revenue_by_segment
+-- ref_id: lucid-ferret-a852
+
+SELECT
+    c.segment,
+    COUNT(DISTINCT o.customer_id) AS customer_count,
+    SUM(o.total) AS total_revenue
+FROM orders o
+JOIN customers c ON o.customer_id = c.id
+WHERE o.status = 'completed'
+GROUP BY c.segment
+ORDER BY total_revenue DESC
+```
+
+## Python File Format (transforms/*.py)
+
+```python
+# name: Customer Cohort Analysis
+# description: Assigns customers to monthly cohorts
+# source_tables: {"orders": 1042, "customers": 1015}
+# target_schema: public
+# target_table: customer_cohorts
+# ref_id: gentle-fox-b123
+
+import pandas as pd
+
+first_purchase = orders_df.groupby('customer_id')['created_at'].min().reset_index()
+first_purchase['cohort'] = pd.to_datetime(first_purchase['created_at']).dt.to_period('M')
+result = customers_df.merge(first_purchase, left_on='id', right_on='customer_id')
+```
+
+## Table Schema (tables/*.yaml)
+
+```yaml
+name: orders
+schema: public
+database_id: 19
+table_id: 1042
+columns:
+  - name: id
+    type: integer
+  - name: customer_id
+    type: integer
+  - name: total
+    type: decimal
+  - name: created_at
+    type: timestamp
+```
+
+## Header Fields
+- `name` (required): Human-readable name
+- `description` (optional): What this does
+- `target_schema` (required): Target schema name
+- `target_table` (required): Target table name
+- `ref_id` (auto): Assigned by API after sync - DO NOT set manually
+- `source_tables` (Python only): JSON map of dataframe name to table ID
+```
+
+### .claude/skills/workspace-sync/SKILL.md
+
+```markdown
+---
+name: workspace-sync
+description: Sync local transform files with the workspace API
+allowed-tools: [Read, Write, Edit, Bash, Glob, Grep]
+---
+
+# Workspace Sync
+
+Syncs local files with the Metabase workspace API.
+
+## Behavior
+- Local files are source of truth
+- New files (no ref_id): POST to create, write back ref_id
+- Existing files (has ref_id): PUT to update
+- Deleted files: DELETE from API
+
+## Reading Configuration
+
+```bash
+API_KEY=$(grep METABASE_API_KEY .env | cut -d= -f2-)
+WS_ID=$(grep '^id:' workspace.yaml | awk '{print $2}')
+DB_ID=$(grep '^database_id:' workspace.yaml | awk '{print $2}')
+URL=$(grep '^metabase_url:' workspace.yaml | awk '{print $2}')
+```
+
+## Syncing a SQL File
+
+```bash
+FILE="questions/q01_revenue.sql"
+NAME=$(grep '^-- name:' "$FILE" | sed 's/^-- name: //')
+SCHEMA=$(grep '^-- target_schema:' "$FILE" | sed 's/^-- target_schema: //')
+TABLE=$(grep '^-- target_table:' "$FILE" | sed 's/^-- target_table: //')
+REF_ID=$(grep '^-- ref_id:' "$FILE" | sed 's/^-- ref_id: //')
+SQL=$(sed -n '/^[^-]/,$p' "$FILE" | jq -Rs .)
+
+JSON='{"name":"'"$NAME"'","source":{"type":"query","query":{"type":"native","native":{"query":'"$SQL"'},"database":'"$DB_ID"'}},"target":{"type":"table","schema":"'"$SCHEMA"'","name":"'"$TABLE"'"}}'
+
+if [ -z "$REF_ID" ]; then
+  RESP=$(curl -s -X POST -H "x-api-key: $API_KEY" -H "Content-Type: application/json" -d "$JSON" "$URL/api/ee/workspace/$WS_ID/transform")
+  NEW_REF=$(echo "$RESP" | jq -r '.ref_id')
+  # Add ref_id to file header
+else
+  curl -s -X PUT -H "x-api-key: $API_KEY" -H "Content-Type: application/json" -d "$JSON" "$URL/api/ee/workspace/$WS_ID/transform/$REF_ID"
+fi
+```
+```
+
+### .claude/skills/data-workspace/SKILL.md
+
+```markdown
+---
+name: data-workspace
+description: Unified workflow for data analysis and pipeline building
+allowed-tools: [Read, Write, Edit, Bash, Glob, Grep, AskUserQuestion]
+---
+
+# Data Workspace
+
+Autonomous workflow. Execute the FULL loop without prompting:
+1. Discover tables - fetch metadata, save to `tables/*.yaml`
+2. Write file with proper headers
+3. Sync to workspace API
+4. Run the transform
+5. Present results
+
+## Configuration
+
+```bash
+API_KEY=$(grep METABASE_API_KEY .env | cut -d= -f2-)
+WS_ID=$(grep '^id:' workspace.yaml | awk '{print $2}')
+DB_ID=$(grep '^database_id:' workspace.yaml | awk '{print $2}')
+URL=$(grep '^metabase_url:' workspace.yaml | awk '{print $2}')
+```
+
+## Step 1: Discover Tables
+
+Before writing queries, fetch metadata for relevant tables:
+
+```bash
+# List all tables
+python3 ~/.claude/skills/workspace-admin/fetch_table.py --list
+
+# Fetch specific table by name
+python3 ~/.claude/skills/workspace-admin/fetch_table.py --name orders
+
+# Fetch all tables
+python3 ~/.claude/skills/workspace-admin/fetch_table.py --all
+```
+
+## File Formats
+
+### SQL
+```sql
+-- name: Revenue by Segment
+-- target_schema: public
+-- target_table: revenue_by_segment
+
+SELECT segment, SUM(total) as revenue FROM orders GROUP BY segment
+```
+
+### Python
+```python
+# name: Customer Cohorts
+# source_tables: {"orders": 1042, "customers": 1015}
+# target_schema: public
+# target_table: customer_cohorts
+
+result = customers_df.merge(orders_df, on='customer_id')
+```
+
+## API
+
+- Create: `POST /api/ee/workspace/:ws/transform`
+- Update: `PUT /api/ee/workspace/:ws/transform/:ref`
+- Run: `POST /api/ee/workspace/:ws/transform/:ref/run`
+```
+
+### .claude/skills/workspace-admin/SKILL.md
+
+Copy this entire file into the project so users can run `/merge-workspace` from within the project.
+
+---
+
+## API Reference
+
+### Create Workspace
+```
+POST /api/ee/workspace/
+Headers: x-api-key: <superuser_key>
+Body: {"name": "...", "database_id": <optional>}
+Response: {
+  "id": 123,
+  "name": "...",
+  "status": "uninitialized",
+  "api_key": "mb_..."  // Only on creation!
+}
+```
+
+### Merge Workspace
+```
+POST /api/ee/workspace/:id/merge
+Headers: x-api-key: <superuser_key>
+Body: {"commit-message": "..."}
+```
+
+### Archive Workspace
+```
+POST /api/ee/workspace/:id/archive
+Headers: x-api-key: <superuser_key>
+```

.claude/skills/workspace-admin/fetch_table.py

@@ -0,0 +1,222 @@
+#!/usr/bin/env python3
+"""Fetch table metadata from Metabase API and save as YAML.
+
+Usage:
+    python fetch_table.py <table_id>              # Fetch by ID
+    python fetch_table.py --name <table_name>     # Fetch by name
+    python fetch_table.py --list                  # List all tables
+    python fetch_table.py --all                   # Fetch all tables
+
+Reads configuration from .env and workspace.yaml in current directory.
+"""
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError
+
+
+def read_env():
+    """Read .env file and return dict of values."""
+    env = {}
+    env_path = Path(".env")
+    if env_path.exists():
+        for line in env_path.read_text().splitlines():
+            if "=" in line and not line.startswith("#"):
+                key, _, value = line.partition("=")
+                env[key.strip()] = value.strip()
+    return env
+
+
+def read_workspace_yaml():
+    """Read workspace.yaml and return dict of values."""
+    config = {}
+    ws_path = Path("workspace.yaml")
+    if ws_path.exists():
+        for line in ws_path.read_text().splitlines():
+            if ":" in line:
+                key, _, value = line.partition(":")
+                config[key.strip()] = value.strip().strip('"').strip("'")
+    return config
+
+
+def api_get(url: str, api_key: str) -> dict:
+    """Make GET request to Metabase API."""
+    req = Request(url, headers={"x-api-key": api_key})
+    try:
+        with urlopen(req) as resp:
+            return json.loads(resp.read().decode())
+    except HTTPError as e:
+        print(f"Error: {e.code} {e.reason}", file=sys.stderr)
+        sys.exit(1)
+
+
+def to_yaml(data: dict, indent: int = 0) -> str:
+    """Convert dict to YAML string (simple implementation)."""
+    lines = []
+    prefix = "  " * indent
+
+    for key, value in data.items():
+        if value is None:
+            lines.append(f"{prefix}{key}:")
+        elif isinstance(value, bool):
+            lines.append(f"{prefix}{key}: {str(value).lower()}")
+        elif isinstance(value, (int, float)):
+            lines.append(f"{prefix}{key}: {value}")
+        elif isinstance(value, str):
+            # Quote if contains special chars
+            if any(c in value for c in ":#{}[]|>&*!?"):
+                lines.append(f'{prefix}{key}: "{value}"')
+            else:
+                lines.append(f"{prefix}{key}: {value}")
+        elif isinstance(value, list):
+            lines.append(f"{prefix}{key}:")
+            for item in value:
+                if isinstance(item, dict):
+                    first = True
+                    for k, v in item.items():
+                        if first:
+                            lines.append(f"{prefix}  - {k}: {v if v is not None else ''}")
+                            first = False
+                        else:
+                            lines.append(f"{prefix}    {k}: {v if v is not None else ''}")
+                else:
+                    lines.append(f"{prefix}  - {item}")
+        elif isinstance(value, dict):
+            lines.append(f"{prefix}{key}:")
+            lines.append(to_yaml(value, indent + 1))
+
+    return "\n".join(lines)
+
+
+def fetch_table_metadata(base_url: str, api_key: str, table_id: int) -> dict:
+    """Fetch table metadata and convert to our YAML schema."""
+    url = f"{base_url}/api/table/{table_id}/query_metadata"
+    data = api_get(url, api_key)
+
+    columns = []
+    for field in data.get("fields", []):
+        col = {
+            "name": field.get("name"),
+            "type": field.get("base_type", "").replace("type/", ""),
+        }
+        if field.get("semantic_type"):
+            col["semantic_type"] = field["semantic_type"].replace("type/", "")
+        if field.get("description"):
+            col["description"] = field["description"]
+        if field.get("fk_target_field_id"):
+            col["fk_target_field_id"] = field["fk_target_field_id"]
+        columns.append(col)
+
+    return {
+        "name": data.get("name"),
+        "schema": data.get("schema"),
+        "database_id": data.get("db_id"),
+        "table_id": data.get("id"),
+        "columns": columns,
+    }
+
+
+def list_tables(base_url: str, api_key: str, database_id: int) -> list:
+    """List all tables in the database."""
+    # Use /api/table?db_id=X instead of /api/database/X/metadata
+    # The metadata endpoint may require different permissions
+    url = f"{base_url}/api/table?db_id={database_id}"
+    return api_get(url, api_key)
+
+
+def save_table_yaml(table_data: dict, output_dir: Path = Path("tables")):
+    """Save table metadata as YAML file."""
+    output_dir.mkdir(exist_ok=True)
+    name = table_data["name"]
+    # Sanitize filename
+    safe_name = re.sub(r'[^\w\-]', '_', name.lower())
+    output_path = output_dir / f"{safe_name}.yaml"
+
+    yaml_content = to_yaml(table_data)
+    output_path.write_text(yaml_content + "\n")
+    print(f"Saved: {output_path}")
+    return output_path
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Fetch table metadata from Metabase")
+    parser.add_argument("table_id", nargs="?", type=int, help="Table ID to fetch")
+    parser.add_argument("--name", "-n", help="Fetch table by name")
+    parser.add_argument("--list", "-l", action="store_true", help="List all tables")
+    parser.add_argument("--all", "-a", action="store_true", help="Fetch all tables")
+    parser.add_argument("--output", "-o", default="tables", help="Output directory")
+    args = parser.parse_args()
+
+    # Read configuration
+    env = read_env()
+    config = read_workspace_yaml()
+
+    api_key = env.get("METABASE_API_KEY")
+    if not api_key:
+        print("Error: METABASE_API_KEY not found in .env", file=sys.stderr)
+        sys.exit(1)
+
+    base_url = config.get("metabase_url", "http://localhost:3000").rstrip("/")
+    database_id = config.get("database_id")
+
+    if not database_id:
+        print("Error: database_id not found in workspace.yaml", file=sys.stderr)
+        sys.exit(1)
+
+    database_id = int(database_id)
+    output_dir = Path(args.output)
+
+    # List tables
+    if args.list:
+        tables = list_tables(base_url, api_key, database_id)
+        print(f"{'ID':<8} {'Schema':<20} {'Name':<40}")
+        print("-" * 68)
+        for t in sorted(tables, key=lambda x: (x.get("schema", ""), x.get("name", ""))):
+            print(f"{t['id']:<8} {t.get('schema', ''):<20} {t['name']:<40}")
+        return
+
+    # Fetch all tables
+    if args.all:
+        tables = list_tables(base_url, api_key, database_id)
+        for t in tables:
+            try:
+                data = fetch_table_metadata(base_url, api_key, t["id"])
+                save_table_yaml(data, output_dir)
+            except Exception as e:
+                print(f"Error fetching {t['name']}: {e}", file=sys.stderr)
+        return
+
+    # Fetch by name
+    if args.name:
+        tables = list_tables(base_url, api_key, database_id)
+        matches = [t for t in tables if t["name"].lower() == args.name.lower()]
+        if not matches:
+            # Try partial match
+            matches = [t for t in tables if args.name.lower() in t["name"].lower()]
+        if not matches:
+            print(f"Error: No table found matching '{args.name}'", file=sys.stderr)
+            sys.exit(1)
+        if len(matches) > 1:
+            print(f"Multiple matches for '{args.name}':")
+            for t in matches:
+                print(f"  {t['id']}: {t.get('schema', '')}.{t['name']}")
+            sys.exit(1)
+        table_id = matches[0]["id"]
+    elif args.table_id:
+        table_id = args.table_id
+    else:
+        parser.print_help()
+        sys.exit(1)
+
+    # Fetch and save
+    data = fetch_table_metadata(base_url, api_key, table_id)
+    save_table_yaml(data, output_dir)
+
+
+if __name__ == "__main__":
+    main()

.claude/skills/workspace-admin/query_transform.py

@@ -0,0 +1,228 @@
+#!/usr/bin/env python3
+"""Run or dry-run a transform and display results.
+
+Usage:
+    python query_transform.py <ref_id>                    # Dry-run by ref_id
+    python query_transform.py <file.sql>                  # Dry-run by file (extracts ref_id)
+    python query_transform.py --run <ref_id>              # Run (persist) by ref_id
+    python query_transform.py --sql "SELECT ..." [--name x] # Quick scratch query
+
+Reads configuration from .env and workspace.yaml in current directory.
+"""
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError
+
+
+def read_env():
+    """Read .env file and return dict of values."""
+    env = {}
+    env_path = Path(".env")
+    if env_path.exists():
+        for line in env_path.read_text().splitlines():
+            if "=" in line and not line.startswith("#"):
+                key, _, value = line.partition("=")
+                env[key.strip()] = value.strip()
+    return env
+
+
+def read_workspace_yaml():
+    """Read workspace.yaml and return dict of values."""
+    config = {}
+    ws_path = Path("workspace.yaml")
+    if ws_path.exists():
+        for line in ws_path.read_text().splitlines():
+            if ":" in line:
+                key, _, value = line.partition(":")
+                config[key.strip()] = value.strip().strip('"').strip("'")
+    return config
+
+
+def api_request(method: str, url: str, api_key: str, data: dict = None) -> dict:
+    """Make request to Metabase API."""
+    body = json.dumps(data).encode() if data else None
+    headers = {"x-api-key": api_key}
+    if data:
+        headers["Content-Type"] = "application/json"
+
+    req = Request(url, data=body, headers=headers, method=method)
+    try:
+        with urlopen(req) as resp:
+            return json.loads(resp.read().decode())
+    except HTTPError as e:
+        error_body = e.read().decode() if e.fp else ""
+        print(f"Error: {e.code} {e.reason}", file=sys.stderr)
+        if error_body:
+            print(f"Response: {error_body}", file=sys.stderr)
+        sys.exit(1)
+
+
+def get_ref_id_from_file(file_path: Path) -> str:
+    """Extract ref_id from a SQL or Python file."""
+    content = file_path.read_text()
+
+    # Try SQL format
+    match = re.search(r'^-- ref_id:\s*(\S+)', content, re.MULTILINE)
+    if match:
+        return match.group(1)
+
+    # Try Python format
+    match = re.search(r'^# ref_id:\s*(\S+)', content, re.MULTILINE)
+    if match:
+        return match.group(1)
+
+    return None
+
+
+def format_table(rows: list, cols: list = None, max_width: int = 40) -> str:
+    """Format rows as an ASCII table."""
+    if not rows:
+        return "(no rows)"
+
+    # Use column names if provided, otherwise generate headers
+    if cols:
+        headers = [c.get("name", f"col{i}") for i, c in enumerate(cols)]
+    else:
+        headers = [f"col{i}" for i in range(len(rows[0]))]
+
+    # Convert all values to strings and truncate
+    def fmt(v):
+        s = str(v) if v is not None else "NULL"
+        return s[:max_width] + "..." if len(s) > max_width else s
+
+    str_rows = [[fmt(v) for v in row] for row in rows]
+
+    # Calculate column widths
+    widths = [max(len(h), max((len(r[i]) for r in str_rows), default=0))
+              for i, h in enumerate(headers)]
+
+    # Build table
+    lines = []
+    header_line = " | ".join(h.ljust(w) for h, w in zip(headers, widths))
+    lines.append(header_line)
+    lines.append("-+-".join("-" * w for w in widths))
+    for row in str_rows:
+        lines.append(" | ".join(v.ljust(w) for v, w in zip(row, widths)))
+
+    return "\n".join(lines)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Run or dry-run transforms")
+    parser.add_argument("target", nargs="?", help="ref_id or file path")
+    parser.add_argument("--run", "-r", action="store_true", help="Run (persist) instead of dry-run")
+    parser.add_argument("--sql", "-s", help="Quick SQL query (creates scratch transform)")
+    parser.add_argument("--name", "-n", default="_scratch", help="Name for scratch query")
+    parser.add_argument("--limit", "-l", type=int, default=20, help="Max rows to display")
+    args = parser.parse_args()
+
+    # Read configuration
+    env = read_env()
+    config = read_workspace_yaml()
+
+    api_key = env.get("METABASE_API_KEY")
+    if not api_key:
+        print("Error: METABASE_API_KEY not found in .env", file=sys.stderr)
+        sys.exit(1)
+
+    base_url = config.get("metabase_url", "http://localhost:3000").rstrip("/")
+    ws_id = config.get("id")
+    db_id = config.get("database_id")
+
+    if not ws_id:
+        print("Error: id not found in workspace.yaml", file=sys.stderr)
+        sys.exit(1)
+
+    ws_id = int(ws_id)
+    db_id = int(db_id) if db_id else None
+
+    # Handle scratch SQL query
+    if args.sql:
+        if not db_id:
+            print("Error: database_id not found in workspace.yaml", file=sys.stderr)
+            sys.exit(1)
+
+        # Create scratch transform
+        payload = {
+            "name": args.name,
+            "source": {
+                "type": "query",
+                "query": {
+                    "type": "native",
+                    "native": {"query": args.sql},
+                    "database": db_id
+                }
+            },
+            "target": {
+                "type": "table",
+                "schema": "public",
+                "name": args.name.replace(" ", "_").lower()
+            }
+        }
+
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform"
+        result = api_request("POST", url, api_key, payload)
+        ref_id = result.get("ref_id")
+        print(f"Created scratch transform: {ref_id}")
+
+        # Dry-run it
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform/{ref_id}/dry-run"
+        result = api_request("POST", url, api_key)
+
+        rows = result.get("rows", [])
+        cols = result.get("cols", [])
+        print(f"\nResults ({len(rows)} rows):\n")
+        print(format_table(rows[:args.limit], cols))
+
+        if len(rows) > args.limit:
+            print(f"\n... and {len(rows) - args.limit} more rows")
+
+        # Delete scratch transform
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform/{ref_id}"
+        api_request("DELETE", url, api_key)
+        print(f"\nDeleted scratch transform: {ref_id}")
+        return
+
+    # Handle ref_id or file
+    if not args.target:
+        parser.print_help()
+        sys.exit(1)
+
+    # Check if it's a file
+    target_path = Path(args.target)
+    if target_path.exists():
+        ref_id = get_ref_id_from_file(target_path)
+        if not ref_id:
+            print(f"Error: No ref_id found in {args.target}. Sync the file first.", file=sys.stderr)
+            sys.exit(1)
+    else:
+        ref_id = args.target
+
+    # Run or dry-run
+    if args.run:
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform/{ref_id}/run"
+        result = api_request("POST", url, api_key)
+        status = result.get("status", "unknown")
+        rows = result.get("rows_affected", "?")
+        print(f"Run complete: {status} ({rows} rows affected)")
+    else:
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform/{ref_id}/dry-run"
+        result = api_request("POST", url, api_key)
+
+        rows = result.get("rows", [])
+        cols = result.get("cols", [])
+        print(f"Dry-run results ({len(rows)} rows):\n")
+        print(format_table(rows[:args.limit], cols))
+
+        if len(rows) > args.limit:
+            print(f"\n... and {len(rows) - args.limit} more rows")
+
+
+if __name__ == "__main__":
+    main()

.claude/skills/workspace-admin/sync_transform.py

@@ -0,0 +1,298 @@
+#!/usr/bin/env python3
+"""Sync a transform file to the Metabase workspace API.
+
+Usage:
+    python sync_transform.py <file.sql>           # Sync a SQL file
+    python sync_transform.py <file.py>            # Sync a Python file
+    python sync_transform.py --all                # Sync all files
+    python sync_transform.py --run <file.sql>     # Sync and run
+
+Reads configuration from .env and workspace.yaml in current directory.
+"""
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError
+
+
+def read_env():
+    """Read .env file and return dict of values."""
+    env = {}
+    env_path = Path(".env")
+    if env_path.exists():
+        for line in env_path.read_text().splitlines():
+            if "=" in line and not line.startswith("#"):
+                key, _, value = line.partition("=")
+                env[key.strip()] = value.strip()
+    return env
+
+
+def read_workspace_yaml():
+    """Read workspace.yaml and return dict of values."""
+    config = {}
+    ws_path = Path("workspace.yaml")
+    if ws_path.exists():
+        for line in ws_path.read_text().splitlines():
+            if ":" in line:
+                key, _, value = line.partition(":")
+                config[key.strip()] = value.strip().strip('"').strip("'")
+    return config
+
+
+def api_request(method: str, url: str, api_key: str, data: dict = None) -> dict:
+    """Make request to Metabase API."""
+    body = json.dumps(data).encode() if data else None
+    headers = {"x-api-key": api_key}
+    if data:
+        headers["Content-Type"] = "application/json"
+
+    req = Request(url, data=body, headers=headers, method=method)
+    try:
+        with urlopen(req) as resp:
+            return json.loads(resp.read().decode())
+    except HTTPError as e:
+        error_body = e.read().decode() if e.fp else ""
+        print(f"Error: {e.code} {e.reason}", file=sys.stderr)
+        if error_body:
+            print(f"Response: {error_body}", file=sys.stderr)
+        sys.exit(1)
+
+
+def parse_sql_file(file_path: Path) -> dict:
+    """Parse a SQL file with header comments."""
+    content = file_path.read_text()
+    lines = content.splitlines()
+
+    metadata = {}
+    body_start = 0
+
+    for i, line in enumerate(lines):
+        if line.startswith("-- ") and ":" in line:
+            key_value = line[3:]  # Remove "-- "
+            key, _, value = key_value.partition(":")
+            metadata[key.strip()] = value.strip()
+        elif line.strip() and not line.startswith("--"):
+            body_start = i
+            break
+        elif line.strip() == "":
+            # Empty line after headers
+            body_start = i + 1
+
+    # Find first non-empty, non-comment line for body start
+    for i in range(body_start, len(lines)):
+        if lines[i].strip() and not lines[i].startswith("--"):
+            body_start = i
+            break
+
+    body = "\n".join(lines[body_start:])
+
+    return {
+        "name": metadata.get("name", file_path.stem),
+        "description": metadata.get("description"),
+        "target_schema": metadata.get("target_schema", "public"),
+        "target_table": metadata.get("target_table", file_path.stem),
+        "ref_id": metadata.get("ref_id"),
+        "body": body,
+        "type": "sql"
+    }
+
+
+def parse_python_file(file_path: Path) -> dict:
+    """Parse a Python file with header comments."""
+    content = file_path.read_text()
+    lines = content.splitlines()
+
+    metadata = {}
+    body_start = 0
+
+    for i, line in enumerate(lines):
+        if line.startswith("# ") and ":" in line:
+            key_value = line[2:]  # Remove "# "
+            key, _, value = key_value.partition(":")
+            metadata[key.strip()] = value.strip()
+        elif line.strip() and not line.startswith("#"):
+            body_start = i
+            break
+
+    body = "\n".join(lines[body_start:])
+
+    # Parse source_tables JSON
+    source_tables = {}
+    if metadata.get("source_tables"):
+        try:
+            source_tables = json.loads(metadata["source_tables"])
+        except json.JSONDecodeError:
+            print(f"Warning: Invalid source_tables JSON in {file_path}", file=sys.stderr)
+
+    return {
+        "name": metadata.get("name", file_path.stem),
+        "description": metadata.get("description"),
+        "target_schema": metadata.get("target_schema", "public"),
+        "target_table": metadata.get("target_table", file_path.stem),
+        "ref_id": metadata.get("ref_id"),
+        "source_tables": source_tables,
+        "body": body,
+        "type": "python"
+    }
+
+
+def build_api_payload(parsed: dict, database_id: int) -> dict:
+    """Build the API request payload from parsed file data."""
+    if parsed["type"] == "sql":
+        source = {
+            "type": "query",
+            "query": {
+                "type": "native",
+                "native": {"query": parsed["body"]},
+                "database": database_id
+            }
+        }
+    else:  # python
+        source = {
+            "type": "python",
+            "source-tables": parsed["source_tables"],
+            "body": parsed["body"]
+        }
+
+    payload = {
+        "name": parsed["name"],
+        "source": source,
+        "target": {
+            "type": "table",
+            "schema": parsed["target_schema"],
+            "name": parsed["target_table"]
+        }
+    }
+
+    if parsed.get("description"):
+        payload["description"] = parsed["description"]
+
+    return payload
+
+
+def add_ref_id_to_file(file_path: Path, ref_id: str):
+    """Add ref_id to file header."""
+    content = file_path.read_text()
+    lines = content.splitlines()
+
+    # Determine comment style
+    is_python = file_path.suffix == ".py"
+    comment_prefix = "# " if is_python else "-- "
+    target_line_prefix = f"{comment_prefix}target_table:"
+
+    # Find target_table line and insert ref_id after it
+    new_lines = []
+    added = False
+    for line in lines:
+        new_lines.append(line)
+        if not added and line.startswith(target_line_prefix):
+            new_lines.append(f"{comment_prefix}ref_id: {ref_id}")
+            added = True
+
+    if not added:
+        # If no target_table found, add at the beginning after other headers
+        for i, line in enumerate(lines):
+            if not line.startswith(comment_prefix) or ":" not in line:
+                new_lines = lines[:i] + [f"{comment_prefix}ref_id: {ref_id}"] + lines[i:]
+                break
+
+    file_path.write_text("\n".join(new_lines) + "\n")
+
+
+def sync_file(file_path: Path, base_url: str, api_key: str, ws_id: int, db_id: int, run: bool = False):
+    """Sync a single file to the workspace."""
+    # Parse file
+    if file_path.suffix == ".sql":
+        parsed = parse_sql_file(file_path)
+    elif file_path.suffix == ".py":
+        parsed = parse_python_file(file_path)
+    else:
+        print(f"Skipping unsupported file: {file_path}", file=sys.stderr)
+        return
+
+    payload = build_api_payload(parsed, db_id)
+    ref_id = parsed.get("ref_id")
+
+    if ref_id:
+        # Update existing transform
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform/{ref_id}"
+        result = api_request("PUT", url, api_key, payload)
+        print(f"Updated: {file_path} ({ref_id})")
+    else:
+        # Create new transform
+        url = f"{base_url}/api/ee/workspace/{ws_id}/transform"
+        result = api_request("POST", url, api_key, payload)
+        new_ref_id = result.get("ref_id")
+        if new_ref_id:
+            add_ref_id_to_file(file_path, new_ref_id)
+            print(f"Created: {file_path} -> {new_ref_id}")
+            ref_id = new_ref_id
+        else:
+            print(f"Warning: No ref_id returned for {file_path}", file=sys.stderr)
+
+    # Run transform if requested
+    if run and ref_id:
+        run_url = f"{base_url}/api/ee/workspace/{ws_id}/transform/{ref_id}/run"
+        run_result = api_request("POST", run_url, api_key)
+        status = run_result.get("status", "unknown")
+        rows = run_result.get("rows_affected", "?")
+        print(f"  Run: {status} ({rows} rows)")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Sync transform files to Metabase workspace")
+    parser.add_argument("files", nargs="*", help="Files to sync")
+    parser.add_argument("--all", "-a", action="store_true", help="Sync all files in questions/ and transforms/")
+    parser.add_argument("--run", "-r", action="store_true", help="Run transforms after syncing")
+    args = parser.parse_args()
+
+    # Read configuration
+    env = read_env()
+    config = read_workspace_yaml()
+
+    api_key = env.get("METABASE_API_KEY")
+    if not api_key:
+        print("Error: METABASE_API_KEY not found in .env", file=sys.stderr)
+        sys.exit(1)
+
+    base_url = config.get("metabase_url", "http://localhost:3000").rstrip("/")
+    ws_id = config.get("id")
+    db_id = config.get("database_id")
+
+    if not ws_id:
+        print("Error: id not found in workspace.yaml", file=sys.stderr)
+        sys.exit(1)
+    if not db_id:
+        print("Error: database_id not found in workspace.yaml", file=sys.stderr)
+        sys.exit(1)
+
+    ws_id = int(ws_id)
+    db_id = int(db_id)
+
+    # Collect files to sync
+    files = []
+    if args.all:
+        for pattern in ["questions/*.sql", "transforms/*.sql", "transforms/*.py"]:
+            files.extend(Path(".").glob(pattern))
+    else:
+        files = [Path(f) for f in args.files]
+
+    if not files:
+        print("No files to sync. Use --all or specify files.", file=sys.stderr)
+        sys.exit(1)
+
+    # Sync each file
+    for file_path in sorted(files):
+        if file_path.exists():
+            sync_file(file_path, base_url, api_key, ws_id, db_id, run=args.run)
+        else:
+            print(f"File not found: {file_path}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

.claude/skills/workspace-files/SKILL.md

@@ -0,0 +1,178 @@
+---
+name: workspace-files
+description: Manage local file representations of transforms and table schemas
+allowed-tools:
+  - Read
+  - Write
+  - Edit
+  - Bash
+  - Glob
+  - Grep
+---
+
+# Workspace Files
+
+Manages the local file representations in a data workspace project.
+
+## Directory Structure
+
+```
+my-analysis/
+├── .env                  # METABASE_API_KEY=...
+├── workspace.yaml        # Workspace metadata
+├── tables/               # Input table schemas
+│   ├── orders.yaml
+│   ├── customers.yaml
+│   └── products.yaml
+├── questions/            # Analysis queries (user-facing)
+│   ├── q01_revenue_by_segment.sql
+│   └── q02_top_customers.sql
+└── transforms/           # Data pipeline transforms
+    ├── enriched_orders.sql
+    └── customer_cohorts.py
+```
+
+## SQL File Format
+
+Use for both `questions/*.sql` and `transforms/*.sql`:
+
+```sql
+-- name: Revenue by Customer Segment
+-- description: Calculates total revenue grouped by customer segment
+-- target_schema: public
+-- target_table: revenue_by_segment
+-- ref_id: lucid-ferret-a852
+
+SELECT
+    c.segment,
+    COUNT(DISTINCT o.customer_id) AS customer_count,
+    SUM(o.total) AS total_revenue,
+    AVG(o.total) AS avg_order_value
+FROM orders o
+JOIN customers c ON o.customer_id = c.id
+WHERE o.status = 'completed'
+GROUP BY c.segment
+ORDER BY total_revenue DESC
+```
+
+### Header Fields (SQL)
+
+| Field | Required | Description |
+|-------|----------|-------------|
+| `name` | Yes | Human-readable name |
+| `description` | No | What this transform does |
+| `target_schema` | Yes | Target schema (e.g., `public`) |
+| `target_table` | Yes | Target table name |
+| `ref_id` | Auto | Assigned by API after sync - never set manually |
+
+## Python File Format
+
+Use for `transforms/*.py`:
+
+```python
+# name: Customer Cohort Analysis
+# description: Assigns customers to monthly cohorts based on first purchase
+# source_tables: {"orders": 1042, "customers": 1015}
+# target_schema: public
+# target_table: customer_cohorts
+# ref_id: gentle-fox-b123
+
+import pandas as pd
+
+# First purchase date per customer
+first_purchase = orders_df.groupby('customer_id')['created_at'].min().reset_index()
+first_purchase.columns = ['customer_id', 'first_purchase_date']
+first_purchase['cohort_month'] = pd.to_datetime(first_purchase['first_purchase_date']).dt.to_period('M')
+
+# Join with customer data
+result = customers_df.merge(first_purchase, left_on='id', right_on='customer_id')
+result = result[['id', 'name', 'email', 'segment', 'cohort_month']]
+```
+
+### Header Fields (Python)
+
+| Field | Required | Description |
+|-------|----------|-------------|
+| `name` | Yes | Human-readable name |
+| `description` | No | What this transform does |
+| `source_tables` | Yes | JSON map: `{"df_name": table_id, ...}` |
+| `target_schema` | Yes | Target schema |
+| `target_table` | Yes | Target table name |
+| `ref_id` | Auto | Assigned by API after sync |
+
+The `source_tables` field maps DataFrame variable names to Metabase table IDs. Each table is loaded as `{name}_df` (e.g., `orders_df`, `customers_df`).
+
+## Table Schema Format
+
+Store in `tables/*.yaml`:
+
+```yaml
+name: orders
+schema: public
+database_id: 19
+table_id: 1042
+columns:
+  - name: id
+    type: integer
+    description: Primary key
+  - name: customer_id
+    type: integer
+    description: Foreign key to customers
+  - name: product_id
+    type: integer
+  - name: total
+    type: decimal
+    description: Order total in USD
+  - name: status
+    type: varchar
+    description: "completed, pending, cancelled"
+  - name: created_at
+    type: timestamp
+```
+
+## workspace.yaml Format
+
+```yaml
+id: 2856
+name: "Customer Analysis"
+database_id: 19
+status: ready
+metabase_url: http://localhost:3000
+```
+
+## .env Format
+
+```
+METABASE_API_KEY=mb_abc123def456...
+```
+
+## Fetching Table Metadata from API
+
+```bash
+API_KEY=$(grep METABASE_API_KEY .env | cut -d= -f2-)
+URL=$(grep '^metabase_url:' workspace.yaml | awk '{print $2}')
+TABLE_ID=1042
+
+curl -s -H "x-api-key: $API_KEY" "$URL/api/table/$TABLE_ID/query_metadata" | \
+  jq '{
+    name: .name,
+    schema: .schema,
+    database_id: .db_id,
+    table_id: .id,
+    columns: [.fields[] | {name: .name, type: .base_type, description: .description}]
+  }'
+```
+
+## Best Practices
+
+1. **Filenames**: Use descriptive names
+   - Questions: `q01_<topic>.sql`, `q02_<topic>.sql`
+   - Transforms: `<descriptive_name>.sql` or `.py`
+
+2. **ref_id**: Never manually set or modify - let the sync process manage it
+
+3. **Questions vs Transforms**:
+   - `questions/` = Analysis outputs, may or may not merge
+   - `transforms/` = Pipeline building blocks, intended for production
+
+4. **Table schemas**: Keep `tables/` updated for context when writing queries

.claude/skills/workspace-sync/SKILL.md

@@ -0,0 +1,189 @@
+---
+name: workspace-sync
+description: Sync local transform files with the workspace API
+allowed-tools:
+  - Read
+  - Write
+  - Edit
+  - Bash
+  - Glob
+  - Grep
+---
+
+# Workspace Sync
+
+Syncs local transform files with the Metabase workspace API.
+
+## Behavior
+
+- **Local files are source of truth** - API state is overwritten
+- **New files** (no `ref_id`): POST to create, then update file with returned `ref_id`
+- **Existing files** (has `ref_id`): PUT to update
+- **Deleted files**: Transforms in API but not locally are deleted
+
+## Reading Configuration
+
+```bash
+# Read .env properly (handles values containing =)
+API_KEY=$(grep METABASE_API_KEY .env | cut -d= -f2-)
+WS_ID=$(grep '^id:' workspace.yaml | awk '{print $2}')
+DB_ID=$(grep '^database_id:' workspace.yaml | awk '{print $2}')
+URL=$(grep '^metabase_url:' workspace.yaml | awk '{print $2}')
+```
+
+## Syncing a SQL File
+
+```bash
+FILE="questions/q01_revenue.sql"
+
+# Extract metadata from header comments
+NAME=$(grep '^-- name:' "$FILE" | sed 's/^-- name: //')
+DESC=$(grep '^-- description:' "$FILE" | sed 's/^-- description: //')
+SCHEMA=$(grep '^-- target_schema:' "$FILE" | sed 's/^-- target_schema: //')
+TABLE=$(grep '^-- target_table:' "$FILE" | sed 's/^-- target_table: //')
+REF_ID=$(grep '^-- ref_id:' "$FILE" | sed 's/^-- ref_id: //')
+
+# Extract SQL body (everything after header comments)
+SQL=$(sed -n '/^[^-]/,$p' "$FILE" | jq -Rs .)
+
+# Build JSON request
+JSON=$(cat <<EOF
+{
+  "name": "$NAME",
+  "description": "$DESC",
+  "source": {
+    "type": "query",
+    "query": {
+      "type": "native",
+      "native": {"query": $SQL},
+      "database": $DB_ID
+    }
+  },
+  "target": {
+    "type": "table",
+    "schema": "$SCHEMA",
+    "name": "$TABLE"
+  }
+}
+EOF
+)
+
+if [ -z "$REF_ID" ]; then
+  # CREATE: POST new transform
+  RESPONSE=$(curl -s -X POST \
+    -H "x-api-key: $API_KEY" \
+    -H "Content-Type: application/json" \
+    -d "$JSON" \
+    "$URL/api/ee/workspace/$WS_ID/transform")
+
+  # Extract ref_id from response and add to file
+  NEW_REF_ID=$(echo "$RESPONSE" | jq -r '.ref_id')
+  if [ "$NEW_REF_ID" != "null" ]; then
+    # Insert ref_id after target_table line
+    sed -i '' "/^-- target_table:/a\\
+-- ref_id: $NEW_REF_ID
+" "$FILE"
+    echo "Created: $FILE -> $NEW_REF_ID"
+  else
+    echo "Error creating $FILE: $RESPONSE"
+  fi
+else
+  # UPDATE: PUT existing transform
+  curl -s -X PUT \
+    -H "x-api-key: $API_KEY" \
+    -H "Content-Type: application/json" \
+    -d "$JSON" \
+    "$URL/api/ee/workspace/$WS_ID/transform/$REF_ID"
+  echo "Updated: $FILE ($REF_ID)"
+fi
+```
+
+## Syncing a Python File
+
+```bash
+FILE="transforms/cohorts.py"
+
+# Extract metadata
+NAME=$(grep '^# name:' "$FILE" | sed 's/^# name: //')
+DESC=$(grep '^# description:' "$FILE" | sed 's/^# description: //')
+SOURCE_TABLES=$(grep '^# source_tables:' "$FILE" | sed 's/^# source_tables: //')
+SCHEMA=$(grep '^# target_schema:' "$FILE" | sed 's/^# target_schema: //')
+TABLE=$(grep '^# target_table:' "$FILE" | sed 's/^# target_table: //')
+REF_ID=$(grep '^# ref_id:' "$FILE" | sed 's/^# ref_id: //')
+
+# Extract Python body (everything after header comments)
+BODY=$(sed -n '/^[^#]/,$p' "$FILE" | jq -Rs .)
+
+JSON=$(cat <<EOF
+{
+  "name": "$NAME",
+  "description": "$DESC",
+  "source": {
+    "type": "python",
+    "source-tables": $SOURCE_TABLES,
+    "body": $BODY
+  },
+  "target": {
+    "type": "table",
+    "schema": "$SCHEMA",
+    "name": "$TABLE"
+  }
+}
+EOF
+)
+
+# Same POST/PUT logic as SQL...
+```
+
+## Deleting Orphaned Transforms
+
+```bash
+# Get all ref_ids from API
+API_REFS=$(curl -s -H "x-api-key: $API_KEY" \
+  "$URL/api/ee/workspace/$WS_ID/transform" | jq -r '.transforms[].ref_id')
+
+# Get all ref_ids from local files
+LOCAL_REFS=$(grep -rh '^-- ref_id:\|^# ref_id:' questions/ transforms/ 2>/dev/null | \
+  sed 's/^-- ref_id: //; s/^# ref_id: //')
+
+# Find orphans (in API but not local)
+for ref in $API_REFS; do
+  if ! echo "$LOCAL_REFS" | grep -q "^$ref$"; then
+    curl -s -X DELETE \
+      -H "x-api-key: $API_KEY" \
+      "$URL/api/ee/workspace/$WS_ID/transform/$ref"
+    echo "Deleted orphan: $ref"
+  fi
+done
+```
+
+## Full Sync Script
+
+```bash
+#!/bin/bash
+set -e
+
+API_KEY=$(grep METABASE_API_KEY .env | cut -d= -f2-)
+WS_ID=$(grep '^id:' workspace.yaml | awk '{print $2}')
+DB_ID=$(grep '^database_id:' workspace.yaml | awk '{print $2}')
+URL=$(grep '^metabase_url:' workspace.yaml | awk '{print $2}')
+
+echo "Syncing to workspace $WS_ID..."
+
+# Sync all SQL files
+for f in questions/*.sql transforms/*.sql 2>/dev/null; do
+  [ -f "$f" ] || continue
+  # ... sync logic
+done
+
+# Sync all Python files
+for f in transforms/*.py 2>/dev/null; do
+  [ -f "$f" ] || continue
+  # ... sync logic
+done
+
+# Delete orphans
+# ... orphan deletion logic
+
+echo "Sync complete"
+```

.claude/skills/workspace-transforms/SKILL.md

@@ -0,0 +1,597 @@
+---
+name: workspace-transforms
+description: Develop and debug workspace transforms, or perform ad-hoc data analysis using the Metabase enterprise API for isolated transform editing, testing, and merging
+---
+
+# Workspace Transforms
+
+This skill covers the Metabase workspace and transform APIs -- creating, testing, running, and debugging transforms in isolated workspaces. Use workspaces for developing transform pipelines before merging to production, or for ad-hoc data analysis and exploration.
+
+## Key Rules
+
+**These rules are mandatory. Follow them for every transform operation.**
+
+1. **User questions → Output transforms.** When the user asks a question that produces an answer, create an `Output N: <Title>` transform, NOT a scratch transform. Scratch is only for exploring what data looks like before you know what to ask.
+
+2. **Always build on existing transforms.** Before creating ANY new transform, list existing transforms and check if any already has the data you need. Read from existing transform outputs instead of re-querying source tables.
+
+3. **Never duplicate queries -- extract intermediates eagerly.** If you're about to write a subquery that duplicates logic from another transform, STOP. Extract the shared computation into an intermediate transform first, then have both transforms read from it.
+
+4. **Delete scratch transforms immediately.** After a scratch transform has served its exploratory purpose, delete it before moving on. Don't leave scratch transforms lying around.
+
+5. **Present results in chat.** After running a transform, show the results as an ASCII table in the conversation. The chat is the primary interface; the workspace view is for auditing.
+
+6. **Always specify schema in targets.** Use `"target": { "type": "table", "schema": "public", "name": "..." }` -- don't rely on defaults.
+
+7. **Never reference isolated schemas.** Always use target names (e.g., `FROM cleaned_reviews`). The isolation schema is an internal detail.
+
+8. **Keep JSON on one line.** The entire `-d '{...}'` body must be on a single line. Don't let the JSON wrap mid-string - use `\n` for newlines in SQL, not actual line breaks.
+
+## Getting Started
+
+### What you need a workspace for
+
+Workspaces provide an isolated environment for:
+- **Ad-hoc data analysis** -- explore data, run queries, and iterate without affecting production
+- **Transform development** -- build and test data pipelines before merging
+
+### Setting up a workspace
+
+**If you don't have a workspace yet:**
+
+1. Check if you have superuser access by attempting to list workspaces:
+   ```bash
+   curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/"
+   ```
+
+2. If you're a superuser, create an empty workspace:
+   ```bash
+   curl -s -X POST -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+     -d '{"name": "My Analysis Workspace"}' \
+     "http://localhost:3000/api/ee/workspace/"
+   ```
+   The database is determined automatically when you create your first transform.
+
+3. If you're not a superuser (403 error), ask an admin to create a workspace for you and share the workspace ID.
+
+**If you have a workspace:**
+
+Ask the user for:
+- Workspace ID
+- API key (the user will provide this explicitly)
+
+### Check workspace status
+
+```bash
+curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/WS"
+```
+
+The workspace should be in `ready` status. Key fields:
+- `database_id` -- the database this workspace operates on (set by first transform)
+- `status` -- should be `ready` (or `uninitialized` if no transforms yet)
+
+## Data Discovery
+
+Before starting analysis, discover what data is available.
+
+### Find the database from workspace
+
+```bash
+# Get workspace info and note the database_id field
+curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/WS" | jq .database_id
+```
+
+### List available tables
+
+```bash
+# Get database metadata with tables (filter to public schema)
+curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/database/DB_ID/metadata" \
+  | jq '.tables[] | select(.schema == "public") | {id, name}'
+```
+
+### Inspect table schema
+
+```bash
+# Get column info for a specific table
+curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/table/TABLE_ID/query_metadata" \
+  | jq '.fields[] | {name, base_type}'
+```
+
+### Explore data with dry-run
+
+Create a quick exploratory transform to preview data:
+
+```bash
+# Create a scratch transform to explore a table
+curl -s -X POST -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+  -d '{"name":"scratch_explore_users","source":{"type":"query","query":{"database":DB_ID,"type":"native","native":{"query":"SELECT * FROM users LIMIT 100"}}},"target":{"type":"table","schema":"public","name":"scratch_explore_users"}}' \
+  "http://localhost:3000/api/ee/workspace/WS/transform"
+
+# Dry-run to see data without persisting
+curl -s -X POST -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID/dry-run"
+```
+
+Use dry-run liberally -- it's fast and doesn't persist anything.
+
+## Data Analysis Workflow
+
+### Exploring with dry-run
+
+Dry-run is your primary exploration tool. It:
+- Executes the query and returns preview rows
+- Rolls back without persisting any data
+- Provides column metadata in the response
+
+Use it to iterate quickly on queries before committing to a transform structure.
+
+### Presenting results
+
+After running a dry-run or transform, **present results directly in chat** as an ASCII table. The conversation is the primary interaction surface -- the workspace view is for auditing and merging.
+
+Example output format:
+```
+| id | name        | total_orders |
+|----|-------------|--------------|
+| 1  | Alice Smith | 42           |
+| 2  | Bob Jones   | 17           |
+| 3  | Carol White | 89           |
+```
+
+Always show the results to the user; don't just link to the workspace.
+
+### Naming conventions
+
+Use consistent naming to keep your workspace organized:
+
+| Type | Convention | Example |
+|------|------------|---------|
+| **Final outputs** | `Output N: <Descriptive Title>` | `Output 1: Rating Variance by Reviewer Initial` |
+| **Intermediate transforms** | Descriptive snake_case | `cleaned_reviews`, `variance_by_prefix` |
+| **Scratch/exploratory** | Prefix with `scratch_` | `scratch_explore_orders`, `scratch_test_join` |
+
+### Always specify schema in targets
+
+When creating transforms, always specify the schema explicitly in the target:
+
+```json
+"target": { "type": "table", "schema": "public", "name": "variance_by_prefix" }
+```
+
+Don't rely on the default schema -- be explicit.
+
+### When to use scratch vs. Output
+
+**CRITICAL:** Use the right transform type:
+
+- **`Output N:`** -- Use when the user asks a question that produces an answer. This is a final deliverable. Most user questions should result in an Output.
+- **`scratch_`** -- Use ONLY for exploratory queries where you're discovering what data looks like before formulating the actual question. Examples: "what columns does this table have?", "what does the raw data look like?", "testing if this join works".
+
+**If the user asks a question, create an Output, not a scratch transform.** The question "what is the mean variance?" is a question with an answer → Output. "Let me see what the reviews table looks like" is exploration → scratch.
+
+### Cleaning up scratch transforms
+
+Delete scratch transforms **immediately after they've served their purpose**, not just at end of session. After you've used a scratch transform to explore and understand the data, delete it before moving on:
+
+```bash
+curl -s -X DELETE -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/transform/SCRATCH_REF_ID"
+```
+
+Scratch transforms should be short-lived. A workspace with many old scratch transforms is poorly maintained.
+
+### Using descriptions for findings
+
+Store analysis notes and findings in transform descriptions:
+
+```bash
+curl -s -X PUT -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+  -d '{"description": "Finding: 23% of reviews have rating variance > 2 std devs."}' \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID"
+```
+
+This keeps your discoveries attached to the transforms that produced them.
+
+### When to create vs. edit transforms
+
+- **New transform**: Create a new transform when asking a semantically different question (different data, different aggregation, different purpose)
+- **Edit in place**: Update an existing transform for bug fixes, tweaks, or refinements to the same question
+
+### Reusing existing transforms (MANDATORY)
+
+**Before creating ANY new transform, you MUST:**
+
+1. **List existing transforms** in the workspace:
+   ```bash
+   curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/WS/transform" \
+     | jq '.[] | {name, ref_id, target: .target.name}'
+   ```
+2. **Check if any existing transform already has the data you need** -- if so, read from its output table instead of re-querying source tables
+
+This is not optional. Don't re-derive data that already exists in a transform output.
+
+### Extract intermediates eagerly (don't duplicate queries)
+
+**Before writing any new transform, check:** Am I about to duplicate a subquery or CTE that exists in another transform?
+
+If yes, **STOP and refactor first:**
+
+1. Extract the shared computation into a new intermediate transform
+2. Update the existing transform to read from the intermediate
+3. Write the new transform to also read from the intermediate
+
+**Never duplicate a subquery across transforms.** If Output 1 computes variance with a complex GROUP BY, and Output 2 needs that same data, don't copy the GROUP BY -- extract it.
+
+Example -- WRONG (duplicated logic):
+```sql
+-- Output 1
+SELECT prefix, AVG(rating), VAR_POP(rating)
+FROM reviews GROUP BY prefix
+
+-- Output 2 (BAD: duplicates the aggregation)
+SELECT AVG(variance) FROM (
+  SELECT prefix, VAR_POP(rating) as variance
+  FROM reviews GROUP BY prefix
+)
+```
+
+Example -- CORRECT (shared intermediate):
+```
+variance_by_prefix              ← intermediate (holds the GROUP BY)
+  ├── Output 1: Variance Details    ← SELECT * FROM variance_by_prefix
+  └── Output 2: Variance Summary    ← SELECT AVG(variance) FROM variance_by_prefix
+```
+
+This keeps Outputs as leaf nodes that present data, while intermediates hold the reusable computations.
+
+### Outputs should not depend on other Outputs
+
+If a new Output needs data computed by an existing Output, don't make Output 2 depend on Output 1. Instead, extract the shared data into an intermediate, then have both Outputs read from it.
+
+### Refactoring shared logic
+
+If multiple transforms share the same cleaning/filtering/joining logic, extract it into a dedicated intermediate transform. Downstream transforms then read from that intermediate.
+
+When referencing another transform's output, use the **target name** (e.g., `cleaned_reviews`), never the isolated schema. The isolation schema is an internal implementation detail.
+
+### Breaking complex analysis into phases
+
+For complex questions, build up in sanity-checkable phases:
+
+1. **Data cleaning** -- Create transforms that clean/normalize source data
+2. **Aggregation** -- Build intermediate aggregations you can verify
+3. **Final analysis** -- Combine intermediates into final outputs
+
+Run each phase and verify results before proceeding. This makes debugging easier when something looks wrong.
+
+Example progression:
+```
+scratch_raw_sample      → Verify source data looks right, then DELETE
+cleaned_reviews         → Verify cleaning logic (intermediate, keep)
+variance_by_prefix      → Verify grouping is correct (intermediate, keep)
+Output 1: Variance Details  → Final answer
+```
+
+### Session cleanup
+
+Before ending an analysis session:
+
+1. List all transforms:
+   ```bash
+   curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/WS/transform" \
+     | jq '.[] | {name, ref_id}'
+   ```
+
+2. Verify no scratch transforms remain (they should have been deleted as you went -- see "Cleaning up scratch transforms" above)
+
+3. If the analysis is complete and you don't need the workspace anymore, archive it:
+   ```bash
+   curl -s -X POST -H "x-api-key: API_KEY" \
+     "http://localhost:3000/api/ee/workspace/WS/archive"
+   ```
+
+## Pipeline Development Workflow
+
+### Create and test a transform
+
+```bash
+# 1. Check workspace is usable
+curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/WS/transform"
+
+# 2. Create a transform (pretty-print SQL, always specify schema)
+curl -s -X POST -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+  -d '{"name":"Active Users","description":"Summarizes active user accounts","source":{"type":"query","query":{"database":19,"type":"native","native":{"query":"SELECT id, email, first_name FROM core_user WHERE is_active = true"}}},"target":{"type":"table","schema":"public","name":"active_users"}}' \
+  "http://localhost:3000/api/ee/workspace/WS/transform"
+# Response includes ref_id, target_isolated, target_stale
+
+# 3. Dry-run to preview results (no persistence)
+curl -s -X POST -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID/dry-run"
+# Returns: { status, data: { rows, cols, results_metadata } }
+
+# 4. Iterate -- update the transform and dry-run again
+curl -s -X PUT -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+  -d '{"source":{"type":"query","query":{"database":19,"type":"native","native":{"query":"SELECT id, email, first_name, last_login FROM core_user WHERE is_active = true ORDER BY last_login DESC"}}}}' \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID"
+
+# 5. Run to persist to isolated schema
+curl -s -X POST -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID/run"
+# Returns: { status, start_time, end_time, table: { name, schema } }
+```
+
+### Debug a failing transform
+
+```bash
+# 1. Dry-run to see the error message
+curl -s -X POST -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID/dry-run"
+# Failed response: { "status": "failed", "message": "ERROR: ..." }
+
+# 2. Check transform details for last run info
+curl -s -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/transform/REF_ID"
+# Look at: last_run_status, last_run_message, target_stale
+
+# 3. Check workspace-level problems
+curl -s -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/problem"
+
+# 4. Inspect the dependency graph
+curl -s -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/graph"
+
+# 5. Check setup/execution logs
+curl -s -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/log"
+```
+
+### Run all and prepare for merge
+
+```bash
+# Run all transforms in dependency order (skip up-to-date ones)
+curl -s -X POST -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/run?stale_only=1"
+# Returns: { succeeded: [...], failed: [...], not_run: [...] }
+
+# Verify no merge-blocking problems exist
+curl -s -H "x-api-key: API_KEY" \
+  "http://localhost:3000/api/ee/workspace/WS/problem"
+
+# Merge to production (superuser only)
+curl -s -X POST -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+  -d '{"commit-message": "Add user analytics transforms"}' \
+  "http://localhost:3000/api/ee/workspace/WS/merge"
+# Returns: { merged: [{ op, global_id, ref_id }], workspace: { id, name } }
+```
+
+### Work with Python transforms
+
+```bash
+curl -s -X POST -H "x-api-key: API_KEY" -H "Content-Type: application/json" \
+  -d '{"name":"User Order Summary","source":{"type":"python","source-database":19,"source-tables":{"users":1015,"orders":1042},"body":"import pandas as pd\nresult = users_df.merge(orders_df, on='\''user_id'\'')"},"target":{"type":"table","schema":"public","name":"user_order_summary"}}' \
+  "http://localhost:3000/api/ee/workspace/WS/transform"
+```
+
+Python transforms receive each source table as a pandas DataFrame named `<table_name>_df`. The result must be a DataFrame assigned to `result`.
+
+## Best Practices
+
+### Keep queries simple and focused
+
+Each transform should do one thing well. If a query is getting complex, break it into multiple transforms.
+
+### Pretty-print SQL
+
+Format SQL queries for readability with proper indentation and line breaks:
+
+```sql
+SELECT
+  u.id,
+  u.name,
+  COUNT(o.id) AS order_count,
+  SUM(o.total) AS total_spent
+FROM users u
+LEFT JOIN orders o ON o.user_id = u.id
+WHERE u.is_active = true
+GROUP BY u.id, u.name
+ORDER BY total_spent DESC
+```
+
+### Factor out data cleaning
+
+Create dedicated cleaning transforms for:
+- Null handling
+- Type conversions
+- Deduplication
+- Filtering invalid records
+
+Other transforms can then depend on the cleaned data.
+
+### Avoid duplicated boilerplate
+
+If you're copying the same CTE or subquery into multiple transforms:
+1. Extract it into its own transform
+2. Have downstream transforms read from that output table (using the target name)
+
+This keeps logic in one place and makes updates easier.
+
+### Always use target names, not isolated schemas
+
+When one transform reads from another's output, reference the **target name** (e.g., `SELECT * FROM cleaned_reviews`). Never query the isolated schema directly -- it's an internal implementation detail that should not appear in transform queries.
+
+### Build reusable intermediates
+
+When patterns emerge across multiple analyses:
+- Create well-named intermediate transforms (not `scratch_*`)
+- Add clear descriptions explaining what they contain
+- Run them (not just dry-run) so downstream transforms can reference their output
+
+## Conceptual Model
+
+### What workspaces are
+
+A workspace is an isolated environment for developing transforms -- like a branch in git. You create or check out transforms inside a workspace, iterate on them with dry-runs and test executions, and merge when ready. What happens in a workspace stays in the workspace until you merge.
+
+### Isolation
+
+Every workspace gets its own database schema (e.g., `mb__isolation_ab493_2849`). When you run a transform in a workspace, the output table is written to this isolated schema, not to the global target location. This means:
+
+- You can freely run transforms without affecting production data
+- Multiple workspaces can work on the same transform independently
+- Dry-runs don't persist anything at all -- they return preview rows and roll back
+
+**Important:** The isolation schema is an internal implementation detail. Always write queries that reference target names (e.g., `cleaned_reviews`), not isolated schema paths. The system handles resolution automatically.
+
+### Transforms
+
+A transform takes a source (SQL query or Python code) and writes the result to a target table. Transforms can depend on each other -- if Transform B reads from Transform A's output table, that's a dependency.
+
+**Source types in workspaces:**
+- **Native SQL** (`"type": "query"` with `"type": "native"` inside) -- most common
+- **Python** (`"type": "python"`) -- for pandas-style data manipulation
+- MBQL and card references are **not supported** in workspaces
+
+**Target types:**
+- `"table"` -- full replacement on each run
+- `"table-incremental"` with checkpoint strategy -- append-only, tracks progress via a checkpoint column
+
+### Dependency graph
+
+Workspaces maintain a dependency graph with three node types:
+- **Input tables** -- source data that transforms read from
+- **Workspace transforms** -- transforms being developed in this workspace
+- **External transforms** -- global transforms that workspace transforms depend on
+
+The graph determines execution order (topological sort) and is used to detect problems like cycles and conflicts.
+
+### Merge
+
+Merging promotes workspace transform definitions to global scope. The merge process:
+1. Validates that no merge-blocking problems exist (cycles, conflicts, broken downstream dependencies)
+2. Creates or updates global transform records from workspace versions
+3. Archives the workspace
+4. Records an audit trail (workspace merge history)
+
+Non-blocking problems (like unused transforms) produce warnings but don't prevent merging.
+
+### Staleness
+
+A transform is "stale" when any of the following is true:
+1. It has never run
+2. Its definition has changed since its last successful run
+3. A table it depends upon has since been updated by another transform
+4. Any of its ancestor transforms is stale
+
+The `stale_only` flag on the run-all endpoint will skip transforms that are already up to date, performing the minimal amount of work to ensure that there are no longer any stale transforms.
+
+### Identity
+
+Workspace transforms are identified by a `ref_id` -- a human-readable slug like `lucid-ferret-a852`. This is workspace-scoped, not globally unique. A transform may also have a `global_id` if it was checked out from an existing global transform.
+
+## Authentication
+
+Pass an API key via header:
+
+```bash
+curl -s -H "x-api-key: API_KEY" "http://localhost:3000/api/ee/workspace/WS"
+```
+
+The user will provide the API key explicitly.
+
+**Access model:**
+- **Superuser** -- full access to everything, required for merge operations
+- **Workspace service user** -- can read workspace state, manage and run transforms (matched by the workspace's `execution_user` field)
+- **Data analyst** -- can manage global transforms (needs read permission on source tables)
+
+## API Reference
+
+### Workspace Endpoints
+
+| Method | Path | Purpose |
+|--------|------|---------|
+| `GET` | `/api/ee/workspace/` | List workspaces (superuser) |
+| `POST` | `/api/ee/workspace/` | Create workspace (superuser) |
+| `GET` | `/api/ee/workspace/:id` | Get workspace |
+| `PUT` | `/api/ee/workspace/:id` | Update workspace (superuser) |
+| `POST` | `/api/ee/workspace/:id/archive` | Archive (superuser) |
+| `POST` | `/api/ee/workspace/:id/unarchive` | Unarchive (superuser) |
+| `DELETE` | `/api/ee/workspace/:id` | Delete archived workspace (superuser) |
+| `GET` | `/api/ee/workspace/:id/table` | List input/output tables |
+| `GET` | `/api/ee/workspace/:id/graph` | Dependency graph |
+| `GET` | `/api/ee/workspace/:id/log` | Setup and execution logs |
+| `GET` | `/api/ee/workspace/:id/problem` | Detect problems |
+| `POST` | `/api/ee/workspace/:id/run` | Run all transforms |
+| `POST` | `/api/ee/workspace/:id/merge` | Merge all to production (superuser) |
+| `GET` | `/api/ee/workspace/:id/external/transform` | Global transforms for checkout |
+| `GET` | `/api/ee/workspace/enabled` | Check workspace support |
+| `GET` | `/api/ee/workspace/database` | Database support info |
+| `GET` | `/api/ee/workspace/checkout?transform-id=N` | Checkout info for a global transform |
+
+### Workspace Transform Endpoints
+
+| Method | Path | Purpose |
+|--------|------|---------|
+| `GET` | `/api/ee/workspace/:ws/transform` | List transforms |
+| `POST` | `/api/ee/workspace/:ws/transform` | Create transform |
+| `GET` | `/api/ee/workspace/:ws/transform/:ref` | Get transform |
+| `PUT` | `/api/ee/workspace/:ws/transform/:ref` | Update transform |
+| `DELETE` | `/api/ee/workspace/:ws/transform/:ref` | Delete transform |
+| `POST` | `/api/ee/workspace/:ws/transform/:ref/run` | Run transform |
+| `POST` | `/api/ee/workspace/:ws/transform/:ref/dry-run` | Preview transform |
+| `POST` | `/api/ee/workspace/:ws/transform/:ref/merge` | Merge single transform (superuser) |
+| `POST` | `/api/ee/workspace/:ws/transform/:ref/archive` | Mark for archival on merge |
+| `POST` | `/api/ee/workspace/:ws/transform/:ref/unarchive` | Recall archived transform |
+| `POST` | `/api/ee/workspace/:ws/transform/validate/target` | Validate target table |
+
+### Global Transform Endpoints
+
+| Method | Path | Purpose |
+|--------|------|---------|
+| `GET` | `/api/ee/transform/` | List transforms (filterable) |
+| `POST` | `/api/ee/transform/` | Create transform |
+| `GET` | `/api/ee/transform/:id` | Get transform |
+| `PUT` | `/api/ee/transform/:id` | Update transform |
+| `DELETE` | `/api/ee/transform/:id` | Delete transform |
+| `DELETE` | `/api/ee/transform/:id/table` | Delete output table |
+| `POST` | `/api/ee/transform/:id/run` | Run (async, returns run_id) |
+| `POST` | `/api/ee/transform/:id/cancel` | Cancel running transform |
+| `GET` | `/api/ee/transform/:id/dependencies` | Get dependencies |
+| `GET` | `/api/ee/transform/:id/merge-history` | Merge audit trail |
+| `GET` | `/api/ee/transform/run` | Query transform runs (filterable) |
+| `POST` | `/api/ee/transform/extract-columns` | Get checkpoint-eligible columns |
+| `POST` | `/api/ee/transform/is-simple-query` | Check if SQL is simple |
+
+### Transform Job Endpoints
+
+| Method | Path | Purpose |
+|--------|------|---------|
+| `GET` | `/api/ee/transform-job/` | List scheduled jobs |
+| `POST` | `/api/ee/transform-job/` | Create job (cron schedule) |
+| `GET` | `/api/ee/transform-job/:id` | Get job |
+| `PUT` | `/api/ee/transform-job/:id` | Update job |
+| `DELETE` | `/api/ee/transform-job/:id` | Delete job |
+| `POST` | `/api/ee/transform-job/:id/run` | Manually trigger job |
+| `GET` | `/api/ee/transform-job/:id/transforms` | List transforms in job |
+
+## Key Codebase Locations
+
+| Area | Path |
+|------|------|
+| Workspace API | `enterprise/backend/src/metabase_enterprise/workspaces/api.clj` |
+| Transform API | `enterprise/backend/src/metabase_enterprise/transforms/api.clj` |
+| Transform Job API | `enterprise/backend/src/metabase_enterprise/transforms/api/transform_job.clj` |
+| Workspace models | `enterprise/backend/src/metabase_enterprise/workspaces/models/` |
+| Transform models | `enterprise/backend/src/metabase_enterprise/transforms/models/` |
+| Workspace types | `enterprise/backend/src/metabase_enterprise/workspaces/types.clj` |
+| Transform schemas | `enterprise/backend/src/metabase_enterprise/transforms/schema.clj` |
+| Route registration | `enterprise/backend/src/metabase_enterprise/api_routes/routes.clj` |
+
+## Gotchas
+
+- **Keep curl commands on one line:** The entire `-d '...'` JSON body must be on a single line (or use `\` continuation properly). Do NOT let the JSON wrap with actual newlines inside string values - this breaks JSON parsing.
+- **Escape single quotes for bash:** Since curl JSON is wrapped in bash single quotes, any single quote in your SQL or Python code needs escaping. Use `'\''` for each single quote. Example: `!= ''` becomes `!= '\'''\''` and `on='user_id'` becomes `on='\''user_id'\''`.
+- **Always include `schema` in targets:** The API may reject targets without an explicit schema. Always specify it: `"target": { "type": "table", "schema": "public", "name": "..." }`.
+- **Workspace auto-initializes:** An `uninitialized` workspace transitions to `ready` when you create the first transform. No manual initialization step is needed.
+- **Database set by first transform:** An empty workspace doesn't have a database yet. The database is determined when you create the first transform (from the query's database field).
+- **Never query isolated schemas:** Always reference target names in your SQL (e.g., `FROM cleaned_reviews`). The isolation schema is an internal detail -- the system resolves references automatically.

CLAUDE.md

@@ -23,6 +23,14 @@ For detailed guidance on writing and reviewing code and documentation, see the s
 
 - **[analytics-events](.claude/skills/analytics-events/SKILL.md)** - Add product analytics events to track user interactions
 
+## Workspaces & Transforms
+
+- **[workspace-admin](.claude/skills/workspace-admin/SKILL.md)** - Initialize analysis projects and merge to production (superuser operations)
+- **[workspace-sync](.claude/skills/workspace-sync/SKILL.md)** - Sync local transform files with the workspace API
+- **[workspace-files](.claude/skills/workspace-files/SKILL.md)** - Manage local file representations of transforms and table schemas
+- **[workspace-transforms](.claude/skills/workspace-transforms/SKILL.md)** - Develop and debug workspace transforms, or perform ad-hoc data analysis using the Metabase enterprise API for isolated transform editing, testing, and merging
+- **[data-workspace](.claude/skills/data-workspace/SKILL.md)** - Unified workflow for data analysis and pipeline building in workspaces
+
 ## Tool Preferences
 
 If `clojure-mcp` tools are available, prefer them over shell-based alternatives for Clojure development.

enterprise/backend/src/metabase_enterprise/workspaces/api.clj

@@ -53,7 +53,10 @@
    [:database_id ::ws.t/appdb-id]
    [:status ::status]
    [:created_at ms/TemporalInstant]
-   [:updated_at ms/TemporalInstant]])
+   [:updated_at ms/TemporalInstant]
+   ;; Only present in POST response - the unmasked API key for workspace service user.
+   ;; This is the only time the key is available; after creation it's hashed and unrecoverable.
+   [:api_key {:optional true} :string]])
 
 ;; Transform-related schemas (adapted from transforms/api.clj)
 ;; TODO (Chris 2026-02-02) -- We should reuse these schemas, by exposing common types from the transforms module. They *can* match exactly.
@@ -170,10 +173,11 @@
   (routes.common/wrap-middleware-for-open-api-spec-generation authorize*))
 
 (defn- ws->response
-  "Transform a workspace record into an API response, computing the backwards-compatible status."
+  "Transform a workspace record into an API response, computing the backwards-compatible status.
+   If :api_key is present (only at creation time), it is preserved in the response."
   [ws]
   (-> ws
-      (select-keys [:id :name :collection_id :database_id :created_at :updated_at])
+      (select-keys [:id :name :collection_id :database_id :created_at :updated_at :api_key])
       (assoc :status (ws.model/computed-status ws))))
 
 ;;; routes
@@ -862,9 +866,12 @@
             [:source {:optional true} ::transform-source]
             [:target {:optional true} ::transform-target]]]
   (t2/with-transaction [_tx]
-    (api/check-404 (t2/select-one :model/WorkspaceTransform :ref_id tx-id :workspace_id ws-id))
-    (let [source-or-target-changed? (or (:source body) (:target body))]
-      (t2/update! :model/WorkspaceTransform {:workspace_id ws-id :ref_id tx-id} body)
+    (let [existing (api/check-404 (t2/select-one :model/WorkspaceTransform :ref_id tx-id :workspace_id ws-id))
+          ;; Merge incoming target with existing to preserve fields like :database when not explicitly provided
+          merged-body (cond-> body
+                        (:target body) (update :target #(merge (:target existing) %)))
+          source-or-target-changed? (or (:source body) (:target body))]
+      (t2/update! :model/WorkspaceTransform {:workspace_id ws-id :ref_id tx-id} merged-body)
       ;; If source or target changed, increment versions for re-analysis
       (when source-or-target-changed?
         (ws.impl/increment-analysis-version! ws-id tx-id)

enterprise/backend/src/metabase_enterprise/workspaces/common.clj

@@ -9,6 +9,7 @@
    [metabase.api.common :as api]
    [metabase.util.log :as log]
    [metabase.util.quick-task :as quick-task]
+   [metabase.util.secret :as u.secret]
    [toucan2.core :as t2]))
 
 (defn- extract-suffix-number
@@ -41,7 +42,8 @@
       (str stripped-name " (" next-num ")"))))
 
 (defn- create-workspace-container!
-  "Create the workspace and its related collection, user, and api key."
+  "Create the workspace and its related collection, user, and api key.
+   Returns the workspace with :api_key attached (unmasked, only available at creation time)."
   [creator-id db-id workspace-name]
   (let [key-name (format "API key for Workspace: %s" workspace-name)
         ;; Ensure the key-name is unique.
@@ -68,9 +70,9 @@
         ws      (assoc ws :collection_id (:id coll))]
     ;; Set the backlink from the workspace to the collection inside it and set the schema.
     (t2/update! :model/Workspace (:id ws) {:collection_id (:id coll)})
-    ;; TODO (Sanya 2025-11-18) -- For now we expose this in logs for manual testing. In future we need a secure channel.
-    (log/infof "Generated API key for workspace: %s" (:unmasked_key api-key))
-    ws))
+    ;; Return the workspace with the unmasked API key attached.
+    ;; This is the only time the key is available - after this it's hashed and unrecoverable.
+    (assoc ws :api_key (u.secret/expose (:unmasked_key api-key)))))
 
 (defn- unique-constraint-violation?
   "Check if an exception is due to a unique constraint violation."