I’m with you. Let’s wire the whole flow so uploads are snapshotted, deduped, and only re-run the AI pipeline when needed. We’ll keep your HF dataset store as the “remote S3,” maintain local snapshots, and call the strict Python Executor API to render charts from Agent 2’s python_snippet.

Overview
- Service A: Storage + Snapshotter (extends your HF uploader)
  - Handles uploads, computes fingerprint (sha256, shape, schema), dedup across renames, keeps a manifest of snapshots and artifacts, pushes to HF.
  - If identical snapshot already exists, it returns the previous AI artifacts without re-running.
  - If changed, it creates a new snapshot and triggers the pipeline.
- Service B: Analytics Orchestrator (Gemini + Qdrant + Chart Plan + Execution)
  - Runs Agent 1 (Profiler) and Agent 2 (Viz Planner) and stores artifacts (profile.json, plan.json, figures.json).
  - Calls the Strict Python Executor API (/execute) to run the python_snippet against the snapshot dataset.
- Service C: Strict Python Executor (already done above)
  - Executes python_snippet with df + plan and returns Plotly figure JSON.

Data and UX notes
- Renamed same file: dedup by content sha256 → reuse snapshot and all AI artifacts.
- Same shape (rows/cols) but data changed: new snapshot → re-run pipeline.
- Completely different dataset: new dataset key or new snapshot under a different key.
- Frontend “click columns”:
  - An API returns preprocessed slices (value counts, stats, crosstabs) for the clicked columns and highlights those in the charts.
  - We serve preprocessed data from the snapshot to avoid recomputing.

Service A: Storage + Snapshotter (FastAPI)
- Replaces/extends your uploader with fingerprinting, dedup, and snapshot registry.
- Still uses CommitScheduler to push to HF.

requirements.txt
- fastapi, uvicorn, pydantic, pandas, pyarrow, orjson, huggingface_hub

app/storage_main.py
```python
import os, io, json, uuid, time, hashlib, asyncio
from pathlib import Path
from typing import Any, Dict, Optional, List
from fastapi import FastAPI, UploadFile, File, HTTPException, Form
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from huggingface_hub import CommitScheduler
import pandas as pd

# ---------------- Config ----------------
HF_TOKEN    = os.environ.get("HF_TOKEN")
BACKUP_REPO = os.environ.get("HF_REPO", "triflix/database")
UPLOAD_DIR  = Path(os.environ.get("UPLOAD_DIR", "/tmp/uploadedfiles"))
DATA_ROOT   = Path(os.environ.get("DATA_ROOT", "/data"))  # local canonical storage
ARTIFACTS   = DATA_ROOT / "artifacts"                     # per-snapshot artifacts
REGISTRY    = DATA_ROOT / "manifest.json"                 # global manifest

UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
DATA_ROOT.mkdir(parents=True, exist_ok=True)
ARTIFACTS.mkdir(parents=True, exist_ok=True)
if not REGISTRY.exists():
    REGISTRY.write_text(json.dumps({"snapshots": {}, "by_sha": {}, "by_dataset": {}}, indent=2))

scheduler = CommitScheduler(
    repo_id      = BACKUP_REPO,
    repo_type    = "dataset",
    folder_path  = DATA_ROOT,    # push entire /data tree (datasets + artifacts)
    path_in_repo = "data",
    every        = 60*24*365,
    token        = HF_TOKEN,
)

app = FastAPI(title="Storage + Snapshotter")

# --------------- Models -----------------
class SnapshotMeta(BaseModel):
    snapshot_id: str
    dataset_key: str
    file_sha256: str
    file_name: str
    fmt: str
    n_rows: int
    n_cols: int
    columns: List[str]
    dtypes: Dict[str, str]
    sample_hash: str
    local_path: str
    created_at: float
    artifacts: Dict[str, str] = Field(default_factory=dict)  # profile.json, plan.json, figures.json

# --------------- Helpers ----------------
def load_manifest() -> Dict[str, Any]:
    try:
        return json.loads(REGISTRY.read_text())
    except Exception:
        return {"snapshots": {}, "by_sha": {}, "by_dataset": {}}

def save_manifest(man: Dict[str, Any]):
    tmp = REGISTRY.with_suffix(".tmp.json")
    tmp.write_text(json.dumps(man, indent=2))
    tmp.replace(REGISTRY)

def sha256_of_bytes(b: bytes) -> str:
    h = hashlib.sha256()
    h.update(b)
    return h.hexdigest()

def sha256_of_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024*1024), b""):
            h.update(chunk)
    return h.hexdigest()

def sniff_fmt(filename: str) -> str:
    lower = filename.lower()
    if lower.endswith(".csv"): return "csv"
    if lower.endswith(".parquet") or lower.endswith(".pq"): return "parquet"
    raise ValueError("Unsupported file type (csv/parquet only)")

def fingerprint_dataset(path: Path, fmt: str, sample_n: int = 5) -> Dict[str, Any]:
    if fmt == "csv":
        # Count rows quickly without loading full DF into memory
        with path.open("rb") as f:
            n_rows = sum(1 for _ in f) - 1  # minus header
        df_head = pd.read_csv(path, nrows=sample_n)
        # tail: read all but keep last 5 (OK for mid-size; switch to chunks for very large)
        df_tail = pd.read_csv(path).tail(sample_n)
        cols = list(df_head.columns)
        dtypes = {}
        for c in cols:
            try:
                dtypes[c] = str(pd.read_csv(path, nrows=1)[c].dtype)
            except Exception:
                dtypes[c] = "unknown"
        n_cols = len(cols)
    else:  # parquet
        df = pd.read_parquet(path)
        n_rows, n_cols = df.shape
        cols = list(df.columns)
        dtypes = {c: str(df[c].dtype) for c in cols}
        df_head = df.head(sample_n)
        df_tail = df.tail(sample_n)

    # sample hash from head+tail to catch trivial renames
    sample_blob = json.dumps({
        "head": df_head.to_dict(orient="records"),
        "tail": df_tail.to_dict(orient="records")
    }, ensure_ascii=False, default=str).encode("utf-8")
    sample_hash = sha256_of_bytes(sample_blob)

    return {
        "n_rows": int(n_rows),
        "n_cols": int(n_cols),
        "columns": cols,
        "dtypes": dtypes,
        "sample_hash": sample_hash
    }

def ensure_snapshot_dirs(snapshot_id: str) -> Path:
    sdir = DATA_ROOT / "snapshots" / snapshot_id
    sdir.mkdir(parents=True, exist_ok=True)
    (sdir / "artifacts").mkdir(parents=True, exist_ok=True)
    return sdir

# --------------- Upload + Snapshot logic ----------------
@app.on_event("shutdown")
def _stop_scheduler():
    scheduler.stop()

@app.post("/upload")
async def upload(
    file: UploadFile = File(...),
    dataset_key: Optional[str] = Form(None),
    trigger_pipeline: bool = Form(True),
):
    """
    - Saves file locally under /data/uploads
    - Computes sha256, shape, schema, sample hash
    - Dedups across renames using sha256
    - Returns snapshot + optionally triggers pipeline by HTTP call (deferred to Orchestrator)
    """
    try:
        # 1) Save temp
        tmp = UPLOAD_DIR / f"{uuid.uuid4()}_{file.filename}"
        content = await file.read()
        tmp.write_bytes(content)

        # 2) File meta
        fmt = sniff_fmt(file.filename)
        fsha = sha256_of_file(tmp)

        # 3) Dedup by sha
        man = load_manifest()
        existing_snapshot_id = man["by_sha"].get(fsha)

        if not dataset_key:
            # Use filename stem as default dataset key
            dataset_key = Path(file.filename).stem.lower().replace(" ", "_")

        if existing_snapshot_id:
            snap = man["snapshots"][existing_snapshot_id]
            # Also ensure dataset_key list contains this snapshot
            byds = man["by_dataset"].setdefault(dataset_key, [])
            if existing_snapshot_id not in byds:
                byds.append(existing_snapshot_id)
                save_manifest(man)
            # Cleanup temp (we already have canonical file)
            try: tmp.unlink()
            except: pass

            # Return existing snapshot (reuse artifacts)
            return {
                "status": "reused",
                "reason": "identical_content_sha",
                "dataset_key": dataset_key,
                "snapshot_id": existing_snapshot_id,
                "artifacts": snap.get("artifacts", {}),
                "meta": snap
            }

        # 4) New snapshot path
        snapshot_id = fsha[:32]
        sdir = ensure_snapshot_dirs(snapshot_id)
        dest = sdir / file.filename
        tmp.replace(dest)

        # 5) Fingerprint
        fp = fingerprint_dataset(dest, fmt)
        meta = SnapshotMeta(
            snapshot_id=snapshot_id,
            dataset_key=dataset_key,
            file_sha256=fsha,
            file_name=file.filename,
            fmt=fmt,
            n_rows=fp["n_rows"],
            n_cols=fp["n_cols"],
            columns=fp["columns"],
            dtypes=fp["dtypes"],
            sample_hash=fp["sample_hash"],
            local_path=str(dest),
            created_at=time.time(),
            artifacts={}
        ).model_dump()

        # 6) Update registry
        man["snapshots"][snapshot_id] = meta
        man["by_sha"][fsha] = snapshot_id
        man["by_dataset"].setdefault(dataset_key, []).append(snapshot_id)
        save_manifest(man)

        # 7) Push to HF (async)
        await asyncio.to_thread(scheduler.trigger)

        # 8) Optionally call the Orchestrator to prepare artifacts
        if trigger_pipeline:
            # POST to orchestrator (configure ORCH_URL env)
            orch = os.environ.get("ORCH_URL")
            if orch:
                import requests
                try:
                    requests.post(f"{orch}/prepare", json={
                        "dataset_key": dataset_key,
                        "snapshot_id": snapshot_id,
                        "local_path": meta["local_path"]
                    }, timeout=5)
                except Exception:
                    pass

        return {
            "status": "snapshot_created",
            "dataset_key": dataset_key,
            "snapshot_id": snapshot_id,
            "meta": meta
        }

    except Exception as e:
        raise HTTPException(500, f"Upload failed: {e}")

@app.get("/datasets/{dataset_key}/snapshots")
def list_snapshots(dataset_key: str):
    man = load_manifest()
    ids = man["by_dataset"].get(dataset_key, [])
    return {"dataset_key": dataset_key, "snapshots": [man["snapshots"][i] for i in ids]}

@app.get("/snapshots/{snapshot_id}")
def get_snapshot(snapshot_id: str):
    man = load_manifest()
    snap = man["snapshots"].get(snapshot_id)
    if not snap:
        raise HTTPException(404, "Snapshot not found")
    return snap
```

Key rules implemented
- Exact same file (sha256) → “reused” snapshot and artifacts. This covers rename uploads too.
- Otherwise we create a new snapshot (even if rows/cols match but values changed).
- Everything is recorded in /data/manifest.json; data is under /data/snapshots/<snapshot_id>/.

Service B: Analytics Orchestrator (FastAPI)
- Runs Agent 1 + Agent 2, stores artifacts, and calls Strict Python Executor.

requirements.txt
- fastapi, uvicorn, pydantic, pandas, numpy, plotly, google-generativeai, qdrant-client, orjson

app/orchestrator_main.py
```python
import os, json, time
from typing import Any, Dict, Optional
from pathlib import Path
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import pandas as pd

import google.generativeai as genai
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

# Reuse your LLM/Qdrant helpers from earlier response or import them as a module
# For brevity, define minimal here

GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
if not GEMINI_API_KEY:
    raise RuntimeError("Set GEMINI_API_KEY")
genai.configure(api_key=GEMINI_API_KEY)
TEXT_MODEL = os.environ.get("GEMINI_TEXT_MODEL", "gemini-1.5-pro")
EMBED_MODEL = "models/embedding-001"
EMBED_DIM = 768

QDRANT_URL = os.environ.get("QDRANT_URL", "http://localhost:6333")
ARTIFACTS_ROOT = Path(os.environ.get("ARTIFACTS_ROOT", "/data/artifacts"))
SNAPSHOTS_ROOT = Path(os.environ.get("SNAPSHOTS_ROOT", "/data/snapshots"))
EXEC_URL = os.environ.get("EXEC_URL", "http://python-exec-sandbox:8000")

app = FastAPI(title="Analytics Orchestrator")

def qdrant():
    c = QdrantClient(url=QDRANT_URL)
    # Ensure collection
    try:
        c.get_collection("dataset_profiles")
    except Exception:
        c.create_collection("dataset_profiles",
            vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.COSINE))
    return c

def embed_text(text: str, task: str) -> list[float]:
    return genai.embed_content(model=EMBED_MODEL, content=text, task_type=task)["embedding"]

def df_schema_summary(df: pd.DataFrame) -> Dict[str, Any]:
    cols = []
    for c in df.columns:
        s = df[c]
        cols.append({
            "name": c,
            "dtype": str(s.dtype),
            "unique_count": int(s.nunique(dropna=True)),
            "missing": int(s.isna().sum()),
            "missing_pct": float(s.isna().mean())
        })
    return {
        "shape": {"rows": int(df.shape[0]), "columns": int(df.shape[1])},
        "duplicates": int(df.duplicated().sum()),
        "columns": cols
    }

def profile_text(dataset_name: str, meta: Dict[str, Any]) -> str:
    lines = [
        f"Dataset: {dataset_name}",
        f"Shape: {meta['shape']['rows']}x{meta['shape']['columns']}",
        f"Duplicate rows: {meta['duplicates']}"
    ]
    for c in meta["columns"]:
        lines.append(f"- {c['name']} ({c['dtype']}) unique={c['unique_count']} missing={c['missing']}({round(100*c['missing_pct'],1)}%)")
    return "\n".join(lines)

def agent1(df: pd.DataFrame, dataset_name: str) -> Dict[str, Any]:
    meta = df_schema_summary(df)
    ctx = profile_text(dataset_name, meta)
    model = genai.GenerativeModel(model_name=TEXT_MODEL,
                                  generation_config={"temperature": 0.2, "response_mime_type": "application/json"})
    prompt = """
You are Agent 1 (Profiler).
Return STRICT JSON:
{
  "dataset_name": "string",
  "shape": {"rows": int, "columns": int},
  "duplicates": int,
  "columns": [{"name": "string", "dtype": "string", "unique_count": int, "missing": int, "missing_pct": float, "notes": "string"}],
  "summary": "short summary",
  "data_quality": {
     "high_missing_columns": [],
     "high_cardinality_columns": [],
     "potential_id_columns": [],
     "warnings": []
  }
}
Be concise. Use only provided columns.
"""
    parts = [
        {"role": "user", "parts": [
            f"Dataset name: {dataset_name}",
            "Compact profile:",
            ctx,
            "Schema JSON:",
            json.dumps(meta, ensure_ascii=False)
        ]}
    ]
    resp = model.generate_content(parts)
    out = json.loads(resp.text)

    # Store in Qdrant
    c = qdrant()
    vec = embed_text(ctx, "retrieval_document")
    c.upsert("dataset_profiles", points=[{"id": f"{dataset_name}-{int(time.time())}", "vector": vec, "payload": {"dataset_name": dataset_name, "profile": out}}])
    return out

def agent2(agent1_out: Dict[str, Any], df: pd.DataFrame) -> Dict[str, Any]:
    schema = [{"name": c["name"], "dtype": c["dtype"], "unique_count": c["unique_count"], "missing_pct": c["missing_pct"]}
              for c in agent1_out["columns"]]
    model = genai.GenerativeModel(model_name=TEXT_MODEL,
                                  generation_config={"temperature": 0.2, "response_mime_type": "application/json"})
    prompt = """
You are Agent 2 (Visualization Planner).
Produce EXACTLY 4 bar charts (count or mean/sum by category).
STRICT JSON:
{
  "charts": [
    {"id":"bar1","kind":"bar","x":"col","y":"count_or_numeric","agg":"count|mean|sum","top_k":10,"title":"t","description":"d"},
    {...},{...},{...}
  ]
}
Only use columns in schema. Avoid high-cardinality free-text.
"""
    parts = [{"role":"user","parts":[
        "Agent 1 output:", json.dumps(agent1_out, ensure_ascii=False),
        "Schema:", json.dumps({"schema": schema}, ensure_ascii=False)
    ]}]
    resp = model.generate_content(parts)
    plan = json.loads(resp.text)

    # Attach python_snippet (same safe snippet as before)
    plan["python_snippet"] = r'''
import json
import pandas as pd
import plotly.express as px
def _safe_bar(df, spec):
    x = spec["x"]; y = spec.get("y","count"); agg = spec.get("agg","count"); k=int(spec.get("top_k",10))
    if x not in df.columns: raise ValueError(f"x not found: {x}")
    if agg=="count" or y=="count":
        vc = df[x].astype("object").value_counts(dropna=False).head(k)
        fig = px.bar(x=[str(i) for i in vc.index], y=vc.values, labels={"x":x,"y":"count"}, title=spec.get("title",""))
        fig.update_xaxes(type="category"); return fig
    else:
        if y not in df.columns: raise ValueError(f"y not found: {y}")
        grouped = df.groupby(x, dropna=False)[y].agg(agg).sort_values(ascending=False).head(k)
        fig = px.bar(x=[str(i) for i in grouped.index], y=grouped.values, labels={"x":x,"y":f"{agg}({y})"}, title=spec.get("title",""))
        fig.update_xaxes(type="category"); return fig
def execute(df, plan):
    figs = []
    for spec in plan.get("charts", []):
        if spec.get("kind")=="bar": figs.append(_safe_bar(df, spec))
    return {"figures":[f.to_plotly_json() for f in figs], "charts_meta": plan.get("charts", [])}
result = execute(df, plan)
'''
    return plan

class PrepareReq(BaseModel):
    dataset_key: str
    snapshot_id: str
    local_path: str

@app.post("/prepare")
def prepare(req: PrepareReq):
    # If artifacts already exist, return them
    sdir = SNAPSHOTS_ROOT / req.snapshot_id
    if not sdir.exists():
        raise HTTPException(404, "Snapshot not found on disk")
    adir = sdir / "artifacts"
    prof_path = adir / "profile.json"
    plan_path = adir / "plan.json"
    figs_path = adir / "figures.json"

    if prof_path.exists() and plan_path.exists() and figs_path.exists():
        return {"status": "reused", "snapshot_id": req.snapshot_id,
                "artifacts": {"profile": str(prof_path), "plan": str(plan_path), "figures": str(figs_path)}}

    # Run Agents
    fmt = "csv" if req.local_path.lower().endswith(".csv") else "parquet"
    df = pd.read_csv(req.local_path) if fmt=="csv" else pd.read_parquet(req.local_path)
    a1 = agent1(df, req.dataset_key)
    a2 = agent2(a1, df)

    # Execute charts via Strict Executor
    import requests
    exec_req = {
        "code": a2["python_snippet"],
        "variables": {"plan": a2},
        "datasets": [{"alias": "df", "path": req.local_path, "fmt": fmt, "read_kwargs": {}}],
        "limits": {"cpu_seconds": 10, "memory_mb": 1024, "wall_seconds": 20}
    }
    exec_resp = requests.post(f"{EXEC_URL}/execute", json=exec_req, timeout=60).json()
    if not exec_resp.get("ok"):
        raise HTTPException(500, f"Executor failed: {exec_resp.get('stderr')}")

    # Persist artifacts
    adir.mkdir(parents=True, exist_ok=True)
    prof_path.write_text(json.dumps(a1, indent=2))
    plan_to_save = dict(a2); plan_to_save["python_snippet"] = "<omitted>"
    plan_path.write_text(json.dumps(plan_to_save, indent=2))
    figs_path.write_text(json.dumps(exec_resp["result"], indent=2))

    return {"status": "created", "snapshot_id": req.snapshot_id,
            "artifacts": {"profile": str(prof_path), "plan": str(plan_path), "figures": str(figs_path)}}

@app.get("/snapshots/{snapshot_id}/artifacts")
def artifacts(snapshot_id: str):
    sdir = SNAPSHOTS_ROOT / snapshot_id / "artifacts"
    prof, plan, figs = sdir / "profile.json", sdir / "plan.json", sdir / "figures.json"
    if not (prof.exists() and plan.exists() and figs.exists()):
        raise HTTPException(404, "Artifacts not ready")
    return {"profile": json.loads(prof.read_text()),
            "plan": json.loads(plan.read_text()),
            "figures": json.loads(figs.read_text())}

@app.get("/snapshots/{snapshot_id}/preprocess")
def preprocess(snapshot_id: str, columns: str):
    cols = [c.strip() for c in columns.split(",") if c.strip()]
    sdir = SNAPSHOTS_ROOT / snapshot_id
    files = list(sdir.glob("*.*"))
    if not files:
        raise HTTPException(404, "Snapshot file missing")
    path = str(files[0])
    df = pd.read_csv(path) if path.lower().endswith(".csv") else pd.read_parquet(path)

    out: Dict[str, Any] = {"columns": cols, "frames": {}, "stats": {}}
    if len(cols) == 1:
        c = cols[0]
        if c not in df.columns: raise HTTPException(400, f"Column not found: {c}")
        if str(df[c].dtype).startswith(("float","int")):
            desc = df[c].describe().to_dict()
            out["stats"][c] = {k: float(v) for k, v in desc.items()}
        vc = df[c].astype("object").value_counts(dropna=False).head(50)
        out["frames"][f"value_counts_{c}"] = {"index": [str(i) for i in vc.index], "values": vc.values.tolist()}
    elif len(cols) == 2:
        a, b = cols
        for col in cols:
            if col not in df.columns: raise HTTPException(400, f"Column not found: {col}")
        ct = pd.crosstab(df[a].astype("object"), df[b].astype("object")).head(50)
        out["frames"]["crosstab"] = {"index": [str(i) for i in ct.index], "columns": [str(i) for i in ct.columns], "values": ct.values.tolist()}
    else:
        # Fallback: per-column value_counts
        for c in cols[:3]:
            vc = df[c].astype("object").value_counts(dropna=False).head(20)
            out["frames"][f"value_counts_{c}"] = {"index": [str(i) for i in vc.index], "values": vc.values.tolist()}
    return out
```

What happens now
- Upload pipeline:
  - POST /upload (Storage): saves to /data/snapshots/<snapshot>/file.ext, fingerprints, dedups by sha256, pushes to HF repo, and returns snapshot info.
  - If new snapshot, it optionally calls Orchestrator /prepare (async fire-and-forget). Otherwise, it returns “reused” and the artifacts that are already there.

- Orchestrator:
  - /prepare loads the local file, runs Agent 1 and Agent 2, calls the Strict Executor /execute to get Plotly figures JSON, and saves artifacts under /data/snapshots/<snapshot>/artifacts/.
  - /snapshots/{id}/artifacts returns the saved profile + chart plan + figures JSON for frontend.
  - /snapshots/{id}/preprocess?columns=... returns preprocessed slices/stats to support your “click columns to show data” UX.

- Strict Python Executor:
  - Already in place; we’re calling it with df alias and plan to render figures.

Docker Compose (3 services + Qdrant)
docker-compose.yml
```yaml
version: "3.9"
services:
  qdrant:
    image: qdrant/qdrant:latest
    ports: ["6333:6333"]
    volumes:
      - qdrant_data:/qdrant/storage

  python-exec-sandbox:
    build: ./python-exec-sandbox
    ports: ["8000:8000"]
    environment:
      - STORAGE_ROOT=/data
    volumes:
      - exec_data:/data
    # recommended in prod:
    # network_mode: "none"
    # read_only: true

  storage:
    build: ./storage
    ports: ["8010:8000"]
    environment:
      - HF_TOKEN=${HF_TOKEN}
      - HF_REPO=triflix/database
      - DATA_ROOT=/data
      - ORCH_URL=http://orchestrator:8000
    volumes:
      - data_root:/data

  orchestrator:
    build: ./orchestrator
    ports: ["8020:8000"]
    environment:
      - GEMINI_API_KEY=${GEMINI_API_KEY}
      - QDRANT_URL=http://qdrant:6333
      - EXEC_URL=http://python-exec-sandbox:8000
      - ARTIFACTS_ROOT=/data/artifacts
      - SNAPSHOTS_ROOT=/data/snapshots
    volumes:
      - data_root:/data
    depends_on:
      - qdrant
      - python-exec-sandbox

volumes:
  data_root:
  exec_data:
  qdrant_data:
```

Front-end flow
- Upload file to Storage /upload with dataset_key:
  - If identical, response: status=reused + artifacts already available.
  - If new, status=snapshot_created; the Orchestrator starts preparing artifacts; poll /snapshots/{id}/artifacts until ready.
- Render charts:
  - Use artifacts.figures (Plotly figure dicts) directly in the frontend.
- “Click columns”:
  - Call Orchestrator /snapshots/{id}/preprocess?columns=colA,colB
  - Show value counts/crosstab/data slice; highlight clicked categories in your UI.

Edge cases fully covered
- Rename same file: same sha256 => reuse previous snapshot and artifacts.
- Same shape (rows/cols) but different values: sha differs => new snapshot => new run.
- Different dataset entirely: new dataset_key or separate snapshot series.
- Repeat uploads of an identical file: instant reuse, no pipeline cost.
- “Cycle that executed code of AI output and another output?”:
  - Agent 2 can be re-run anytime to generate new chart plans (e.g., different params), and you can /execute again on the sandbox. Store these as additional artifacts (plan_v2.json, figures_v2.json) if needed. The orchestrator code can accept a plan override endpoint if you want to drive this.

Do you want me to:
- Add a /upload_json variant for small DataFrames (base64 or records) so we skip file IO?
- Add binary quantization in Qdrant to scale profiles?
- Add JWT auth to Storage/Orchestrator/Executor?
- Add “plan override” endpoint that takes a new chart plan and executes it against an existing snapshot?