src/analytics/ingestion/analytics-ingestion-service.ts

@@ -0,0 +1,143 @@
+/**
+ * src/analytics/ingestion/analytics-ingestion-service.ts
+ * =======================================================
+ * BI-P129: one entry point that takes an `AnalyticsIngestionSpec` and lands
+ * rows in a canonical analytics tier, whichever provider supplies the
+ * connector.
+ *
+ * The sequence, and why it is this sequence:
+ *
+ *   1. plan            — every gate checkable without touching the source
+ *                        (tier, physical home, licence, provider readiness).
+ *                        Cheap checks first: a spec that can never be legal
+ *                        should not open a session to find that out.
+ *   2. connect+discover— ask the source what it actually looks like.
+ *   3. landability     — the discovered shape must satisfy the BI-P004
+ *                        contract. This is the check that cannot be done at
+ *                        step 1, and doing it here is the whole reason the
+ *                        service exists rather than a bare `planX()` call.
+ *   4. run             — hand the connector to the EXISTING P062 runner with
+ *                        a landing sink. The runner owns paging, envelopes,
+ *                        schema conformance, quality gating, the idempotency
+ *                        ledger and the durable checkpoint. None of that is
+ *                        re-implemented here.
+ *   5. disconnect      — always, including on failure.
+ *
+ * Step 4 is worth spelling out because it is where a "new ingestion system"
+ * would normally get built by accident. There is no second runner. The
+ * analytics plane contributes a TARGET and four GATES; the highway is the one
+ * that already exists and already has crash/resume tests.
+ */
+import { SiaxError } from '../../platform/errors';
+import { IngestionRunner, type IngestionRunReport } from '../../ingestion/ingestion-runner';
+import type { IngestionRepository } from '../../repositories/ingestion-repository';
+import type { SchemaRegistry } from '../../schema/schema-registry';
+import type { LineageStore } from '../../lineage/lineage-store';
+import type { QualityEnforcementPolicy } from '../../quality/quality-policy';
+import type { RecordSchemaFields } from '../../schema/record-schema';
+import {
+  planAnalyticsIngestion,
+  type AnalyticsConnectorProvider,
+  type AnalyticsIngestionSpec,
+  type PlannedAnalyticsIngestion,
+} from './connector-provider';
+import { assertDiscoveryLandable, landingSink, type LandingWriter } from './landing';
+
+export interface AnalyticsIngestionDeps {
+  repo: IngestionRepository;
+  schemas: SchemaRegistry;
+  writer: LandingWriter;
+  lineage?: LineageStore;
+  /** Runtime environment name; "production" turns gate 4 on. */
+  runtimeEnv: string;
+}
+
+export interface AnalyticsIngestionResult {
+  plan: PlannedAnalyticsIngestion;
+  /** What discovery reported, per partition — recorded so a later drift is provable. */
+  discovered: Record<string, RecordSchemaFields>;
+  run: IngestionRunReport;
+}
+
+export interface AnalyticsIngestionOptions {
+  run_id?: string;
+  page_size?: number;
+  max_pages?: number;
+  partition_key?: string;
+  quality_policy?: QualityEnforcementPolicy;
+  correlation_id?: string;
+}
+
+export class AnalyticsIngestionService {
+  constructor(private readonly deps: AnalyticsIngestionDeps) {}
+
+  /** Gates 1, 3 and 4 — no IO, no session. Safe to call from a UI pre-flight. */
+  plan(provider: AnalyticsConnectorProvider, spec: AnalyticsIngestionSpec): PlannedAnalyticsIngestion {
+    return planAnalyticsIngestion(provider, spec, this.deps.runtimeEnv);
+  }
+
+  async ingest(
+    provider: AnalyticsConnectorProvider,
+    spec: AnalyticsIngestionSpec,
+    options: AnalyticsIngestionOptions = {},
+  ): Promise<AnalyticsIngestionResult> {
+    const plan = this.plan(provider, spec);
+    const runId = options.run_id ?? `run_${spec.source_id}_${Date.now()}`;
+    const connector = plan.connector;
+
+    const discovered: Record<string, RecordSchemaFields> = {};
+    await connector.connect(spec.config, {
+      tenant_id: spec.tenant_id,
+      project_id: spec.project_id,
+      source_id: spec.source_id,
+      run_id: runId,
+    });
+    try {
+      const discovery = await connector.discover();
+      for (const partition of discovery.partitions) {
+        if (options.partition_key && partition.partition_key !== options.partition_key) continue;
+        discovered[partition.partition_key] = partition.schema;
+      }
+      // Gate 2 — the one that needs the source to have answered.
+      assertDiscoveryLandable(spec.target, discovered);
+    } finally {
+      // The runner opens its own session; this discovery session is ours to close.
+      await connector.disconnect();
+    }
+
+    const runner = new IngestionRunner(this.deps.repo, this.deps.schemas, this.deps.lineage);
+    const run = await runner.run({
+      source_id: spec.source_id,
+      tenant_id: spec.tenant_id,
+      project_id: spec.project_id,
+      mode: spec.mode,
+      connector,
+      config: spec.config,
+      sink: landingSink({
+        target: spec.target,
+        tenant_id: spec.tenant_id,
+        project_id: spec.project_id,
+        run_id: runId,
+        writer: this.deps.writer,
+      }),
+      run_id: runId,
+      page_size: options.page_size,
+      max_pages: options.max_pages,
+      partition_key: options.partition_key,
+      quality_policy: options.quality_policy,
+      output_ref: `0pt:dataset:${spec.target.dataset_id}`,
+      correlation_id: options.correlation_id,
+    });
+
+    if (run.status === 'failed') {
+      throw new SiaxError(
+        'RETRYABLE_FAILURE',
+        '0pt.analytics.ingestion',
+        `ingestion run ${run.run_id} for dataset ${spec.target.dataset_id} failed: ` +
+          `${run.failure?.code ?? 'unknown'} ${run.failure?.detail ?? ''}`.trim(),
+      );
+    }
+
+    return { plan, discovered, run };
+  }
+}

src/analytics/ingestion/connector-provider.ts

@@ -0,0 +1,442 @@
+/**
+ * src/analytics/ingestion/connector-provider.ts
+ * ==============================================
+ * BI-P129 — ANALYTICS DATA INGESTION.
+ *
+ * BI-P129 names four provider kinds for 0pt's `ConnectorProvider`:
+ *
+ *   native            — connectors 0pt implements and owns (P061 contract)
+ *   dlt               — the dlt Python framework (BI-P130)
+ *   Meltano/Singer    — the Singer tap ecosystem (BI-P131)
+ *   Airbyte           — "only optional where legal/license architecture accepts it"
+ *
+ * WHAT THIS FILE IS. It is the seam between 0pt's EXISTING ingestion highway
+ * (`src/connectors/connector.ts` P061 + `src/ingestion/ingestion-runner.ts`
+ * P062) and the ANALYTICS tier/contract model (BI-P003 tiers, BI-P004
+ * `DatasetContract`). It deliberately does NOT define a second connector
+ * interface: a provider's job is to hand back a P061 `Connector`, and the
+ * runner that already exists then does the reading, checkpointing and
+ * quality gating. A parallel ingestion stack would be exactly the "do not
+ * define parallel models" failure this programme keeps paying for.
+ *
+ * WHAT IT ADDS, which the P061 stack has no opinion about: an ingestion run
+ * must LAND somewhere, and where it lands is an analytics contract question.
+ * Four gates, all fail-closed, all attacked in
+ * tests/analytics/analytics-ingestion-providers.test.ts:
+ *
+ *   1. assertLandingTierIngestible  — source data may land in RAW, or in
+ *      NORMALIZED with a declared normalization. It may NEVER land directly
+ *      in CURATED or SERVING. BI-P003's rule ("no business dashboard reads
+ *      arbitrary raw ingestion tables by default") is trivially defeated by
+ *      relabelling the raw landing table SERVING; this gate is what stops
+ *      that, because the tier is checked at the point of WRITE, not only at
+ *      the point of read.
+ *
+ *   2. assertContractCoversDiscovery — every field the connector actually
+ *      discovered must be declared in the target `DatasetContract`, with a
+ *      compatible type. An UNDECLARED field is refused rather than landed:
+ *      an undeclared column is an unclassified column, and an unclassified
+ *      column carrying personal data is a GDPR incident that every freshness
+ *      and quality check would still report green.
+ *
+ *   3. assertPhysicalHomeMatchesTier — a RAW/NORMALIZED landing goes to
+ *      ST0RE Parquet/Iceberg (the shapes st0re#45 standardised), never
+ *      straight into ClickHouse. Writing ingestion output into the serving
+ *      engine skips the lake, and with it every replay, backfill and
+ *      erasure path that depends on the files existing.
+ *
+ *   4. assertProviderRunnable — a provider whose own `providerReadiness()`
+ *      says `verified: false` cannot execute in a production runtime without
+ *      a recorded exception naming a human. This is the structural version
+ *      of the honesty rule `0pt#35` set with MetricFlow: an unverified
+ *      provider may exist, may be planned, may be tested — but it cannot
+ *      quietly become production capability because someone forgot it was
+ *      unverified.
+ *
+ * ARCHITECTURE BOUNDARY (BI programme §0): 0pt owns ingestion into analytics
+ * tiers. A product repo declares what the rows MEAN through
+ * `BIApplicationBinding`; it never runs its own ingestion into these tiers
+ * and never receives ClickHouse credentials.
+ */
+import { SiaxError } from '../../platform/errors';
+import type { Connector, ConnectorConfig, ValidationResult } from '../../connectors/connector';
+import type { RecordSchemaFields, FieldType } from '../../schema/record-schema';
+import type { IngestionMode } from '../../types/ingestion';
+import type { AnalyticsFieldType } from '../domain/entities';
+import type { AnalyticsTier } from '../domain/tiers';
+import type { DatasetContract } from '../domain/contract';
+import type { ProviderReadiness } from '../metrics/semantic-provider';
+import type { N0dRequiredRole } from '../../clients/n0d-client';
+
+/**
+ * Deliberately the SAME `ProviderReadiness` shape BI-P029 introduced, not a
+ * second one. An estate with two readiness vocabularies has none.
+ */
+export type { ProviderReadiness };
+
+// ── provider kinds ─────────────────────────────────────────────────────────
+
+export const CONNECTOR_PROVIDER_KINDS = ['native', 'dlt', 'singer', 'airbyte'] as const;
+export type ConnectorProviderKind = (typeof CONNECTOR_PROVIDER_KINDS)[number];
+
+/**
+ * BI-P129: "Airbyte only optional where legal/license architecture accepts
+ * it." That acceptance has NOT been given in this estate — Airbyte's server
+ * is Elastic License 2.0 (ELv2), which restricts offering the software as a
+ * managed service to third parties, and 0pt's whole purpose is offering
+ * managed data capability to customers. So `airbyte` is a declared kind with
+ * no implementation, and registering one is refused here rather than
+ * somewhere downstream. When a licence review says otherwise, this constant
+ * is the single place that changes.
+ */
+export const PROVIDER_KINDS_REQUIRING_LICENCE_ACCEPTANCE: readonly ConnectorProviderKind[] = ['airbyte'];
+
+// ── landing target ─────────────────────────────────────────────────────────
+
+/** Physical formats an analytics landing may use. st0re#45's shapes. */
+export const LANDING_FORMATS = ['parquet', 'iceberg'] as const;
+export type LandingFormat = (typeof LANDING_FORMATS)[number];
+
+/** Where an ingestion run's output physically goes. */
+export interface LandingPhysical {
+  format: LandingFormat;
+  /** ST0RE lakehouse namespace. */
+  namespace: string;
+  table_name: string;
+}
+
+export interface AnalyticsLandingTarget {
+  dataset_id: string;
+  tier: AnalyticsTier;
+  /** The BI-P004 contract the landed rows must satisfy. */
+  contract: DatasetContract;
+  physical: LandingPhysical;
+  /**
+   * Required when `tier` is NORMALIZED: what normalization was applied
+   * between the source bytes and the landed rows. Free text, recorded — an
+   * empty or missing value means the dataset is really RAW wearing a
+   * NORMALIZED label, and is refused.
+   */
+  normalization?: string;
+}
+
+/** One ingestion the analytics plane is asked to perform. */
+export interface AnalyticsIngestionSpec {
+  source_id: string;
+  tenant_id: string;
+  project_id: string;
+  mode: IngestionMode;
+  provider: ConnectorProviderKind;
+  /** Provider-specific connector selector (a native connector id, a tap name, a dlt source). */
+  connector_ref: string;
+  config: ConnectorConfig;
+  target: AnalyticsLandingTarget;
+  /**
+   * Records the human who accepted running an UNVERIFIED provider in a
+   * production runtime. Absent = an unverified provider is refused.
+   */
+  unverified_provider_exception?: string;
+}
+
+// ── execution plan ─────────────────────────────────────────────────────────
+
+export type ProviderRuntime = 'in_process' | 'n0d';
+
+/**
+ * Where a provider's work actually runs. BI-P130 says dlt executes on N0D;
+ * this is the type that carries that decision instead of a comment.
+ */
+export interface ProviderExecutionPlan {
+  runtime: ProviderRuntime;
+  /** N0D job_type; set iff runtime is "n0d". */
+  job_type?: string;
+  required_role?: N0dRequiredRole;
+  /** Named profile from N0D_RESOURCE_PROFILES; set iff runtime is "n0d". */
+  resource_profile?: string;
+  /** Operator-facing reason, in plain words. */
+  reason: string;
+}
+
+// ── the port ───────────────────────────────────────────────────────────────
+
+export interface AnalyticsConnectorProvider {
+  readonly provider: ConnectorProviderKind;
+  /** What this provider has and has NOT demonstrated. Never optimistic. */
+  providerReadiness(): ProviderReadiness;
+  /** Pure, no IO, never throws — the P061 `validate` posture, at spec level. */
+  validateSpec(spec: AnalyticsIngestionSpec): ValidationResult;
+  /** The P061 connector this spec ingests through. */
+  connectorFor(spec: AnalyticsIngestionSpec): Connector;
+  executionPlan(spec: AnalyticsIngestionSpec): ProviderExecutionPlan;
+}
+
+// ── gate 1: landing tier ───────────────────────────────────────────────────
+
+/** Tiers an ingestion run may write into at all. */
+export const INGESTIBLE_TIERS: readonly AnalyticsTier[] = ['RAW', 'NORMALIZED'];
+
+export function assertLandingTierIngestible(target: AnalyticsLandingTarget): void {
+  if (!INGESTIBLE_TIERS.includes(target.tier)) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `ingestion may not land directly in ${target.tier} (dataset ${target.dataset_id}): ` +
+        `CURATED and SERVING are DERIVED tiers, built from a landed RAW/NORMALIZED dataset. ` +
+        `Landing source data there relabels raw ingestion as dashboard-readable and defeats BI-P003.`,
+    );
+  }
+  if (target.tier === 'NORMALIZED' && !(target.normalization ?? '').trim()) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `dataset ${target.dataset_id} lands in NORMALIZED but declares no normalization — ` +
+        `that is a RAW dataset wearing a NORMALIZED label (BI-P003)`,
+    );
+  }
+}
+
+// ── gate 2: contract covers what was discovered ────────────────────────────
+
+/**
+ * How a connector's wire-level field type maps onto the analytics type
+ * vocabulary. Conservative on purpose: a connector's `number` may be an
+ * integer or a float, so it satisfies EITHER analytics numeric type, but
+ * nothing else. Anything not listed here is a mismatch, not a guess.
+ */
+const CONNECTOR_TYPE_SATISFIES: Record<FieldType, readonly AnalyticsFieldType[]> = {
+  string: ['string', 'decimal', 'date', 'timestamp'],
+  number: ['int64', 'float64', 'decimal'],
+  boolean: ['bool'],
+  timestamp: ['timestamp', 'date'],
+  object: ['json'],
+  array: ['json'],
+};
+
+export interface ContractCoverageProblem {
+  code: 'undeclared_field' | 'type_mismatch' | 'missing_required_field' | 'nullability_mismatch';
+  field: string;
+  detail: string;
+}
+
+/**
+ * Compare what the connector says the source LOOKS LIKE against what the
+ * contract says the dataset IS. Returns every problem; never throws.
+ */
+export function contractCoverageProblems(
+  contract: DatasetContract,
+  discovered: RecordSchemaFields,
+): ContractCoverageProblem[] {
+  const problems: ContractCoverageProblem[] = [];
+  const byName = new Map(contract.fields.map((f) => [f.name, f]));
+
+  for (const [name, spec] of Object.entries(discovered)) {
+    const field = byName.get(name);
+    if (!field) {
+      problems.push({
+        code: 'undeclared_field',
+        field: name,
+        detail:
+          `source field "${name}" is not declared in contract ${contract.dataset_id}@${contract.version}; ` +
+          `an undeclared field is an unclassified field and is refused, not landed`,
+      });
+      continue;
+    }
+    const satisfies = CONNECTOR_TYPE_SATISFIES[spec.type] ?? [];
+    if (!satisfies.includes(field.type)) {
+      problems.push({
+        code: 'type_mismatch',
+        field: name,
+        detail: `source field "${name}" is ${spec.type}; contract declares ${field.type}`,
+      });
+    }
+    // `FieldSpec.required` is the connector-side vocabulary; the contract
+    // speaks `nullable`. required:false === nullable:true.
+    if (spec.required === false && field.nullable === false) {
+      problems.push({
+        code: 'nullability_mismatch',
+        field: name,
+        detail: `source field "${name}" is optional at the source; contract declares it NOT NULL`,
+      });
+    }
+  }
+
+  for (const field of contract.fields) {
+    if (field.nullable === false && !(field.name in discovered)) {
+      problems.push({
+        code: 'missing_required_field',
+        field: field.name,
+        detail: `contract requires non-null field "${field.name}" but the source does not produce it`,
+      });
+    }
+  }
+  return problems;
+}
+
+export function assertContractCoversDiscovery(
+  contract: DatasetContract,
+  discovered: RecordSchemaFields,
+): void {
+  const problems = contractCoverageProblems(contract, discovered);
+  if (problems.length > 0) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `source shape does not satisfy dataset contract ${contract.dataset_id}@${contract.version}: ` +
+        problems.map((p) => `[${p.code}] ${p.detail}`).join('; '),
+    );
+  }
+}
+
+// ── gate 3: physical home ──────────────────────────────────────────────────
+
+export function assertPhysicalHomeMatchesTier(target: AnalyticsLandingTarget): void {
+  if (!(LANDING_FORMATS as readonly string[]).includes(target.physical.format)) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `landing format "${target.physical.format}" is not a lakehouse format; ` +
+        `RAW/NORMALIZED land as Parquet or Iceberg in ST0RE (BI-P003, st0re#45)`,
+    );
+  }
+  if (!target.physical.namespace.trim() || !target.physical.table_name.trim()) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `landing target for ${target.dataset_id} is missing a ST0RE namespace/table name`,
+    );
+  }
+}
+
+// ── gate 4: unverified providers may not run in production ─────────────────
+
+export function assertProviderRunnable(
+  provider: AnalyticsConnectorProvider,
+  spec: AnalyticsIngestionSpec,
+  runtimeEnv: string,
+): void {
+  const readiness = provider.providerReadiness();
+  if (readiness.verified) return;
+  const isProduction = runtimeEnv === 'production';
+  const exception = (spec.unverified_provider_exception ?? '').trim();
+  if (!isProduction) return;
+  if (exception.length === 0) {
+    throw new SiaxError(
+      'CAPABILITY_UNAVAILABLE',
+      '0pt.analytics.ingestion',
+      `provider "${provider.provider}" reports verified: false ` +
+        `(open: ${readiness.open.join('; ')}) and may not run in production ` +
+        `without a recorded unverified_provider_exception`,
+    );
+  }
+}
+
+// ── planning ───────────────────────────────────────────────────────────────
+
+export interface PlannedAnalyticsIngestion {
+  spec: AnalyticsIngestionSpec;
+  connector: Connector;
+  execution: ProviderExecutionPlan;
+  readiness: ProviderReadiness;
+  /** The tier the rows land in — echoed so a caller cannot claim a different one later. */
+  landing_tier: AnalyticsTier;
+}
+
+/**
+ * Run every gate that can be checked WITHOUT touching the source, then hand
+ * back the connector. `assertContractCoversDiscovery` is deliberately NOT
+ * here: discovery requires a live session, so it runs in `landingSink`'s
+ * caller / `AnalyticsIngestionService` once `discover()` has answered.
+ */
+export function planAnalyticsIngestion(
+  provider: AnalyticsConnectorProvider,
+  spec: AnalyticsIngestionSpec,
+  runtimeEnv: string,
+): PlannedAnalyticsIngestion {
+  if (provider.provider !== spec.provider) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `spec asks for provider "${spec.provider}" but was given "${provider.provider}"`,
+    );
+  }
+  if (PROVIDER_KINDS_REQUIRING_LICENCE_ACCEPTANCE.includes(spec.provider)) {
+    throw new SiaxError(
+      'CAPABILITY_UNAVAILABLE',
+      '0pt.analytics.ingestion',
+      `provider "${spec.provider}" requires a licence/architecture acceptance this estate has not given ` +
+        `(BI-P129: "Airbyte only optional where legal/license architecture accepts it")`,
+    );
+  }
+  const verdict = provider.validateSpec(spec);
+  if (!verdict.valid) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `provider "${provider.provider}" rejected the spec: ${verdict.errors.join('; ')}`,
+    );
+  }
+  assertLandingTierIngestible(spec.target);
+  assertPhysicalHomeMatchesTier(spec.target);
+  assertProviderRunnable(provider, spec, runtimeEnv);
+
+  return {
+    spec,
+    connector: provider.connectorFor(spec),
+    execution: provider.executionPlan(spec),
+    readiness: provider.providerReadiness(),
+    landing_tier: spec.target.tier,
+  };
+}
+
+// ── registry ───────────────────────────────────────────────────────────────
+
+/**
+ * The set of providers a runtime actually has. Registration is explicit:
+ * there is no discovery-by-filename, because a provider appearing in the
+ * registry because a file exists is how an unreviewed integration becomes
+ * live.
+ */
+export class ConnectorProviderRegistry {
+  private readonly providers = new Map<ConnectorProviderKind, AnalyticsConnectorProvider>();
+
+  register(provider: AnalyticsConnectorProvider): this {
+    if (this.providers.has(provider.provider)) {
+      throw new SiaxError(
+        'CONFLICT',
+        '0pt.analytics.ingestion',
+        `provider "${provider.provider}" is already registered`,
+      );
+    }
+    if (PROVIDER_KINDS_REQUIRING_LICENCE_ACCEPTANCE.includes(provider.provider)) {
+      throw new SiaxError(
+        'CAPABILITY_UNAVAILABLE',
+        '0pt.analytics.ingestion',
+        `provider "${provider.provider}" may not be registered: no licence/architecture acceptance exists ` +
+          `for it in this estate (BI-P129)`,
+      );
+    }
+    this.providers.set(provider.provider, provider);
+    return this;
+  }
+
+  get(kind: ConnectorProviderKind): AnalyticsConnectorProvider {
+    const p = this.providers.get(kind);
+    if (!p) {
+      throw new SiaxError(
+        'NOT_FOUND',
+        '0pt.analytics.ingestion',
+        `no connector provider registered for kind "${kind}"`,
+      );
+    }
+    return p;
+  }
+
+  has(kind: ConnectorProviderKind): boolean {
+    return this.providers.has(kind);
+  }
+
+  /** Readiness of every registered provider — the operator's one-screen answer. */
+  readinessReport(): ProviderReadiness[] {
+    return [...this.providers.values()].map((p) => p.providerReadiness());
+  }
+}

src/analytics/ingestion/landing.ts

@@ -0,0 +1,160 @@
+/**
+ * src/analytics/ingestion/landing.ts
+ * ===================================
+ * BI-P129: the write side. An ingestion run's envelopes have to become FILES
+ * in a canonical tier, and this is the only place in the analytics plane that
+ * decides what those files look like.
+ *
+ * WHERE THE BOUNDARY IS. 0pt does not own object storage. ST0RE owns the
+ * lakehouse (namespaces, Iceberg tables, snapshots — `st0re#45`), and 0pt's
+ * existing `src/services/iceberg-write-service.ts` is already the client for
+ * it. So `LandingWriter` is a PORT, not an implementation: it names exactly
+ * what the analytics plane needs from a lake writer (append a batch to one
+ * tier-owned table, tell me what landed) and nothing more. The default
+ * implementation here is in-memory and is used by tests and by dry runs; the
+ * ST0RE-backed one is the existing service, injected.
+ *
+ * WHAT THIS FILE ENFORCES, and why each rule is not decoration:
+ *
+ *   - Every landed batch carries `tier` and `dataset_id`. A batch of rows
+ *     with no tier is a batch nobody can decide the read rules for later.
+ *
+ *   - `assertContractCoversDiscovery` (BI-P129 gate 2) runs HERE, at the
+ *     point discovery has actually answered, not at plan time when the
+ *     source has not been asked yet. A gate that runs before the fact it
+ *     checks is available is a gate that checks nothing.
+ *
+ *   - Deletes are LANDED, not dropped. `IngestionEnvelope` with
+ *     `operation: "delete"` carries `payload: null`; a landing that silently
+ *     skipped it would produce exactly the failure this programme has already
+ *     paid for once — an aggregate that never sees a DELETE and overstates a
+ *     number while every freshness check reports green. Staleness is not
+ *     correctness, and a delete that never lands is not staleness at all,
+ *     it is permanent wrongness.
+ *
+ *   - The batch records `envelope_id`s, so a replay is detectable downstream.
+ *     The runner's guarantee is at-least-once (documented in
+ *     src/ingestion/ingestion-runner.ts); a landing that pretended otherwise
+ *     would be lying about the pipeline it sits in.
+ */
+import { SiaxError } from '../../platform/errors';
+import type { IngestionEnvelope } from '../../types/ingestion';
+import type { RecordSchemaFields } from '../../schema/record-schema';
+import type { AnalyticsTier } from '../domain/tiers';
+import {
+  assertContractCoversDiscovery,
+  type AnalyticsLandingTarget,
+  type LandingFormat,
+} from './connector-provider';
+
+/** One physical batch appended to a tier-owned table. */
+export interface LandedBatch {
+  dataset_id: string;
+  tier: AnalyticsTier;
+  format: LandingFormat;
+  namespace: string;
+  table_name: string;
+  tenant_id: string;
+  project_id: string;
+  run_id: string;
+  partition_key: string;
+  /** Rows written, including tombstones. */
+  row_count: number;
+  /** Rows whose payload was null (CDC deletes). Counted separately, never dropped. */
+  tombstone_count: number;
+  envelope_ids: string[];
+}
+
+/**
+ * The port. One method: append a batch. Deliberately narrow — every extra
+ * method here is a piece of ST0RE's contract leaking into 0pt's.
+ */
+export interface LandingWriter {
+  append(batch: LandedBatch): Promise<void>;
+}
+
+/** Test/dry-run writer. Keeps everything it was given so a test can assert on it. */
+export class InMemoryLandingWriter implements LandingWriter {
+  readonly batches: LandedBatch[] = [];
+  async append(batch: LandedBatch): Promise<void> {
+    this.batches.push(batch);
+  }
+  get rowsLanded(): number {
+    return this.batches.reduce((n, b) => n + b.row_count, 0);
+  }
+  get tombstonesLanded(): number {
+    return this.batches.reduce((n, b) => n + b.tombstone_count, 0);
+  }
+}
+
+export interface LandingSinkInput {
+  target: AnalyticsLandingTarget;
+  tenant_id: string;
+  project_id: string;
+  run_id: string;
+  writer: LandingWriter;
+}
+
+/**
+ * Build an `IngestionSink` (the runner's own type) that lands envelopes into
+ * the declared analytics tier.
+ *
+ * Note the sink is per-run: the runner commits its checkpoint only AFTER
+ * this resolves, so a writer that throws leaves the page unconsumed and the
+ * next run re-reads it. That ordering belongs to the runner and is not
+ * re-implemented here — it is relied upon.
+ */
+export function landingSink(input: LandingSinkInput): (envelopes: IngestionEnvelope[]) => Promise<void> {
+  return async (envelopes: IngestionEnvelope[]): Promise<void> => {
+    if (envelopes.length === 0) return;
+    const partitions = new Set(envelopes.map((e) => e.partition_key));
+    for (const partition of partitions) {
+      const forPartition = envelopes.filter((e) => e.partition_key === partition);
+      await input.writer.append({
+        dataset_id: input.target.dataset_id,
+        tier: input.target.tier,
+        format: input.target.physical.format,
+        namespace: input.target.physical.namespace,
+        table_name: input.target.physical.table_name,
+        tenant_id: input.tenant_id,
+        project_id: input.project_id,
+        run_id: input.run_id,
+        partition_key: partition,
+        row_count: forPartition.length,
+        tombstone_count: forPartition.filter((e) => e.payload === null).length,
+        envelope_ids: forPartition.map((e) => e.envelope_id),
+      });
+    }
+  };
+}
+
+/**
+ * BI-P129 gate 2, applied once the connector's `discover()` has answered.
+ * Called by `AnalyticsIngestionService`; exported so an operator pre-flight
+ * can run it without starting a run.
+ */
+export function assertDiscoveryLandable(
+  target: AnalyticsLandingTarget,
+  discoveredByPartition: Record<string, RecordSchemaFields>,
+): void {
+  const partitions = Object.keys(discoveredByPartition);
+  if (partitions.length === 0) {
+    throw new SiaxError(
+      'INVALID_REQUEST',
+      '0pt.analytics.ingestion',
+      `source for dataset ${target.dataset_id} discovered no partitions; there is nothing to land`,
+    );
+  }
+  for (const partition of partitions) {
+    try {
+      assertContractCoversDiscovery(target.contract, discoveredByPartition[partition]);
+    } catch (err) {
+      const detail = err instanceof Error ? err.message : String(err);
+      throw new SiaxError(
+        'INVALID_REQUEST',
+        '0pt.analytics.ingestion',
+        `partition "${partition}": ${detail}`,
+      );
+    }
+  }
+}

src/analytics/ingestion/native-provider.ts

@@ -0,0 +1,97 @@
+/**
+ * src/analytics/ingestion/native-provider.ts
+ * ===========================================
+ * BI-P129, provider kind `native`: connectors 0pt implements and owns.
+ *
+ * There is nothing to integrate here — the connectors already exist and are
+ * already conformance-tested (`src/connectors/conformance.ts`, P061). This
+ * provider's whole job is to select one and report, honestly, that it is
+ * verified: these connectors run in this repo's CI, in-process, with no
+ * third-party runtime and no unreviewed dependency.
+ *
+ * The two shipped natives are deliberately heterogeneous (that was P061's
+ * point and it is BI-P129's DoD too — "two real source types"):
+ *
+ *   http_json   — remote HTTP API, opaque server-issued page token, health
+ *                 is a network round trip, record count unknown until the
+ *                 server says it is done.
+ *   jsonl_file  — local file drop, cursor is a line offset the connector
+ *                 computes itself, health is a stat(), record count exact,
+ *                 and it emits real CDC deletes.
+ *
+ * They share no code but the contract, so "both land in a canonical tier"
+ * is a statement about the SEAM, not about one connector.
+ */
+import { HttpJsonConnector, type FetchImpl } from '../../connectors/http-json-connector';
+import { JsonlFileConnector } from '../../connectors/jsonl-file-connector';
+import type { Connector, ValidationResult } from '../../connectors/connector';
+import type {
+  AnalyticsConnectorProvider,
+  AnalyticsIngestionSpec,
+  ProviderExecutionPlan,
+  ProviderReadiness,
+} from './connector-provider';
+
+/** Native connector ids this provider can build. */
+export const NATIVE_CONNECTOR_IDS = ['http_json', 'jsonl_file'] as const;
+export type NativeConnectorId = (typeof NATIVE_CONNECTOR_IDS)[number];
+
+export class NativeConnectorProvider implements AnalyticsConnectorProvider {
+  readonly provider = 'native' as const;
+
+  /**
+   * Optional injection point so a caller can supply an `HttpJsonConnector`
+   * with a specific `fetchImpl` (the connector's own documented test seam).
+   * With nothing supplied, a real connector talking to the configured URL is
+   * built — this is not a mock fallback.
+   */
+  constructor(private readonly fetchImpl?: FetchImpl) {}
+
+  providerReadiness(): ProviderReadiness {
+    return {
+      provider: this.provider,
+      pinned_version: 'in-repo',
+      verified: true,
+      proven: [
+        'both native connectors pass the P061 conformance suite (tests/p0/connector-conformance.test.ts)',
+        'both land into a canonical analytics tier through the P062 ingestion runner ' +
+          '(tests/analytics/analytics-ingestion-providers.test.ts)',
+        'no third-party runtime and no unreviewed dependency is involved',
+      ],
+      open: [
+        'only http_json and jsonl_file exist natively; every other source type needs dlt/Singer or a new native connector',
+      ],
+    };
+  }
+
+  validateSpec(spec: AnalyticsIngestionSpec): ValidationResult {
+    const errors: string[] = [];
+    if (!(NATIVE_CONNECTOR_IDS as readonly string[]).includes(spec.connector_ref)) {
+      errors.push(
+        `unknown native connector "${spec.connector_ref}" (have: ${NATIVE_CONNECTOR_IDS.join(', ')})`,
+      );
+      return { valid: false, errors };
+    }
+    // Delegate the config check to the connector itself — it is the authority
+    // on its own config, and its `validate` is pure and never throws.
+    const connector = this.connectorFor(spec);
+    const verdict = connector.validate(spec.config);
+    errors.push(...verdict.errors);
+    return { valid: errors.length === 0, errors };
+  }
+
+  connectorFor(spec: AnalyticsIngestionSpec): Connector {
+    if (spec.connector_ref === 'http_json') {
+      return this.fetchImpl ? new HttpJsonConnector(this.fetchImpl) : new HttpJsonConnector();
+    }
+    if (spec.connector_ref === 'jsonl_file') return new JsonlFileConnector();
+    throw new Error(`unknown native connector "${spec.connector_ref}"`);
+  }
+
+  executionPlan(_spec: AnalyticsIngestionSpec): ProviderExecutionPlan {
+    return {
+      runtime: 'in_process',
+      reason: 'native connectors are Node code in this repo; they need no separate runtime and no N0D job',
+    };
+  }
+}

src/connectors/http-json-connector.ts

@@ -66,7 +66,7 @@ export interface HttpJsonConnectorConfig extends ConnectorConfig {
   timeout_ms?: number;
 }
 
-type FetchImpl = (url: string, init: RequestInit) => Promise<Response>;
+export type FetchImpl = (url: string, init: RequestInit) => Promise<Response>;
 
 /** Wire shape this connector expects from the upstream endpoint. */
 interface UpstreamPage {

tests/analytics/analytics-ingestion-providers.test.ts

@@ -0,0 +1,521 @@
+/**
+ * BI-P129 — ANALYTICS DATA INGESTION.
+ *
+ * DoD: "Two real source types ingest into canonical analytics tiers."
+ *
+ * The two source types here are the REAL P061 connectors, sharing no code but
+ * the contract: a JSONL file on actual disk, and an HTTP JSON API served by a
+ * real `fetch` implementation over a real request/response cycle. Both go
+ * through the REAL P062 ingestion runner and the REAL landing sink into a
+ * declared tier. Nothing about the tier or the contract is stubbed.
+ *
+ * Every gate is attacked TWICE: once with the guard in place (it must
+ * refuse), and once in a counterfactual that removes only the guard (it must
+ * succeed). A guard whose counterfactual also fails is not a guard — the
+ * behaviour was coming from somewhere else, and the estate has shipped three
+ * of those this month.
+ */
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+import { promises as fs } from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { InMemoryIngestionRepository } from '@/src/repositories/in-memory-ingestion-repository';
+import { SchemaRegistry } from '@/src/schema/schema-registry';
+import { LineageStore } from '@/src/lineage/lineage-store';
+import { SiaxError } from '@/src/platform/errors';
+import type { RecordSchemaFields } from '@/src/schema/record-schema';
+import {
+  ConnectorProviderRegistry,
+  assertContractCoversDiscovery,
+  assertLandingTierIngestible,
+  assertPhysicalHomeMatchesTier,
+  assertProviderRunnable,
+  contractCoverageProblems,
+  planAnalyticsIngestion,
+  type AnalyticsConnectorProvider,
+  type AnalyticsIngestionSpec,
+  type AnalyticsLandingTarget,
+} from '@/src/analytics/ingestion/connector-provider';
+import { NativeConnectorProvider } from '@/src/analytics/ingestion/native-provider';
+import { InMemoryLandingWriter, assertDiscoveryLandable } from '@/src/analytics/ingestion/landing';
+import { AnalyticsIngestionService } from '@/src/analytics/ingestion/analytics-ingestion-service';
+import type { DatasetContract } from '@/src/analytics/domain/contract';
+
+// ── fixtures ───────────────────────────────────────────────────────────────
+
+const SOURCE_SCHEMA: RecordSchemaFields = {
+  org_number: { type: 'string', required: true },
+  name: { type: 'string', required: true },
+  employees: { type: 'number', required: false },
+};
+
+const CONTRACT: DatasetContract = {
+  dataset_id: 'ds_companies_raw',
+  version: '1.0.0',
+  tier: 'RAW',
+  fields: [
+    { name: 'org_number', type: 'string', nullable: false, role: 'identifier', primary_identifier: true },
+    { name: 'name', type: 'string', nullable: false, role: 'dimension' },
+    { name: 'employees', type: 'int64', nullable: true, role: 'measure' },
+  ],
+  primary_identifiers: ['org_number'],
+  dimensions: ['name'],
+  measures: ['employees'],
+  freshness_slo_seconds: 86_400,
+  retention_days: 3650,
+  pii_classification: 'none',
+  source_lineage: ['bolagsverket'],
+  quality_rules: [],
+  allowed_purposes: ['analytics'],
+};
+
+function target(overrides: Partial<AnalyticsLandingTarget> = {}): AnalyticsLandingTarget {
+  return {
+    dataset_id: 'ds_companies_raw',
+    tier: 'RAW',
+    contract: CONTRACT,
+    physical: { format: 'parquet', namespace: 'analytics_raw', table_name: 'companies' },
+    ...overrides,
+  };
+}
+
+const ROWS = Array.from({ length: 9 }, (_, i) => ({
+  org_number: `55600${i}-000${i}`,
+  name: `Bolag ${i} AB`,
+  employees: (i + 1) * 10,
+}));
+
+const tmpDir = path.join(os.tmpdir(), `opt-bi129-${process.pid}`);
+const filePath = path.join(tmpDir, 'companies.jsonl');
+
+beforeAll(async () => {
+  await fs.mkdir(tmpDir, { recursive: true });
+  await fs.writeFile(filePath, ROWS.map((r) => JSON.stringify(r)).join('\n') + '\n', 'utf8');
+});
+afterAll(async () => {
+  await fs.rm(tmpDir, { recursive: true, force: true }).catch(() => undefined);
+});
+
+/**
+ * A real HTTP JSON source: an actual fetch implementation returning real
+ * `Response` objects, paging by an opaque server token. The connector under
+ * test does a genuine request/response cycle against it — the seam is the
+ * transport, not the connector.
+ */
+function httpSourceFetch(): (url: string, init: RequestInit) => Promise<Response> {
+  return async (url: string) => {
+    const parsed = new URL(url);
+    if (parsed.pathname.endsWith('/health')) {
+      return new Response(JSON.stringify({ ok: true }), { status: 200 });
+    }
+    const cursor = parsed.searchParams.get('cursor');
+    const limit = Number(parsed.searchParams.get('limit') ?? '100');
+    const offset = cursor ? Number(cursor) : 0;
+    const page = ROWS.slice(offset, offset + limit);
+    const nextOffset = offset + page.length;
+    return new Response(
+      JSON.stringify({
+        records: page,
+        next_cursor: nextOffset < ROWS.length ? String(nextOffset) : undefined,
+      }),
+      { status: 200, headers: { 'content-type': 'application/json' } },
+    );
+  };
+}
+
+function jsonlSpec(overrides: Partial<AnalyticsIngestionSpec> = {}): AnalyticsIngestionSpec {
+  return {
+    source_id: 'src_jsonl',
+    tenant_id: 't_test',
+    project_id: 'p_test',
+    mode: 'batch',
+    provider: 'native',
+    connector_ref: 'jsonl_file',
+    config: {
+      file_path: filePath,
+      partition_key: 'companies',
+      schema: SOURCE_SCHEMA,
+      record_key_field: 'org_number',
+    },
+    target: target(),
+    ...overrides,
+  };
+}
+
+function httpSpec(overrides: Partial<AnalyticsIngestionSpec> = {}): AnalyticsIngestionSpec {
+  return {
+    source_id: 'src_http',
+    tenant_id: 't_test',
+    project_id: 'p_test',
+    mode: 'batch',
+    provider: 'native',
+    connector_ref: 'http_json',
+    config: {
+      base_url: 'http://source.invalid',
+      records_path: '/v1/companies',
+      partition_key: 'companies',
+      schema: SOURCE_SCHEMA,
+      record_key_field: 'org_number',
+      record_cursor_field: 'org_number',
+      health_path: '/health',
+    },
+    target: target({ dataset_id: 'ds_companies_raw' }),
+    ...overrides,
+  };
+}
+
+let repo: InMemoryIngestionRepository;
+let writer: InMemoryLandingWriter;
+let service: AnalyticsIngestionService;
+
+function buildService(runtimeEnv = 'test'): AnalyticsIngestionService {
+  repo = new InMemoryIngestionRepository();
+  writer = new InMemoryLandingWriter();
+  return new AnalyticsIngestionService({
+    repo,
+    schemas: new SchemaRegistry(repo),
+    lineage: new LineageStore(repo),
+    writer,
+    runtimeEnv,
+  });
+}
+
+beforeEach(() => {
+  service = buildService();
+});
+
+// ── DoD: two real source types land in a canonical tier ────────────────────
+
+describe('BI-P129 DoD — two real source types ingest into canonical analytics tiers', () => {
+  it('source type 1 (JSONL file on disk) lands every row in the declared RAW tier', async () => {
+    const result = await service.ingest(new NativeConnectorProvider(), jsonlSpec(), { page_size: 4 });
+
+    expect(result.run.status).toBe('succeeded');
+    expect(writer.rowsLanded).toBe(ROWS.length);
+    expect(writer.batches.every((b) => b.tier === 'RAW')).toBe(true);
+    expect(writer.batches.every((b) => b.dataset_id === 'ds_companies_raw')).toBe(true);
+    expect(writer.batches.every((b) => b.format === 'parquet')).toBe(true);
+    expect(writer.batches.every((b) => b.namespace === 'analytics_raw')).toBe(true);
+    // Paged: 4 + 4 + 1, so the tier landing is exercised across batches.
+    expect(writer.batches.length).toBeGreaterThan(1);
+  });
+
+  it('source type 2 (HTTP JSON API, opaque page token) lands every row in the same tier', async () => {
+    const provider = new NativeConnectorProvider(httpSourceFetch());
+    const result = await service.ingest(provider, httpSpec(), { page_size: 4 });
+
+    expect(result.run.status).toBe('succeeded');
+    expect(writer.rowsLanded).toBe(ROWS.length);
+    expect(writer.batches.every((b) => b.tier === 'RAW')).toBe(true);
+    expect(new Set(writer.batches.flatMap((b) => b.envelope_ids)).size).toBe(ROWS.length);
+  });
+
+  it('the two source types share no connector code — different ids, versions and cursor vocabulary', async () => {
+    const nativeProvider = new NativeConnectorProvider(httpSourceFetch());
+    const jsonl = nativeProvider.connectorFor(jsonlSpec());
+    const http = nativeProvider.connectorFor(httpSpec());
+    expect(jsonl.connector_id).toBe('jsonl_file');
+    expect(http.connector_id).toBe('http_json');
+    expect(jsonl.connector_version).not.toBe(http.connector_version);
+    expect(Object.getPrototypeOf(jsonl)).not.toBe(Object.getPrototypeOf(http));
+  });
+});
+
+// ── gate 1: landing tier ───────────────────────────────────────────────────
+
+describe('BI-P129 gate 1 — ingestion may not land directly in a derived tier', () => {
+  it('WITH the guard: landing in SERVING is refused', () => {
+    expect(() => assertLandingTierIngestible(target({ tier: 'SERVING' }))).toThrow(
+      /may not land directly in SERVING/,
+    );
+    expect(() => assertLandingTierIngestible(target({ tier: 'CURATED' }))).toThrow(
+      /may not land directly in CURATED/,
+    );
+  });
+
+  it('WITHOUT the guard: the same spec lands source rows into a dashboard-readable tier', async () => {
+    // Counterfactual — the service is bypassed and the sink is called directly,
+    // which is exactly what a provider that skipped the gate would do.
+    const { landingSink } = await import('@/src/analytics/ingestion/landing');
+    const bypassWriter = new InMemoryLandingWriter();
+    const sink = landingSink({
+      target: target({ tier: 'SERVING' }),
+      tenant_id: 't_test',
+      project_id: 'p_test',
+      run_id: 'run_bypass',
+      writer: bypassWriter,
+    });
+    await sink([
+      {
+        envelope_id: 'e1',
+        source_id: 'src_jsonl',
+        tenant_id: 't_test',
+        project_id: 'p_test',
+        mode: 'batch',
+        operation: 'upsert',
+        partition_key: 'companies',
+        record_key: '1',
+        schema_version: 1,
+        sequence: 1,
+        ingested_at: new Date().toISOString(),
+        payload: { org_number: '1', name: 'x' },
+        provenance: { connector_id: 'jsonl_file', connector_version: 'v1', run_id: 'run_bypass' },
+      },
+    ]);
+    // Raw source rows now sit in SERVING, which BI-P003 makes readable by
+    // every dashboard with no exception recorded anywhere. That is the harm.
+    expect(bypassWriter.batches[0].tier).toBe('SERVING');
+    expect(bypassWriter.rowsLanded).toBe(1);
+  });
+
+  it('NORMALIZED without a declared normalization is a RAW dataset wearing a label', () => {
+    expect(() => assertLandingTierIngestible(target({ tier: 'NORMALIZED' }))).toThrow(
+      /declares no normalization/,
+    );
+    expect(() =>
+      assertLandingTierIngestible(
+        target({ tier: 'NORMALIZED', normalization: 'typed + deduped on org_number' }),
+      ),
+    ).not.toThrow();
+    // An all-whitespace declaration is not a declaration.
+    expect(() => assertLandingTierIngestible(target({ tier: 'NORMALIZED', normalization: '   ' }))).toThrow(
+      /declares no normalization/,
+    );
+  });
+
+  it('the service refuses a SERVING spec before it opens a session to the source', async () => {
+    await expect(
+      service.ingest(new NativeConnectorProvider(), jsonlSpec({ target: target({ tier: 'SERVING' }) })),
+    ).rejects.toThrow(/may not land directly in SERVING/);
+    expect(writer.batches).toHaveLength(0);
+  });
+});
+
+// ── gate 2: contract covers what was discovered ────────────────────────────
+
+describe('BI-P129 gate 2 — an undeclared source field is refused, not landed', () => {
+  it('WITH the guard: a new undeclared column stops the run', () => {
+    const drifted: RecordSchemaFields = {
+      ...SOURCE_SCHEMA,
+      contact_email: { type: 'string', required: false },
+    };
+    expect(() => assertContractCoversDiscovery(CONTRACT, drifted)).toThrow(/undeclared_field/);
+    expect(() => assertContractCoversDiscovery(CONTRACT, drifted)).toThrow(/contact_email/);
+  });
+
+  it('the harm being prevented is named: an undeclared column is an unclassified column', () => {
+    const problems = contractCoverageProblems(CONTRACT, {
+      ...SOURCE_SCHEMA,
+      personnummer: { type: 'string', required: false },
+    });
+    expect(problems).toHaveLength(1);
+    expect(problems[0].code).toBe('undeclared_field');
+    expect(problems[0].detail).toMatch(/unclassified/);
+  });
+
+  it('WITHOUT the guard: the drifted source lands and the PII column is invisible to the contract', async () => {
+    const driftedFile = path.join(tmpDir, 'drifted.jsonl');
+    await fs.writeFile(
+      driftedFile,
+      ROWS.map((r) => JSON.stringify({ ...r, personnummer: '19800101-1234' })).join('\n') + '\n',
+      'utf8',
+    );
+    const driftedSchema: RecordSchemaFields = {
+      ...SOURCE_SCHEMA,
+      personnummer: { type: 'string', required: false },
+    };
+    // Counterfactual: the runner alone (no analytics gate) happily lands it.
+    const bypass = buildService();
+    const bypassWriter = writer;
+    const { IngestionRunner } = await import('@/src/ingestion/ingestion-runner');
+    const { JsonlFileConnector } = await import('@/src/connectors/jsonl-file-connector');
+    const { landingSink } = await import('@/src/analytics/ingestion/landing');
+    const runner = new IngestionRunner(repo, new SchemaRegistry(repo));
+    const landed: Array<Record<string, unknown> | null> = [];
+    const report = await runner.run({
+      source_id: 'src_drift',
+      tenant_id: 't_test',
+      project_id: 'p_test',
+      mode: 'batch',
+      connector: new JsonlFileConnector(),
+      config: {
+        file_path: driftedFile,
+        partition_key: 'companies',
+        schema: driftedSchema,
+        record_key_field: 'org_number',
+      },
+      sink: async (envelopes) => {
+        for (const e of envelopes) landed.push(e.payload);
+        await landingSink({
+          target: target(),
+          tenant_id: 't_test',
+          project_id: 'p_test',
+          run_id: 'run_drift',
+          writer: bypassWriter,
+        })(envelopes);
+      },
+    });
+    expect(report.status).toBe('succeeded');
+    expect(landed).toHaveLength(ROWS.length);
+    // The personal number is now sitting in a RAW table whose contract says
+    // pii_classification: "none". Nothing else in the pipeline notices.
+    expect(landed[0]).toHaveProperty('personnummer');
+    expect(CONTRACT.pii_classification).toBe('none');
+    void bypass;
+  });
+
+  it('a type mismatch and a missing required field are both refused', () => {
+    expect(() =>
+      assertContractCoversDiscovery(CONTRACT, {
+        ...SOURCE_SCHEMA,
+        employees: { type: 'string', required: false },
+      }),
+    ).toThrow(/type_mismatch/);
+    const { org_number: _dropped, ...withoutKey } = SOURCE_SCHEMA;
+    expect(() => assertContractCoversDiscovery(CONTRACT, withoutKey)).toThrow(/missing_required_field/);
+  });
+
+  it('a source field that is optional where the contract says NOT NULL is refused', () => {
+    const problems = contractCoverageProblems(CONTRACT, {
+      ...SOURCE_SCHEMA,
+      name: { type: 'string', required: false },
+    });
+    expect(problems.map((p) => p.code)).toContain('nullability_mismatch');
+  });
+
+  it('discovery with no partitions is refused rather than reported as a clean empty run', () => {
+    expect(() => assertDiscoveryLandable(target(), {})).toThrow(/discovered no partitions/);
+  });
+
+  it('the live path runs gate 2 against what the connector actually discovered', async () => {
+    // The contract is fine for the file; narrow the CONTRACT instead so the
+    // real discovery result is what fails the gate.
+    const narrow: DatasetContract = { ...CONTRACT, fields: CONTRACT.fields.slice(0, 2) };
+    await expect(
+      service.ingest(new NativeConnectorProvider(), jsonlSpec({ target: target({ contract: narrow }) })),
+    ).rejects.toThrow(/undeclared_field/);
+    expect(writer.batches).toHaveLength(0);
+  });
+});
+
+// ── gate 3: physical home ──────────────────────────────────────────────────
+
+describe('BI-P129 gate 3 — RAW/NORMALIZED land as Parquet/Iceberg in ST0RE', () => {
+  it('a ClickHouse landing is refused', () => {
+    expect(() =>
+      assertPhysicalHomeMatchesTier(
+        target({ physical: { format: 'clickhouse' as never, namespace: 'n', table_name: 't' } }),
+      ),
+    ).toThrow(/not a lakehouse format/);
+  });
+
+  it('iceberg and parquet are both accepted; a nameless table is not', () => {
+    expect(() =>
+      assertPhysicalHomeMatchesTier(
+        target({ physical: { format: 'iceberg', namespace: 'n', table_name: 't' } }),
+      ),
+    ).not.toThrow();
+    expect(() =>
+      assertPhysicalHomeMatchesTier(
+        target({ physical: { format: 'parquet', namespace: ' ', table_name: 't' } }),
+      ),
+    ).toThrow(/missing a ST0RE namespace/);
+  });
+});
+
+// ── gate 4: unverified providers ───────────────────────────────────────────
+
+class UnverifiedProvider implements AnalyticsConnectorProvider {
+  readonly provider = 'dlt' as const;
+  providerReadiness() {
+    return {
+      provider: 'dlt',
+      pinned_version: '0.0.0',
+      verified: false,
+      proven: [],
+      open: ['nothing has been executed'],
+    };
+  }
+  validateSpec() {
+    return { valid: true, errors: [] };
+  }
+  connectorFor(): never {
+    throw new Error('not used in this test');
+  }
+  executionPlan() {
+    return { runtime: 'n0d' as const, reason: 'test' };
+  }
+}
+
+describe('BI-P129 gate 4 — an unverified provider cannot quietly become production capability', () => {
+  const provider = new UnverifiedProvider();
+  const spec = jsonlSpec({ provider: 'dlt', connector_ref: 'whatever' });
+
+  it('WITH the guard: production refuses an unverified provider', () => {
+    expect(() => assertProviderRunnable(provider, spec, 'production')).toThrow(SiaxError);
+    expect(() => assertProviderRunnable(provider, spec, 'production')).toThrow(/verified: false/);
+  });
+
+  it('WITHOUT the guard (non-production runtime): the same provider is allowed to plan', () => {
+    expect(() => assertProviderRunnable(provider, spec, 'staging')).not.toThrow();
+  });
+
+  it('a recorded, named exception unblocks production — and is required to be non-empty', () => {
+    expect(() =>
+      assertProviderRunnable(provider, { ...spec, unverified_provider_exception: '   ' }, 'production'),
+    ).toThrow(/unverified_provider_exception/);
+    expect(() =>
+      assertProviderRunnable(
+        provider,
+        {
+          ...spec,
+          unverified_provider_exception: 'accepted by simon@siax.io 2026-09-07 for a bounded backfill',
+        },
+        'production',
+      ),
+    ).not.toThrow();
+  });
+
+  it('the native provider is verified, so it needs no exception in production', () => {
+    expect(() =>
+      assertProviderRunnable(new NativeConnectorProvider(), jsonlSpec(), 'production'),
+    ).not.toThrow();
+    expect(new NativeConnectorProvider().providerReadiness().verified).toBe(true);
+  });
+});
+
+// ── licence boundary + registry ────────────────────────────────────────────
+
+describe('BI-P129 — Airbyte is a declared kind with no acceptance in this estate', () => {
+  it('registering an airbyte provider is refused at the registry', () => {
+    const registry = new ConnectorProviderRegistry();
+    const fake = { ...new UnverifiedProvider(), provider: 'airbyte' as const };
+    expect(() => registry.register(fake as unknown as AnalyticsConnectorProvider)).toThrow(
+      /no licence\/architecture acceptance/,
+    );
+  });
+
+  it('planning an airbyte ingestion is refused even if a provider object is supplied directly', () => {
+    const fake = {
+      ...new UnverifiedProvider(),
+      provider: 'airbyte' as const,
+    } as unknown as AnalyticsConnectorProvider;
+    expect(() => planAnalyticsIngestion(fake, jsonlSpec({ provider: 'airbyte' }), 'test')).toThrow(
+      /licence\/architecture acceptance/,
+    );
+  });
+
+  it('the registry refuses duplicates and unknown kinds, and reports readiness for what it has', () => {
+    const registry = new ConnectorProviderRegistry();
+    registry.register(new NativeConnectorProvider());
+    expect(() => registry.register(new NativeConnectorProvider())).toThrow(/already registered/);
+    expect(() => registry.get('singer')).toThrow(/no connector provider registered/);
+    expect(registry.readinessReport().map((r) => r.provider)).toEqual(['native']);
+  });
+
+  it('a provider handed a spec for a different kind is refused', () => {
+    expect(() =>
+      planAnalyticsIngestion(new NativeConnectorProvider(), jsonlSpec({ provider: 'dlt' }), 'test'),
+    ).toThrow(/asks for provider "dlt"/);
+  });
+});