# Shopify Support Chatbot — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Build an embeddable Shopify support chatbot that answers product/info questions (live catalog + a knowledge base of PDFs/URLs) and order/tracking questions (GraphQL Admin API, identity-verified), in the customer's language, on free LLMs (Groq → Cloudflare failover) with local embeddings.

**Architecture:** A Python FastAPI backend exposes a chat endpoint reached from a JS widget through a Shopify App Proxy (no CORS, HMAC-verified). The backend orchestrates a provider-agnostic LLM (Groq primary, Cloudflare Workers AI failover) with tool-calling over three tools: `search_knowledge` (RAG via local embeddings + pgvector), `search_products` (Shopify GraphQL), `lookup_order` (Shopify GraphQL + tiered identity verification). An admin panel manages knowledge sources. Postgres+pgvector stores vectors, config, sessions. Deployed on Railway.

**Tech Stack:** Python 3.12, FastAPI, Uvicorn, SQLAlchemy 2.x, Alembic, pgvector, psycopg, `sentence-transformers` (multilingual-e5-small) for local embeddings, `httpx` (Groq/Cloudflare/Shopify HTTP), `pypdf` + `openpyxl` + `python-docx` + `selectolax` (doc/URL extraction), `pytest` + `pytest-asyncio` + `respx` (HTTP mocking), vanilla JS widget, Shopify Theme App Extension. Package manager: `uv`.

**Open decisions locked by owner (2026-06-09, full autonomy):**
1. Start on free LLM tiers; document paid upgrade path. No paid keys required for v1.
2. v1 covers orders ≤60 days (`read_orders`). `read_all_orders` left as a config flag (`SHOPIFY_READ_ALL_ORDERS`) to enable after Shopify approval.
3. Human escalation = minimal: capture the user's email + question and email the store's support address.

**Conventions:**
- All code under `app/`, tests under `tests/` mirroring module paths.
- TDD: write the failing test, run it (red), implement minimal code, run (green), commit.
- Every commit message: `<type>: <summary>` then trailer `Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>`.
- Async everywhere for I/O (httpx.AsyncClient, async SQLAlchemy). Embeddings model runs sync in a threadpool.
- Tests must assert positive outcomes (never a vacuous pass). Run the FULL suite green before each phase-end commit.
- Secrets only via env (`app/config.py` Settings). Never commit secrets.

---

## File Structure

```
shopify-support-bot/
├── pyproject.toml                  # uv project, deps, pytest config
├── .env.example                    # documented env vars (no secrets)
├── .gitignore
├── README.md
├── alembic.ini
├── migrations/                     # alembic
│   ├── env.py
│   └── versions/
├── app/
│   ├── __init__.py
│   ├── main.py                     # FastAPI app factory, routers, lifespan
│   ├── config.py                   # Settings (pydantic-settings)
│   ├── db.py                       # async engine/session, Base
│   ├── models.py                   # ORM: Config, KnowledgeSource, KnowledgeChunk, ChatSession, ChatMessage
│   ├── schemas.py                  # pydantic request/response models
│   ├── llm/
│   │   ├── __init__.py
│   │   ├── base.py                 # LLMProvider protocol, ChatResult, ToolCall, ToolSpec
│   │   ├── groq.py                 # GroqProvider
│   │   ├── cloudflare.py           # CloudflareProvider
│   │   └── router.py               # LLMRouter: model routing + Groq→Cloudflare failover
│   ├── embeddings.py               # local multilingual embedder (singleton, threadpool)
│   ├── rag/
│   │   ├── __init__.py
│   │   ├── extract.py              # file/URL -> text
│   │   ├── chunk.py                # text -> chunks
│   │   └── index.py                # index/reindex/delete sources; search
│   ├── shopify/
│   │   ├── __init__.py
│   │   ├── token.py                # ShopifyTokenManager (client_credentials, 24h refresh)
│   │   ├── client.py               # ShopifyGraphQLClient (httpx)
│   │   ├── proxy.py                # App Proxy HMAC verification
│   │   ├── products.py             # search_products
│   │   └── orders.py               # lookup_order_raw + tracking parsing
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── registry.py             # ToolSpec list + dispatch
│   │   ├── knowledge_tool.py       # search_knowledge
│   │   ├── products_tool.py        # search_products
│   │   └── order_tool.py           # lookup_order + verification + escalation
│   ├── verification.py             # identity verification state machine + lockout
│   ├── orchestrator.py             # chat turn: history + LLM tool loop -> reply
│   ├── escalation.py               # capture email + question -> send to store
│   ├── mailer.py                   # SMTP send (escalation)
│   ├── prompts.py                  # system prompt builder (multilingual, persona, rules)
│   ├── routes/
│   │   ├── __init__.py
│   │   ├── chat.py                 # POST /apps/chat (proxy-verified)
│   │   ├── admin.py                # admin API (auth): sources CRUD, reindex, config
│   │   └── health.py               # GET /healthz
│   └── admin_ui/
│       └── index.html              # minimal admin SPA (vanilla)
├── extension/                      # Shopify Theme App Extension
│   ├── shopify.app.toml            # app config incl. app_proxy + scopes
│   └── extensions/chat-widget/
│       ├── shopify.extension.toml
│       ├── blocks/chat.liquid      # app embed block (settings + loader)
│       └── assets/widget.js        # chat widget
└── tests/
    ├── conftest.py                 # db fixture, app client, fakes
    ├── llm/...                     # provider + router + failover tests
    ├── rag/...                     # extract/chunk/index/search tests
    ├── shopify/...                 # token/proxy/products/orders tests
    ├── tools/...                   # tool dispatch tests
    ├── test_verification.py
    ├── test_orchestrator.py
    ├── test_escalation.py
    └── routes/...                  # chat/admin/health endpoint tests
```

---

## PHASE 0 — Project scaffold & CI-ready test harness

### Task 0.1: Initialize uv project & dependencies

**Files:** Create `pyproject.toml`, `.gitignore`, `.env.example`, `README.md`

- [ ] **Step 1:** Create `pyproject.toml`:

```toml
[project]
name = "shopify-support-bot"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
  "fastapi>=0.115",
  "uvicorn[standard]>=0.32",
  "pydantic>=2.9",
  "pydantic-settings>=2.6",
  "sqlalchemy[asyncio]>=2.0",
  "psycopg[binary]>=3.2",
  "pgvector>=0.3",
  "alembic>=1.14",
  "httpx>=0.28",
  "sentence-transformers>=3.3",
  "pypdf>=5.1",
  "openpyxl>=3.1",
  "python-docx>=1.1",
  "selectolax>=0.3.25",
  "python-multipart>=0.0.12",
  "tenacity>=9.0",
]

[dependency-groups]
dev = [
  "pytest>=8.3",
  "pytest-asyncio>=0.24",
  "respx>=0.22",
  "aiosqlite>=0.20",
  "ruff>=0.8",
]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
filterwarnings = ["ignore::DeprecationWarning"]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
```

- [ ] **Step 2:** Create `.gitignore` (Python, `.env`, `.venv`, `__pycache__`, `*.db`, `.pytest_cache`, model cache `.hf_cache/`, `node_modules`).
- [ ] **Step 3:** Create `.env.example` documenting every var (see Task 1.1 Settings) with placeholder values and comments.
- [ ] **Step 4:** Run `uv sync` and confirm the venv builds.

Run: `uv run python -c "import fastapi, sqlalchemy, httpx, sentence_transformers; print('ok')"`
Expected: prints `ok`.

- [ ] **Step 5: Commit** `chore: scaffold uv project and dependencies`.

### Task 0.2: Test harness with async DB fixture (sqlite for unit, pg for integration)

**Files:** Create `tests/conftest.py`, `app/db.py`, `app/__init__.py`

Decision: unit tests use in-memory sqlite (`aiosqlite`) with a vector fallback (store embeddings as JSON + cosine in Python when the dialect isn't postgres); pgvector path is exercised in Phase 2 integration tests guarded by `DATABASE_URL` pointing at postgres. This keeps the suite runnable with zero external services.

- [ ] **Step 1:** Write `app/db.py`:

```python
from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

_engine = None
_sessionmaker = None

def init_engine(url: str) -> None:
    global _engine, _sessionmaker
    _engine = create_async_engine(url, future=True)
    _sessionmaker = async_sessionmaker(_engine, expire_on_commit=False, class_=AsyncSession)

def get_sessionmaker() -> async_sessionmaker:
    assert _sessionmaker is not None, "engine not initialized"
    return _sessionmaker

async def get_session() -> AsyncIterator[AsyncSession]:
    async with get_sessionmaker()() as session:
        yield session
```

- [ ] **Step 2:** Write `tests/conftest.py` with fixtures: `db_session` (creates all tables on a fresh in-memory sqlite engine per test), `client` (httpx ASGITransport against the app with DB dependency overridden). Include a `_is_postgres` helper. Provide an `anyio_backend`/asyncio config via pytest-asyncio auto mode.
- [ ] **Step 3:** Write a trivial `tests/test_smoke.py::test_db_fixture_creates_tables` that inserts and reads back a row from a throwaway model to prove the fixture works.
- [ ] **Step 4:** Run `uv run pytest tests/test_smoke.py -v` → PASS.
- [ ] **Step 5: Commit** `test: async db test harness`.

---

## PHASE 1 — Config, models, and the LLM layer (Groq → Cloudflare failover)

### Task 1.1: Settings

**Files:** Create `app/config.py`, Test `tests/test_config.py`

- [ ] **Step 1 (test):** assert `Settings` reads env vars with defaults and that `llm_provider_order` defaults to `["groq", "cloudflare"]`.

```python
def test_settings_defaults(monkeypatch):
    monkeypatch.setenv("DATABASE_URL", "sqlite+aiosqlite://")
    from app.config import Settings
    s = Settings()
    assert s.llm_provider_order == ["groq", "cloudflare"]
    assert s.model_small and s.model_large
    assert s.order_verify_max_attempts == 5
```

- [ ] **Step 2:** Run → FAIL.
- [ ] **Step 3 (impl):** `Settings(BaseSettings)` fields: `database_url`, `groq_api_key`, `groq_base_url="https://api.groq.com/openai/v1"`, `cloudflare_account_id`, `cloudflare_api_token`, `model_large="llama-3.3-70b-versatile"`, `model_small="llama-3.1-8b-instant"`, `cf_model_large="@cf/meta/llama-3.3-70b-instruct-fp8-fast"`, `cf_model_small="@cf/meta/llama-3.1-8b-instruct"`, `llm_provider_order=["groq","cloudflare"]`, `shopify_shop`, `shopify_client_id`, `shopify_client_secret`, `shopify_api_version="2026-01"`, `shopify_app_proxy_secret`, `shopify_read_all_orders=False`, `admin_token`, `embedding_model="intfloat/multilingual-e5-small"`, `order_verify_max_attempts=5`, `order_verify_lockout_seconds=900`, `session_retention_days=30`, `smtp_*`, `support_email`. `model_config = SettingsConfigDict(env_file=".env", extra="ignore")`.
- [ ] **Step 4:** Run → PASS. **Step 5: Commit** `feat: settings`.

### Task 1.2: ORM models + schemas

**Files:** Create `app/models.py`, `app/schemas.py`, Test `tests/test_models.py`

- [ ] **Step 1 (test):** create a `ChatSession`, add a `ChatMessage`, flush, and assert relationship + defaults (`verified=False`, `verify_attempts=0`).
- [ ] **Step 2:** FAIL.
- [ ] **Step 3 (impl):** Models on `Base`:
  - `Config(id, key UNIQUE, value JSON)` — singleton-ish store for branding/flags.
  - `KnowledgeSource(id, kind[file|url], name, location, status[pending|indexed|error], error, created_at)`
  - `KnowledgeChunk(id, source_id FK, ordinal, text, embedding[Vector(384) on pg / JSON on sqlite via a TypeDecorator], meta JSON)`
  - `ChatSession(id[str uuid], shop, lang, verified bool=False, verify_attempts int=0, locked_until datetime|None, created_at, last_seen)`
  - `ChatMessage(id, session_id FK, role, content, created_at)`
  Implement an `EmbeddingType` `TypeDecorator` that uses `pgvector.sqlalchemy.Vector(384)` on postgres and `JSON` (list of floats) on other dialects, so unit tests run on sqlite.
  `schemas.py`: `ChatRequest{session_id?, message}`, `ChatResponse{session_id, reply, lang, used_tools[]}`, `SourceIn{kind, location?}`, `SourceOut{...}`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: orm models and schemas`.

### Task 1.3: LLM base protocol

**Files:** Create `app/llm/base.py`, Test `tests/llm/test_base.py`

- [ ] **Step 1 (test):** assert dataclasses construct: `ToolSpec(name, description, parameters)`, `ToolCall(id, name, arguments: dict)`, `ChatResult(content: str|None, tool_calls: list[ToolCall], finish_reason)`. Assert `LLMProvider` is a `typing.Protocol` with async `chat(messages, tools, model, temperature)`.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** dataclasses + Protocol + a `ProviderError(Exception)` and `RateLimitError(ProviderError)`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: llm base protocol`.

### Task 1.4: GroqProvider (OpenAI-compatible chat completions, tool calling)

**Files:** Create `app/llm/groq.py`, Test `tests/llm/test_groq.py`

- [ ] **Step 1 (test):** with `respx` mock POST `https://api.groq.com/openai/v1/chat/completions`:
  - returns assistant message content → `ChatResult.content == "hola"`, `tool_calls == []`.
  - returns a `tool_calls` array → parsed into `ToolCall(name=..., arguments=dict)` (arguments JSON-decoded).
  - returns HTTP 429 → raises `RateLimitError`.
  - returns HTTP 500 → raises `ProviderError`.

```python
@respx.mock
async def test_groq_parses_tool_calls():
    respx.post("https://api.groq.com/openai/v1/chat/completions").mock(
        return_value=httpx.Response(200, json={
          "choices":[{"finish_reason":"tool_calls","message":{"content":None,"tool_calls":[
            {"id":"c1","type":"function","function":{"name":"lookup_order","arguments":"{\"email\":\"a@b.c\"}"}}]}}]}))
    p = GroqProvider(api_key="k", base_url="https://api.groq.com/openai/v1")
    r = await p.chat(messages=[{"role":"user","content":"x"}], tools=[], model="llama-3.1-8b-instant")
    assert r.tool_calls[0].name == "lookup_order"
    assert r.tool_calls[0].arguments == {"email":"a@b.c"}
```

- [ ] **Step 2:** FAIL. **Step 3 (impl):** `GroqProvider.chat` builds OpenAI-style payload (`messages`, `tools=[{"type":"function","function":ToolSpec}]`, `tool_choice="auto"`, `model`, `temperature`), posts via shared `httpx.AsyncClient`, maps 429→`RateLimitError`, other ≥400→`ProviderError`, parses choices[0].
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: groq provider`.

### Task 1.5: CloudflareProvider (Workers AI, OpenAI-compatible endpoint)

**Files:** Create `app/llm/cloudflare.py`, Test `tests/llm/test_cloudflare.py`

- [ ] **Step 1 (test):** respx mock POST `https://api.cloudflare.com/client/v4/accounts/{acct}/ai/v1/chat/completions` (OpenAI-compatible) → same parsing contract as Groq (content + tool_calls + 429→RateLimitError). Use account id `acct` and Bearer token header asserted.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `CloudflareProvider(account_id, api_token)` posting to the `/ai/v1/chat/completions` OpenAI-compatible route with `Authorization: Bearer`. Same payload/response mapping as Groq (share a small `_parse_openai_choice` helper in `base.py`).
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: cloudflare workers ai provider`.

### Task 1.6: LLMRouter — model routing + failover

**Files:** Create `app/llm/router.py`, Test `tests/llm/test_router.py`

- [ ] **Step 1 (test):**
  - `tier="small"` routes to `model_small`/`cf_model_small`; `tier="large"` to large models.
  - Primary (Groq) raises `RateLimitError` → router falls back to Cloudflare and returns its result. Assert the fake Cloudflare provider was called once and Groq once.
  - All providers fail → raises `ProviderError` after exhausting `llm_provider_order`.

```python
async def test_router_failover_on_ratelimit():
    groq = FakeProvider(raise_exc=RateLimitError())
    cf = FakeProvider(result=ChatResult(content="ok", tool_calls=[], finish_reason="stop"))
    router = LLMRouter(providers={"groq":groq,"cloudflare":cf}, order=["groq","cloudflare"],
                       models={"groq":{"small":"gs","large":"gl"},"cloudflare":{"small":"cs","large":"cl"}})
    r = await router.chat(messages=[{"role":"user","content":"x"}], tools=[], tier="small")
    assert r.content == "ok"
    assert groq.calls == 1 and cf.calls == 1
```

- [ ] **Step 2:** FAIL. **Step 3 (impl):** `LLMRouter.chat(messages, tools, tier)` iterates `order`, picks each provider's model for the tier, calls `chat`, returns first success; on `ProviderError`/`RateLimitError` logs and tries next; raises if all fail. `tenacity` retry (2 attempts, expo backoff) per provider before moving on.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: llm router with failover`.

---

## PHASE 2 — Embeddings, extraction, RAG index & search

### Task 2.1: Local embedder

**Files:** Create `app/embeddings.py`, Test `tests/test_embeddings.py`

- [ ] **Step 1 (test):** `embed_texts(["hola mundo","adios"])` returns 2 vectors of equal length >0; identical text yields identical vectors; cosine(self,self)≈1. Mark `@pytest.mark.slow` (loads the model once).
- [ ] **Step 2:** FAIL. **Step 3 (impl):** lazy-loaded `SentenceTransformer(settings.embedding_model)` singleton; `embed_texts(list[str]) -> list[list[float]]` and `embed_query(str)->list[float]` (e5 wants `"query: "`/`"passage: "` prefixes — apply them). Run inference in `anyio.to_thread`. Cache model in module global. Set `HF_HOME=.hf_cache`.
- [ ] **Step 4:** PASS (downloads model on first run). **Step 5: Commit** `feat: local multilingual embedder`.

### Task 2.2: Text extraction (file + URL)

**Files:** Create `app/rag/extract.py`, Test `tests/rag/test_extract.py` + fixtures `tests/rag/fixtures/sample.pdf|.xlsx|.docx|.txt`

- [ ] **Step 1 (test):** `extract_file(path)` returns non-empty text for each fixture type with a known substring. `extract_url(html_bytes)` (pass raw HTML to a pure function `extract_html(bytes)->str`) strips tags and returns visible text containing a known phrase. Generate fixtures programmatically in a fixture-builder step.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** dispatch by extension: `.pdf`→pypdf, `.xlsx`→openpyxl (join non-empty cells), `.docx`→python-docx, `.txt`→read. `extract_html(bytes)` via `selectolax` (`.text()` after removing script/style). `extract_url(url)` fetches via httpx then calls `extract_html`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: text extraction for files and urls`.

### Task 2.3: Chunking

**Files:** Create `app/rag/chunk.py`, Test `tests/rag/test_chunk.py`

- [ ] **Step 1 (test):** `chunk_text("a"*2500, size=1000, overlap=200)` → 3 chunks, each ≤1000 chars, consecutive chunks overlap by 200, concatenation order preserved. Empty/whitespace → `[]`.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** sliding window by characters with overlap; trim whitespace; drop empty. (Keep simple/deterministic; token-accurate chunking is YAGNI.)
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: text chunking`.

### Task 2.4: Index/reindex/delete + search

**Files:** Create `app/rag/index.py`, Test `tests/rag/test_index.py`

- [ ] **Step 1 (test):** using `db_session` + a fake embedder (monkeypatch `embed_texts`/`embed_query` to deterministic small vectors): `index_source(session, source)` extracts→chunks→embeds→stores `KnowledgeChunk` rows and sets source `status="indexed"`. `search(session, "query", k=2)` returns the most similar chunks (cosine), highest first. `delete_source(session, id)` removes chunks. On extraction error, source `status="error"` and `error` populated.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `index_source` (uses extract+chunk+embed_texts), `search` computes cosine in-Python over stored vectors when dialect != postgres, else uses pgvector `<=>` operator ordering; returns `[(chunk, score)]`. `reindex_all`. Wrap extraction in try/except → status error.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: rag index and search`.

### Task 2.5: pgvector integration test (guarded)

**Files:** Test `tests/rag/test_pgvector.py`

- [ ] **Step 1 (test):** skip unless `TEST_DATABASE_URL` (postgres) set. Create extension `vector`, run a real index+search, assert ordering matches the in-Python path. (Run locally/CI with a postgres service.)
- [ ] **Step 2-4:** Implement guard + assertions; PASS or SKIP. **Step 5: Commit** `test: pgvector integration (guarded)`.

---

## PHASE 3 — Shopify integration (token, client, proxy, products, orders, verification)

### Task 3.1: ShopifyTokenManager (client_credentials, 24h refresh)

**Files:** Create `app/shopify/token.py`, Test `tests/shopify/test_token.py`

- [ ] **Step 1 (test):** respx mock POST `https://{shop}/admin/oauth/access_token` returns `{access_token, expires_in: 86399}`. `await mgr.get_token()` returns the token and caches it; a second call within expiry does NOT re-POST (assert respx call count == 1). Simulate clock past expiry (inject a `now()` callable) → re-POSTs. On 401 invalidate + refetch.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `ShopifyTokenManager(shop, client_id, client_secret, now=...)`: caches `(token, expires_at)`, refreshes when `now() >= expires_at - 60s`; async lock to avoid stampede; `invalidate()`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: shopify token manager with refresh`.

### Task 3.2: ShopifyGraphQLClient

**Files:** Create `app/shopify/client.py`, Test `tests/shopify/test_client.py`

- [ ] **Step 1 (test):** respx mock POST `https://{shop}/admin/api/{ver}/graphql.json` with `X-Shopify-Access-Token` header asserted; `await client.execute(query, variables)` returns `data`; GraphQL `errors` in body → raises `ShopifyError`; HTTP 401 → calls token manager `invalidate` then retries once.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `ShopifyGraphQLClient(shop, api_version, token_manager)`: posts query+variables with token header; on `errors` raise; on 401 invalidate+retry once; on `THROTTLED`/429 backoff via tenacity reading `extensions.cost.throttleStatus`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: shopify graphql client`.

### Task 3.3: App Proxy HMAC verification

**Files:** Create `app/shopify/proxy.py`, Test `tests/shopify/test_proxy.py`

- [ ] **Step 1 (test):** `verify_proxy_signature(query_params: dict, secret)` — build params, compute the correct Shopify proxy signature (sorted params, concatenated without `signature`, HMAC-SHA256 hex), assert valid passes and a tampered signature fails. Missing `signature` → False.

```python
def test_proxy_signature_valid():
    secret = "shh"
    params = {"shop":"x.myshopify.com","path_prefix":"/apps/chat","timestamp":"1700000000"}
    sig = _compute(params, secret)  # helper in test mirrors Shopify spec
    assert verify_proxy_signature({**params, "signature": sig}, secret) is True
    assert verify_proxy_signature({**params, "signature": "bad"}, secret) is False
```

- [ ] **Step 2:** FAIL. **Step 3 (impl):** per Shopify App Proxy spec: remove `signature`, sort keys, join as `k=v` (comma-join multi-values) concatenated, HMAC-SHA256 with secret, `hmac.compare_digest` against provided hex.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: app proxy hmac verification`.

### Task 3.4: search_products

**Files:** Create `app/shopify/products.py`, Test `tests/shopify/test_products.py`

- [ ] **Step 1 (test):** fake GraphQL client returns a `products` connection; `await search_products(client, "treadmill")` returns list of `{title, price, available, url, description}` mapped from nodes. Empty → `[]`.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** GraphQL `products(first:5, query:$q)` selecting title, handle, onlineStoreUrl, description(truncate), variants(price, availableForSale). Map to dicts; compute min price + any-available.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: shopify product search`.

### Task 3.5: orders — lookup + tracking parsing

**Files:** Create `app/shopify/orders.py`, Test `tests/shopify/test_orders.py`

- [ ] **Step 1 (test):** fake client returns an `orders(query:"email:...")` payload with one order incl. fulfillments/trackingInfo/displayFulfillmentStatus. `await find_orders_by_email(client, "a@b.c")` returns parsed orders incl. `name` (order number), `financial_status`, `fulfillment_status`. `parse_tracking(order)` returns `{status, tracking:[{company,number,url}], estimated_delivery, delivered_at}`. Double-quote the email in the query (assert query string contains `email:"a@b.c"`).
- [ ] **Step 2:** FAIL. **Step 3 (impl):** GraphQL query from spec §using `orders(first:10, query:$q, sortKey:CREATED_AT, reverse:true)` with fulfillments{displayStatus, estimatedDeliveryAt, deliveredAt, inTransitAt, trackingInfo{company number url}}. If `SHOPIFY_READ_ALL_ORDERS` false, rely on default 60-day window. Helpers `find_orders_by_email`, `find_order_by_email_and_number`, `parse_tracking`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: shopify order lookup and tracking`.

### Task 3.6: Identity verification state machine

**Files:** Create `app/verification.py`, Test `tests/test_verification.py`

- [ ] **Step 1 (test):**
  - `verify(session, email, order_number, orders)` where orders contains a match (email+number) → returns `VerifyResult(ok=True)`, sets `session.verified=True`.
  - No match → `ok=False`, increments `verify_attempts`, generic message (no field-specific leak), and after `max_attempts` sets `locked_until` and returns `locked=True`.
  - `is_locked(session, now)` True while `locked_until` in future.
  - Logged-in trust: `mark_trusted(session)` sets verified without attempts.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** pure functions over `ChatSession`; case-insensitive email compare; order number normalized (strip leading `#`); generic failure text constant; lockout via settings.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: identity verification state machine`.

---

## PHASE 4 — Tools, prompts, orchestrator, escalation

### Task 4.1: Tool registry + ToolSpecs

**Files:** Create `app/tools/registry.py`, `app/tools/knowledge_tool.py`, `app/tools/products_tool.py`, `app/tools/order_tool.py`, Tests `tests/tools/...`

- [ ] **Step 1 (tests, one per tool):**
  - `knowledge_tool`: given fake `search` returning chunks, `run(args)` returns concatenated context + sources list.
  - `products_tool`: wraps `search_products`, returns compact product list.
  - `order_tool`: requires `session`; if not verified and args lack email/order_number → returns a "need verification" structured result asking for email+order number; with valid args and a matching fake order → returns tracking summary; on lockout → returns generic locked message; never returns address/payment fields.
  - `registry.specs()` returns 3 `ToolSpec`s with correct JSON-schema params; `registry.dispatch(name, args, ctx)` routes to the right tool.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** each tool a small async callable `run(args, ctx)` where `ctx` carries `session`, db `session`, shopify client. `registry` holds specs + name→callable. JSON-schema params: `search_knowledge{query}`, `search_products{query}`, `lookup_order{email?, order_number?}`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: tool registry and tools`.

### Task 4.2: Prompt builder

**Files:** Create `app/prompts.py`, Test `tests/test_prompts.py`

- [ ] **Step 1 (test):** `build_system_prompt(brand)` includes: reply-in-customer-language rule, persona/brand name, the verification policy (ask email+order number, reveal only shipment status), honesty + escalation rule, and tool-use guidance. Assert key phrases present and brand name interpolated.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** template returning a string with the rules above; pt-PT hint; "never reveal which field failed verification"; "if you cannot help, offer to pass the question to the team and ask for their email".
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: system prompt builder`.

### Task 4.3: Escalation + mailer

**Files:** Create `app/escalation.py`, `app/mailer.py`, Test `tests/test_escalation.py`

- [ ] **Step 1 (test):** `create_escalation(session, email, question, sender)` calls a fake `sender` with the store `support_email`, subject incl. session id, body incl. question+customer email; returns confirmation text. Mailer tested via a fake SMTP (assert `sendmail` called with right args) — inject SMTP client.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `escalation.create_escalation(...)` builds message and calls injected `sender`. `mailer.SmtpMailer(settings)` with `send(to, subject, body)`; `sender` default wires to it. No-op/log if SMTP unconfigured.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: human escalation via email`.

### Task 4.4: Orchestrator (chat turn + tool loop)

**Files:** Create `app/orchestrator.py`, Test `tests/test_orchestrator.py`

- [ ] **Step 1 (test):** with a `FakeRouter` scripted to: (turn 1) return a `tool_calls` for `search_knowledge`, (turn 2) return final content. `run_turn(ctx, session, "¿envíos?")`:
  - persists user + assistant messages,
  - executes the tool via registry (fake search returns context),
  - feeds tool result back, gets final reply,
  - returns `ChatResponse(reply, lang, used_tools=["search_knowledge"])`.
  - Second test: order intent unverified → tool returns need-verification → final reply asks for email+order number; `used_tools` includes `lookup_order`.
  - Guard: max 4 tool iterations then force final answer.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** load short history (last N msgs), build messages = [system, ...history, user]; loop: call `router.chat(tier="large", tools=registry.specs())`; if tool_calls, dispatch each, append tool results as `{"role":"tool",...}`, continue; else return content. Detect language is delegated to the model (system rule); store `lang` if model annotates, else heuristic. Cap iterations.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: chat orchestrator with tool loop`.

---

## PHASE 5 — HTTP routes, app wiring, admin, widget, deploy

### Task 5.1: health + app factory + lifespan

**Files:** Create `app/main.py`, `app/routes/health.py`, Test `tests/routes/test_health.py`

- [ ] **Step 1 (test):** `GET /healthz` → 200 `{"status":"ok"}`.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `create_app()` sets up settings, `init_engine`, includes routers, lifespan warms embedder (skippable via env in tests). **GDPR retention:** add `purge_old_sessions(session, days)` (deletes `ChatSession`/`ChatMessage` older than `session_retention_days`), invoked lazily on startup and opportunistically per chat turn. Add `tests/routes/test_retention.py` asserting old sessions are purged and recent ones kept. 
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: app factory and health route`.

### Task 5.2: POST /apps/chat (proxy-verified)

**Files:** Create `app/routes/chat.py`, Test `tests/routes/test_chat.py`

- [ ] **Step 1 (test):** with proxy verification monkeypatched/valid signature + a fake orchestrator bound via dependency override:
  - valid signature + body `{message}` → 200 `ChatResponse`; creates a session if no `session_id`.
  - invalid signature → 401.
  - logged-in customer signal (`logged_in_customer_id` query param present & non-empty) → session marked trusted (assert verification trust path used).
- [ ] **Step 2:** FAIL. **Step 3 (impl):** verify proxy signature from query params; load/create session; if `logged_in_customer_id` present → `mark_trusted`; call orchestrator; return response. Dependencies (`get_router`, `get_shopify_client`, `get_session`) overridable for tests.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: chat endpoint behind app proxy`.

### Task 5.3: Admin API (auth) — sources CRUD + reindex + config

**Files:** Create `app/routes/admin.py`, Test `tests/routes/test_admin.py`

- [ ] **Step 1 (test):** requests without `Authorization: Bearer <admin_token>` → 401. With token:
  - `POST /admin/sources` (url) → creates source, kicks index (background or sync in test) → 201 with id.
  - `POST /admin/sources` multipart file upload → stores + indexes.
  - `GET /admin/sources` → lists with status.
  - `DELETE /admin/sources/{id}` → removes + deletes chunks.
  - `POST /admin/reindex` → reindexes all.
  - `PUT /admin/config` → updates branding/flags.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** `Depends(require_admin)` checks token; endpoints use rag.index functions; file saved to a temp/work dir then indexed; indexing runs via `BackgroundTasks` in prod, synchronously when `TESTING`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: admin api for knowledge sources and config`.

### Task 5.4: Alembic migrations + pgvector

**Files:** Create `alembic.ini`, `migrations/env.py`, initial revision

- [ ] **Step 1:** `uv run alembic init migrations`; wire `env.py` to `Base.metadata` + async engine + `DATABASE_URL`.
- [ ] **Step 2:** Author initial migration: `CREATE EXTENSION IF NOT EXISTS vector;` then all tables (embedding column `Vector(384)` on pg).
- [ ] **Step 3:** Test (guarded by `TEST_DATABASE_URL`): run `upgrade head` against postgres, assert tables exist.
- [ ] **Step 4:** Run migration locally if postgres available, else document. **Step 5: Commit** `feat: alembic migrations with pgvector`.

### Task 5.5: Admin UI (minimal vanilla SPA)

**Files:** Create `app/admin_ui/index.html`, serve via a route; Test `tests/routes/test_admin_ui.py`

- [ ] **Step 1 (test):** `GET /admin/` (with token via query for the page load, then JS uses Bearer) returns HTML containing the words "Fuentes" and a file input + URL input.
- [ ] **Step 2:** FAIL. **Step 3 (impl):** single HTML page with fetch() calls to the admin API (token entered in a field, stored in memory): list/add/delete sources, paste URL, upload file, edit branding. No build step.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: minimal admin UI`.

### Task 5.6: Chat widget + Theme App Extension

**Files:** Create `extension/shopify.app.toml`, `extension/extensions/chat-widget/shopify.extension.toml`, `blocks/chat.liquid`, `assets/widget.js`; Test `tests/test_widget_assets.py`

- [ ] **Step 1 (test):** a Python test asserts `widget.js` exists, contains a `fetch("/apps/chat"` call and posts `{message, session_id}`, persists `session_id` in `localStorage`, and reads block settings (data attributes). `chat.liquid` contains a `{% schema %}` with `target: "body"` (app embed) and settings: `backend_path` (default `/apps/chat`), `brand_name`, `brand_color`, `welcome`. Assert these keys present.
- [ ] **Step 2:** FAIL. **Step 3 (impl):**
  - `chat.liquid`: app embed block rendering a container with data-* from settings and `{{ 'widget.js' | asset_url | script_tag }}`; passes `logged_in_customer_id: {{ customer.id }}` into the widget config.
  - `widget.js`: render bubble + panel; on send POST to `backend_path` with `{message, session_id}` and append `?logged_in_customer_id=` when known; localStorage session; typing indicator; language is whatever the user types (backend handles).
  - `shopify.app.toml`: name, scopes `read_orders,read_products,read_customers`, `[app_proxy] url, subpath="chat", prefix="apps"`.
- [ ] **Step 4:** PASS. **Step 5: Commit** `feat: chat widget and theme app extension`.

### Task 5.7: README, .env.example finalize, run docs

**Files:** Modify `README.md`, `.env.example`

- [ ] **Step 1:** Document: local run (`uv run uvicorn app.main:app`), tests (`uv run pytest`), Shopify app setup (Dev Dashboard, scopes, app proxy, install, client credentials), Railway deploy, env vars, admin usage, LLM keys (Groq + Cloudflare), paid upgrade path, `read_all_orders` flag.
- [ ] **Step 2: Commit** `docs: readme and env docs`.

### Task 5.8: Railway deployment config

**Files:** Create `railway.json`/`Procfile`/`Dockerfile` as needed, `nixpacks` config

- [ ] **Step 1:** Provide start command `uvicorn app.main:app --host 0.0.0.0 --port $PORT`, ensure Postgres provisioned + `DATABASE_URL`, run `alembic upgrade head` on release. Pin model download at build (optional) or warm at first boot.
- [ ] **Step 2:** Document Railway steps in README (use the use-railway skill at deploy time). **Step 3: Commit** `chore: railway deployment config`.

---

## Final verification (run before declaring complete)

- [ ] `uv run ruff check .` → clean.
- [ ] `uv run pytest -q` → **full suite green** (assertions positive, no skips except the postgres-guarded ones when no PG).
- [ ] Run the app locally; `curl /healthz` ok; with a signed proxy request, a chat turn returns a reply using a mocked/real LLM key.
- [ ] Spec coverage check: every spec §1-§9 requirement maps to a task (see self-review below).
- [ ] Code review pass (requesting-code-review skill) before final commit/tag.

---

## Self-Review (plan vs spec)

**Spec coverage:**
- §1/§2 info+order flows → Tasks 4.1 (tools), 4.4 (orchestrator), 5.2 (chat route). ✓
- §2 GraphQL/token/scopes/proxy/LLM/embeddings/Railway → Tasks 3.1–3.3, 1.4–1.6, 2.1, 5.8. ✓
- §3 components → all mapped (widget 5.6, proxy 3.3, orchestrator 4.4, tools 4.1, indexer 2.4, admin 5.3/5.5, token 3.1, db 1.2/5.4). ✓
- §4 flows incl. logged-in trust + email+order# → Tasks 3.6, 4.1 (order_tool), 5.2. ✓
- §5 security/GDPR (HMAC, admin auth, generic errors, lockout, retention, no-train LLM, local embeddings) → Tasks 3.3, 5.3, 3.6, (retention: add purge in 5.1 lifespan note), 1.4–1.6, 2.1. ✓
- §6 resilience (token refresh, failover, 60-day, throttle, escalation, free caps/routing) → Tasks 3.1, 1.6, 3.5, 3.2, 4.3, 4.4(tier routing). ✓
- §7 data model → Task 1.2 + 5.4. ✓
- §8 deploy → 5.4, 5.7, 5.8. ✓
- §9 testing → tests in every task + Final verification. ✓

**Gaps fixed inline:** Session retention purge — add to Task 5.1 lifespan (a periodic/lazy purge of sessions older than `session_retention_days`). Add a note there.

**Type consistency:** `ChatResult`/`ToolCall`/`ToolSpec` used identically across 1.3–1.6, 4.1, 4.4. `search(session, query, k)` signature consistent 2.4↔4.1. `verify(...)`/`mark_trusted` consistent 3.6↔5.2↔4.1. ✓