Build durable extraction backend with recovery, OAuth and MCP

.env.example

@@ -1,11 +1,32 @@
-# Required — Postgres connection string (Neon, Supabase, local, whatever).
-DATABASE_URL="postgresql://user:password@host:5432/dbname"
+# Copy this file to .env. Node 22+ is required.
+# Local Docker setup: docker compose up -d db
+DATABASE_URL="postgresql://lippy:lippy@localhost:5432/lippy?schema=public"
+APP_URL="http://localhost:3000"
 
-# Optional — omit either to run that client in mock mode.
-BRIGHTDATA_API_KEY=
+# google = sign in + session/API-token authentication.
+# development = explicit local auth bypass for curl/demo; NEVER works in production.
+AUTH_MODE="google"
+GOOGLE_CLIENT_ID=""
+GOOGLE_CLIENT_SECRET=""
+# Register this exact authorized redirect URI in Google Cloud:
+# http://localhost:3000/api/auth/google/callback
+# For deployment, set APP_URL to the HTTPS origin and register its callback URI.
+
+# fixture = deterministic HTML extraction + selector repair, no paid API calls.
+# brightdata = real Scraper Studio API calls; no silent fallback to fixture mode.
+EXTRACTION_PROVIDER="fixture"
+BRIGHTDATA_API_KEY=""
 BRIGHTDATA_API_BASE="https://api.brightdata.com"
+# Required ONLY when creating new Bright Data collectors; provider emails results here.
+# Leave empty if every contract supplies an existing, dedicated collectorId.
+BRIGHTDATA_DELIVERY_EMAIL=""
 
-PARALLEL_API_KEY=
-PARALLEL_API_BASE="https://api.parallel.ai"
+# Worker / provider polling and bounded recovery
+WORKER_POLL_MS=1000
+PROVIDER_POLL_MS=5000
+MAX_HEAL_ATTEMPTS=3
+RUN_TIMEOUT_MS=1800000
 
-WEBHOOK_SIGNING_SECRET="dev-secret-change-me"
+# Optional: API token used by npm run demo when AUTH_MODE=google.
+# Issue one using POST /api/auth/tokens after Google sign-in (see README).
+LIPPY_API_TOKEN=""

.github/workflows/ci.yml

@@ -0,0 +1,39 @@
+name: Backend CI
+on:
+  pull_request:
+  push:
+    branches: [main]
+jobs:
+  verify:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: lippy
+          POSTGRES_PASSWORD: lippy
+          POSTGRES_DB: lippy_test
+        ports: ["5432:5432"]
+        options: >-
+          --health-cmd "pg_isready -U lippy"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      DATABASE_URL: postgresql://lippy:lippy@localhost:5432/lippy_test
+      AUTH_MODE: google
+      EXTRACTION_PROVIDER: fixture
+      NEXT_TELEMETRY_DISABLED: "1"
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "22"
+          cache: npm
+      - run: npm ci
+      - run: npm run db:generate
+      - run: npm run db:deploy
+      - run: npm run lint
+      - run: npm run typecheck
+      - run: npm test
+        env:
+          RUN_DB_TESTS: "1"
+      - run: npm run build

README.md

@@ -1,67 +1,114 @@
 # Lippy Agent
 
+Backend infrastructure for reliable structured web extraction. Define a URL and typed contract; a separate worker extracts data, validates it, repairs broken selectors, verifies changes and publishes durable events. REST and MCP share the same authenticated service layer.
 
-Websites change. Your API shouldn't. See [`SPEC.md`](./SPEC.md) for the full
-architecture, judging-criteria mapping, and demo script.
+## Run locally
 
-## Setup
+Requires Node.js 22+ and PostgreSQL. Docker is optional if you already have PostgreSQL.
 
-```bash
-npm install
-cp .env.example .env      # works out of the box in mock mode, no keys needed
+```sh
+npm ci
+cp .env.example .env
+docker compose up -d db
 npm run db:generate
-npm run db:migrate
+npm run db:deploy
 npm run dev
+# In another terminal:
+npm run worker
 ```
 
-Runs fully in **mock mode** with zero API keys — `BrightDataClient` and
-`ParallelClient` both fall back to deterministic mock responses so the
-create → run → violate → heal → verify → semantic-diff loop is runnable
-end to end out of the box. Set `BRIGHTDATA_API_KEY` / `PARALLEL_API_KEY` in
-`.env` to hit the real APIs.
+Fill `.env` using the explanations in `.env.example`. For a zero-API-key local demonstration, set `AUTH_MODE=development` and `EXTRACTION_PROVIDER=fixture`, then run `npm run demo` with the API and worker running. Development authentication never works with `NODE_ENV=production`. The demo creates a disabled contract, verifies a baseline, changes its HTML, proves selector recovery, and quarantines a suspicious price change while retaining accepted data. Fixture data is isolated per contract and stored in PostgreSQL, not in process-global flags.
 
-## Project layout
+For real extraction set `EXTRACTION_PROVIDER=brightdata` and `BRIGHTDATA_API_KEY`. Supply an existing **dedicated batch-mode** Scraper Studio `collectorId` per contract, or set `BRIGHTDATA_DELIVERY_EMAIL` to let the worker create and generate a collector. Newly created collectors also deliver results to that email. Field names/types must match an existing collector's output. Contracts currently represent one page and exactly one output record.
 
+## Google OAuth and API tokens
+
+1. Create a Google OAuth web application and copy its client ID/secret into `.env`.
+2. Register `http://localhost:3000/api/auth/google/callback` as an authorized redirect URI. In deployment use the HTTPS `APP_URL` and its corresponding callback.
+3. With `AUTH_MODE=google`, open `/api/auth/google` in your browser. A successful login redirects to `/api/auth/me` and sets an HttpOnly session cookie.
+4. From that same origin, request an API token:
+
+```js
+await fetch('/api/auth/tokens', { method: 'POST' }).then(r => r.json())
 ```
-src/
-  app/                    Next.js App Router — routes are thin adapters only
-    api/contracts/        REST surface (create, get, run, live event stream)
-    api/mcp/              MCP server over Streamable HTTP
-  lib/
-    contracts/            schema (Zod), validator, service layer
-    brightdata/           typed client (create/run/heal/approve), mock-capable
-    parallel/              semantic-diff client, mock-capable
-    selfheal/              the orchestrator state machine
-    events/                typed event union + pub/sub bus
-    env.ts errors.ts logger.ts db.ts
-  mcp/server.ts            tool definitions (create_contract, watch_contract, get_data)
-  tests/                   vitest unit tests
-prisma/schema.prisma       Contract / Run / ContractEvent models
+
+The returned token is shown once; store it securely. REST and MCP clients send `Authorization: Bearer TOKEN`. API tokens expire after 30 days, sessions after 7 days. Delete a token with `DELETE /api/auth/tokens` and JSON `{ "id": "TOKEN_ID" }`. Log out with `POST /api/auth/logout`. Cookie-authenticated writes require the matching `Origin`; bearer clients do not. Cross-origin requests are rejected. Google login uses one-time state, PKCE, nonce and signature/issuer/audience verification. Only token hashes are stored.
+
+## REST example
+
+In development-auth mode omit the Authorization header. Set TOKEN in your shell for Google mode.
+
+```sh
+curl -X POST http://localhost:3000/api/contracts \
+  -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
+  -d '{"url":"https://example.com/product","enabled":false,"fields":[
+    {"key":"product","type":"string","expectedValue":"Example product"},
+    {"key":"price","type":"number","minimum":0,"maxRelativeChange":0.5}
+  ]}'
+
+curl -X POST http://localhost:3000/api/contracts/CONTRACT_ID/run \
+  -H "Authorization: Bearer $TOKEN" -H 'Idempotency-Key: first-run'
+# 202 + {runId, state}; Location points to /api/runs/RUN_ID.
+
+curl http://localhost:3000/api/runs/RUN_ID -H "Authorization: Bearer $TOKEN"
+curl http://localhost:3000/api/contracts/CONTRACT_ID/data -H "Authorization: Bearer $TOKEN"
+curl -N http://localhost:3000/api/contracts/CONTRACT_ID/events \
+  -H "Authorization: Bearer $TOKEN" -H 'Accept: text/event-stream'
 ```
 
-## Commands
-
-| Command | What it does |
-|---|---|
-| `npm run dev` | Start the dev server |
-| `npm test` | Run the vitest suite |
-| `npm run typecheck` | `tsc --noEmit` |
-| `npm run db:studio` | Browse the SQLite DB visually |
-
-## Try it
-
-```bash
-# create a contract
-curl -X POST localhost:3000/api/contracts -H 'content-type: application/json' -d '{
-  "url": "https://example.com/product",
-  "fields": [
-    { "key": "product", "type": "string", "required": true },
-    { "key": "price", "type": "number", "required": true },
-    { "key": "stock", "type": "boolean", "required": false }
-  ]
-}'
-
-# trigger a run (mock mode breaks the extraction every 3rd call, so re-run
-# a few times to see contract.violated -> contract.healing -> contract.healed)
-curl -X POST localhost:3000/api/contracts/<id>/run
+| Endpoint | Behavior |
+| --- | --- |
+| `GET/POST /api/contracts` | List/create owned contracts |
+| `GET /api/contracts/:id` | Contract, recent runs and initial event history |
+| `PATCH /api/contracts/:id` | Pause/resume scheduling using `{"enabled":false/true}` |
+| `POST /api/contracts/:id/run` | Enqueue; optional `Idempotency-Key` deduplicates retries |
+| `GET /api/runs/:id` | State, raw/validated output, violations, provider job ID and repair history |
+| `GET /api/contracts/:id/data` | Last accepted data and its run/timestamp; includes current health |
+| `GET /api/contracts/:id/events?after=SEQUENCE` | Durable event pages; SSE with `Accept: text/event-stream` |
+| `POST /api/runs/:id/review` | `{"decision":"accept"}` or `{"decision":"reject"}` for quarantined results |
+| `POST /api/runs/:id/reconcile` | Record manual reconciliation of an indeterminate provider submission |
+| `POST /api/dev/break` | Non-production fixture mutation: `{contractId, scenario:"layout"}`; also `value` with `field`/`value`, or `unrepairable` |
+| `GET /api/health` | Database readiness |
+
+Review and reconciliation leave scheduling paused. Explicitly resume it after resolving the problem. The reconciliation request requires a descriptive `note`; it can attach `collectorId` after you identify a scraper created during an interrupted request. Verify/cancel outstanding provider work in Scraper Studio before reconciling.
+
+## MCP
+
+Connect a Streamable HTTP MCP client to `/api/mcp` with an API bearer token. Available tools: `create_contract`, `list_contracts`, `run_contract`, `get_run`, `get_data`, `watch_contract`. `watch_contract` now reads durable events after a cursor; it does not trigger extraction. Tool calls are stateless JSON responses. Event streaming uses the REST SSE endpoint, with `Last-Event-ID` for reconnect replay. API tokens are configured explicitly; this server does not implement MCP OAuth discovery/dynamic client registration.
+
+## Verification and guarantees
+
+- PostgreSQL stores queued work, state transitions, raw output, repair attempts and events. Multiple workers claim jobs with `FOR UPDATE SKIP LOCKED`, expiring leases and fencing tokens. One active run per contract is serialized with a database row lock.
+- Each bounded step releases its lease. Polling states survive restarts and read failures use bounded backoff. All runs have a deadline and repair budget. Repeated or unrecoverable failures pause scheduling.
+- State and event writes share a transaction. Publishers serialize per contract before allocating event sequences, so a cursor cannot skip a later-committing earlier event. Each SSE/MCP subscriber owns its cursor. This is PostgreSQL-backed pub/sub; it does not depend on Redis or process-local fanout.
+- A crash between a non-idempotent provider request and storing its response is **indeterminate**, not silently retried. Exactly-once external side effects are not claimed.
+- Schema verification supports required fields, types, numeric bounds and expected identity/value. Semantic difference verification uses explicit normalization (Unicode/whitespace, optional case folding, URL canonicalization) and numeric relative-change limits. It reports initialization, meaningful changes and suspicious changes under those rules. It is not general-purpose AI truth verification. `confidence: 1` means the rule matched deterministically, not that a model proved source accuracy.
+- Bright Data repair is approved and saved upstream, then the extraction is run again. Only verified data is promoted locally. The integration does not offer isolated candidate scrapers or automatic upstream rollback. Fields without identity/bounds constraints can still contain semantically wrong but type-valid data.
+- No live Bright Data or Google login is claimed as tested without real credentials. Provider protocol tests use documented response fixtures. Removed the previous undocumented Parallel AI diff endpoint and silent "no change" fallback.
+
+See [architecture and failure walkthrough](docs/architecture.md).
+
+## Tests and deployment
+
+```sh
+npm run lint
+npm run typecheck
+npm test
+# Explicitly use a dedicated migrated TEST database:
+DATABASE_URL=postgresql://user:password@localhost:5432/lippy_test RUN_DB_TESTS=1 npm test
+npm run build
+npm start
+# Separate long-running worker process, with the same .env:
+npm run worker
 ```
+
+The CI workflow provisions PostgreSQL, applies migrations, runs unit/integration tests and builds the API. Deploy the API and at least one worker against the same database. Use HTTPS, `AUTH_MODE=google`, and configure secrets on both processes. Polling subscriptions need a hosting environment that allows streaming connections; long extraction jobs execute only in workers.
+
+For an existing database already matching the old schema, back it up and mark the baseline migration applied before deploying the second migration:
+
+```sh
+npx prisma migrate resolve --applied 202609140001_baseline
+npm run db:deploy
+```
+
+Legacy history is preserved, legacy interrupted runs are closed, and unowned contracts are paused and inaccessible until deliberately assigned to a user. Legacy mock collector IDs are cleared. Duplicate real collector IDs must be resolved before migration because each contract now requires a dedicated collector.

app/api/auth/google/callback/route.ts

@@ -0,0 +1,12 @@
+import { NextResponse } from "next/server";
+import { completeGoogleLogin, cookie, cookieOptions, STATE_COOKIE, SESSION_COOKIE } from "@/lib/auth/service";
+import { endpoint } from "@/lib/http";
+import { env } from "@/lib/env";
+export async function GET(req: Request) { return endpoint(async () => {
+  const url = new URL(req.url);
+  const session = await completeGoogleLogin(url.searchParams.get("code") ?? "", url.searchParams.get("state") ?? "", cookie(req, STATE_COOKIE));
+  const response = NextResponse.redirect(`${env.APP_URL}/api/auth/me`);
+  response.cookies.set(SESSION_COOKIE, session.token, { ...cookieOptions, expires: session.expiresAt });
+  response.cookies.set(STATE_COOKIE, "", { ...cookieOptions, maxAge: 0 });
+  return response;
+}); }

app/api/auth/google/route.ts

@@ -0,0 +1,10 @@
+import { NextResponse } from "next/server";
+import { beginGoogleLogin, STATE_COOKIE, cookieOptions } from "@/lib/auth/service";
+import { endpoint } from "@/lib/http";
+export const dynamic = "force-dynamic";
+export async function GET() { return endpoint(async () => {
+  const login = await beginGoogleLogin();
+  const response = NextResponse.redirect(login.url);
+  response.cookies.set(STATE_COOKIE, login.state, { ...cookieOptions, maxAge: 600 });
+  return response;
+}); }

app/api/auth/logout/route.ts

@@ -0,0 +1,10 @@
+import { NextResponse } from "next/server";
+import { requireUser, cookie, SESSION_COOKIE, hashToken, cookieOptions } from "@/lib/auth/service";
+import { db } from "@/lib/db";
+import { endpoint } from "@/lib/http";
+export async function POST(req: Request) { return endpoint(async () => {
+  const userId = await requireUser(req), token = cookie(req, SESSION_COOKIE);
+  if (token) await db.session.deleteMany({ where: { userId, tokenHash: hashToken(token) } });
+  const response = NextResponse.json({ signedOut: true });
+  response.cookies.set(SESSION_COOKIE, "", { ...cookieOptions, maxAge: 0 }); return response;
+}); }

app/api/auth/me/route.ts

@@ -0,0 +1,4 @@
+import { requireUser } from "@/lib/auth/service";
+import { db } from "@/lib/db";
+import { endpoint } from "@/lib/http";
+export async function GET(req: Request) { return endpoint(async () => Response.json({ user: await db.user.findUnique({ where: { id: await requireUser(req) }, select: { id: true, email: true } }) }, { headers: { "Cache-Control": "no-store" } })); }

app/api/auth/tokens/route.ts

@@ -0,0 +1,10 @@
+import { z } from "zod";
+import { requireUser, issueToken } from "@/lib/auth/service";
+import { db } from "@/lib/db";
+import { endpoint, body } from "@/lib/http";
+export async function POST(req: Request) { return endpoint(async () => Response.json(await issueToken(await requireUser(req), "api"), { status: 201, headers: { "Cache-Control": "no-store" } })); }
+export async function DELETE(req: Request) { return endpoint(async () => {
+  const owner = await requireUser(req), { id } = z.object({ id: z.string() }).parse(await body(req));
+  await db.session.deleteMany({ where: { id, userId: owner, kind: "api" } });
+  return new Response(null, { status: 204 });
+}); }

app/api/contracts/[id]/data/route.ts

@@ -0,0 +1,4 @@
+import { requireUser } from "@/lib/auth/service";
+import { getData } from "@/lib/contracts/service";
+import { endpoint } from "@/lib/http";
+export async function GET(req: Request, ctx: { params: Promise<{ id: string }> }) { return endpoint(async () => Response.json(await getData((await ctx.params).id, await requireUser(req)))); }