qtalker---/docs/superpowers/plans/2026-07-20-quantumancy-02-frontend-llm-pipeline.md

Quantumancy Plan 2/7: Frontend, LLM & Realtime Pipeline — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Get an actual browsable website live (React frontend served by the FastAPI backend, wired to Plan 1's auth API), plus the shared backend plumbing every spirit mode in Plans 3-5 will need: a bounded Ollama request queue, local Piper TTS, and a per-session WebSocket channel.

Architecture: React + TypeScript + Vite SPA, built to static assets and served directly by FastAPI (no separate web server). Task order is deliberately frontend-first — Tasks 1-2 produce a real, visitable site before Tasks 3-5 add backend-only plumbing that isn't visible in the browser yet (that lands in Plan 3, when the Ouija/Wire Ghost UI consumes it).

Tech Stack: React 18, TypeScript, Vite, Vitest + React Testing Library (frontend); httpx, Piper (via the piper-tts PyPI package), SQLAlchemy (backend, extending Plan 1's stack).

Global Constraints

• No containers — same as Plan 1. Frontend is built to static files (npm run build) and served by the existing FastAPI process; no separate frontend server process in production.
• App still serves plain HTTP on port 7777; TLS remains the Cloudflare Tunnel's job.
• Ollama is CPU-only and shared (10.30.20.107:11434) — all LLM calls must go through a bounded-concurrency queue, never called directly.
• app.db.Base/engine/async_session_maker/get_db(), app.deps.get_current_user/SESSION_COOKIE_NAME, and backend/tests/conftest.py's client/test_engine fixtures (from Plan 1) are stable interfaces — do not modify their existing behavior, only add to them where a task explicitly says so.
• Piper TTS and the Ollama queue are infrastructure only in this plan — no spirit-mode logic (anomaly detection, prompt content, entity persistence) is built here. That's Plans 3-6.

Plan Series

This is 2 of 7 plans implementing the Quantumancy website spec (docs/superpowers/specs/2026-07-20-quantumancy-website-design.md). Plan 1 (Foundation & Auth) is complete and live on this CT via systemctl status quantumancy.

1. Foundation & Auth (complete)
2. Frontend, LLM & Realtime Pipeline (this plan)
3. Wire Ghost mode + Ouija/Planchette UI
4. EVP Listening mode
5. Spirit Radio mode
6. Codex & entity persistence
7. Internationalization (EN/ES)

---

Task 1: Frontend Scaffold (Vite + React + TS)

Files:

• Create: frontend/package.json
• Create: frontend/vite.config.ts
• Create: frontend/tsconfig.json
• Create: frontend/tsconfig.node.json
• Create: frontend/index.html
• Create: frontend/src/main.tsx
• Create: frontend/src/App.tsx
• Create: frontend/src/App.css
• Create: frontend/src/api.ts
• Create: frontend/src/setupTests.ts
• Test: frontend/src/App.test.tsx

Interfaces:

• Consumes: Plan 1's POST /auth/register, POST /auth/login, POST /auth/logout, GET /auth/me (paths and JSON shapes only — called via fetch, no backend code imported).

• Produces: a built frontend/dist/ directory (via npm run build) that Task 2 serves; register/login/logout/me functions and a User type in frontend/src/api.ts for later tasks' UI work to reuse.

• Step 1: Scaffold the project files

frontend/package.json:

{
  "name": "quantumancy-frontend",
  "private": true,
  "version": "0.1.0",
  "type": "module",
  "scripts": {
    "dev": "vite",
    "build": "tsc -b && vite build",
    "test": "vitest run"
  },
  "dependencies": {
    "react": "^18.3.1",
    "react-dom": "^18.3.1"
  },
  "devDependencies": {
    "@testing-library/jest-dom": "^6.6.3",
    "@testing-library/react": "^16.0.1",
    "@testing-library/user-event": "^14.5.2",
    "@types/react": "^18.3.12",
    "@types/react-dom": "^18.3.1",
    "@vitejs/plugin-react": "^4.3.3",
    "jsdom": "^25.0.1",
    "typescript": "^5.6.3",
    "vite": "^5.4.10",
    "vitest": "^2.1.4"
  }
}

frontend/vite.config.ts:

/// <reference types="vitest/config" />
import { defineConfig } from 'vite'
import react from '@vitejs/plugin-react'

export default defineConfig({
  plugins: [react()],
  build: {
    outDir: 'dist',
  },
  server: {
    proxy: {
      '/auth': 'http://localhost:7777',
      '/healthz': 'http://localhost:7777',
    },
  },
  test: {
    environment: 'jsdom',
    globals: true,
    setupFiles: './src/setupTests.ts',
  },
})

frontend/tsconfig.json:

{
  "compilerOptions": {
    "target": "ES2020",
    "useDefineForClassFields": true,
    "lib": ["ES2020", "DOM", "DOM.Iterable"],
    "module": "ESNext",
    "skipLibCheck": true,
    "moduleResolution": "bundler",
    "allowImportingTsExtensions": true,
    "resolveJsonModule": true,
    "isolatedModules": true,
    "noEmit": true,
    "jsx": "react-jsx",
    "strict": true,
    "types": ["vitest/globals", "@testing-library/jest-dom"]
  },
  "include": ["src"],
  "references": [{ "path": "./tsconfig.node.json" }]
}

frontend/tsconfig.node.json:

{
  "compilerOptions": {
    "composite": true,
    "skipLibCheck": true,
    "module": "ESNext",
    "moduleResolution": "bundler",
    "allowSyntheticDefaultImports": true
  },
  "include": ["vite.config.ts"]
}

frontend/index.html:

<!doctype html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Quantumancy</title>
  </head>
  <body>
    <div id="root"></div>
    <script type="module" src="/src/main.tsx"></script>
  </body>
</html>

frontend/src/setupTests.ts:

import '@testing-library/jest-dom'

frontend/src/api.ts:

export type User = {
  id: string
  username: string
}

async function request<T>(path: string, options: RequestInit = {}): Promise<T> {
  const response = await fetch(path, {
    ...options,
    credentials: 'include',
    headers: {
      'Content-Type': 'application/json',
      ...options.headers,
    },
  })
  if (!response.ok) {
    const body = await response.json().catch(() => ({ detail: response.statusText }))
    throw new Error(body.detail ?? `Request failed: ${response.status}`)
  }
  if (response.status === 204) {
    return undefined as T
  }
  return response.json() as Promise<T>
}

export function register(username: string, password: string): Promise<User> {
  return request<User>('/auth/register', {
    method: 'POST',
    body: JSON.stringify({ username, password }),
  })
}

export function login(username: string, password: string): Promise<User> {
  return request<User>('/auth/login', {
    method: 'POST',
    body: JSON.stringify({ username, password }),
  })
}

export function logout(): Promise<void> {
  return request<void>('/auth/logout', { method: 'POST' })
}

export function me(): Promise<User> {
  return request<User>('/auth/me')
}

frontend/src/App.css:

:root {
  color-scheme: dark;
}

body {
  margin: 0;
  min-height: 100vh;
  background: #0a0a0f;
  color: #e6e6f0;
  font-family: Georgia, serif;
  display: flex;
  align-items: center;
  justify-content: center;
}

.app {
  width: min(90vw, 420px);
  padding: 2rem;
  text-align: center;
}

h1 {
  font-size: 2.5rem;
  letter-spacing: 0.1em;
  margin-bottom: 0.25rem;
}

.tagline {
  opacity: 0.6;
  margin-bottom: 2rem;
  font-style: italic;
}

form {
  display: flex;
  flex-direction: column;
  gap: 1rem;
  text-align: left;
}

label {
  display: flex;
  flex-direction: column;
  gap: 0.25rem;
  font-size: 0.9rem;
  opacity: 0.8;
}

input {
  background: #16161f;
  border: 1px solid #333;
  color: inherit;
  padding: 0.6rem;
  border-radius: 4px;
  font-size: 1rem;
}

button {
  background: #2a1f3d;
  border: 1px solid #5a3f8f;
  color: inherit;
  padding: 0.6rem 1rem;
  border-radius: 4px;
  cursor: pointer;
  font-size: 1rem;
}

button:hover {
  background: #3a2a55;
}

.mode-toggle {
  display: flex;
  gap: 0.5rem;
}

.mode-toggle button.active {
  background: #5a3f8f;
}

.error {
  color: #ff6b6b;
  font-size: 0.9rem;
}

frontend/src/main.tsx:

import { StrictMode } from 'react'
import { createRoot } from 'react-dom/client'
import App from './App'
import './App.css'

createRoot(document.getElementById('root')!).render(
  <StrictMode>
    <App />
  </StrictMode>,
)

• Step 2: Write the failing test

frontend/src/App.test.tsx:

import { render, screen, waitFor } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { describe, expect, it, vi } from 'vitest'
import App from './App'
import * as api from './api'

vi.mock('./api')

describe('App', () => {
  it('renders the Quantumancy title', async () => {
    vi.mocked(api.me).mockRejectedValue(new Error('not authenticated'))
    render(<App />)
    await waitFor(() => expect(screen.getByText('Quantumancy')).toBeInTheDocument())
  })

  it('logs in and shows the contact-established state', async () => {
    vi.mocked(api.me).mockRejectedValue(new Error('not authenticated'))
    vi.mocked(api.login).mockResolvedValue({ id: '1', username: 'medium1' })
    render(<App />)

    await waitFor(() => expect(screen.getByText('Quantumancy')).toBeInTheDocument())

    await userEvent.type(screen.getByLabelText('Username'), 'medium1')
    await userEvent.type(screen.getByLabelText('Password'), 'spookyspooky')
    await userEvent.click(screen.getByRole('button', { name: 'Enter' }))

    await waitFor(() =>
      expect(screen.getByText('Contact established as medium1.')).toBeInTheDocument(),
    )
  })
})

• Step 3: Install dependencies and run test to verify it fails

Run:

cd frontend
npm install
npm test

Expected: FAIL — Failed to resolve import "./App" (or similar), since App.tsx doesn't exist yet.

• Step 4: Write minimal implementation

frontend/src/App.tsx:

import { useEffect, useState } from 'react'
import { login, logout, me, register, type User } from './api'

function App() {
  const [user, setUser] = useState<User | null>(null)
  const [username, setUsername] = useState('')
  const [password, setPassword] = useState('')
  const [mode, setMode] = useState<'login' | 'register'>('login')
  const [error, setError] = useState<string | null>(null)
  const [checkingSession, setCheckingSession] = useState(true)

  useEffect(() => {
    me()
      .then(setUser)
      .catch(() => setUser(null))
      .finally(() => setCheckingSession(false))
  }, [])

  async function handleSubmit(event: React.FormEvent) {
    event.preventDefault()
    setError(null)
    try {
      if (mode === 'register') {
        await register(username, password)
      }
      const loggedInUser = await login(username, password)
      setUser(loggedInUser)
    } catch (err) {
      setError(err instanceof Error ? err.message : 'Something went wrong')
    }
  }

  async function handleLogout() {
    await logout()
    setUser(null)
  }

  if (checkingSession) {
    return (
      <main className="app">
        <p>Listening for a signal...</p>
      </main>
    )
  }

  return (
    <main className="app">
      <h1>Quantumancy</h1>
      <p className="tagline">Something is listening on the other side.</p>

      {user ? (
        <section>
          <p>Contact established as {user.username}.</p>
          <button onClick={handleLogout}>End session</button>
        </section>
      ) : (
        <form onSubmit={handleSubmit}>
          <div className="mode-toggle">
            <button
              type="button"
              className={mode === 'login' ? 'active' : ''}
              onClick={() => setMode('login')}
            >
              Reconnect
            </button>
            <button
              type="button"
              className={mode === 'register' ? 'active' : ''}
              onClick={() => setMode('register')}
            >
              First contact
            </button>
          </div>
          <label>
            Username
            <input value={username} onChange={(e) => setUsername(e.target.value)} required />
          </label>
          <label>
            Password
            <input
              type="password"
              value={password}
              onChange={(e) => setPassword(e.target.value)}
              required
              minLength={8}
            />
          </label>
          {error && <p className="error">{error}</p>}
          <button type="submit">{mode === 'login' ? 'Enter' : 'Begin'}</button>
        </form>
      )}
    </main>
  )
}

export default App

• Step 5: Run test to verify it passes

Run: cd frontend && npm test Expected: PASS (2/2)

• Step 6: Build the production bundle

Run: cd frontend && npm run build Expected: succeeds, produces frontend/dist/index.html and frontend/dist/assets/*. Task 2 requires this directory to exist.

• Step 7: Commit

git add frontend
git commit -m "feat: scaffold React frontend with login/register UI"

---

Task 2: Backend Serves the Frontend

Files:

• Modify: backend/app/main.py
• Test: backend/tests/test_frontend_serving.py

Interfaces:

• Consumes: frontend/dist/ (built in Task 1 — this task's tests will fail if it doesn't exist; that's expected, not a bug in this task).

• Produces: nothing new for later tasks — this is the terminal "make it visitable" step for the frontend.

• Step 1: Write the failing test

backend/tests/test_frontend_serving.py:

import pytest


@pytest.mark.asyncio
async def test_unknown_path_serves_spa_index(client):
    response = await client.get("/some/client/side/route")
    assert response.status_code == 200
    assert '<div id="root">' in response.text


@pytest.mark.asyncio
async def test_auth_routes_still_work_alongside_spa_fallback(client):
    response = await client.post(
        "/auth/register", json={"username": "frontendcheck", "password": "spookyspooky"}
    )
    assert response.status_code == 201

• Step 2: Run test to verify it fails

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_frontend_serving.py -v

Expected: FAIL — assert 404 == 200 (no catch-all route exists yet, unknown paths 404).

• Step 3: Write minimal implementation

backend/app/main.py (replace entirely):

from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

import app.models  # noqa: F401 — registers models on Base.metadata before create_all
from app.db import Base, engine
from app.routes.auth import router as auth_router

FRONTEND_DIST = Path(__file__).resolve().parent.parent.parent / "frontend" / "dist"


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield


app = FastAPI(title="Quantumancy", lifespan=lifespan)
app.include_router(auth_router)


@app.get("/healthz")
async def healthz():
    return {"status": "ok"}


app.mount("/assets", StaticFiles(directory=FRONTEND_DIST / "assets"), name="frontend-assets")


@app.get("/{full_path:path}")
async def serve_spa(full_path: str):
    return FileResponse(FRONTEND_DIST / "index.html")

Note: the catch-all serve_spa route is registered last, after /healthz and the auth_router include, so those routes still take precedence — FastAPI matches in registration order.

• Step 4: Run test to verify it passes

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_frontend_serving.py -v

Expected: PASS (2/2)

Then the full suite:

DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest -v

Expected: all pass (14 total: 12 from Plan 1 + 2 new).

• Step 5: Commit

git add backend/app/main.py backend/tests/test_frontend_serving.py
git commit -m "feat: serve built frontend from backend with SPA fallback"

---

Task 3: Ollama Client & Bounded Request Queue

Files:

• Create: backend/app/llm/__init__.py
• Create: backend/app/llm/client.py
• Create: backend/app/llm/queue.py
• Test: backend/tests/test_llm_queue.py

Interfaces:

• Consumes: settings.ollama_base_url (Plan 1's app.config).

• Produces: OllamaClient with async generate(model: str, prompt: str, system: str | None = None) -> str in app.llm.client; LLMQueue(max_concurrency: int, max_queue_depth: int) with async submit(coro_factory: Callable[[], Awaitable[T]]) -> T and class QueueFullError(Exception) in app.llm.queue — Plans 3-6 depend on LLMQueue.submit to serialize every spirit-mode LLM call.

• Step 1: Write the failing tests

backend/tests/test_llm_queue.py:

import asyncio

import pytest

from app.llm.queue import LLMQueue, QueueFullError


@pytest.mark.asyncio
async def test_queue_runs_calls_up_to_concurrency_limit():
    queue = LLMQueue(max_concurrency=2, max_queue_depth=10)
    concurrent_count = 0
    max_observed = 0

    async def slow_call():
        nonlocal concurrent_count, max_observed
        concurrent_count += 1
        max_observed = max(max_observed, concurrent_count)
        await asyncio.sleep(0.05)
        concurrent_count -= 1
        return "done"

    results = await asyncio.gather(*(queue.submit(slow_call) for _ in range(5)))

    assert results == ["done"] * 5
    assert max_observed == 2


@pytest.mark.asyncio
async def test_queue_raises_when_depth_exceeded():
    queue = LLMQueue(max_concurrency=1, max_queue_depth=1)

    async def slow_call():
        await asyncio.sleep(0.1)
        return "done"

    task1 = asyncio.create_task(queue.submit(slow_call))
    task2 = asyncio.create_task(queue.submit(slow_call))
    await asyncio.sleep(0.01)  # let task1 start running, task2 start waiting

    with pytest.raises(QueueFullError):
        await queue.submit(slow_call)

    await asyncio.gather(task1, task2)

• Step 2: Run test to verify it fails

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_llm_queue.py -v

Expected: FAIL with ModuleNotFoundError: No module named 'app.llm'

• Step 3: Write minimal implementation

backend/app/llm/__init__.py: (empty file)

backend/app/llm/client.py:

import httpx

from app.config import settings


class OllamaClient:
    def __init__(self, base_url: str | None = None):
        self._base_url = base_url or settings.ollama_base_url

    async def generate(self, model: str, prompt: str, system: str | None = None) -> str:
        payload = {"model": model, "prompt": prompt, "stream": False}
        if system is not None:
            payload["system"] = system

        async with httpx.AsyncClient(base_url=self._base_url, timeout=120.0) as http_client:
            response = await http_client.post("/api/generate", json=payload)
            response.raise_for_status()
            return response.json()["response"]

backend/app/llm/queue.py:

import asyncio
from collections.abc import Awaitable, Callable
from typing import TypeVar

T = TypeVar("T")


class QueueFullError(Exception):
    pass


class LLMQueue:
    """Bounds concurrent Ollama calls and rejects work once too much is queued."""

    def __init__(self, max_concurrency: int, max_queue_depth: int):
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self._max_queue_depth = max_queue_depth
        self._waiting = 0
        self._lock = asyncio.Lock()

    async def submit(self, coro_factory: Callable[[], Awaitable[T]]) -> T:
        async with self._lock:
            if self._waiting >= self._max_queue_depth:
                raise QueueFullError("too many seekers right now")
            self._waiting += 1

        try:
            async with self._semaphore:
                return await coro_factory()
        finally:
            async with self._lock:
                self._waiting -= 1

• Step 4: Run test to verify it passes

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_llm_queue.py -v

Expected: PASS (2/2)

• Step 5: Commit

git add backend/app/llm backend/tests/test_llm_queue.py
git commit -m "feat: add Ollama client and bounded request queue"

---

Task 4: Piper TTS Wrapper & Effects Chain

Files:

• Create: backend/app/tts/__init__.py
• Create: backend/app/tts/piper.py
• Create: backend/app/tts/effects.py
• Modify: backend/requirements.txt
• Test: backend/tests/test_tts_effects.py

Interfaces:

• Produces: PiperTTS(voice_model_path: str) with synthesize(text: str) -> bytes (returns WAV bytes) in app.tts.piper; apply_static_effect(wav_bytes: bytes, noise_level: float = 0.02) -> bytes in app.tts.effects — Plans 3-5 call these to render spirit voices.

Note: PiperTTS.synthesize requires a real Piper voice model file and the piper CLI, which this task does not install a model for (that's a deployment step for whichever plan first uses a mode with audio). This task's automated tests exercise only app.tts.effects, which is pure signal processing with no external dependency — piper.py's wrapper is written and typed correctly but is not covered by an automated test here, since doing so would require downloading a voice model as part of CI/test setup. Flag this honestly rather than faking a test against a mocked subprocess that wouldn't catch real integration issues.

• Step 1: Add numpy-based effects dependency check

numpy is already in backend/requirements.txt from Plan 1 (pre-installed for this purpose) — confirm it's there, no new entry needed. Add piper-tts for the CLI/runtime:

Append to backend/requirements.txt:

piper-tts==1.2.0

Run: cd backend && source venv/bin/activate && pip install -r requirements.txt

• Step 2: Write the failing test

backend/tests/test_tts_effects.py:

import struct
import wave
from io import BytesIO

from app.tts.effects import apply_static_effect


def _make_silent_wav(duration_seconds: float = 0.1, sample_rate: int = 22050) -> bytes:
    num_samples = int(duration_seconds * sample_rate)
    buffer = BytesIO()
    with wave.open(buffer, "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(struct.pack(f"<{num_samples}h", *([0] * num_samples)))
    return buffer.getvalue()


def test_apply_static_effect_returns_valid_wav_of_same_duration():
    original = _make_silent_wav()
    processed = apply_static_effect(original)

    with wave.open(BytesIO(original)) as original_wav:
        original_frames = original_wav.getnframes()
        original_rate = original_wav.getframerate()

    with wave.open(BytesIO(processed)) as processed_wav:
        assert processed_wav.getnframes() == original_frames
        assert processed_wav.getframerate() == original_rate
        assert processed_wav.getnchannels() == 1


def test_apply_static_effect_actually_adds_noise():
    original = _make_silent_wav()
    processed = apply_static_effect(original, noise_level=0.5)

    with wave.open(BytesIO(processed)) as processed_wav:
        frames = processed_wav.readframes(processed_wav.getnframes())

    # A silent input run through noise injection should no longer be all-zero.
    assert any(byte != 0 for byte in frames)

• Step 3: Run test to verify it fails

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_tts_effects.py -v

Expected: FAIL with ModuleNotFoundError: No module named 'app.tts'

• Step 4: Write minimal implementation

backend/app/tts/__init__.py: (empty file)

backend/app/tts/effects.py:

import wave
from io import BytesIO

import numpy as np


def apply_static_effect(wav_bytes: bytes, noise_level: float = 0.02) -> bytes:
    """Adds white noise to a mono 16-bit PCM WAV, simulating spirit-box static."""
    with wave.open(BytesIO(wav_bytes)) as wav_in:
        params = wav_in.getparams()
        frames = wav_in.readframes(wav_in.getnframes())

    samples = np.frombuffer(frames, dtype=np.int16).astype(np.float32)
    noise = np.random.normal(0, noise_level * 32767, size=samples.shape)
    noisy_samples = np.clip(samples + noise, -32768, 32767).astype(np.int16)

    output = BytesIO()
    with wave.open(output, "wb") as wav_out:
        wav_out.setparams(params)
        wav_out.writeframes(noisy_samples.tobytes())
    return output.getvalue()

backend/app/tts/piper.py:

import subprocess


class PiperTTS:
    """Wraps the `piper` CLI to synthesize speech locally, no cloud calls."""

    def __init__(self, voice_model_path: str):
        self._voice_model_path = voice_model_path

    def synthesize(self, text: str) -> bytes:
        result = subprocess.run(
            ["piper", "--model", self._voice_model_path, "--output-raw"],
            input=text.encode("utf-8"),
            capture_output=True,
            check=True,
        )
        return result.stdout

• Step 5: Run test to verify it passes

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_tts_effects.py -v

Expected: PASS (2/2)

• Step 6: Commit

git add backend/app/tts backend/requirements.txt backend/tests/test_tts_effects.py
git commit -m "feat: add Piper TTS wrapper and static-noise effects chain"

---

Task 5: WebSocket Session Channel

Files:

• Create: backend/app/models/contact_session.py
• Modify: backend/app/models/__init__.py
• Create: backend/app/ws.py
• Modify: backend/app/main.py
• Test: backend/tests/test_ws_session.py

Interfaces:

• Consumes: get_current_user (Plan 1's app.deps), Base/get_db (Plan 1's app.db).

• Produces: ContactSession model (id, user_id, mode, started_at, ended_at) in app.models.contact_session; a WS /ws/session endpoint — Plans 3-5 build their per-mode message handling on top of this connection.

• Step 1: Write the failing test

backend/tests/test_ws_session.py:

import pytest
from sqlalchemy import select

from app.models.contact_session import ContactSession


@pytest.mark.asyncio
async def test_websocket_requires_authentication(client):
    with pytest.raises(Exception):
        async with client.websocket_connect("/ws/session") as ws:
            await ws.receive_json()


@pytest.mark.asyncio
async def test_websocket_ping_pong_and_session_lifecycle(client, db_session):
    await client.post("/auth/register", json={"username": "wsmedium", "password": "spookyspooky"})
    await client.post("/auth/login", json={"username": "wsmedium", "password": "spookyspooky"})

    with client.websocket_connect("/ws/session") as ws:
        ws.send_json({"type": "ping"})
        response = ws.receive_json()
        assert response == {"type": "pong"}

        sessions = (await db_session.execute(select(ContactSession))).scalars().all()
        assert len(sessions) == 1
        assert sessions[0].ended_at is None

    sessions = (await db_session.execute(select(ContactSession))).scalars().all()
    assert sessions[0].ended_at is not None

Note: httpx.AsyncClient (used by the existing async client fixture) does not support WebSocket testing — websocket_connect is a starlette.testclient.TestClient (sync) method. This test uses a second, differently-named fixture, sync_client, plus a db_session fixture for direct DB assertions after the socket closes. Do not modify or shadow the existing async client fixture — add these as new, additional fixtures alongside it.

Replace backend/tests/test_ws_session.py's contents with this final version (uses sync_client throughout, since TestClient also handles regular HTTP calls, so there's no need to mix it with the async client fixture in this file):

import pytest
from sqlalchemy import select

from app.models.contact_session import ContactSession


def test_websocket_requires_authentication(sync_client):
    with pytest.raises(Exception):
        with sync_client.websocket_connect("/ws/session"):
            pass


@pytest.mark.asyncio
async def test_websocket_ping_pong_and_session_lifecycle(sync_client, db_session):
    sync_client.post("/auth/register", json={"username": "wsmedium", "password": "spookyspooky"})
    sync_client.post("/auth/login", json={"username": "wsmedium", "password": "spookyspooky"})

    with sync_client.websocket_connect("/ws/session") as ws:
        ws.send_json({"type": "ping"})
        response = ws.receive_json()
        assert response == {"type": "pong"}

        sessions = (await db_session.execute(select(ContactSession))).scalars().all()
        assert len(sessions) == 1
        assert sessions[0].ended_at is None

    sessions = (await db_session.execute(select(ContactSession))).scalars().all()
    assert sessions[0].ended_at is not None

Add these two new fixtures to backend/tests/conftest.py (append — do not touch the existing async client/_reset_db fixtures):

from fastapi.testclient import TestClient


@pytest.fixture
def sync_client():
    return TestClient(app)


@pytest_asyncio.fixture
async def db_session():
    async with TestSessionLocal() as session:
        yield session

TestClient and pytest need importing at the top of conftest.py — pytest itself isn't imported yet there (only pytest_asyncio is), so add import pytest alongside the existing imports.

• Step 2: Run test to verify it fails

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_ws_session.py -v

Expected: FAIL — connection to /ws/session fails outright (no route registered), or ModuleNotFoundError for app.models.contact_session.

• Step 3: Write minimal implementation

backend/app/models/contact_session.py:

import uuid
from datetime import datetime, timezone

from sqlalchemy import DateTime, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base


class ContactSession(Base):
    __tablename__ = "contact_sessions"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
    mode: Mapped[str] = mapped_column(String(32), default="unknown")
    started_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
    )
    ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

backend/app/models/__init__.py (replace entirely):

from app.models.auth_session import AuthSession
from app.models.contact_session import ContactSession
from app.models.user import User

__all__ = ["User", "AuthSession", "ContactSession"]

backend/app/ws.py:

from datetime import datetime, timezone

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from sqlalchemy import select

from app.db import async_session_maker
from app.deps import SESSION_COOKIE_NAME
from app.models.auth_session import AuthSession, hash_token
from app.models.contact_session import ContactSession

router = APIRouter()


async def _authenticate(websocket: WebSocket):
    raw_token = websocket.cookies.get(SESSION_COOKIE_NAME)
    if raw_token is None:
        return None

    token_hash = hash_token(raw_token)
    async with async_session_maker() as db:
        result = await db.execute(select(AuthSession).where(AuthSession.token_hash == token_hash))
        session = result.scalar_one_or_none()
        if session is None or session.expires_at < datetime.now(timezone.utc):
            return None
        return session.user_id


@router.websocket("/ws/session")
async def session_socket(websocket: WebSocket):
    user_id = await _authenticate(websocket)
    if user_id is None:
        await websocket.close(code=4401)
        return

    await websocket.accept()

    async with async_session_maker() as db:
        contact_session = ContactSession(user_id=user_id)
        db.add(contact_session)
        await db.commit()
        await db.refresh(contact_session)

    try:
        while True:
            message = await websocket.receive_json()
            if message.get("type") == "ping":
                await websocket.send_json({"type": "pong"})
    except WebSocketDisconnect:
        pass
    finally:
        async with async_session_maker() as db:
            session_to_close = await db.get(ContactSession, contact_session.id)
            session_to_close.ended_at = datetime.now(timezone.utc)
            await db.commit()

backend/app/main.py — add the WebSocket router (insert alongside the existing auth_router include, before the catch-all SPA route):

from app.ws import router as ws_router
...
app.include_router(auth_router)
app.include_router(ws_router)

• Step 4: Run test to verify it passes

Run:

cd backend && source venv/bin/activate
DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest tests/test_ws_session.py -v

Expected: PASS (2/2)

Then the full suite:

DATABASE_URL=postgresql+asyncpg://quantumancy:quantumancy@localhost:5432/quantumancy_test \
OLLAMA_BASE_URL=http://10.30.20.107:11434 \
SESSION_SECRET=test-secret \
python -m pytest -v

Expected: all pass (20 total: 14 from Tasks 1-2 range + 2 from Task 3 + 2 from Task 4 + 2 from Task 5).

• Step 5: Commit

git add backend/app/models/contact_session.py backend/app/models/__init__.py backend/app/ws.py backend/app/main.py backend/tests/test_ws_session.py backend/tests/conftest.py
git commit -m "feat: add WebSocket session channel with ContactSession lifecycle"

---

Self-Review

Spec coverage: Frontend React/Vite SPA served by backend (spec §2) → Tasks 1-2. Ollama request queue (spec §2 "known operational constraint") → Task 3. Local Piper TTS (spec §6) → Task 4 (effects chain fully tested; piper.py's subprocess wrapper is honestly flagged as untested pending a real voice model, deferred to whichever Plan 3-5 task first plays audio). WebSocket per-session channel implied by spec §3's "streamed to the frontend as they occur" real-time requirement → Task 5. Rate limiting enforcement onto these endpoints is explicitly NOT in this plan — no LLM-triggering HTTP/WS endpoint with real spirit-mode content exists yet; that lands in Plan 3+ alongside the RateLimiter wiring.

Placeholder scan: none — every step has complete, runnable code, and the one deliberate test gap (Piper's subprocess wrapper) is explicitly justified rather than faked with a hollow mock.

Type consistency: LLMQueue.submit(coro_factory) signature matches its two test usages (queue.submit(slow_call) passing a zero-arg async callable). OllamaClient.generate signature (model, prompt, system=None) is what Plans 3-5 will call. ContactSession fields (id, user_id, mode, started_at, ended_at) match the spec's contact_sessions naming resolution from Plan 1. apply_static_effect(wav_bytes, noise_level=0.02) signature matches both test calls (one default, one explicit noise_level=0.5).

---

Plan complete and saved to docs/superpowers/plans/2026-07-20-quantumancy-02-frontend-llm-pipeline.md. Two execution options:

1. Subagent-Driven (recommended) - I dispatch a fresh subagent per task, review between tasks, fast iteration

2. Inline Execution - Execute tasks in this session using executing-plans, batch execution with checkpoints

Which approach?