P060-P068: durable ingestion plane — source classification gate, connector contract, common envelope, schema evolution, quality enforcement, queryable lineage, idempotent backfill #31

Merged

admin merged 1 commits from feat/p060-p068-ingestion-plane into main 2026-09-04 16:11:21 +00:00

Conversation 0 Commits 1 Files Changed 43

+5506 -20

admin commented 2026-09-04 15:04:13 +00:00

Owner

Copy Link

Copy Source

BEFORE_STATE (step 1, with evidence)

Req	Classification	Evidence
P060 Source Registry	PARTIAL / BROKEN	Registry + Postgres persistence existed (sql/002_source_registry.sql, src/repositories/postgres-source-registry-repository.ts) with rights/purpose/provenance/health/freshness. No authorization, no schedule, no checkpoint, no schema, no owner distinct from created_by. The activation DoD was inverted: src/repositories/postgres-source-repository.ts:70 wrote ${input.status ?? "active"} and no activation operation existed — an unclassified source went live on registration
P061 Connector contract	MISSING	services/api/sources/connectors.py = per-source fetch_*() functions. No lifecycle, cursor, resume, health, disconnect or conformance suite anywhere
P062 CDC/batch/stream envelope	MISSING	Already documented honestly in services/api/tests/test_known_gaps.py:10-19 — no CDC pipeline exists; workers/pipeline.py has no persistence at all
P063 Schema evolution	MISSING	schema_version existed as a free string on datasets/Iceberg writes; no comparison, no policy, no versioned source schema
P064 Data quality	PARTIAL	src/quality/* had rules/runs/results/issues/scores + a GE-subset engine, but a failure only opened an ISSUE — the data flowed on. grep -ri quarantin src/ → one unrelated hit
P065 Normalization/ER	DUPLICATE (confirmed)	Two engines: src/er/entity-resolution-service.ts (canonical) and services/api/workers/resolve.py (simplified), the latter called from workers/pipeline.py:110
P067 Lineage	PARTIAL	src/lineage/lineage-emitter.ts emitted to AUD0; nothing in this repo could ANSWER a lineage question — no store, no query, no route
P068 Backfill	MISSING	Documented in test_known_gaps.py:44-52: "no delivery-store persistence layer exists to dedupe a backfill against"
P066 Iceberg	out of scope (ST0RE, other workflow)	contract read only

Only PARTIAL / BROKEN / MISSING items were built.

What was built

P060 — sql/007_ingestion_plane.sql adds authorization, schedule, versioned schemas, checkpoints, ledger, quarantine and lineage-edge tables, owner_id, and flips opt_sources.status DEFAULT to 'pending' so even a raw psql INSERT cannot create an unclassified-but-active source. SourceRegistryService gains classifySource / activateSource / deactivateSource. Registration never activates — asking for status:"active" goes through the gate, not around it — and a service identity (svc_*, system, …) cannot be the authorizing actor. Routes: activate (GET = classification, POST = activate), authorization, schedule.

P061 — src/connectors/connector.ts defines all nine operations on one interface. src/connectors/conformance.ts is a runnable suite (15 checks, no "skip" verdict — a connector that cannot honestly serve a check fails it). Two genuinely heterogeneous connectors pass the identical suite: HttpJsonConnector (real node:http server, opaque server cursor, network health check) and JsonlFileConnector (real file, self-computed line offsets, filesystem health, CDC deletes). The suite earned its keep while being written: it caught that an API which drops next_cursor at the end leaves the connector unable to resume — hence cursor_field.

P062 — one IngestionEnvelope for batch/stream/cdc; durable per-(source, partition) checkpoints; IngestionRunner with the ordering that makes resume real.

P063 — checkCompatibility + SchemaRegistry, enforced at two points both upstream of any consumer write: a discovery gate before the first read(), and per-record conformance before the page reaches the sink. A producer cannot retroactively widen its own policy to none.

P064 — QualityGate composes the existing ExpectationEngine (not a second engine) and adds the missing consequence: block (checkpoint does not advance) / quarantine (durable, full envelope preserved) / warn. No policy is reported as none_declared, never as a pass. An expectation the engine cannot execute is a violation, never a silent pass.

P067 — LineageStore writes 0pt's own queryable projection of the OpenLineage events it already emits (AUD0 remains the system of record) and GET /api/0pt/lineage answers both directions. An unfiltered dump is refused — that is an export, not a query.

P068 — opt_ingestion_ledger + IngestionRunner.backfill: bounded, idempotent on repeat, leaves the live checkpoint alone, still re-emits a genuine content change.

A bug this PR wrote and then caught

The first IngestionRepository draft combined the idempotency check with its recording into one atomic call placed before the sink. The crash/resume test then delivered 7 of 10 records: the page in flight when the sink died had already been marked delivered, so the resumed run skipped it as a duplicate — permanent, silent data loss. Fixed by splitting peekDeliveries / commitDeliveries, with both the ledger commit and the checkpoint commit after the sink resolves. The failure mode is documented in the interface, in the runner, and in STATUS.md.

What was NOT done

• P065 is not met. services/api/workers/resolve.py is still a second simplified production ER engine. Closing it means rewriting it as an adapter over the canonical TypeScript service, which cannot be verified from this repo today: services/api is not built by this repo's Dockerfile (it builds only the Next.js app), api.0pt.app answers 404, and no CI workflow runs its pytest suite (grep -rn pytest .gitea/workflows .github/workflows → nothing). A blind rewrite would ship untested into a path nobody can exercise. The duplication is now stated in code at both sites and the 4-step removal plan is in STATUS.md. The new ingestion plane performs no ER, so it adds no third engine.
• P066 untouched (ST0RE, other workflow).
• Delivery is at-least-once, not exactly-once — no per-(source, partition) run lock.
• No CDC source exists. The envelope, operations and runner handle change events; a Debezium/WAL consumer is still missing.
• 0pt does not schedule itself; opt_source_schedule is a declaration ACT0 would read.

Verified (commands actually run, not "should pass")

• pnpm run typecheck — clean
• pnpm run test — 365 passed / 25 skipped (base: 311 / 18)
• pnpm run build — clean; all four new routes in the route table
• pnpm run lint — 12 errors / 5 warnings, byte-identical count to the base commit (advisory in CI). This PR adds none
• Real Postgres 16 container: INGESTION_TEST_DATABASE_URL=… pnpm vitest run tests/p0/ingestion-plane-postgres.realinfra.test.ts — 7 passed
• Real container restart: docker restart the Postgres container, re-read through the production repositories. Identical before/after — 2 sources (one active with owner, one pending), 1 checkpoint (cursor 6, 6 committed), 6 ledger rows, 1 lineage edge, 4 quarantined records, 1 schema version. That is P060's "a restart preserves all sources", measured

Deployment reality (curl, 2026-09-04)

0pt.app/api/health → 200 from Vercel, {"db":"not_configured"}. 0pt.app/api/0pt/sources → 404 (the deployed build predates the whole app/api/0pt/* layer). api.0pt.app/health → 404. .gitea/workflows/cd.yml still guards on COOLIFY_APP_ID = TODO-no-coolify-app-provisioned-yet. None of the capability layer — this PR's or P010/P014's — runs in production. Nothing here claims LIVE_VERIFIED.

🤖 Generated with Claude Code

## BEFORE_STATE (step 1, with evidence) | Req | Classification | Evidence | |---|---|---| | P060 Source Registry | **PARTIAL / BROKEN** | Registry + Postgres persistence existed (`sql/002_source_registry.sql`, `src/repositories/postgres-source-registry-repository.ts`) with rights/purpose/provenance/health/freshness. **No** authorization, **no** schedule, **no** checkpoint, **no** schema, **no** owner distinct from `created_by`. The activation DoD was inverted: `src/repositories/postgres-source-repository.ts:70` wrote `${input.status ?? "active"}` and no activation operation existed — an unclassified source went live on registration | | P061 Connector contract | **MISSING** | `services/api/sources/connectors.py` = per-source `fetch_*()` functions. No lifecycle, cursor, resume, health, disconnect or conformance suite anywhere | | P062 CDC/batch/stream envelope | **MISSING** | Already documented honestly in `services/api/tests/test_known_gaps.py:10-19` — no CDC pipeline exists; `workers/pipeline.py` has no persistence at all | | P063 Schema evolution | **MISSING** | `schema_version` existed as a free string on datasets/Iceberg writes; no comparison, no policy, no versioned source schema | | P064 Data quality | **PARTIAL** | `src/quality/*` had rules/runs/results/issues/scores + a GE-subset engine, but a failure only opened an ISSUE — the data flowed on. `grep -ri quarantin src/` → one unrelated hit | | P065 Normalization/ER | **DUPLICATE (confirmed)** | Two engines: `src/er/entity-resolution-service.ts` (canonical) and `services/api/workers/resolve.py` (simplified), the latter called from `workers/pipeline.py:110` | | P067 Lineage | **PARTIAL** | `src/lineage/lineage-emitter.ts` emitted to AUD0; nothing in this repo could ANSWER a lineage question — no store, no query, no route | | P068 Backfill | **MISSING** | Documented in `test_known_gaps.py:44-52`: "no delivery-store persistence layer exists to dedupe a backfill against" | | P066 Iceberg | out of scope (ST0RE, other workflow) | contract read only | Only PARTIAL / BROKEN / MISSING items were built. ## What was built **P060** — `sql/007_ingestion_plane.sql` adds authorization, schedule, versioned schemas, checkpoints, ledger, quarantine and lineage-edge tables, `owner_id`, and flips `opt_sources.status` DEFAULT to `'pending'` so even a raw psql `INSERT` cannot create an unclassified-but-active source. `SourceRegistryService` gains `classifySource` / `activateSource` / `deactivateSource`. Registration never activates — asking for `status:"active"` goes *through* the gate, not around it — and a service identity (`svc_*`, `system`, …) cannot be the authorizing actor. Routes: `activate` (GET = classification, POST = activate), `authorization`, `schedule`. **P061** — `src/connectors/connector.ts` defines all nine operations on one interface. `src/connectors/conformance.ts` is a runnable suite (15 checks, no "skip" verdict — a connector that cannot honestly serve a check fails it). Two genuinely heterogeneous connectors pass the identical suite: `HttpJsonConnector` (real `node:http` server, opaque server cursor, network health check) and `JsonlFileConnector` (real file, self-computed line offsets, filesystem health, CDC deletes). The suite earned its keep while being written: it caught that an API which drops `next_cursor` at the end leaves the connector unable to resume — hence `cursor_field`. **P062** — one `IngestionEnvelope` for batch/stream/cdc; durable per-(source, partition) checkpoints; `IngestionRunner` with the ordering that makes resume real. **P063** — `checkCompatibility` + `SchemaRegistry`, enforced at two points both upstream of any consumer write: a discovery gate before the first `read()`, and per-record conformance before the page reaches the sink. A producer cannot retroactively widen its own policy to `none`. **P064** — `QualityGate` composes the **existing** `ExpectationEngine` (not a second engine) and adds the missing consequence: `block` (checkpoint does not advance) / `quarantine` (durable, full envelope preserved) / `warn`. No policy is reported as `none_declared`, never as a pass. An expectation the engine cannot execute is a violation, never a silent pass. **P067** — `LineageStore` writes 0pt's own queryable projection of the OpenLineage events it already emits (AUD0 remains the system of record) and `GET /api/0pt/lineage` answers both directions. An unfiltered dump is refused — that is an export, not a query. **P068** — `opt_ingestion_ledger` + `IngestionRunner.backfill`: bounded, idempotent on repeat, leaves the live checkpoint alone, still re-emits a genuine content change. ## A bug this PR wrote and then caught The first `IngestionRepository` draft combined the idempotency check with its recording into one atomic call placed **before** the sink. The crash/resume test then delivered **7 of 10** records: the page in flight when the sink died had already been marked delivered, so the resumed run skipped it as a duplicate — permanent, silent data loss. Fixed by splitting `peekDeliveries` / `commitDeliveries`, with both the ledger commit and the checkpoint commit after the sink resolves. The failure mode is documented in the interface, in the runner, and in `STATUS.md`. ## What was NOT done - **P065 is not met.** `services/api/workers/resolve.py` is still a second simplified production ER engine. Closing it means rewriting it as an adapter over the canonical TypeScript service, which **cannot be verified from this repo today**: `services/api` is not built by this repo's Dockerfile (it builds only the Next.js app), `api.0pt.app` answers 404, and no CI workflow runs its pytest suite (`grep -rn pytest .gitea/workflows .github/workflows` → nothing). A blind rewrite would ship untested into a path nobody can exercise. The duplication is now stated **in code at both sites** and the 4-step removal plan is in `STATUS.md`. The new ingestion plane performs no ER, so it adds no third engine. - **P066** untouched (ST0RE, other workflow). - Delivery is **at-least-once**, not exactly-once — no per-(source, partition) run lock. - **No CDC *source*** exists. The envelope, operations and runner handle change events; a Debezium/WAL consumer is still missing. - 0pt does not schedule itself; `opt_source_schedule` is a declaration ACT0 would read. ## Verified (commands actually run, not "should pass") - `pnpm run typecheck` — clean - `pnpm run test` — **365 passed / 25 skipped** (base: 311 / 18) - `pnpm run build` — clean; all four new routes in the route table - `pnpm run lint` — 12 errors / 5 warnings, byte-identical count to the base commit (advisory in CI). This PR adds none - Real **Postgres 16** container: `INGESTION_TEST_DATABASE_URL=… pnpm vitest run tests/p0/ingestion-plane-postgres.realinfra.test.ts` — **7 passed** - **Real container restart**: `docker restart` the Postgres container, re-read through the production repositories. Identical before/after — 2 sources (one `active` with owner, one `pending`), 1 checkpoint (cursor 6, 6 committed), 6 ledger rows, 1 lineage edge, 4 quarantined records, 1 schema version. That is P060's "a restart preserves all sources", measured ## Deployment reality (curl, 2026-09-04) `0pt.app/api/health` → **200** from **Vercel**, `{"db":"not_configured"}`. `0pt.app/api/0pt/sources` → **404** (the deployed build predates the whole `app/api/0pt/*` layer). `api.0pt.app/health` → **404**. `.gitea/workflows/cd.yml` still guards on `COOLIFY_APP_ID = TODO-no-coolify-app-provisioned-yet`. **None** of the capability layer — this PR's or P010/P014's — runs in production. Nothing here claims LIVE_VERIFIED. 🤖 Generated with [Claude Code](https://claude.com/claude-code)

admin added 1 commit 2026-09-04 15:04:14 +00:00

feat(data-os): P060-P068 durable ingestion plane — source classification gate, connector contract, common envelope, schema evolution, quality enforcement, queryable lineage, idempotent backfill ... 9267576b7c

BEFORE_STATE (verified, with evidence):

* P060 PARTIAL/BROKEN. The registry persisted rights/provenance/health/
  freshness to Postgres, but carried no authorization, no schedule, no
  checkpoint, no schema and no owner distinct from created_by — and the
  activation DoD was inverted: postgres-source-repository.ts wrote
  `${input.status ?? "active"}`, so a source with no rights, no provenance
  and no authorization was ACTIVE the instant it was registered. No
  activation operation existed at all.
* P061 MISSING. services/api/sources/connectors.py had per-source fetch_*()
  functions only — no lifecycle, no cursor, no resume, no health, no
  conformance suite.
* P062 MISSING, already documented as such in
  services/api/tests/test_known_gaps.py:10-19 (no CDC pipeline anywhere;
  workers/pipeline.py has no persistence at all).
* P063 MISSING. schema_version existed as a free string; no comparison,
  no policy, no versioned source schema.
* P064 PARTIAL. Rules/runs/issues/scores existed, but a failing rule only
  opened an issue and the data flowed on. No quarantine anywhere.
* P065 DUPLICATE (confirmed). Two ER engines: src/er/entity-resolution-
  service.ts (canonical) and services/api/workers/resolve.py (simplified,
  still called by pipeline.py:110).
* P067 PARTIAL. Lineage was emitted to AUD0 but nothing here could ANSWER a
  lineage question — no store, no query, no route.
* P068 MISSING, documented in test_known_gaps.py:44-52 ("no delivery-store
  persistence layer exists to dedupe a backfill against").

BUILT (only what was PARTIAL/BROKEN/MISSING):

* P060 sql/007_ingestion_plane.sql adds authorization/schedule/schemas/
  checkpoints/ledger/quarantine/lineage tables, owner_id, and flips the
  opt_sources status DEFAULT to 'pending' so even a raw psql INSERT cannot
  produce an unclassified-but-active source. SourceRegistryService gains
  classifySource/activateSource/deactivateSource + authorization/schedule
  accessors; registration NEVER activates (asking for status:"active" goes
  through the gate, not around it); a service identity cannot authorize a
  source. Routes: activate (GET classification / POST activate),
  authorization, schedule.
* P061 src/connectors/connector.ts — all nine operations on one interface.
  src/connectors/conformance.ts — a runnable suite (15 checks, no "skip"
  verdict). HttpJsonConnector (real HTTP, opaque cursor) and
  JsonlFileConnector (real file, line offsets, CDC deletes) both pass it.
  The suite caught a real defect while being written: an API that drops
  next_cursor at the end left the connector unable to resume, which is why
  HttpJsonConnectorConfig has cursor_field.
* P062 src/types/ingestion.ts + src/ingestion/{envelope,ingestion-runner}.ts.
  One envelope for batch/stream/cdc; durable per-(source,partition)
  checkpoints; resume proven by killing a run inside the sink.
* P063 src/schema/{record-schema,compatibility,schema-registry}.ts. Two
  enforcement points, both upstream of any consumer write: a discovery gate
  before the first read, and per-record conformance before the page reaches
  the sink. A producer cannot widen its own policy retroactively.
* P064 src/quality/quality-policy.ts composes the EXISTING ExpectationEngine
  (not a second engine) and adds the missing consequence: block /
  quarantine / warn, with quarantine durable and recoverable. No policy is
  reported as "none_declared", never as a pass; an unexecutable expectation
  counts as a violation, never a silent pass.
* P067 src/lineage/{lineage-edge,lineage-store}.ts + GET /api/0pt/lineage.
  AUD0 stays the system of record; this is 0pt's own queryable projection.
* P068 opt_ingestion_ledger + IngestionRunner.backfill: bounded, idempotent
  on repeat, leaves the live checkpoint alone, still re-emits real changes.

A BUG THIS PASS WROTE AND CAUGHT: the first IngestionRepository draft
combined the idempotency check and its recording into one atomic call
BEFORE the sink. The crash/resume test then delivered 7 of 10 records — the
page in flight when the sink died was already marked delivered, so the
resumed run skipped it as a duplicate. Permanent, silent data loss. Fixed by
splitting peekDeliveries/commitDeliveries and committing both the ledger and
the checkpoint only after the sink resolves.

NOT DONE, stated rather than deferred quietly:

* P065 is NOT met. resolve.py is still a second simplified production
  engine. Closing it means rewriting it as an adapter over the canonical
  service, which cannot be verified from this repo today: services/api is
  not built by this repo's Dockerfile, api.0pt.app answers 404, and NO CI
  workflow runs its pytest suite. The duplication is now stated at both
  sites in code and the removal plan is in STATUS.md. The new ingestion
  plane performs no ER, so it adds no third engine.
* P066 (Iceberg) untouched — ST0RE is owned by another workflow. Contract
  read only.
* Delivery is AT-LEAST-ONCE, not exactly-once (no per-partition run lock).
* No CDC *source* exists; the contract is met, a CDC connector is not built.
* 0pt does not schedule itself — the schedule row is a declaration ACT0
  would read.

VERIFIED: pnpm typecheck clean; pnpm test 365 passed / 25 skipped (was
311/18); pnpm build clean with all four new routes in the route table; lint
unchanged at 12 pre-existing errors. Real Postgres 16
(tests/p0/ingestion-plane-postgres.realinfra.test.ts, 7 passed) plus a real
`docker restart` of that container with an identical read-back through the
production code path — that is P060's "a restart preserves all sources",
measured.

DEPLOYMENT REALITY (curl, 2026-09-04): 0pt.app/api/health returns 200 from
Vercel with db:not_configured; 0pt.app/api/0pt/sources returns 404;
api.0pt.app/health returns 404; cd.yml still guards on a placeholder Coolify
app id. None of the capability layer runs in production. Nothing here is
LIVE_VERIFIED and STATUS.md says so.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

admin merged commit into main 2026-09-04 16:11:21 +00:00

Sign in to join this conversation.

Reviewers

No Reviewers

Labels

No items

No labels

Milestone

No items

No Milestone

Projects

Clear projects

No projects

Assignees

Clear assignees

siax-bot siax-ci siax-healer (siax-healer (CI self-healing agent)) admin (Simon Axelsson)

No Assignees

1 Participants

Notifications

Subscribe

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: admin/0pt#31