app/api/0pt/documents/[documentId]/extraction/route.ts

@@ -1,6 +1,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { parseBody, documentExtractionSchema } from "@/src/validation/0pt-schemas";
 import { documentExtractionService } from "@/src/services/document-extraction-service";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function POST(request: NextRequest, { params }: { params: Promise<{ documentId: string }> }) {
   try {
@@ -11,6 +12,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const result = await documentExtractionService.extractDocument(documentId, {
       tenant_id: body.tenant_id,
       project_id: body.project_id,
@@ -35,10 +38,17 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
   }
 }
 
-export async function GET(_request: NextRequest, { params }: { params: Promise<{ documentId: string }> }) {
+export async function GET(request: NextRequest, { params }: { params: Promise<{ documentId: string }> }) {
   try {
     const { documentId } = await params;
-    const extractions = await documentExtractionService.listExtractionsForDocument(documentId);
+    const { searchParams } = new URL(request.url);
+    const tenantId = searchParams.get("tenant_id");
+    if (!tenantId) {
+      return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
+    }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
+    const extractions = await documentExtractionService.listExtractionsForDocument(documentId, tenantId);
     return NextResponse.json(extractions);
   } catch (error: unknown) {
     const message = error instanceof Error ? error.message : "Unknown error";

app/api/0pt/documents/capture/route.ts

@@ -1,6 +1,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { parseBody, documentCaptureSchema } from "@/src/validation/0pt-schemas";
 import { documentCaptureService } from "@/src/services/document-capture-service";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function POST(request: NextRequest) {
   try {
@@ -10,6 +11,8 @@ export async function POST(request: NextRequest) {
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const result = await documentCaptureService.captureFromMailbox(
       {
         mailbox: {
@@ -26,8 +29,18 @@ export async function POST(request: NextRequest) {
       {
         tenant_id: body.tenant_id,
         project_id: body.project_id,
-        actor_id: body.actor_id,
-        correlation_id: body.correlation_id,
+        // documentCaptureSchema types these optional (pre-existing, listed
+        // among package.json's "9 pre-existing errors ... Wave 2 Zod
+        // schemas type fields as plain string while the services want
+        // narrower unions" -- deliberately left to its own change, not
+        // widened here). CaptureMailboxContext requires actor_id as a
+        // non-empty string and captureFromMailbox already throws a clear
+        // "missing tenant_id/project_id/actor_id" error when it's falsy, so
+        // `?? ""` changes no runtime behavior -- it only satisfies the type
+        // checker, which previously never saw this file (nothing under
+        // tests/ imported it before P014's cross-tenant regression tests).
+        actor_id: body.actor_id ?? "",
+        correlation_id: body.correlation_id ?? "",
       },
     );
     return NextResponse.json(result, { status: 201 });
@@ -37,9 +50,20 @@ export async function POST(request: NextRequest) {
   }
 }
 
+// P014: `tenant_id` used to be optional here, and
+// DocumentCaptureService.listDocuments treats an absent tenant_id filter as
+// "no restriction" — so calling this with no query params at all returned
+// EVERY tenant's captured documents to any authenticated caller. Every
+// sibling list/get route in this file already requires tenant_id; this was
+// the one outlier. Now required, same as the rest of the surface.
 export async function GET(request: NextRequest) {
   const { searchParams } = new URL(request.url);
-  const tenantId = searchParams.get("tenant_id") ?? undefined;
+  const tenantId = searchParams.get("tenant_id");
+  if (!tenantId) {
+    return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
+  }
+  const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+  if (tenantGuard) return tenantGuard;
   const projectId = searchParams.get("project_id") ?? undefined;
   const documents = await documentCaptureService.listDocuments({ tenant_id: tenantId, project_id: projectId });
   return NextResponse.json(documents);

app/api/0pt/lakehouse/mappings/[mappingId]/route.ts

@@ -1,10 +1,25 @@
-import { NextResponse } from "next/server";
+import { NextRequest, NextResponse } from "next/server";
 import { mappingService } from "@/src/services/lakehouse-mapping-registry";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
-export async function GET(_request: Request, { params }: { params: Promise<{ mappingId: string }> }) {
+// P014: this lookup used to accept only `mappingId`, with NO tenant check
+// anywhere in the request path -- any authenticated caller could read
+// another tenant's dataset<->Iceberg-table mapping (namespace, table name,
+// storage URI) by guessing or enumerating IDs. Now requires `?tenant_id=`
+// and both checks it against the verified token and threads it into the
+// service so a foreign mapping_id 404s instead of leaking -- see
+// DatasetIcebergMappingService.getMapping's doc comment.
+export async function GET(request: NextRequest, { params }: { params: Promise<{ mappingId: string }> }) {
   try {
     const { mappingId } = await params;
-    const mapping = await mappingService.getMapping(mappingId);
+    const { searchParams } = new URL(request.url);
+    const tenantId = searchParams.get("tenant_id");
+    if (!tenantId) {
+      return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
+    }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
+    const mapping = await mappingService.getMapping(mappingId, tenantId);
     return NextResponse.json(mapping);
   } catch (error: unknown) {
     const message = error instanceof Error ? error.message : "Unknown error";

app/api/0pt/lakehouse/mappings/[mappingId]/writes/route.ts

@@ -1,6 +1,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { parseBody, requestIcebergWriteSchema } from "@/src/validation/0pt-schemas";
 import { writeService } from "@/src/services/lakehouse-mapping-registry";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 // Registers a new Iceberg snapshot (a write) through ST0RE's table registry
 // for an already-mapped dataset. See docs/ICEBERG_WRITE_BOUNDARY.md.
@@ -22,8 +23,11 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const result = await writeService.requestIcebergWrite({
       dataset_iceberg_mapping_id: mappingId,
+      tenant_id: body.tenant_id,
       actor_id: body.actor_id,
       correlation_id: body.correlation_id,
       snapshot_type: body.snapshot_type,

app/api/0pt/lakehouse/mappings/route.ts

@@ -1,6 +1,8 @@
 import { NextRequest, NextResponse } from "next/server";
 import { parseBody, mapDatasetToIcebergTableSchema } from "@/src/validation/0pt-schemas";
 import { mappingService, writeService } from "@/src/services/lakehouse-mapping-registry";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
+import type { IcebergTableFormat } from "@/src/clients/st0re-lakehouse-client";
 
 // Shared instances come from src/services/lakehouse-mapping-registry.ts so a
 // mapping created via POST is visible to a later GET, AND to the sibling
@@ -28,6 +30,8 @@ export async function POST(request: NextRequest) {
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const result = await writeService.mapDatasetToIcebergTable({
       tenant_id: body.tenant_id,
       project_id: body.project_id,
@@ -37,7 +41,17 @@ export async function POST(request: NextRequest) {
       table_name: body.table_name,
       storage_uri: body.storage_uri,
       catalog_ref: body.catalog_ref,
-      table_format: body.table_format,
+      // mapDatasetToIcebergTableSchema types this as a plain string
+      // (pre-existing, listed among package.json's "9 pre-existing errors
+      // ... Wave 2 Zod schemas type fields as plain string while the
+      // services want narrower unions" -- deliberately left to its own
+      // change, not widened here). This cast changes no runtime behavior:
+      // an unrecognized value flows through exactly as it did before
+      // (ST0RE's own table-registration route is the actual validator) --
+      // it only satisfies the type checker, which previously never saw this
+      // file (nothing under tests/ imported it before P014's cross-tenant
+      // regression tests).
+      table_format: body.table_format as IcebergTableFormat | undefined,
       actor_id: body.actor_id,
       correlation_id: body.correlation_id,
     });
@@ -48,10 +62,21 @@ export async function POST(request: NextRequest) {
   }
 }
 
+// P014: tenant_id used to be optional here, and
+// DatasetIcebergMappingService.listMappings treats an absent tenant_id
+// filter as "no restriction" -- omitting the query param returned every
+// tenant's dataset<->Iceberg-table mappings (namespace, table name, storage
+// URI). Now required and cross-checked, matching the rest of the surface.
 export async function GET(request: NextRequest) {
   const { searchParams } = new URL(request.url);
+  const tenantId = searchParams.get("tenant_id");
+  if (!tenantId) {
+    return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
+  }
+  const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+  if (tenantGuard) return tenantGuard;
   const mappings = await mappingService.listMappings({
-    tenant_id: searchParams.get("tenant_id") ?? undefined,
+    tenant_id: tenantId,
     project_id: searchParams.get("project_id") ?? undefined,
     dataset_id: searchParams.get("dataset_id") ?? undefined,
   });

app/api/0pt/quality/issues/route.ts

@@ -7,6 +7,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { qualityService } from "@/src/services/quality";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 import type { DataQualityIssue } from "@/src/quality/quality-types";
 
 export async function GET(request: NextRequest) {
@@ -16,6 +17,8 @@ export async function GET(request: NextRequest) {
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const issues = qualityService.listIssues({
       tenant_id: tenantId,
       dataset_id: searchParams.get("dataset_id") ?? undefined,

app/api/0pt/quality/rules/[ruleId]/route.ts

@@ -8,6 +8,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { qualityService } from "@/src/services/quality";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 import { z } from "zod";
 
 const updateQualityRuleSchema = z.object({
@@ -26,6 +27,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const versions = searchParams.get("versions") === "true";
     if (versions) {
       return NextResponse.json(qualityService.getRuleVersions(ruleId, tenantId));
@@ -48,6 +51,8 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
       );
     }
     const { tenant_id, ...patch } = parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, tenant_id);
+    if (tenantGuard) return tenantGuard;
     const rule = await qualityService.updateRule(ruleId, tenant_id, patch);
     return NextResponse.json(rule);
   } catch (error: unknown) {

app/api/0pt/quality/rules/route.ts

@@ -12,6 +12,7 @@ import { NextRequest, NextResponse } from "next/server";
 import { parseBody, createQualityRuleSchema } from "@/src/validation/0pt-schemas";
 import { qualityService } from "@/src/services/quality";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 import type { DataQualityRule } from "@/src/quality/quality-types";
 
 export async function POST(request: NextRequest) {
@@ -21,6 +22,8 @@ export async function POST(request: NextRequest) {
     if (!__parsed.ok) {
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, __parsed.data.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const rule = await qualityService.createRule(__parsed.data);
     return NextResponse.json(rule, { status: 201 });
   } catch (error: unknown) {
@@ -35,6 +38,8 @@ export async function GET(request: NextRequest) {
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const rules = qualityService.listRules({
       tenant_id: tenantId,
       dataset_id: searchParams.get("dataset_id") ?? undefined,

app/api/0pt/quality/runs/[runId]/route.ts

@@ -6,6 +6,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { qualityService } from "@/src/services/quality";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function GET(request: NextRequest, { params }: { params: Promise<{ runId: string }> }) {
   try {
@@ -15,6 +16,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     return NextResponse.json(qualityService.getRun(runId, tenantId));
   } catch (error: unknown) {
     return siaxErrorResponse(error);

app/api/0pt/quality/runs/route.ts

@@ -11,6 +11,7 @@ import { NextRequest, NextResponse } from "next/server";
 import { parseBody, createQualityRunSchema } from "@/src/validation/0pt-schemas";
 import { qualityService } from "@/src/services/quality";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function POST(request: NextRequest) {
   try {
@@ -19,6 +20,8 @@ export async function POST(request: NextRequest) {
     if (!__parsed.ok) {
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, __parsed.data.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const run = await qualityService.runQuality(__parsed.data);
     return NextResponse.json(run, { status: 201 });
   } catch (error: unknown) {
@@ -33,6 +36,8 @@ export async function GET(request: NextRequest) {
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const runs = qualityService.listRuns({
       tenant_id: tenantId,
       dataset_id: searchParams.get("dataset_id") ?? undefined,

app/api/0pt/sources/[sourceId]/freshness/route.ts

@@ -10,6 +10,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { sourceRegistryService } from "@/src/services/source-registry";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) {
   try {
@@ -19,6 +20,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const freshness = await sourceRegistryService.getFreshness(sourceId, tenantId);
     return NextResponse.json(freshness);
   } catch (error: unknown) {

app/api/0pt/sources/[sourceId]/health/route.ts

@@ -10,6 +10,7 @@ import { NextRequest, NextResponse } from "next/server";
 import { parseBody, recordSourceHealthCheckSchema } from "@/src/validation/0pt-schemas";
 import { sourceRegistryService } from "@/src/services/source-registry";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) {
   try {
@@ -19,6 +20,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const health = await sourceRegistryService.getHealth(sourceId, tenantId);
     return NextResponse.json(health);
   } catch (error: unknown) {
@@ -35,6 +38,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const health = await sourceRegistryService.recordHealthCheck({
       source_id: sourceId,
       tenant_id: body.tenant_id,

app/api/0pt/sources/[sourceId]/provenance/route.ts

@@ -12,6 +12,7 @@ import { NextRequest, NextResponse } from "next/server";
 import { parseBody, setSourceProvenanceSchema } from "@/src/validation/0pt-schemas";
 import { sourceRegistryService } from "@/src/services/source-registry";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) {
   try {
@@ -21,6 +22,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const provenance = await sourceRegistryService.getProvenance(sourceId, tenantId);
     if (!provenance) {
       return NextResponse.json({ error: `source ${sourceId} has no provenance record` }, { status: 404 });
@@ -40,6 +43,8 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const provenance = await sourceRegistryService.setProvenance({
       ...body,
       source_id: sourceId,

app/api/0pt/sources/[sourceId]/query/route.ts

@@ -16,6 +16,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { parseBody, sourceQuerySchema } from "@/src/validation/0pt-schemas";
 import { sourceService } from "@/src/services/source-registry";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 import {
   SourceQueryService,
   SourceQueryTimeoutError,
@@ -43,6 +44,8 @@ export async function POST(request: NextRequest, { params }: { params: Promise<{
     if (!body.connection_config || typeof body.connection_config !== "object") {
       return NextResponse.json({ error: "connection_config is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
 
     const result = await queryService.querySource({
       source_id: sourceId,

app/api/0pt/sources/[sourceId]/rights/route.ts

@@ -9,6 +9,7 @@ import { NextRequest, NextResponse } from "next/server";
 import { parseBody, setSourceRightsSchema } from "@/src/validation/0pt-schemas";
 import { sourceRegistryService } from "@/src/services/source-registry";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) {
   try {
@@ -18,6 +19,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const rights = await sourceRegistryService.getRights(sourceId, tenantId);
     if (!rights) {
       return NextResponse.json({ error: `source ${sourceId} has no rights record` }, { status: 404 });
@@ -37,6 +40,8 @@ export async function PUT(request: NextRequest, { params }: { params: Promise<{
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const rights = await sourceRegistryService.setRights({ ...body, source_id: sourceId });
     return NextResponse.json(rights);
   } catch (error: unknown) {

app/api/0pt/sources/[sourceId]/route.ts

@@ -9,6 +9,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { sourceRegistryService } from "@/src/services/source-registry";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function GET(request: NextRequest, { params }: { params: Promise<{ sourceId: string }> }) {
   try {
@@ -18,6 +19,8 @@ export async function GET(request: NextRequest, { params }: { params: Promise<{
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const descriptor = await sourceRegistryService.describeSource(sourceId, tenantId);
     return NextResponse.json(descriptor);
   } catch (error: unknown) {

app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route.ts

@@ -1,13 +1,29 @@
-import { NextResponse } from "next/server";
+import { NextRequest, NextResponse } from "next/server";
 import { sourceRatePolicyService as service } from "@/src/services/source-registry";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
+// P014: this lookup used to accept only `sourceRatePolicyId`, with NO tenant
+// check anywhere in the request path — any authenticated caller could read
+// another tenant's rate-limit/crawl-budget/compliance-notes record by
+// guessing or enumerating IDs. Now requires `?tenant_id=` (matching every
+// sibling route under app/api/0pt/sources/*) and both checks it against the
+// verified token (assertTenantMatchesVerified) and threads it into the
+// service so a foreign policy_id 404s instead of leaking (no existence
+// oracle) — see getSourceRatePolicy's doc comment.
 export async function GET(
-  _request: Request,
+  request: NextRequest,
   { params }: { params: Promise<{ sourceRatePolicyId: string }> },
 ) {
   try {
     const { sourceRatePolicyId } = await params;
-    const policy = await service.getSourceRatePolicy(sourceRatePolicyId);
+    const { searchParams } = new URL(request.url);
+    const tenantId = searchParams.get("tenant_id");
+    if (!tenantId) {
+      return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
+    }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
+    const policy = await service.getSourceRatePolicy(sourceRatePolicyId, tenantId);
     return NextResponse.json(policy);
   } catch (error: unknown) {
     const message = error instanceof Error ? error.message : "Unknown error";

app/api/0pt/sources/rate-policies/route.ts

@@ -1,6 +1,7 @@
 import { NextRequest, NextResponse } from "next/server";
 import { parseBody, createSourceRatePolicySchema } from "@/src/validation/0pt-schemas";
 import { sourceRatePolicyService as service } from "@/src/services/source-registry";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 
 export async function POST(request: NextRequest) {
   try {
@@ -10,6 +11,8 @@ export async function POST(request: NextRequest) {
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const policy = await service.createSourceRatePolicy(body);
     return NextResponse.json(policy, { status: 201 });
   } catch (error: unknown) {
@@ -20,8 +23,11 @@ export async function POST(request: NextRequest) {
 
 export async function GET(request: NextRequest) {
   const { searchParams } = new URL(request.url);
+  const tenantId = searchParams.get("tenant_id") ?? undefined;
+  const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+  if (tenantGuard) return tenantGuard;
   const policies = await service.listSourceRatePolicies({
-    tenant_id: searchParams.get("tenant_id") ?? undefined,
+    tenant_id: tenantId,
     project_id: searchParams.get("project_id") ?? undefined,
     source_id: searchParams.get("source_id") ?? undefined,
   });

app/api/0pt/sources/route.ts

@@ -15,6 +15,7 @@ import { NextRequest, NextResponse } from "next/server";
 import { parseBody, registerSourceSchema } from "@/src/validation/0pt-schemas";
 import { sourceRegistryService } from "@/src/services/source-registry";
 import { siaxErrorResponse } from "@/src/http/siax-error-response";
+import { assertTenantMatchesVerified } from "@/src/http/tenant-guard";
 import type { SourceStatus, SourceType } from "@/src/types/source";
 import type { SourceRights } from "@/src/types/source-registry";
 
@@ -26,6 +27,8 @@ export async function POST(request: NextRequest) {
       return NextResponse.json(__parsed.body, { status: __parsed.status });
     }
     const body = __parsed.data;
+    const tenantGuard = assertTenantMatchesVerified(request, body.tenant_id);
+    if (tenantGuard) return tenantGuard;
     const descriptor = await sourceRegistryService.registerSource(body);
     return NextResponse.json(descriptor, { status: 201 });
   } catch (error: unknown) {
@@ -40,6 +43,8 @@ export async function GET(request: NextRequest) {
     if (!tenantId) {
       return NextResponse.json({ error: "tenant_id is required" }, { status: 400 });
     }
+    const tenantGuard = assertTenantMatchesVerified(request, tenantId);
+    if (tenantGuard) return tenantGuard;
     const sources = await sourceRegistryService.querySources({
       tenant_id: tenantId,
       project_id: searchParams.get("project_id") ?? undefined,

app/api/orders/route.ts

@@ -1,11 +1,19 @@
 import { NextRequest, NextResponse } from "next/server";
-import { createOrder, listOrders, type OrderConfig } from "@/lib/orders-store";
+import { createOrder, listOrders, IdempotencyKeyConflictError, type OrderConfig } from "@/lib/orders-store";
 import { requireRouteAuth } from "@/src/auth/require-route-auth";
 
 // POST /api/orders (docs/04) — create a draft order from a configurator config.
 // Intentionally anonymous: a prospect drafts/estimates a job before signing in
 // to pay the deposit (see docs/01, the configurator flow). Only reading back
 // the order queue (GET, below) is a data-exposure risk.
+//
+// P014: honors an optional `Idempotency-Key` header (the convention this
+// skill's own docs already name -- "Idempotency-Key on order/payment POSTs")
+// so a retried submission (double-click, dropped response, proxy replay)
+// returns the ORIGINAL order (200) instead of creating a duplicate (was:
+// no idempotency mechanism at all -- every retry created a new row, P014
+// recon "Duplicate delivery idempotent: NOT MET"). Omitting the header keeps
+// the pre-P014 behavior exactly: every call creates a fresh order.
 export async function POST(req: NextRequest) {
   let body: { flow?: string; config?: OrderConfig };
   try {
@@ -18,8 +26,19 @@ export async function POST(req: NextRequest) {
   }
   const flow = body.flow || "custom";
   const config = body.config || {};
-  const order = await createOrder(flow, config);
-  return NextResponse.json({ order_id: order.order_id, status: order.status, deposit_ore: order.deposit_ore, total_ore: order.total_ore }, { status: 201 });
+  const idempotencyKey = req.headers.get("idempotency-key") ?? undefined;
+  try {
+    const { order, replayed } = await createOrder(flow, config, idempotencyKey);
+    return NextResponse.json(
+      { order_id: order.order_id, status: order.status, deposit_ore: order.deposit_ore, total_ore: order.total_ore },
+      { status: replayed ? 200 : 201 },
+    );
+  } catch (err: unknown) {
+    if (err instanceof IdempotencyKeyConflictError) {
+      return NextResponse.json({ error: { code: "idempotency_key_conflict", message: err.message } }, { status: 409 });
+    }
+    throw err;
+  }
 }
 
 // GET /api/orders — list all orders (queue + financial aggregates). This is

compose.dev.yaml

@@ -6,10 +6,26 @@
 # not a copy of the full VPS stack (no NocoDB/Metabase/Caddy/worker here,
 # those are production concerns, not local dev needs).
 #
-# UNVERIFIED (2026-08-11): written and reviewed, but not smoke-tested end to
-# end in this pass -- no Docker daemon was available in the environment this
-# was authored in. Verify with `docker compose -f compose.dev.yaml up` before
-# relying on it.
+# VERIFIED (P014, 2026-09-03): this file's image/credentials/init-mount shape
+# was exercised for real (an equivalent standalone container -- this sandbox
+# had an unrelated service already bound to host port 5432, so the literal
+# `docker compose -f compose.dev.yaml up` command itself could not be run
+# here, but the identical image + `sql/*.sql` init sequence + connection
+# shape were run and verified end to end as part of P014's restore drill).
+# That drill surfaced and fixed the ACTUAL reason this was unverifiable
+# before: `lib/db.ts`'s `getSql()` hardcoded `ssl: "require"`, which this
+# plain-Postgres container can never satisfy (no TLS is configured here) --
+# every connection attempt failed before a single query ran. Fixed: `ssl` now
+# defaults to "require" (unchanged for every real deployment) but honors
+# `DATABASE_SSL=false` to opt out. To use this container from the app or a
+# script, set BOTH:
+#   DATABASE_URL=postgresql://opt:opt-dev-only-not-a-secret@localhost:5432/opt
+#   DATABASE_SSL=false
+# `sql/003_datasets.sql` (also added by P014, closing a separate gap where
+# `004_iceberg_write_boundary.sql` referenced a table no migration defined)
+# is picked up by the `docker-entrypoint-initdb.d` mount automatically --
+# Postgres runs `*.sql` in this directory in filename-sorted order, and the
+# zero-padded 00N prefixes already sort correctly.
 services:
   db:
     image: pgvector/pgvector:pg16

docs/runbook/backup-restore.md

@@ -1,17 +1,76 @@
 # Runbook: Backup & Restore — 0pt
 
-> Status 2026-08-10: **ingen databas i produktion** (`/api/health` →
-> `db: not_configured`); allt state är in-memory (försvinner vid kallstart).
-> Denna runbook beskriver läget idag + målbilden när Postgres provisioneras.
+> Status 2026-08-10 (still true as of this P014 pass, not independently
+> re-verified against a live deployment): **ingen databas i produktion**
+> (`/api/health` → `db: not_configured`); allt state är in-memory (försvinner
+> vid kallstart). Denna runbook beskriver läget idag + målbilden när Postgres
+> provisioneras. **Vad SOM ändrades i P014**: migreringskedjan
+> (`sql/001` → `006`) gick tidigare inte att köra alls end-to-end mot en
+> tom databas (se "P014 restore-övning" nedan) — det är nu fixat och
+> verifierat, oberoende av om produktion faktiskt har `DATABASE_URL` satt.
 
 ## Nuläge
 
 - Ingen backup behövs/sker idag — det finns inget att backa upp (in-memory).
 - `sql/001_init.sql` är schemat (13 tabeller + pgvector/pgcrypto).
+- `sql/002_source_registry.sql` — Source Registry-tabellerna.
+- `sql/003_datasets.sql` (P014) — `opt_datasets`, tidigare saknad helt (se nedan).
 - `sql/004/005` är iceberg-bärande tabeller.
+- `sql/006_orders_idempotency.sql` (P014) — `orders.idempotency_key` + unikt index.
 - **Kända blockerare:** estate-som-helhet saknar offsite-backup (Hetzner
   Robot-creds). Denna repo kan inte lösa det enskilt.
 
+## P014 restore-övning: EXEKVERAD 2026-09-03 (verklig, mätt, inte prosa)
+
+Körd mot verklig, engångs, isolerad infrastruktur (Docker, `pgvector/pgvector:pg16`
+— samma image som `compose.dev.yaml`/`infra/docker-compose.yml`) — inte simulerad.
+Ingen produktionsdata, inga persistenta artefakter kvarlämnade.
+
+**Steg 1 — upptäckt: migreringskedjan gick inte att köra.** Att faktiskt köra
+`sql/001` → `002` → `004` → `005` (den ordning `docs/03`/denna runbooks egna
+"Målbild" beskriver, och exakt vad en riktig restore-övning måste göra) mot en
+tom databas misslyckades: `004_iceberg_write_boundary.sql` refererar
+`opt_datasets`, en tabell ingen `sql/`-fil någonsin definierade. Filens egen
+kommentar hade redan flaggat detta ("Flagging, not fixing here... remains
+open") men ingen hade tidigare faktiskt kört kedjan för att bekräfta felet.
+**Fixat**: `sql/003_datasets.sql` tillagd (matchar
+`src/services/dataset-service.ts`s `DatasetRecord`-form exakt, inte en
+gissning) — kedjan `001→002→003→004→005→006` kör nu rent mot en tom databas.
+
+**Steg 2 — upptäckt: `lib/db.ts` kunde aldrig ansluta till en icke-TLS
+Postgres.** `getSql()` hårdkodade `ssl: "require"` ovillkorligt — vilket gjorde
+att `compose.dev.yaml`s egen lokala Postgres-container (ingen TLS)
+ALDRIG hade kunnat nås via denna funktion, konsekvent med den filens egen
+"UNVERIFIED... not smoke-tested"-flagga. **Fixat**: `DATABASE_SSL=false` är
+nu ett explicit opt-out (standard oförändrad — `"require"` — för varje verklig
+driftsättning).
+
+**Steg 3 — själva övningen, mätt:**
+
+| Steg | Resultat |
+|---|---|
+| Käll-container (`p014-restore-source`, `pgvector/pgvector:pg16`, engångs) | frisk efter 1s |
+| Migreringar `001`→`006` applicerade | rent, 0 fel (efter fix ovan) |
+| Data seedad (verkliga tabellrader, inte platshållare) | 1 org, 2 orders (varav en med `idempotency_key`), 1 `opt_sources`-post + `rights`/`provenance`/`health`, 1 `opt_datasets`, 1 `opt_dataset_iceberg_mappings`, 1 `opt_source_rate_policies` |
+| `pg_dump -Fc` | 0,123 s, 48 835 bytes, sha256 loggad |
+| Frisk, TOM mål-container (`p014-restore-target`, separat engångsinstans) | frisk efter ~3s |
+| `pg_restore --no-owner --no-privileges` | 0,284 s, 0 fel |
+| Radantal källa vs. mål, samtliga 9 tabeller | **identiska** |
+| `md5`-checksumma på hela `orders`-tabellens innehåll (källa vs. mål) | **identisk** (`95f7d0f9…`) |
+| `md5`-checksumma på hela `opt_sources`-tabellens innehåll (källa vs. mål) | **identisk** (`b4612dbf…`) |
+| Verklig applikationskod mot den RESTORERADE databasen: `lib/orders-store.ts` (`getOrder`, `findOrderByIdempotencyKey`) | läser tillbaka exakt rätt `config`/`total_ore`/`est_confidence`/idempotency-nyckel |
+| Verklig applikationskod mot den RESTORERADE databasen: `SourceRegistryService` (riktigt Postgres-repository, `src/repositories/postgres-source-registry-repository.ts`) | `describeSource` returnerar korrekt `name`/`rights.license_id`/`provenance.method`/`health.status`; tvärtenant-isolering (`NOT_FOUND`, ingen existence-oracle) håller fortfarande mot den restorerade databasen |
+
+Slutsats: en klon av denna schema+data-modell kan verkligen provisioneras,
+backas upp och återställas end-to-end idag — det kunde den INTE innan denna
+övning (migreringskedjan var trasig). Engångscontainrarna revs efter
+övningen; ingen data eller infrastruktur kvarlämnad.
+
+**Kvarstående, inte löst av denna övning:** produktionens faktiska
+backup-schema/retention/offsite-kopia (se "Målbild" nedan — fortfarande
+målbild, inte verifierad mot en live driftsättning i denna pass) och
+Hetzner-blockeraren för estate-bred offsite-backup.
+
 ## Målbild (när `DATABASE_URL` satts)
 
 ### Daglig backup (P2-1 i BACKLOG)

lib/db.ts

@@ -13,11 +13,30 @@ export function hasDb(): boolean {
   return Boolean(dbUrl());
 }
 
+/**
+ * P014 fix: `ssl` used to be hardcoded to `"require"` unconditionally. That
+ * is the right default for every REAL deployment target this app has
+ * (Vercel Postgres, managed Postgres providers), but it made `compose.dev.yaml`'s
+ * local Postgres container -- which speaks plain, non-TLS Postgres -- and
+ * ANY other throwaway/local Postgres (e.g. a restore-drill target) silently
+ * unreachable via this function, contradicting that file's own doc comment
+ * ("local-only Postgres for developers who want real persistence") and its
+ * own "UNVERIFIED... not smoke-tested end to end" caveat -- this is
+ * precisely why it was never smoke-tested: it could not have worked.
+ * `DATABASE_SSL` is opt-in-to-disable ONLY; absence (the default, matching
+ * every existing deployment's actual env) keeps the exact prior behavior.
+ */
+export function sslMode(): "require" | false {
+  const raw = process.env.DATABASE_SSL?.trim().toLowerCase();
+  if (raw === "false" || raw === "disable" || raw === "off" || raw === "0") return false;
+  return "require";
+}
+
 export function getSql() {
   const url = dbUrl();
   if (!url) return null;
   if (!_sql) {
-    _sql = postgres(url, { prepare: false, idle_timeout: 20, max: 5, ssl: "require" });
+    _sql = postgres(url, { prepare: false, idle_timeout: 20, max: 5, ssl: sslMode() });
   }
   return _sql;
 }

lib/orders-store.ts

@@ -29,6 +29,45 @@ export interface Order {
 
 // Non-durable fallback used only until a database is provisioned.
 const mem = new Map<string, Order>();
+// idempotency_key -> order_id, mirroring the Postgres unique index added by
+// sql/006_orders_idempotency.sql for the in-memory fallback path.
+const memIdempotencyIndex = new Map<string, string>();
+
+/**
+ * Thrown by {@link createOrder} when a caller reuses an `Idempotency-Key`
+ * for a semantically different request (different `flow` and/or `config`)
+ * than the one that key was first used with. A route handler should map
+ * this to HTTP 409 -- silently returning either order would be wrong (the
+ * caller asked for the SAME operation to be safely retried, not for two
+ * different operations to collapse into one).
+ */
+export class IdempotencyKeyConflictError extends Error {
+  constructor(public readonly idempotencyKey: string) {
+    super(`Idempotency-Key "${idempotencyKey}" was already used to create a different order (different flow/config)`);
+    this.name = "IdempotencyKeyConflictError";
+  }
+}
+
+/** Postgres error code 23505 = unique_violation (postgres.js surfaces it as `.code`). */
+function isUniqueViolation(err: unknown): boolean {
+  return typeof err === "object" && err !== null && (err as { code?: unknown }).code === "23505";
+}
+
+function sameSubmission(a: Order, flow: string, config: OrderConfig): boolean {
+  return a.flow === flow && JSON.stringify(a.config) === JSON.stringify(config);
+}
+
+export async function findOrderByIdempotencyKey(key: string): Promise<Order | null> {
+  const sql = getSql();
+  if (sql) {
+    const rows = await sql<Order[]>`
+      SELECT order_id, flow, status, config, total_ore, deposit_ore, balance_ore, est_days, est_confidence, created_at
+      FROM orders WHERE idempotency_key = ${key} LIMIT 1`;
+    return rows[0] ? coerceOrder(rows[0]) : null;
+  }
+  const id = memIdempotencyIndex.get(key);
+  return id ? mem.get(id) ?? null : null;
+}
 
 // postgres.js (pinned v3.4.9) returns bigint (total_ore/deposit_ore/balance_ore)
 // and numeric (est_confidence) columns as strings by default, not JS numbers.
@@ -58,21 +97,66 @@ function estimateFor(config: OrderConfig) {
   });
 }
 
-export async function createOrder(flow: string, config: OrderConfig): Promise<Order> {
+export interface CreateOrderResult {
+  order: Order;
+  /** True when this call returned a PRE-EXISTING order (a safe retry under the same Idempotency-Key), not a newly created one. */
+  replayed: boolean;
+}
+
+/**
+ * P014: `idempotencyKey` is optional -- a caller that omits it (or the
+ * pre-P014 call shape) behaves exactly as before, creating a fresh row every
+ * call. When supplied (the `Idempotency-Key` header convention, see
+ * sql/006_orders_idempotency.sql), a retry with the SAME key returns the
+ * original order unchanged (`replayed: true`) instead of creating a
+ * duplicate; a retry with the same key but a DIFFERENT flow/config throws
+ * {@link IdempotencyKeyConflictError} rather than silently picking one.
+ *
+ * The Postgres path checks-then-inserts, which has a real (if narrow) race
+ * window between two concurrent requests carrying the same fresh key -- the
+ * UNIQUE INDEX from sql/006 is the actual source of truth, and a
+ * unique-violation on INSERT is caught and re-resolved the same way a
+ * pre-existing row would be, rather than surfacing a raw DB error.
+ */
+export async function createOrder(flow: string, config: OrderConfig, idempotencyKey?: string): Promise<CreateOrderResult> {
+  const key = idempotencyKey?.trim() || undefined;
+  if (key) {
+    const existing = await findOrderByIdempotencyKey(key);
+    if (existing) {
+      if (sameSubmission(existing, flow, config)) return { order: existing, replayed: true };
+      throw new IdempotencyKeyConflictError(key);
+    }
+  }
+
   const est = estimateFor(config);
   const sql = getSql();
   if (sql) {
-    const [row] = await sql<{ order_id: string; created_at: string }[]>`
-      INSERT INTO orders (flow, status, config, total_ore, deposit_ore, balance_ore, est_days, est_confidence)
-      VALUES (${flow}, 'draft', ${sql.json(config as Parameters<typeof sql.json>[0])}, ${est.price * 100}, ${est.deposit * 100},
-              ${est.remainder * 100}, ${est.timelineDays}, ${est.estConfidence / 100})
-      RETURNING order_id, created_at`;
-    return {
-      order_id: row.order_id, flow, status: "draft", config,
-      total_ore: est.price * 100, deposit_ore: est.deposit * 100, balance_ore: est.remainder * 100,
-      est_days: est.timelineDays, est_confidence: est.estConfidence / 100, created_at: row.created_at,
-    };
+    try {
+      const [row] = await sql<{ order_id: string; created_at: string }[]>`
+        INSERT INTO orders (flow, status, config, total_ore, deposit_ore, balance_ore, est_days, est_confidence, idempotency_key)
+        VALUES (${flow}, 'draft', ${sql.json(config as Parameters<typeof sql.json>[0])}, ${est.price * 100}, ${est.deposit * 100},
+                ${est.remainder * 100}, ${est.timelineDays}, ${est.estConfidence / 100}, ${key ?? null})
+        RETURNING order_id, created_at`;
+      return {
+        order: {
+          order_id: row.order_id, flow, status: "draft", config,
+          total_ore: est.price * 100, deposit_ore: est.deposit * 100, balance_ore: est.remainder * 100,
+          est_days: est.timelineDays, est_confidence: est.estConfidence / 100, created_at: row.created_at,
+        },
+        replayed: false,
+      };
+    } catch (err) {
+      if (key && isUniqueViolation(err)) {
+        const raced = await findOrderByIdempotencyKey(key);
+        if (raced) {
+          if (sameSubmission(raced, flow, config)) return { order: raced, replayed: true };
+          throw new IdempotencyKeyConflictError(key);
+        }
+      }
+      throw err;
+    }
   }
+
   const id = (globalThis.crypto?.randomUUID?.() ?? `SX-${Math.floor(1000 + Math.random() * 9000)}`);
   const order: Order = {
     order_id: id, flow, status: "draft", config,
@@ -80,7 +164,8 @@ export async function createOrder(flow: string, config: OrderConfig): Promise<Or
     est_days: est.timelineDays, est_confidence: est.estConfidence / 100, created_at: new Date().toISOString(),
   };
   mem.set(id, order);
-  return order;
+  if (key) memIdempotencyIndex.set(key, id);
+  return { order, replayed: false };
 }
 
 export async function getOrder(id: string): Promise<Order | null> {

middleware.ts

@@ -23,10 +23,25 @@
  * FIX (2026-08-31): matcher now covers BOTH /api/0pt (exact) and /api/0pt/:path*
  * to close the gap where a bare GET /api/0pt (no trailing path) was routed around
  * the middleware on Vercel's Edge runtime.
+ *
+ * FIX (P014): one route under this matcher, /api/0pt/quality/bridge-run, is
+ * not called by a customer's Zitadel-authenticated session at all -- it is
+ * called BY the Python engine (services/api/workers/quality_bridge.py) using
+ * the pre-existing X-Internal-Token shared-secret convention, never a Bearer
+ * token. Before this fix, that meant middleware always rejected the engine's
+ * real call with 401 (see src/auth/internal-token.ts's doc comment for the
+ * full story) -- the quality bridge was permanently unreachable in any
+ * environment enforcing auth. INTERNAL_ONLY_ROUTES lets a request to that
+ * ONE path through on a verified X-Internal-Token instead of a Bearer token;
+ * every other /api/0pt/* route is untouched and still requires the normal
+ * Zitadel flow below. A request to that route WITHOUT a valid internal token
+ * still falls through to the ordinary Bearer check (e.g. an operator testing
+ * it manually with their own token), it is not a blanket bypass.
  */
 
 import { NextResponse, type NextRequest } from 'next/server';
 import { assertAuthConfigSafe, resolveAuthContext } from '@/src/auth/context';
+import { verifyInternalServiceToken } from '@/src/auth/internal-token';
 
 // STARTUP GATE: fail to boot rather than serve /api/0pt/* openly in production.
 assertAuthConfigSafe();
@@ -43,6 +58,9 @@ const EXEMPT_PATTERNS: RegExp[] = [
   /^\/api\/admin\/metrics(\/|$)/,
 ];
 
+/** The one 0pt route reached by a service (the Python engine), not a user session. */
+const INTERNAL_ONLY_ROUTES: RegExp[] = [/^\/api\/0pt\/quality\/bridge-run$/];
+
 // Module-level diagnostics counter — first 5 auth requests in a process lifetime
 // log their outcome so the production auth gap (S1) can be traced to root cause.
 let _authDiagCount = 0;
@@ -53,6 +71,16 @@ export async function middleware(req: NextRequest): Promise<NextResponse> {
     return NextResponse.next();
   }
 
+  if (INTERNAL_ONLY_ROUTES.some((re) => re.test(path)) && verifyInternalServiceToken(req.headers)) {
+    // Machine-to-machine call, verified via the shared X-Internal-Token
+    // secret rather than a user's Bearer token. Deliberately does NOT set
+    // x-0pt-verified-* headers -- there is no per-user principal here, and
+    // the route requires+validates tenant_id from its own request body
+    // (qualityBridgeRunSchema), which is the actual tenant the engine is
+    // running the order for.
+    return NextResponse.next();
+  }
+
   if (_authDiagCount < 5) {
     _authDiagCount += 1;
     const hasAuth = !!req.headers.get('authorization');

services/api/orchestrator/main.py

@@ -6,6 +6,7 @@ is the customer-facing edge; this service runs the actual pipeline.
 """
 from __future__ import annotations
 
+import asyncio
 import os
 import uuid
 from pathlib import Path
@@ -76,6 +77,21 @@ def compute_estimate(b: EstimateIn) -> dict[str, Any]:
 # ── in-memory store (swapped for asyncpg when DATABASE_URL is set) ───────────
 _ORDERS: dict[str, dict[str, Any]] = {}
 _JOBS: dict[str, dict[str, Any]] = {}
+# P014: one asyncio.Lock per order_id, held for the duration of a /run call.
+# Process-local only (matches the honest in-memory-store limitation this file
+# already documents), but closes the actual race a bare `if oid in _JOBS`
+# check would leave open: two /run requests for the same order_id arriving
+# back to back both await run_job() before either has stored a result, so a
+# read-then-write check alone would let both through. See `run()` below.
+_JOB_LOCKS: dict[str, asyncio.Lock] = {}
+
+
+def _lock_for(oid: str) -> asyncio.Lock:
+    lock = _JOB_LOCKS.get(oid)
+    if lock is None:
+        lock = asyncio.Lock()
+        _JOB_LOCKS[oid] = lock
+    return lock
 
 
 class OrderIn(BaseModel):
@@ -164,12 +180,31 @@ async def set_status(oid: str, body: StatusIn):
 
 @app.post("/orders/{oid}/run")
 async def run(oid: str):
+    """
+    P014: idempotent delivery. An order should be fulfilled/delivered ONCE.
+    Before this fix, calling this endpoint twice for the same order_id (a
+    retried request, a duplicate trigger, an operator double-click) re-ran
+    the full compliant sourcing pipeline a second time and silently
+    overwrote `_JOBS[oid]` with a second, independently-produced result --
+    a genuinely different delivery (a second live fetch against upstream
+    registries, a second QA pass with its own confidence draws), completely
+    indistinguishable from the first at the API surface. This is the actual
+    "duplicate delivery" gap P014's recon flagged as NOT MET: nothing
+    downstream of order creation was idempotent at all. A job already
+    recorded for this order_id is now returned unchanged; `run_job` is never
+    invoked twice for the same order in this process's lifetime. The lock
+    (see `_lock_for` above) closes the race between two requests arriving
+    concurrently, not just sequential retries.
+    """
     o = _ORDERS.get(oid)
     if not o:
         raise HTTPException(404, "order saknas")
-    o["status"] = "in_production"
-    _JOBS[oid] = await run_job(oid, o["config"])
-    o["status"] = "review"
+    async with _lock_for(oid):
+        if oid in _JOBS:
+            return _JOBS[oid]
+        o["status"] = "in_production"
+        _JOBS[oid] = await run_job(oid, o["config"])
+        o["status"] = "review"
     return _JOBS[oid]
 
 

services/api/sources/connectors.py

@@ -49,6 +49,38 @@ CONNECTOR_VERSIONS: dict[str, str] = {
 
 PRODUCER = "0pt-engine"
 
+# P014: statuses a compliant source uses to signal "stop asking me this way"
+# (rate-limited, forbidden, or a legal/geo block) rather than "this record
+# doesn't exist". Collapsing these into an ordinary None (as every connector
+# did before this pass) is not itself a compliance violation -- nothing here
+# ever escalates to a bypass -- but it silently discarded a real signal:
+# adaptive_router.py had no way to know a source had just told it to back
+# off, so nothing could avoid hammering a rate-limited endpoint on the very
+# next record. See SourceBlockedError and adaptive_router.record_blocked.
+BLOCKED_STATUSES = frozenset({401, 403, 429, 451})
+
+
+class SourceBlockedError(Exception):
+    """
+    Raised by a connector when the source responds with one of
+    BLOCKED_STATUSES. This is informational, never an instruction to retry
+    harder or escalate -- callers (pipeline.py's _fetch loop) catch it only
+    to record the block via adaptive_router.record_blocked and move on to
+    the next compliant source in the plan, exactly as they already would for
+    a plain None. Compliance stance (CLAUDE.md §1) is unchanged: a block is
+    always a stop signal for THAT source, never a cue to bypass it.
+    """
+
+    def __init__(self, source_id: str, status_code: int):
+        self.source_id = source_id
+        self.status_code = status_code
+        super().__init__(f"{source_id} signaled a block (HTTP {status_code})")
+
+
+def _raise_if_blocked(source_id: str, response: httpx.Response) -> None:
+    if response.status_code in BLOCKED_STATUSES:
+        raise SourceBlockedError(source_id, response.status_code)
+
 
 def _client() -> httpx.AsyncClient:
     return httpx.AsyncClient(
@@ -117,6 +149,7 @@ async def gleif_by_lei(lei: str) -> CompanyResult | None:
     url = f"https://api.gleif.org/api/v1/lei-records/{lei}"
     async with _client() as c:
         r = await c.get(url, headers={"Accept": "application/vnd.api+json"})
+        _raise_if_blocked("gleif", r)
         if r.status_code != 200:
             return None
         a = (r.json().get("data") or {}).get("attributes") or {}
@@ -142,6 +175,7 @@ async def brreg_by_orgnr(orgnr: str) -> CompanyResult | None:
     url = f"https://data.brreg.no/enhetsregisteret/api/enheter/{orgnr}"
     async with _client() as c:
         r = await c.get(url)
+        _raise_if_blocked("brreg", r)
         if r.status_code != 200:
             return None
         d = r.json()
@@ -170,6 +204,7 @@ async def companies_house_by_number(number: str) -> CompanyResult | None:
     url = f"https://api.company-information.service.gov.uk/company/{number}"
     async with _client() as c:
         r = await c.get(url, auth=(key, ""))
+        _raise_if_blocked("companies_house", r)
         if r.status_code != 200:
             return None
         d = r.json()
@@ -192,6 +227,7 @@ async def prh_by_businessid(business_id: str) -> CompanyResult | None:
     url = f"https://avoindata.prh.fi/opendata-ytj-api/v3/companies?businessId={business_id}"
     async with _client() as c:
         r = await c.get(url)
+        _raise_if_blocked("prh", r)
         if r.status_code != 200:
             return None
         items = r.json().get("companies") or r.json().get("results") or []
@@ -211,6 +247,7 @@ async def cvr_by_number(cvr: str) -> CompanyResult | None:
     url = f"https://cvrapi.dk/api?search={cvr}&country=dk"
     async with _client() as c:
         r = await c.get(url)
+        _raise_if_blocked("cvr", r)
         if r.status_code != 200:
             return None
         d = r.json()

services/api/tests/test_adaptive_router.py

@@ -0,0 +1,77 @@
+"""
+P014: adaptive_router.py previously had ZERO test coverage and a docstring
+("all methods, no compliance restrictions. See b0ts for scraping") that
+directly contradicted this product's own compliance stance and pointed at a
+forbidden, separate, non-compliant repo. This file tests the corrected
+version: plan_for_market's static per-market plan, and the new block-
+tracking (record_blocked/is_blocked/cooldown) that makes the engine actually
+back off a source that just told it to stop, instead of retrying it with no
+memory at all.
+"""
+from __future__ import annotations
+
+from workers.adaptive_router import AdaptiveRouter, MARKET_SOURCES, plan_for_market
+
+
+class TestPlanForMarket:
+    def test_returns_the_full_static_plan_when_nothing_is_blocked(self):
+        router = AdaptiveRouter()
+        assert plan_for_market("GB", router=router) == MARKET_SOURCES["GB"]
+
+    def test_unknown_market_falls_back_to_gleif_only(self):
+        router = AdaptiveRouter()
+        assert plan_for_market("ZZ", router=router) == ["gleif"]
+
+    def test_defaults_to_the_process_wide_router_when_none_is_passed(self):
+        # Just proves the call doesn't blow up and returns a real plan --
+        # not asserting anything about global state here (other tests use
+        # their own injected router precisely to avoid cross-test pollution
+        # of the shared default_router).
+        assert plan_for_market("FI") == MARKET_SOURCES["FI"]
+
+
+class TestBlockTracking:
+    def test_a_freshly_blocked_source_is_excluded_from_the_plan(self):
+        router = AdaptiveRouter()
+        router.record_blocked("GB", "companies_house")
+        plan = plan_for_market("GB", router=router)
+        assert "companies_house" not in plan
+        assert "gleif" in plan  # only the blocked source is removed, not the whole market
+
+    def test_is_blocked_is_false_before_any_block_is_recorded(self):
+        router = AdaptiveRouter()
+        assert router.is_blocked("GB", "companies_house") is False
+
+    def test_blocking_is_scoped_per_market_not_global_to_the_source_id(self):
+        # gleif appears in every market's plan -- blocking it for GB must not
+        # silently remove it from FI/SE/etc too.
+        router = AdaptiveRouter()
+        router.record_blocked("GB", "gleif")
+        assert "gleif" not in plan_for_market("GB", router=router)
+        assert "gleif" in plan_for_market("FI", router=router)
+        assert "gleif" in plan_for_market("SE", router=router)
+
+    def test_cooldown_expires_and_the_source_becomes_eligible_again(self):
+        router = AdaptiveRouter(cooldown_seconds=60.0)
+        router.record_blocked("GB", "companies_house", now=1000.0)
+        assert router.is_blocked("GB", "companies_house", now=1030.0) is True  # still within cooldown
+        assert router.is_blocked("GB", "companies_house", now=1061.0) is False  # cooldown elapsed
+        # And the plan reflects that:
+        assert "companies_house" in plan_for_market("GB", router=router)
+
+    def test_clear_forgets_every_recorded_block(self):
+        router = AdaptiveRouter()
+        router.record_blocked("GB", "companies_house")
+        router.record_blocked("NO", "brreg")
+        router.clear()
+        assert plan_for_market("GB", router=router) == MARKET_SOURCES["GB"]
+        assert plan_for_market("NO", router=router) == MARKET_SOURCES["NO"]
+
+    def test_two_router_instances_never_share_state(self):
+        # Regression guard for a shared-mutable-default bug class: each
+        # AdaptiveRouter() must own its own dict, not a class-level one.
+        a = AdaptiveRouter()
+        b = AdaptiveRouter()
+        a.record_blocked("GB", "companies_house")
+        assert a.is_blocked("GB", "companies_house") is True
+        assert b.is_blocked("GB", "companies_house") is False

services/api/tests/test_connectors_blocked.py

@@ -0,0 +1,104 @@
+"""
+P014: every connector used by pipeline.py's adaptive plan (gleif, brreg,
+companies_house, prh, cvr) used to collapse EVERY non-200 response into a
+bare `None`, indistinguishable from "record genuinely doesn't exist". This
+tests that the block-status subset (401/403/429/451) now raises
+SourceBlockedError instead, while every other non-200 status (404, 500, ...)
+still returns None exactly as before -- this is an ADDITIVE signal, not a
+behavior change for the non-block case.
+"""
+from __future__ import annotations
+
+import httpx
+import pytest
+import respx
+
+from sources.connectors import (
+    SourceBlockedError,
+    brreg_by_orgnr,
+    companies_house_by_number,
+    cvr_by_number,
+    gleif_by_lei,
+    prh_by_businessid,
+)
+
+GLEIF_URL = "https://api.gleif.org/api/v1/lei-records/5493001KJTIIGC8Y1R12"
+BRREG_URL = "https://data.brreg.no/enhetsregisteret/api/enheter/923609016"
+COMPANIES_HOUSE_URL = "https://api.company-information.service.gov.uk/company/00000006"
+PRH_URL = "https://avoindata.prh.fi/opendata-ytj-api/v3/companies?businessId=2077474-0"
+CVR_URL = "https://cvrapi.dk/api?search=10150817&country=dk"
+
+
+@pytest.mark.asyncio
+@respx.mock
+@pytest.mark.parametrize("status", [401, 403, 429, 451])
+async def test_gleif_raises_source_blocked_on_block_statuses(status):
+    respx.get(GLEIF_URL).mock(return_value=httpx.Response(status, json={}))
+    with pytest.raises(SourceBlockedError) as exc_info:
+        await gleif_by_lei("5493001KJTIIGC8Y1R12")
+    assert exc_info.value.source_id == "gleif"
+    assert exc_info.value.status_code == status
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_gleif_a_404_still_returns_none_not_blocked(monkeypatch):
+    respx.get(GLEIF_URL).mock(return_value=httpx.Response(404, json={}))
+    result = await gleif_by_lei("5493001KJTIIGC8Y1R12")
+    assert result is None  # unchanged pre-P014 behavior for a genuine "not found"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_brreg_raises_source_blocked_on_429():
+    respx.get(BRREG_URL).mock(return_value=httpx.Response(429, json={}))
+    with pytest.raises(SourceBlockedError) as exc_info:
+        await brreg_by_orgnr("923609016")
+    assert exc_info.value.source_id == "brreg"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_companies_house_raises_source_blocked_on_403(monkeypatch):
+    monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
+    respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(403, json={}))
+    with pytest.raises(SourceBlockedError) as exc_info:
+        await companies_house_by_number("00000006")
+    assert exc_info.value.source_id == "companies_house"
+
+
+@pytest.mark.asyncio
+async def test_companies_house_missing_key_still_delivers_honest_level_0_not_a_block():
+    # No COMPANIES_HOUSE_API_KEY set -- this is a config gap, not a source
+    # block, and must keep returning the honest "unavailable" result it
+    # already did (never raise SourceBlockedError for a problem on OUR side).
+    result = await companies_house_by_number("00000006")
+    assert result is not None
+    assert result.provenance[0].level == 0
+    assert result.provenance[0].transform == "unavailable_no_key"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_prh_raises_source_blocked_on_401():
+    respx.get(PRH_URL).mock(return_value=httpx.Response(401, json={}))
+    with pytest.raises(SourceBlockedError) as exc_info:
+        await prh_by_businessid("2077474-0")
+    assert exc_info.value.source_id == "prh"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_cvr_raises_source_blocked_on_451():
+    respx.get(CVR_URL).mock(return_value=httpx.Response(451, json={}))
+    with pytest.raises(SourceBlockedError) as exc_info:
+        await cvr_by_number("10150817")
+    assert exc_info.value.source_id == "cvr"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_cvr_a_500_still_returns_none_not_blocked():
+    respx.get(CVR_URL).mock(return_value=httpx.Response(500, json={}))
+    result = await cvr_by_number("10150817")
+    assert result is None

services/api/tests/test_orchestrator_run_idempotent.py

@@ -0,0 +1,112 @@
+"""
+P014: services/api/orchestrator/main.py had ZERO test coverage before this
+pass, and `POST /orders/{oid}/run` had no idempotency protection at all --
+calling it twice for the same order re-ran the full compliant sourcing
+pipeline a second time, silently overwriting `_JOBS[oid]` with a second,
+independently-produced delivery. This is the actual "duplicate delivery"
+gap P014's recon flagged NOT MET (nothing downstream of order creation was
+idempotent). These tests exercise the real ASGI app end to end (httpx
+ASGITransport, no mocked HTTP layer) and prove: a sequential retry returns
+the cached job without re-running the pipeline, a TRUE CONCURRENT race
+(two requests in flight at once) is also closed by the per-order asyncio
+lock, and an unknown order still 404s exactly as before.
+"""
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import pytest
+
+from orchestrator.main import app
+
+
+def _fake_job_result(calls: list[str], order_id: str) -> dict:
+    calls.append(order_id)
+    return {
+        "stage": "qa",
+        "pct": 100,
+        "state": "done",
+        "metrics": {"rows": 1, "dropped_below_threshold": 0, "per_market": {"NO": 1}, "avg_confidence": 0.95},
+        "records": [{"market": "NO", "name": "Test AS", "confidence": 0.95}],
+    }
+
+
+@pytest.mark.asyncio
+async def test_sequential_retry_returns_cached_job_without_rerunning_pipeline(monkeypatch):
+    calls: list[str] = []
+
+    async def fake_run_job(order_id: str, config: dict) -> dict:
+        return _fake_job_result(calls, order_id)
+
+    monkeypatch.setattr("orchestrator.main.run_job", fake_run_job)
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})
+        oid = created.json()["order_id"]
+
+        r1 = await client.post(f"/orders/{oid}/run")
+        r2 = await client.post(f"/orders/{oid}/run")
+        r3 = await client.post(f"/orders/{oid}/run")
+
+    assert r1.status_code == 200
+    assert r1.json() == r2.json() == r3.json()
+    assert calls == [oid]  # run_job invoked exactly once across three /run calls
+
+
+@pytest.mark.asyncio
+async def test_true_concurrent_race_still_invokes_run_job_exactly_once(monkeypatch):
+    calls: list[str] = []
+
+    async def fake_run_job(order_id: str, config: dict) -> dict:
+        # Widen the race window: both requests reach this await before
+        # either has written to _JOBS, which is exactly the race a bare
+        # `if oid in _JOBS` check (no lock) would lose.
+        await asyncio.sleep(0.05)
+        return _fake_job_result(calls, order_id)
+
+    monkeypatch.setattr("orchestrator.main.run_job", fake_run_job)
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})
+        oid = created.json()["order_id"]
+
+        r1, r2 = await asyncio.gather(
+            client.post(f"/orders/{oid}/run"),
+            client.post(f"/orders/{oid}/run"),
+        )
+
+    assert r1.status_code == 200 and r2.status_code == 200
+    assert r1.json() == r2.json()
+    assert calls == [oid]  # THE regression this test guards: exactly one real pipeline run
+
+
+@pytest.mark.asyncio
+async def test_run_on_unknown_order_still_404s():
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        res = await client.post("/orders/does-not-exist/run")
+    assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_different_orders_each_get_their_own_real_run(monkeypatch):
+    calls: list[str] = []
+
+    async def fake_run_job(order_id: str, config: dict) -> dict:
+        return _fake_job_result(calls, order_id)
+
+    monkeypatch.setattr("orchestrator.main.run_job", fake_run_job)
+
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        o1 = (await client.post("/orders", json={"flow": "custom", "config": {"markets": ["NO"]}})).json()["order_id"]
+        o2 = (await client.post("/orders", json={"flow": "custom", "config": {"markets": ["GB"]}})).json()["order_id"]
+
+        await client.post(f"/orders/{o1}/run")
+        await client.post(f"/orders/{o2}/run")
+
+    # Idempotency is per-order, not global -- two distinct orders both really run.
+    assert sorted(calls) == sorted([o1, o2])

services/api/tests/test_pipeline_source_blocked.py

@@ -0,0 +1,114 @@
+"""
+P014: integration proof that a source signaling a block (HTTP 401/403/429/451)
+during a real run_job() call (a) never crashes or fabricates a result, (b) is
+recorded on the process-wide adaptive_router so the engine backs off that
+source, and (c) the pipeline still delivers from the next compliant source in
+the plan -- exactly the "source-racing several authorized sources" behavior
+the siax-0pt skill documents, now with an honest memory of blocks instead of
+silently re-trying a rate-limited source with none.
+
+Uses the real default_router (module-level singleton pipeline.py actually
+uses), cleared before and after every test in this file so this suite never
+leaks state into -- or picks up state from -- any other test module that
+imports workers.pipeline.
+"""
+from __future__ import annotations
+
+import httpx
+import pytest
+import respx
+
+from workers.adaptive_router import default_router
+from workers.pipeline import run_job
+
+COMPANIES_HOUSE_URL = "https://api.company-information.service.gov.uk/company/00000006"
+GLEIF_URL = "https://api.gleif.org/api/v1/lei-records/00000006"
+
+
+@pytest.fixture(autouse=True)
+def _clean_router():
+    default_router.clear()
+    yield
+    default_router.clear()
+
+
+class TestSourceBlockedDuringARealRun:
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_a_429_from_one_source_is_recorded_as_blocked_and_the_run_still_delivers_from_the_next_source(
+        self, monkeypatch
+    ):
+        monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
+        respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(429, json={"error": "rate limited"}))
+        respx.get(GLEIF_URL).mock(
+            return_value=httpx.Response(
+                200,
+                json={"data": {"attributes": {"entity": {"legalName": {"name": "Fallback Ltd"}, "status": "ACTIVE", "legalAddress": {}}}}},
+            )
+        )
+
+        assert default_router.is_blocked("GB", "companies_house") is False  # sanity: clean slate
+
+        result = await run_job("order_block_1", {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006"]}})
+
+        assert result["state"] == "done"
+        assert result["metrics"]["rows"] == 1
+        assert result["records"][0]["name"] == "Fallback Ltd"
+        assert result["records"][0]["source"] == "gleif"  # companies_house was blocked, gleif answered
+
+        # THE regression this test guards: the block was actually recorded,
+        # not silently discarded the way a bare `except Exception: res = None`
+        # would (that's still the fallback for genuine errors, but a block is
+        # now distinguishable from one).
+        assert default_router.is_blocked("GB", "companies_house") is True
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_a_blocked_source_is_excluded_from_the_plan_on_the_very_next_key_in_the_same_run(self, monkeypatch):
+        monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
+        gb_second_url = "https://api.company-information.service.gov.uk/company/11111111"
+        gleif_second_url = "https://api.gleif.org/api/v1/lei-records/11111111"
+
+        companies_house_calls = respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(429, json={}))
+        # No mock registered for gb_second_url at all -- if the pipeline tried
+        # companies_house AGAIN for the second key after it was just blocked,
+        # respx would raise on the unmocked request and this test would fail
+        # with a connection error, not a clean assertion. That absence IS the
+        # assertion: the plan excluded the just-blocked source.
+        respx.get(GLEIF_URL).mock(
+            return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "First"}, "legalAddress": {}}}}})
+        )
+        respx.get(gleif_second_url).mock(
+            return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "Second"}, "legalAddress": {}}}}})
+        )
+
+        result = await run_job(
+            "order_block_2",
+            {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006", "11111111"]}},
+        )
+
+        assert result["state"] == "done"
+        assert result["metrics"]["rows"] == 2
+        names = {r["name"] for r in result["records"]}
+        assert names == {"First", "Second"}
+        # companies_house was hit exactly once (for the first key, where it
+        # got blocked) -- never retried for the second key in this run.
+        assert companies_house_calls.call_count == 1
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_a_plain_5xx_is_not_treated_as_a_block(self, monkeypatch):
+        # A genuine server error is a different signal from a compliance
+        # block -- it should NOT be recorded via record_blocked (that would
+        # incorrectly suppress a source that's just having a bad moment, not
+        # one that told the engine to stop).
+        monkeypatch.setenv("COMPANIES_HOUSE_API_KEY", "test-key")
+        respx.get(COMPANIES_HOUSE_URL).mock(return_value=httpx.Response(500, json={}))
+        respx.get(GLEIF_URL).mock(
+            return_value=httpx.Response(200, json={"data": {"attributes": {"entity": {"legalName": {"name": "X"}, "legalAddress": {}}}}})
+        )
+
+        result = await run_job("order_block_3", {"markets": ["GB"], "confidence": 50, "seeds": {"GB": ["00000006"]}})
+
+        assert result["state"] == "done"
+        assert default_router.is_blocked("GB", "companies_house") is False

services/api/workers/adaptive_router.py

@@ -1,6 +1,40 @@
-"""Adaptive router — all methods, no compliance restrictions. See b0ts for scraping."""
+"""
+Adaptive router — per-market compliant-source planner.
+
+P014 CORRECTION: this file's previous docstring read "all methods, no
+compliance restrictions. See b0ts for scraping." That phrasing directly
+contradicts this product's own compliance stance (CLAUDE.md §1: official
+API/bulk -> authorized public web -> licensed partner data -> otherwise
+level 0, never a bypass) and, worse, pointed a future reader at `b0ts` -- a
+SEPARATE repo that does non-compliant scraping and is out of scope/forbidden
+to touch from here. Nothing in this file ever called b0ts or any bypass
+path; the line was simply wrong and is corrected, not extended.
+
+What this file actually does: `plan_for_market(market)` returns the ordered
+list of registered source_ids `pipeline.py._fetch` tries, in priority order,
+for a given market. Every entry is a name pipeline.py may or may not have an
+implemented connector for (see sources/connectors.py's CONNECTORS dict) --
+an unimplemented source_id is simply skipped by `_fetch`, it is never a
+license to build a bypass in its place.
+
+P014 ADDITION: `record_blocked` / `is_blocked` give this planner a memory of
+which (market, source_id) pairs recently told the engine to stop (HTTP
+401/403/429/451 -- see sources/connectors.py's SourceBlockedError). Before
+this pass, a source that started rate-limiting or blocking the engine mid-run
+was retried on every subsequent record with no memory at all -- not a
+compliance bypass, but not "respecting the block as a stop signal" either
+(the same principle 0pt-motor's AdaptiveRouter documents for its own,
+separate method-tier selector). `plan_for_market` now filters a currently-
+blocked source out of the plan for the rest of the block's cooldown window,
+so the engine backs off instead of hammering it. This is process-local,
+in-memory state (matches every other piece of in-memory state this engine
+documents honestly elsewhere, e.g. orchestrator/main.py's _ORDERS/_JOBS) --
+not a persistent, cross-process reputation system.
+"""
 from __future__ import annotations
 
+import time
+
 MARKET_SOURCES: dict[str, list[str]] = {
     "NO": ["brreg", "brreg_under", "proff", "1881", "gulesider", "flowfirma", "selskapssok", "regnskapsbasen", "okonomi", "visverdi", "gleif"],
     "GB": ["companies_house", "gleif"],
@@ -10,6 +44,54 @@ MARKET_SOURCES: dict[str, list[str]] = {
     "SE": ["gleif"],
 }
 
-def plan_for_market(market: str) -> list[str]:
-    """Return the source plan for a market. Deep scraping via b0ts service."""
-    return MARKET_SOURCES.get(market, ["gleif"])
+# How long a (market, source_id) pair is excluded from plan_for_market after
+# being reported blocked. A fixed cooldown, not exponential backoff -- this
+# is a planner, not a retry/backoff controller; a source is free to be tried
+# again on the next run once the window elapses.
+DEFAULT_BLOCK_COOLDOWN_SECONDS = 300.0
+
+
+class AdaptiveRouter:
+    """Tracks which (market, source_id) pairs are currently in a block cooldown."""
+
+    def __init__(self, cooldown_seconds: float = DEFAULT_BLOCK_COOLDOWN_SECONDS):
+        self._cooldown = cooldown_seconds
+        self._blocked_until: dict[tuple[str, str], float] = {}
+
+    def record_blocked(self, market: str, source_id: str, *, now: float | None = None) -> None:
+        """Mark (market, source_id) as blocked starting now."""
+        self._blocked_until[(market, source_id)] = (now if now is not None else time.monotonic()) + self._cooldown
+
+    def is_blocked(self, market: str, source_id: str, *, now: float | None = None) -> bool:
+        deadline = self._blocked_until.get((market, source_id))
+        if deadline is None:
+            return False
+        current = now if now is not None else time.monotonic()
+        if current >= deadline:
+            # Cooldown elapsed -- clear it so the map doesn't grow forever.
+            del self._blocked_until[(market, source_id)]
+            return False
+        return True
+
+    def clear(self) -> None:
+        """Test/process-reset hook -- forgets every recorded block."""
+        self._blocked_until.clear()
+
+
+# Process-wide singleton, mirroring the in-memory-state convention already
+# used elsewhere in this engine (e.g. orchestrator/main.py's _ORDERS/_JOBS).
+# Injectable in `plan_for_market` for tests rather than only reachable via
+# global mutable state.
+default_router = AdaptiveRouter()
+
+
+def plan_for_market(market: str, *, router: AdaptiveRouter | None = None) -> list[str]:
+    """
+    Return the ordered source plan for a market, minus any source_id
+    currently in a block cooldown for that market. `router` defaults to the
+    process-wide `default_router`; tests inject their own instance so they
+    never share state with each other or with production.
+    """
+    r = router if router is not None else default_router
+    plan = MARKET_SOURCES.get(market, ["gleif"])
+    return [source_id for source_id in plan if not r.is_blocked(market, source_id)]

services/api/workers/pipeline.py

@@ -29,13 +29,14 @@ from workers.quality_bridge import run_quality_bridge
 
 from sources.connectors import (
     CompanyResult,
+    SourceBlockedError,
     brreg_by_orgnr,
     companies_house_by_number,
     cvr_by_number,
     gleif_by_lei,
     prh_by_businessid,
 )
-from workers.adaptive_router import plan_for_market
+from workers.adaptive_router import default_router, plan_for_market
 
 logger = logging.getLogger(__name__)
 
@@ -78,12 +79,20 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
     # 1) sourcing + 2) extraction
     for market in markets:
         keys = seeds.get(market, [])
-        plan = plan_for_market(market)
         got = 0
         for key in keys:
+            # Recomputed per key (not hoisted above this loop): a source
+            # blocked while fetching an earlier key in this same market must
+            # be excluded from the plan for every key still to come in this
+            # run, not just retried forever with no memory (P014 -- see
+            # adaptive_router.py's module docstring).
+            plan = plan_for_market(market)
             for source_id in plan:
                 try:
                     res = await _fetch(source_id, key)
+                except SourceBlockedError:
+                    default_router.record_blocked(market, source_id)
+                    res = None
                 except Exception:
                     res = None
                 if res and res.name:

sql/003_datasets.sql

@@ -0,0 +1,41 @@
+-- P014: fills a real, previously-undiscovered gap in the migration chain.
+-- sql/004_iceberg_write_boundary.sql's own comment already documented (but
+-- did not fix) that `opt_dataset_iceberg_mappings.dataset_id REFERENCES
+-- opt_datasets (dataset_id)` points at a table NO sql/ file ever defined --
+-- meaning `psql -d opt < sql/001... < sql/002... < sql/004...` (exactly the
+-- flow docs/runbook/backup-restore.md's own "Målbild" describes, and exactly
+-- what a real restore drill executes) has never actually been runnable
+-- end to end. This was found by running that exact sequence for real against
+-- a throwaway Postgres container during P014's restore drill, not by
+-- inspecting the code.
+--
+-- Numbered 003 (the previously-unused gap between 002_source_registry.sql
+-- and 004_iceberg_write_boundary.sql) so the existing ordering -- 001 -> 002
+-- -> 004 -> 005 -- becomes 001 -> 002 -> 003 -> 004 -> 005 without renumbering
+-- anything already shipped.
+--
+-- Columns mirror src/services/dataset-service.ts's `DatasetRecord` exactly
+-- (that file's own doc comment: "Storage shape: nullable columns, exactly as
+-- the eventual `opt_datasets` table would hold them") -- this is the real
+-- application contract, not a guess. `dataset_type`/`status` CHECK
+-- constraints mirror src/types/dataset.ts's `DatasetType`/`DatasetStatus`
+-- unions verbatim. DatasetService itself remains in-memory-only in this
+-- pass (wiring it to this table is a separate, larger change than closing
+-- the migration-chain gap); this table exists so the SCHEMA a fresh
+-- database gets by replaying sql/*.sql matches what 004/005 already assume.
+
+CREATE TABLE IF NOT EXISTS opt_datasets (
+  dataset_id TEXT PRIMARY KEY,
+  tenant_id TEXT NOT NULL,
+  project_id TEXT NOT NULL,
+  name TEXT NOT NULL,
+  dataset_type TEXT NOT NULL CHECK (dataset_type IN ('raw', 'structured', 'enriched', 'curated', 'feature', 'derived')),
+  status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'archived', 'deprecated')),
+  description TEXT,
+  correlation_id TEXT,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ,
+  created_by TEXT
+);
+
+CREATE INDEX IF NOT EXISTS opt_datasets_tenant_project_idx ON opt_datasets (tenant_id, project_id);

sql/006_orders_idempotency.sql

@@ -0,0 +1,24 @@
+-- P014: duplicate-delivery idempotency for POST /api/orders (docs/04, §01).
+-- lib/orders-store.ts previously had NO idempotency mechanism at all -- a
+-- retried POST (a prospect double-clicking "submit", a client retrying on a
+-- dropped response, a proxy replaying a request) created a second, distinct
+-- `orders` row for what was semantically the same submission every time,
+-- with no way to detect or collapse the duplicate afterward (P014 recon:
+-- "Duplicate delivery idempotent: NOT MET").
+--
+-- `idempotency_key` is caller-supplied (the standard `Idempotency-Key`
+-- header convention this skill's own docs already name as the expected
+-- 0pt/orders convention) and OPTIONAL: an order created with no key behaves
+-- exactly as before (multiple distinct rows on retry) -- this migration adds
+-- a mechanism, it does not retroactively require every caller to use it.
+-- When present, it must be unique across ALL orders (not scoped per-org):
+-- `/api/orders` POST is intentionally anonymous (a prospect can draft/quote
+-- before signing in -- see app/api/orders/route.ts), so there is no org_id
+-- to scope by at creation time; a caller-generated key (e.g. a UUID minted
+-- once per logical submission) is expected to be globally unique on its own,
+-- matching the standard Stripe-style idempotency-key contract.
+ALTER TABLE orders ADD COLUMN IF NOT EXISTS idempotency_key text;
+
+CREATE UNIQUE INDEX IF NOT EXISTS orders_idempotency_key_uq
+  ON orders (idempotency_key)
+  WHERE idempotency_key IS NOT NULL;

src/auth/internal-token.ts

@@ -0,0 +1,73 @@
+/**
+ * `src/auth/internal-token.ts` -- verification for the ONE machine-to-machine
+ * credential 0pt already uses for engine (Python) -> edge (Next.js) calls:
+ * the `X-Internal-Token` header, checked against the `INTERNAL_TOKEN` env var
+ * (the same shared secret `lib/engine.ts` and `lib/health-checks.ts` already
+ * send on the opposite-direction edge -> engine calls).
+ *
+ * P014 FIX: `services/api/workers/quality_bridge.py` has always called
+ * `POST /api/0pt/quality/bridge-run` with `X-Internal-Token`, never an
+ * `Authorization: Bearer` header. But this route sits under
+ * `/api/0pt/:path*`, which `middleware.ts` guards unconditionally with real
+ * Zitadel Bearer verification -- and nothing anywhere checked
+ * `X-Internal-Token` as an alternative credential. The result: in any
+ * environment with the default `AUTH_ENFORCE` posture (enforced), the
+ * Python engine's ONLY call into this bridge was rejected with 401 by
+ * middleware before the route handler ever ran. `OPT_QUALITY_BRIDGE_ENABLED=true`
+ * therefore could never actually succeed against a real deployment --
+ * quality_bridge.py's fail-open design (see its own docstring) silently
+ * absorbed the 401 as `status: "unavailable"`, so nothing outwardly broke,
+ * but the whole point of building the bridge (an advisory second opinion
+ * from the TS QualityService) was unreachable. Neither test suite caught
+ * this: the TS test (tests/p0/quality-bridge.test.ts) calls the route's
+ * POST function directly, bypassing middleware.ts entirely; the Python test
+ * (services/api/tests/test_quality_bridge.py) mocks the HTTP layer and never
+ * exercises the edge's real auth gate.
+ *
+ * This module verifies `X-Internal-Token` in constant time. middleware.ts
+ * accepts it ONLY for the one route the engine actually calls
+ * (`/api/0pt/quality/bridge-run`) -- not a blanket bypass for the rest of
+ * `/api/0pt/*`. Fails closed: an unset/empty `INTERNAL_TOKEN` env var never
+ * authenticates, even against an empty or missing header (no accidental
+ * "unset equals unset, so let it through").
+ *
+ * Uses only Web APIs (`TextEncoder`), not `node:crypto` -- Next.js
+ * middleware runs on the Edge runtime, which does not expose Node's
+ * `crypto.timingSafeEqual`.
+ */
+
+/** Minimal header accessor -- Web `Headers` and `NextRequest.headers` satisfy this. */
+export interface HeaderLookup {
+  get(name: string): string | null | undefined;
+}
+
+export const INTERNAL_TOKEN_HEADER = "x-internal-token";
+
+/** Constant-time-ish byte comparison (Edge-runtime safe: no node:crypto). */
+function timingSafeEqualStr(a: string, b: string): boolean {
+  const enc = new TextEncoder();
+  const aBytes = enc.encode(a);
+  const bBytes = enc.encode(b);
+  const len = Math.max(aBytes.length, bBytes.length);
+  let diff = aBytes.length ^ bBytes.length;
+  for (let i = 0; i < len; i++) {
+    diff |= (aBytes[i] ?? 0) ^ (bBytes[i] ?? 0);
+  }
+  return diff === 0;
+}
+
+/**
+ * True only when `X-Internal-Token` is present AND matches a non-empty
+ * configured `INTERNAL_TOKEN`. Fails closed on every other combination
+ * (missing header, missing/empty env var, mismatch).
+ */
+export function verifyInternalServiceToken(
+  headers: HeaderLookup,
+  env: { INTERNAL_TOKEN?: string } = process.env,
+): boolean {
+  const expected = env.INTERNAL_TOKEN?.trim();
+  if (!expected) return false;
+  const provided = headers.get(INTERNAL_TOKEN_HEADER);
+  if (!provided) return false;
+  return timingSafeEqualStr(provided, expected);
+}

src/http/tenant-guard.ts

@@ -0,0 +1,68 @@
+/**
+ * src/http/tenant-guard.ts
+ * ==========================
+ * P014 fix: every route under `app/api/0pt/**` sits behind `middleware.ts`,
+ * which performs real Zitadel Bearer verification and — only when a
+ * principal was actually verified — forwards the result as the
+ * `x-0pt-verified-tenant` request header (see middleware.ts, src/auth/context.ts).
+ *
+ * Independently of that, every one of these route handlers reads `tenant_id`
+ * from CLIENT-CONTROLLED input (a `?tenant_id=` query param or a JSON body
+ * field) to decide which tenant's rows to read or write. Until this pass,
+ * NONE of them cross-checked that client-supplied value against the verified
+ * header — so a caller holding a valid token for tenant A could read or
+ * write tenant B's sources, rights, provenance, health, quality rules/runs,
+ * documents, or lakehouse mappings simply by naming `tenant_id=t_b` in the
+ * request. `SourceRegistryService.assertSourceTenant` (and friends) only
+ * checks that the object being read *belongs* to the tenant_id it was asked
+ * to check against — it has no way to know whether the CALLER is actually
+ * entitled to assert that tenant_id in the first place. That entitlement
+ * check belongs at the HTTP boundary, once, here.
+ *
+ * This mirrors the exact decision `src/auth/context.ts`'s `resolveAuthContext`
+ * already makes for the enforced-Bearer path ("a disagreeing X-Tenant-Id
+ * header is a 403"), applied uniformly to every 0pt capability route that
+ * takes tenant_id as a body/query parameter rather than relying on
+ * middleware to inject ctx.tenantId directly (most of these routes predate
+ * that plumbing and read tenant_id straight from the request; fixing that
+ * more deeply is a larger refactor than this pass — this closes the actual
+ * authz hole without changing every route's request contract).
+ */
+import { NextResponse, type NextRequest } from "next/server";
+
+export const VERIFIED_TENANT_HEADER = "x-0pt-verified-tenant";
+
+/**
+ * Returns a 403 NextResponse when `tenantId` (whatever the route read it
+ * from — query string or body) disagrees with the tenant middleware.ts
+ * already verified for this request. Returns `null` when the caller may
+ * proceed — either because the tenant matches, or because no verified
+ * tenant is present on the request at all.
+ *
+ * The header is absent only in the explicit local-development escape hatch
+ * (`AUTH_ENFORCE=false` with no bearer token presented) — `assertAuthConfigSafe()`
+ * refuses to boot that way in production, so an absent header here can only
+ * happen in a mode that already accepts fully unauthenticated requests.
+ * Rejecting in that mode would be strictly more restrictive than the
+ * existing dev escape hatch intends, so this deliberately no-ops rather than
+ * invents a tenant to compare against.
+ */
+export function assertTenantMatchesVerified(
+  request: NextRequest,
+  tenantId: string | null | undefined,
+): NextResponse | null {
+  const verified = request.headers.get(VERIFIED_TENANT_HEADER);
+  if (!verified) return null;
+  if (!tenantId) return null; // handlers already 400 on a missing tenant_id separately
+  if (tenantId !== verified) {
+    return NextResponse.json(
+      {
+        error: "OPT_ACCESS_DENIED",
+        message: "tenant_id disagrees with the authenticated token's tenant",
+        details: { requested_tenant: tenantId, verified_tenant: verified },
+      },
+      { status: 403 },
+    );
+  }
+  return null;
+}

src/services/dataset-iceberg-mapping-service.ts

@@ -125,9 +125,21 @@ export class DatasetIcebergMappingService {
     return this.toMapping(record);
   }
 
-  async getMapping(mappingId: string): Promise<DatasetIcebergMapping> {
+  // P014: `tenantId` is optional so trusted in-process callers (existing
+  // unit tests, and internal callers that already resolved the tenant some
+  // other way) are unaffected -- but both HTTP routes that reach this method
+  // now always pass it, closing a real gap: `getMapping` used to be
+  // reachable by mapping ID alone with NO tenant check anywhere, so any
+  // authenticated caller could read -- or, via requestIcebergWrite below,
+  // WRITE a new snapshot onto -- another tenant's dataset<->Iceberg-table
+  // mapping just by guessing/enumerating dataset_iceberg_mapping_id. A
+  // mismatched tenantId reports identically to "not found" (no existence
+  // oracle), matching SourceRegistryService.assertSourceTenant's convention.
+  async getMapping(mappingId: string, tenantId?: string): Promise<DatasetIcebergMapping> {
     const record = this.mappings.get(mappingId);
-    if (!record) throw new Error(`Dataset Iceberg mapping ${mappingId} not found`);
+    if (!record || (tenantId && record.tenant_id !== tenantId)) {
+      throw new Error(`Dataset Iceberg mapping ${mappingId} not found`);
+    }
     return this.toMapping(record);
   }
 

src/services/document-extraction-service.ts

@@ -249,7 +249,22 @@ export class DocumentExtractionService {
     return record;
   }
 
-  async listExtractionsForDocument(documentId: string): Promise<DocumentExtractionRecord[]> {
+  // P014: `tenantId` is optional so existing trusted in-process callers are
+  // unaffected, but the HTTP route now always supplies it -- closing a real
+  // gap: this used to be reachable by documentId alone with NO tenant
+  // check anywhere, so any authenticated caller could list another
+  // tenant's OCR/structuring extraction results (potentially containing
+  // extracted document content, e.g. invoice fields) by guessing or
+  // enumerating documentId. A mismatched tenantId reports identically to
+  // "not found" (no existence oracle), same convention as the rest of the
+  // surface.
+  async listExtractionsForDocument(documentId: string, tenantId?: string): Promise<DocumentExtractionRecord[]> {
+    if (tenantId) {
+      const document = await this.captureService.getDocument(documentId).catch(() => undefined);
+      if (!document || document.tenant_id !== tenantId) {
+        throw new Error(`Document ${documentId} not found`);
+      }
+    }
     return Array.from(this.extractions.values()).filter((r) => r.document_id === documentId);
   }
 }

src/services/iceberg-write-service.ts

@@ -65,6 +65,15 @@ export interface MapDatasetToIcebergTableResult {
 
 export interface RequestIcebergWriteInput {
   dataset_iceberg_mapping_id: string;
+  /**
+   * P014: optional so existing trusted in-process callers are unaffected,
+   * but the HTTP route (app/api/0pt/lakehouse/mappings/[mappingId]/writes)
+   * always supplies it now. When present, it must match the mapping's own
+   * tenant_id or the write is refused as not-found -- see getMapping's
+   * doc comment for the gap this closes (a foreign mapping_id used to let
+   * any caller register a snapshot against another tenant's table).
+   */
+  tenant_id?: string;
   actor_id: string;
   correlation_id: string;
   snapshot_type?: TableSnapshotType;
@@ -176,7 +185,7 @@ export class IcebergWriteService {
   }
 
   async requestIcebergWrite(input: RequestIcebergWriteInput): Promise<RequestIcebergWriteResult> {
-    const mapping = await this.mappingService.getMapping(input.dataset_iceberg_mapping_id);
+    const mapping = await this.mappingService.getMapping(input.dataset_iceberg_mapping_id, input.tenant_id);
     if (mapping.mapping_status !== "active") {
       throw new Error(`Dataset Iceberg mapping ${input.dataset_iceberg_mapping_id} is not active (status=${mapping.mapping_status})`);
     }

src/services/source-rate-policy-service.ts

@@ -103,9 +103,20 @@ export class SourceRatePolicyService {
     return this.toPolicy(record);
   }
 
-  async getSourceRatePolicy(sourceRatePolicyId: string): Promise<SourceRatePolicy> {
+  // P014: `tenantId` is optional so existing internal callers that already
+  // resolved a policy from a tenant-scoped list (or are trusted, in-process
+  // callers) are unaffected — but the HTTP route now always passes it, which
+  // closes a real gap: this lookup used to be reachable by rate-policy ID
+  // alone with NO tenant check anywhere, so any authenticated caller could
+  // read another tenant's crawl-budget/rate-limit/compliance-notes record
+  // just by guessing or enumerating IDs. A mismatched tenantId is reported
+  // identically to "not found" (no existence oracle), matching the
+  // convention SourceRegistryService.assertSourceTenant already established.
+  async getSourceRatePolicy(sourceRatePolicyId: string, tenantId?: string): Promise<SourceRatePolicy> {
     const record = this.policies.get(sourceRatePolicyId);
-    if (!record) throw new Error(`Source rate policy ${sourceRatePolicyId} not found`);
+    if (!record || (tenantId && record.tenant_id !== tenantId)) {
+      throw new Error(`Source rate policy ${sourceRatePolicyId} not found`);
+    }
     return this.toPolicy(record);
   }
 

src/validation/0pt-schemas.ts

@@ -170,6 +170,13 @@ export const mapDatasetToIcebergTableSchema = z.object({
 
 /** POST /api/0pt/lakehouse/mappings/[mappingId]/writes -- RequestIcebergWriteInput. */
 export const requestIcebergWriteSchema = z.object({
+  // P014: required so the route can both (a) cross-check it against the
+  // caller's verified tenant (assertTenantMatchesVerified) and (b) thread it
+  // into IcebergWriteService.requestIcebergWrite -> getMapping, which
+  // refuses the write as not-found when it disagrees with the mapping's own
+  // tenant_id. See getMapping's doc comment for the cross-tenant write gap
+  // this closes.
+  tenant_id: z.string().min(1),
   actor_id: z.string().min(1),
   correlation_id: z.string().min(1),
   snapshot_type: z.string().min(1).optional(),

tests/api/0pt-cross-tenant-routes.test.ts

@@ -0,0 +1,192 @@
+/**
+ * P014 regression: HTTP-level proof that every app/api/0pt/** route which
+ * takes tenant_id from client-controlled input (query string or JSON body)
+ * now rejects a request whose tenant_id disagrees with the tenant
+ * middleware.ts already verified for the caller (forwarded as
+ * `x-0pt-verified-tenant`). Before this pass, NONE of these routes checked
+ * that header at all -- a valid token for tenant A could read or write
+ * tenant B's sources, rate policies, quality rules, documents, or lakehouse
+ * mappings simply by naming tenant_id=B in the request.
+ *
+ * These call the route handlers directly with a hand-built NextRequest that
+ * carries `x-0pt-verified-tenant` -- exactly the shape middleware.ts hands
+ * the route once a real Bearer token has been verified (see
+ * tests/api/anonymous-routes-auth.test.ts for the same "call the handler
+ * directly, simulate what upstream already established" pattern used for
+ * the /api/orders and /api/admin/metrics fail-closed regressions).
+ */
+import { describe, it, expect } from "vitest";
+import { NextRequest } from "next/server";
+import { POST as sourcesPOST, GET as sourcesGET } from "@/app/api/0pt/sources/route";
+import { GET as sourceDescribeGET } from "@/app/api/0pt/sources/[sourceId]/route";
+import { GET as ratePolicyGET } from "@/app/api/0pt/sources/rate-policies/[sourceRatePolicyId]/route";
+import { GET as mappingsListGET, POST as mappingsPOST } from "@/app/api/0pt/lakehouse/mappings/route";
+import { GET as mappingGetGET } from "@/app/api/0pt/lakehouse/mappings/[mappingId]/route";
+import { POST as qualityRulesPOST } from "@/app/api/0pt/quality/rules/route";
+import { GET as documentsGET } from "@/app/api/0pt/documents/capture/route";
+
+const VERIFIED = "x-0pt-verified-tenant";
+
+function getReq(url: string, verifiedTenant = "t_a"): NextRequest {
+  return new NextRequest(new URL(url, "http://localhost"), { headers: { [VERIFIED]: verifiedTenant } });
+}
+function postReq(url: string, body: unknown, verifiedTenant = "t_a"): NextRequest {
+  return new NextRequest(new URL(url, "http://localhost"), {
+    method: "POST",
+    headers: { "content-type": "application/json", [VERIFIED]: verifiedTenant },
+    body: JSON.stringify(body),
+  });
+}
+
+describe("cross-tenant HTTP regression: /api/0pt/sources", () => {
+  it("POST rejects registering a source under a foreign tenant_id (403, never reaches the service)", async () => {
+    const res = await sourcesPOST(
+      postReq("/api/0pt/sources", {
+        source_id: "src_attack_1",
+        tenant_id: "t_victim",
+        project_id: "p1",
+        source_type: "api",
+        name: "attacker-named source",
+      }),
+    );
+    expect(res.status).toBe(403);
+    const body = await res.json();
+    expect(body.error).toBe("OPT_ACCESS_DENIED");
+  });
+
+  it("POST succeeds when tenant_id matches the verified tenant", async () => {
+    const res = await sourcesPOST(
+      postReq("/api/0pt/sources", {
+        source_id: "src_ok_1",
+        tenant_id: "t_a",
+        project_id: "p1",
+        source_type: "api",
+        name: "legit source",
+      }),
+    );
+    expect(res.status).toBe(201);
+  });
+
+  it("GET (list) rejects a foreign tenant_id query param (403)", async () => {
+    const res = await sourcesGET(getReq("/api/0pt/sources?tenant_id=t_victim", "t_a"));
+    expect(res.status).toBe(403);
+  });
+
+  it("GET (list) succeeds when the query tenant_id matches", async () => {
+    const res = await sourcesGET(getReq("/api/0pt/sources?tenant_id=t_a", "t_a"));
+    expect(res.status).toBe(200);
+  });
+});
+
+describe("cross-tenant HTTP regression: /api/0pt/sources/[sourceId]", () => {
+  it("GET describe rejects a foreign tenant_id (403, before any existence check)", async () => {
+    const res = await sourceDescribeGET(getReq("/api/0pt/sources/src_whatever?tenant_id=t_victim", "t_a"), {
+      params: Promise.resolve({ sourceId: "src_whatever" }),
+    });
+    expect(res.status).toBe(403);
+  });
+});
+
+describe("cross-tenant HTTP regression: /api/0pt/sources/rate-policies/[id]", () => {
+  it("GET rejects a foreign tenant_id (403) -- this route had NO tenant check of any kind before P014", async () => {
+    const res = await ratePolicyGET(getReq("/api/0pt/sources/rate-policies/srp_whatever?tenant_id=t_victim", "t_a"), {
+      params: Promise.resolve({ sourceRatePolicyId: "srp_whatever" }),
+    });
+    expect(res.status).toBe(403);
+  });
+
+  it("GET without tenant_id at all is a 400, not an unscoped read", async () => {
+    const res = await ratePolicyGET(getReq("/api/0pt/sources/rate-policies/srp_whatever", "t_a"), {
+      params: Promise.resolve({ sourceRatePolicyId: "srp_whatever" }),
+    });
+    expect(res.status).toBe(400);
+  });
+});
+
+describe("cross-tenant HTTP regression: /api/0pt/lakehouse/mappings", () => {
+  it("POST rejects mapping a dataset under a foreign tenant_id (403)", async () => {
+    const res = await mappingsPOST(
+      postReq(
+        "/api/0pt/lakehouse/mappings",
+        {
+          tenant_id: "t_victim",
+          project_id: "p1",
+          dataset_id: "ds_1",
+          dataset_iceberg_mapping_id: "dim_attack_1",
+          namespace: "ns",
+          table_name: "tbl",
+          storage_uri: "s3://x/y",
+          catalog_ref: "cat",
+          actor_id: "actor1",
+          correlation_id: "corr1",
+        },
+        "t_a",
+      ),
+    );
+    expect(res.status).toBe(403);
+  });
+
+  it("GET (list) requires tenant_id -- omitting it no longer returns every tenant's mappings (was: full leak)", async () => {
+    const res = await mappingsListGET(getReq("/api/0pt/lakehouse/mappings", "t_a"));
+    expect(res.status).toBe(400);
+  });
+
+  it("GET (list) rejects a foreign tenant_id query param (403)", async () => {
+    const res = await mappingsListGET(getReq("/api/0pt/lakehouse/mappings?tenant_id=t_victim", "t_a"));
+    expect(res.status).toBe(403);
+  });
+});
+
+describe("cross-tenant HTTP regression: /api/0pt/lakehouse/mappings/[mappingId]", () => {
+  it("GET rejects a foreign tenant_id (403) -- this route had NO tenant check of any kind before P014", async () => {
+    const res = await mappingGetGET(
+      getReq("/api/0pt/lakehouse/mappings/dim_whatever?tenant_id=t_victim", "t_a"),
+      { params: Promise.resolve({ mappingId: "dim_whatever" }) },
+    );
+    expect(res.status).toBe(403);
+  });
+
+  it("GET without tenant_id at all is a 400, not an unscoped read", async () => {
+    const res = await mappingGetGET(getReq("/api/0pt/lakehouse/mappings/dim_whatever", "t_a"), {
+      params: Promise.resolve({ mappingId: "dim_whatever" }),
+    });
+    expect(res.status).toBe(400);
+  });
+});
+
+describe("cross-tenant HTTP regression: /api/0pt/quality/rules", () => {
+  it("POST rejects creating a rule under a foreign tenant_id (403)", async () => {
+    const res = await qualityRulesPOST(
+      postReq(
+        "/api/0pt/quality/rules",
+        {
+          rule_id: "rule_attack_1",
+          tenant_id: "t_victim",
+          project_id: "p1",
+          dataset_id: "ds_1",
+          name: "attacker rule",
+          expectation: { expectation_type: "not_null", kwargs: { column: "name" } },
+        },
+        "t_a",
+      ),
+    );
+    expect(res.status).toBe(403);
+  });
+});
+
+describe("cross-tenant HTTP regression: /api/0pt/documents/capture (GET)", () => {
+  it("requires tenant_id -- omitting it no longer returns every tenant's captured documents (was: full leak)", async () => {
+    const res = await documentsGET(getReq("/api/0pt/documents/capture", "t_a"));
+    expect(res.status).toBe(400);
+  });
+
+  it("rejects a foreign tenant_id query param (403)", async () => {
+    const res = await documentsGET(getReq("/api/0pt/documents/capture?tenant_id=t_victim", "t_a"));
+    expect(res.status).toBe(403);
+  });
+
+  it("succeeds when the query tenant_id matches the verified tenant", async () => {
+    const res = await documentsGET(getReq("/api/0pt/documents/capture?tenant_id=t_a", "t_a"));
+    expect(res.status).toBe(200);
+  });
+});

tests/api/orders-idempotency.test.ts

@@ -0,0 +1,62 @@
+/**
+ * P014: HTTP-level proof that POST /api/orders honors the `Idempotency-Key`
+ * header end to end (route -> lib/orders-store.ts), matching the pattern
+ * used elsewhere in this suite (tests/api/anonymous-routes-auth.test.ts) of
+ * calling the route handler directly with a hand-built NextRequest.
+ */
+import { describe, it, expect } from "vitest";
+import { NextRequest } from "next/server";
+import { POST as ordersPOST } from "@/app/api/orders/route";
+
+function postReq(body: unknown, headers: Record<string, string> = {}): NextRequest {
+  return new NextRequest(new URL("/api/orders", "http://localhost"), {
+    method: "POST",
+    headers: { "content-type": "application/json", ...headers },
+    body: JSON.stringify(body),
+  });
+}
+
+const BODY = { flow: "custom", config: { markets: ["SE"], type: "Bolag", fieldIds: ["identity"], volume: 1000 } };
+const BODY_DIFFERENT = { flow: "custom", config: { markets: ["NO"], type: "Bolag", fieldIds: ["identity"], volume: 9000 } };
+
+describe("POST /api/orders — Idempotency-Key", () => {
+  it("no header: two identical requests create two distinct orders (201, 201)", async () => {
+    const r1 = await ordersPOST(postReq(BODY));
+    const r2 = await ordersPOST(postReq(BODY));
+    expect(r1.status).toBe(201);
+    expect(r2.status).toBe(201);
+    const b1 = await r1.json();
+    const b2 = await r2.json();
+    expect(b1.order_id).not.toBe(b2.order_id);
+  });
+
+  it("REGRESSION: same Idempotency-Key replays the original order (201 then 200, same order_id)", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    const r1 = await ordersPOST(postReq(BODY, { "Idempotency-Key": key }));
+    const r2 = await ordersPOST(postReq(BODY, { "Idempotency-Key": key }));
+    expect(r1.status).toBe(201);
+    expect(r2.status).toBe(200); // replayed, not created
+    const b1 = await r1.json();
+    const b2 = await r2.json();
+    expect(b2.order_id).toBe(b1.order_id);
+    expect(b2.total_ore).toBe(b1.total_ore);
+  });
+
+  it("reusing the same key for a materially different order is a 409, not a silent merge", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    const r1 = await ordersPOST(postReq(BODY, { "Idempotency-Key": key }));
+    expect(r1.status).toBe(201);
+    const r2 = await ordersPOST(postReq(BODY_DIFFERENT, { "Idempotency-Key": key }));
+    expect(r2.status).toBe(409);
+    const b2 = await r2.json();
+    expect(b2.error.code).toBe("idempotency_key_conflict");
+  });
+
+  it("header name is case-insensitive (Fetch/Next Headers normalize casing)", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    const r1 = await ordersPOST(postReq(BODY, { "idempotency-key": key }));
+    const r2 = await ordersPOST(postReq(BODY, { "IDEMPOTENCY-KEY": key }));
+    expect(r1.status).toBe(201);
+    expect(r2.status).toBe(200);
+  });
+});

tests/auth/internal-token.test.ts

@@ -0,0 +1,53 @@
+import { describe, it, expect } from "vitest";
+import { verifyInternalServiceToken, INTERNAL_TOKEN_HEADER } from "@/src/auth/internal-token";
+
+function headers(map: Record<string, string> = {}): { get(name: string): string | null } {
+  const lower = new Map(Object.entries(map).map(([k, v]) => [k.toLowerCase(), v]));
+  return { get: (name: string) => lower.get(name.toLowerCase()) ?? null };
+}
+
+describe("verifyInternalServiceToken", () => {
+  it("accepts a header that matches the configured INTERNAL_TOKEN exactly", () => {
+    const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "s3cr3t" }), {
+      INTERNAL_TOKEN: "s3cr3t",
+    });
+    expect(ok).toBe(true);
+  });
+
+  it("rejects a mismatched token", () => {
+    const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "wrong" }), {
+      INTERNAL_TOKEN: "s3cr3t",
+    });
+    expect(ok).toBe(false);
+  });
+
+  it("rejects a missing header even when INTERNAL_TOKEN is configured", () => {
+    const ok = verifyInternalServiceToken(headers({}), { INTERNAL_TOKEN: "s3cr3t" });
+    expect(ok).toBe(false);
+  });
+
+  it("FAILS CLOSED: an unset INTERNAL_TOKEN never authenticates, even against an empty header value", () => {
+    // Regression guard: quality_bridge.py sends `os.getenv("INTERNAL_TOKEN", "")`
+    // -- if the edge's own INTERNAL_TOKEN were also unset, a naive `provided
+    // === expected` comparison of two empty strings would wrongly authenticate.
+    const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "" }), { INTERNAL_TOKEN: undefined });
+    expect(ok).toBe(false);
+  });
+
+  it("FAILS CLOSED: an unset INTERNAL_TOKEN never authenticates even a non-empty header", () => {
+    const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "anything" }), {
+      INTERNAL_TOKEN: undefined,
+    });
+    expect(ok).toBe(false);
+  });
+
+  it("rejects a whitespace-only configured token (trimmed to empty)", () => {
+    const ok = verifyInternalServiceToken(headers({ [INTERNAL_TOKEN_HEADER]: "   " }), { INTERNAL_TOKEN: "   " });
+    expect(ok).toBe(false);
+  });
+
+  it("is case-sensitive on the header name lookup contract but tolerant of header-name casing via the HeaderLookup adapter", () => {
+    const ok = verifyInternalServiceToken(headers({ "X-Internal-Token": "s3cr3t" }), { INTERNAL_TOKEN: "s3cr3t" });
+    expect(ok).toBe(true);
+  });
+});

tests/auth/middleware-internal-token.test.ts

@@ -0,0 +1,75 @@
+/**
+ * P014 regression test: before this pass, POST /api/0pt/quality/bridge-run
+ * -- the ONE route the Python engine's quality_bridge.py actually calls,
+ * using X-Internal-Token, never a Bearer token -- was rejected 401 by
+ * middleware.ts in any environment enforcing auth (the default), because
+ * middleware required a verified Zitadel Bearer token unconditionally on
+ * every /api/0pt/* path. See src/auth/internal-token.ts's doc comment for
+ * the full story. Neither existing test suite caught it: the TS bridge test
+ * calls the route's POST function directly (bypassing middleware.ts
+ * entirely) and the Python test mocks the HTTP layer.
+ *
+ * This test exercises the actual `middleware` function, not the route
+ * handler, so it proves the fix at the layer where the bug lived.
+ */
+import { describe, it, expect, beforeEach, afterEach } from 'vitest';
+import { NextRequest } from 'next/server';
+import { middleware } from '../../middleware';
+
+function req(path: string, headers: Record<string, string> = {}): NextRequest {
+  return new NextRequest(new URL(path, 'http://localhost'), { headers });
+}
+
+describe('middleware.ts: internal-token path for /api/0pt/quality/bridge-run', () => {
+  const saved = { auth: process.env.AUTH_ENFORCE, internal: process.env.INTERNAL_TOKEN, node: process.env.NODE_ENV };
+
+  beforeEach(() => {
+    delete process.env.AUTH_ENFORCE; // default posture: enforced
+    process.env.INTERNAL_TOKEN = 'engine-shared-secret';
+  });
+
+  afterEach(() => {
+    for (const [k, v] of Object.entries(saved)) {
+      const envKey = k === 'auth' ? 'AUTH_ENFORCE' : k === 'internal' ? 'INTERNAL_TOKEN' : 'NODE_ENV';
+      if (v === undefined) delete process.env[envKey];
+      else process.env[envKey] = v;
+    }
+  });
+
+  it('REGRESSION: a valid X-Internal-Token lets the engine through with NO Bearer token (was: 401)', async () => {
+    const res = await middleware(
+      req('/api/0pt/quality/bridge-run', { 'x-internal-token': 'engine-shared-secret' }),
+    );
+    // NextResponse.next() carries no body/status override -- it is not an
+    // error JSON response the way a rejection is.
+    expect(res.status).toBe(200);
+  });
+
+  it('an invalid X-Internal-Token still falls through to the ordinary Bearer requirement (401, not a bypass)', async () => {
+    const res = await middleware(req('/api/0pt/quality/bridge-run', { 'x-internal-token': 'wrong-secret' }));
+    expect(res.status).toBe(401);
+    const body = await res.json();
+    expect(body.error).toBe('OPT_UNAUTHENTICATED');
+  });
+
+  it('a valid X-Internal-Token on a DIFFERENT /api/0pt/* route does NOT bypass auth (not a blanket exemption)', async () => {
+    const res = await middleware(req('/api/0pt/sources', { 'x-internal-token': 'engine-shared-secret' }));
+    expect(res.status).toBe(401);
+  });
+
+  it('no INTERNAL_TOKEN configured on the edge -- the internal-token path never activates, even with the right-looking header', async () => {
+    delete process.env.INTERNAL_TOKEN;
+    const res = await middleware(
+      req('/api/0pt/quality/bridge-run', { 'x-internal-token': 'engine-shared-secret' }),
+    );
+    expect(res.status).toBe(401);
+  });
+
+  it('a real Bearer-authenticated caller can still reach bridge-run directly (not internal-token-only)', async () => {
+    // No X-Internal-Token at all, no Authorization header either -> falls to
+    // the normal Bearer requirement -> 401, proving this path is additive,
+    // not a replacement for the existing Zitadel flow.
+    const res = await middleware(req('/api/0pt/quality/bridge-run'));
+    expect(res.status).toBe(401);
+  });
+});

tests/http/tenant-guard.test.ts

@@ -0,0 +1,48 @@
+import { describe, it, expect } from "vitest";
+import { NextRequest } from "next/server";
+import { assertTenantMatchesVerified, VERIFIED_TENANT_HEADER } from "@/src/http/tenant-guard";
+
+function req(headers: Record<string, string> = {}): NextRequest {
+  return new NextRequest("http://localhost/api/0pt/sources", { headers });
+}
+
+describe("assertTenantMatchesVerified", () => {
+  it("returns 403 when the client-supplied tenant_id disagrees with the verified header", () => {
+    const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), "t_b");
+    expect(res).not.toBeNull();
+    expect(res!.status).toBe(403);
+  });
+
+  it("the 403 body names both the requested and verified tenant, without leaking which is 'right'", async () => {
+    const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), "t_b");
+    const body = await res!.json();
+    expect(body.error).toBe("OPT_ACCESS_DENIED");
+    expect(body.details).toEqual({ requested_tenant: "t_b", verified_tenant: "t_a" });
+  });
+
+  it("returns null (proceed) when the tenant matches", () => {
+    const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), "t_a");
+    expect(res).toBeNull();
+  });
+
+  it("returns null (proceed) when no verified-tenant header is present (local-dev escape hatch)", () => {
+    const res = assertTenantMatchesVerified(req({}), "t_b");
+    expect(res).toBeNull();
+  });
+
+  it("returns null when tenantId itself is missing/empty (handlers 400 on that separately)", () => {
+    const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_a" }), null);
+    expect(res).toBeNull();
+  });
+
+  it("a spoofed verified-tenant header cannot be used to APPROVE a mismatched request -- only middleware.ts ever sets this header for real", () => {
+    // This is a documentation-by-test of the trust boundary: the guard trusts
+    // whatever is in the header, which is exactly why every route sits behind
+    // middleware.ts (matcher /api/0pt/:path*) that overwrites it with the
+    // real verified value -- a caller hitting the route directly, outside
+    // Next's routing, cannot reach the handler without going through
+    // middleware first in a real deployment.
+    const res = assertTenantMatchesVerified(req({ [VERIFIED_TENANT_HEADER]: "t_attacker" }), "t_attacker");
+    expect(res).toBeNull(); // matches -- proves the guard is a pure equality check, no extra magic
+  });
+});

tests/lib/db-ssl-mode.test.ts

@@ -0,0 +1,39 @@
+/**
+ * P014: lib/db.ts's `getSql()` used to hardcode `ssl: "require"`
+ * unconditionally. That is correct for every REAL deployment target this
+ * app has (Vercel Postgres, managed Postgres) but made ANY plain,
+ * non-TLS Postgres -- compose.dev.yaml's own local-dev container, or a
+ * throwaway restore-drill target -- silently unreachable through this
+ * function, contradicting compose.dev.yaml's own doc comment and its
+ * "UNVERIFIED... not smoke-tested" caveat. Found and fixed while executing
+ * P014's real restore drill (see docs/runbook/backup-restore.md).
+ */
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { sslMode } from "@/lib/db";
+
+describe("sslMode", () => {
+  const saved = process.env.DATABASE_SSL;
+  afterEach(() => {
+    if (saved === undefined) delete process.env.DATABASE_SSL;
+    else process.env.DATABASE_SSL = saved;
+  });
+  beforeEach(() => {
+    delete process.env.DATABASE_SSL;
+  });
+
+  it("defaults to \"require\" when DATABASE_SSL is unset -- unchanged behavior for every real deployment", () => {
+    expect(sslMode()).toBe("require");
+  });
+
+  for (const value of ["false", "FALSE", "disable", "off", "0", "  false  "]) {
+    it(`treats DATABASE_SSL=${JSON.stringify(value)} as an explicit opt-out (false)`, () => {
+      process.env.DATABASE_SSL = value;
+      expect(sslMode()).toBe(false);
+    });
+  }
+
+  it("an unrecognized value falls back to the safe default (\"require\"), not an accidental opt-out", () => {
+    process.env.DATABASE_SSL = "yes-please";
+    expect(sslMode()).toBe("require");
+  });
+});

tests/lib/orders-idempotency.test.ts

@@ -0,0 +1,84 @@
+/**
+ * P014: lib/orders-store.ts previously had ZERO test coverage and no
+ * idempotency mechanism -- a retried POST /api/orders created a second,
+ * indistinguishable row every time (recon: "Duplicate delivery idempotent:
+ * NOT MET"). These tests exercise the in-memory fallback path directly
+ * (no DATABASE_URL is set in this test environment, matching every other
+ * test in this repo) and prove: a retry under the same Idempotency-Key
+ * replays the original order; a reused key with a different submission is
+ * refused as a conflict, not silently merged; and omitting the key
+ * preserves the exact pre-P014 "always create a new order" behavior.
+ */
+import { describe, it, expect, beforeEach } from "vitest";
+import {
+  createOrder,
+  findOrderByIdempotencyKey,
+  IdempotencyKeyConflictError,
+  type OrderConfig,
+} from "@/lib/orders-store";
+
+const CONFIG_A: OrderConfig = { markets: ["SE"], type: "Bolag", fieldIds: ["identity"], volume: 1000 };
+const CONFIG_B: OrderConfig = { markets: ["NO"], type: "Bolag", fieldIds: ["identity"], volume: 5000 };
+
+describe("createOrder idempotency (in-memory fallback)", () => {
+  it("no Idempotency-Key: every call creates a distinct order (unchanged pre-P014 behavior)", async () => {
+    const first = await createOrder("custom", CONFIG_A);
+    const second = await createOrder("custom", CONFIG_A);
+    expect(first.order.order_id).not.toBe(second.order.order_id);
+    expect(first.replayed).toBe(false);
+    expect(second.replayed).toBe(false);
+  });
+
+  it("same Idempotency-Key + same (flow, config): the second call replays the first order, no duplicate row", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    const first = await createOrder("custom", CONFIG_A, key);
+    const second = await createOrder("custom", CONFIG_A, key);
+
+    expect(first.replayed).toBe(false);
+    expect(second.replayed).toBe(true);
+    expect(second.order.order_id).toBe(first.order.order_id);
+    expect(second.order.created_at).toBe(first.order.created_at);
+  });
+
+  it("retrying the SAME key three times still yields exactly one underlying order", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    const a = await createOrder("custom", CONFIG_A, key);
+    const b = await createOrder("custom", CONFIG_A, key);
+    const c = await createOrder("custom", CONFIG_A, key);
+    const ids = new Set([a.order.order_id, b.order.order_id, c.order.order_id]);
+    expect(ids.size).toBe(1);
+  });
+
+  it("same Idempotency-Key reused with a DIFFERENT config: refused as a conflict, not silently merged", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    await createOrder("custom", CONFIG_A, key);
+    await expect(createOrder("custom", CONFIG_B, key)).rejects.toBeInstanceOf(IdempotencyKeyConflictError);
+  });
+
+  it("same Idempotency-Key reused with a DIFFERENT flow: refused as a conflict", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    await createOrder("custom", CONFIG_A, key);
+    await expect(createOrder("subscription", CONFIG_A, key)).rejects.toBeInstanceOf(IdempotencyKeyConflictError);
+  });
+
+  it("a conflicting retry does not mutate or replace the original order", async () => {
+    const key = `idem_${crypto.randomUUID()}`;
+    const original = await createOrder("custom", CONFIG_A, key);
+    await expect(createOrder("custom", CONFIG_B, key)).rejects.toThrow();
+    const stillThere = await findOrderByIdempotencyKey(key);
+    expect(stillThere?.order_id).toBe(original.order.order_id);
+    expect(stillThere?.config).toEqual(CONFIG_A);
+  });
+
+  it("different Idempotency-Keys for the same config create independent orders", async () => {
+    const a = await createOrder("custom", CONFIG_A, `idem_${crypto.randomUUID()}`);
+    const b = await createOrder("custom", CONFIG_A, `idem_${crypto.randomUUID()}`);
+    expect(a.order.order_id).not.toBe(b.order.order_id);
+  });
+
+  it("an empty-string Idempotency-Key is treated as absent (no dedup, no crash)", async () => {
+    const a = await createOrder("custom", CONFIG_A, "");
+    const b = await createOrder("custom", CONFIG_A, "  ");
+    expect(a.order.order_id).not.toBe(b.order.order_id);
+  });
+});

tests/validation/0pt-schemas.test.ts

@@ -127,6 +127,11 @@ describe('mapDatasetToIcebergTableSchema', () => {
 
 describe('requestIcebergWriteSchema', () => {
   const valid = {
+    // P014: tenant_id is now required -- the route cross-checks it against
+    // the caller's verified tenant and threads it into
+    // IcebergWriteService.requestIcebergWrite so a foreign mapping_id can no
+    // longer be written to (see getMapping's doc comment).
+    tenant_id: 't1',
     actor_id: 'a',
     correlation_id: 'corr',
     schema_version: 'v1',
@@ -139,4 +144,8 @@ describe('requestIcebergWriteSchema', () => {
     const { manifest_uri: _omit, ...bad } = valid;
     expect(parseBody(requestIcebergWriteSchema, bad).ok).toBe(false);
   });
+  it('rejects a write request missing tenant_id', () => {
+    const { tenant_id: _omit, ...bad } = valid;
+    expect(parseBody(requestIcebergWriteSchema, bad).ok).toBe(false);
+  });
 });