docs/superpowers/plans/2026-06-09-shopify-support-chatbot.md

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:

[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:

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"].

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.

@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.

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.

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 ToolSpecs 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. ✓