Merge commit '3edc56d' into HEAD

STATUS.md

@@ -70,15 +70,31 @@ accurate.
 
 ## Global gaps blocking PRODUCTION_READY (§3, tracked, not hidden)
 
-1. Persistent stores: the new registries (sources/rights/quality
-   rules/dataset versions/index definitions/feature views) are per-process
-   in-memory. `resolveStoreMode` guarantees production boot FAILS rather
-   than serving Maps, but the Postgres stores themselves are not built.
+1. Persistent stores REMAINING: quality runs/rules and feature views are
+   still per-process in-memory (`resolveStoreMode` guarantees production
+   boot FAILS rather than serving Maps). CLOSED since the last revision:
+   dataset registry (`opt_datasets`) + immutable versions
+   (`opt_dataset_versions`, monotonic under transaction lock, idempotency
+   with payload-drift detection, cross-tenant ⇒ NOT_FOUND) and versioned
+   search index definitions (`opt_search_index_definitions`, sequential
+   schema versions, replay-idempotent, same-version-different-schema ⇒
+   CONFLICT; OpenSearch remains a rebuildable projection, never canonical
+   truth) are Postgres-canonical with real-Postgres tests.
 2. Capability registration in CL0UD + API0 discovery for the new
-   capabilities (§3 "Capability är registrerad i CL0UD Capability Registry").
-3. AUD0 outbox for evidence durability under AUD0 outage (§70).
-4. Surface exposure: no `app/api/0pt/*` routes yet for
-   search/quality/datasets-versions/ER (services + contracts only).
+   capabilities (§3 "Capability är registrerad i CL0UD Capability
+   Registry") — PARTIALLY CLOSED: the collaboration/meeting domain was
+   canonicalized in CL0UD's registry (cl0ud #73/#74); 0pt's own
+   capability set still needs the same registration.
+3. CLOSED: AUD0 outbox for evidence durability under AUD0 outage (§70) —
+   durable Postgres outbox + dispatcher (at-least-once, bounded batch,
+   backoff, dead-letter, SKIP LOCKED leasing, tenant binding, idempotency)
+   + real-Postgres tests; dispatcher uses explicit `durable:false` so it
+   never re-enqueues its own delivery evidence.
+4. CLOSED for datasets/versions/search/ER: `app/api/0pt/datasets*`,
+   `app/api/0pt/search/definitions` + tenant-safed query route, and
+   `app/api/0pt/er/{candidates,batch}` (+ internal ER route) are live;
+   clients cannot raise `rights_tag` via request body (fail-closed to
+   `internal`). Quality routes exist (`bridge-run/issues/rules/runs`).
 5. 0pt-motor executors for the delegated jobs
    (`0pt.quality.ge_run`, `0pt.features.*`, `0pt.er.batch_resolution`).
 6. Observability: no metrics/traces emitted by the new services (§36).
@@ -97,7 +113,7 @@ verified, and by which command.
 | **P062** CDC/batch/stream normalized to one envelope; restart resumes from a durable checkpoint without silent gaps | **MISSING**, and documented as such: `services/api/tests/test_known_gaps.py:10-19` skipped the CDC case because no CDC pipeline existed anywhere, and `workers/pipeline.py` had no persistence at all | **INTEGRATION_TESTED** | `src/types/ingestion.ts` (one envelope for all three modes); `src/ingestion/ingestion-runner.ts`; `tests/p0/ingestion-runner.test.ts` kills a run inside the sink and proves the resumed run covers all 10 records exactly once; same proof against real Postgres across fresh processes |
 | **P063** Schema versioning + compatibility policy; a breaking schema is detected before consumer corruption | **MISSING.** `schema_version` existed as a free-form string on datasets/Iceberg writes; there was no comparison, no policy, no versioned source schema | **INTEGRATION_TESTED** | `src/schema/compatibility.ts`, `src/schema/schema-registry.ts`; `tests/p0/schema-evolution.test.ts` (12); the runner's schema gate runs before the first read — `tests/p0/ingestion-runner.test.ts` asserts `sinkCalls === 0` on a break |
 | **P064** Quality rules behind the SIAX abstraction; failures block or quarantine per declared policy | **PARTIAL.** `src/quality/*` had rules, runs, results, issues, scores and a GE-compatible engine, but a failing rule only opened an ISSUE — the data flowed on regardless. `grep -ri quarantin src/` found one unrelated hit | **INTEGRATION_TESTED** | `src/quality/quality-policy.ts` composes the EXISTING `ExpectationEngine` (not a second engine) and adds the consequence; block / quarantine / warn each proved in `tests/p0/ingestion-runner.test.ts`; quarantine durability proved against real Postgres |
-| **P065** Normalization/ER: only the canonical highway; no second simplified production engine | **DUPLICATE (confirmed).** `src/er/entity-resolution-service.ts` (canonical: hard rules, review band, Splink delegation) AND `services/api/workers/resolve.py` (simplified fuzzy threshold, auto-merge) — the latter still called by `workers/pipeline.py:110` | **NOT DONE — documented, not built.** The duplication is now stated at BOTH sites in code with the removal plan. The new ingestion plane deliberately performs no ER, so it adds no third engine. See "P065 — why it was not closed" below |
+| **P065** Normalization/ER: only the canonical highway; no second simplified production engine | **CLOSED.** `services/api/workers/resolve.py` is now a thin adapter: Python no longer scores identity — it POSTs opaque records to `/api/0pt/internal/er/candidates` and applies ONLY explicit hard-rule `match` decisions (score/fuzzy can never merge records, even from a compromised bridge); canonical engine unreachable ⇒ records return UNMERGED, never fallback-scored. Company semantics `stable-identifiers.v3`: LEI global; org_nr hard identity only within the same registration_country; punctuation normalized; negative stable identity dominates every positive; fuzzy alone ⇒ human review, never auto-merge. Cluster-wide hard-negative constraints: A≈B ∧ B≈C never forces A≈C when canonical identity says A≠C | **INTEGRATION_TESTED** | `tests/p0/entity-resolution.test.ts` (v3 semantics incl. hard-negative transitivity), `tests/p0/entity-resolution-http.test.ts` (internal route fails closed: empty/wrong token ⇒ 401), `services/api/tests/test_resolve_determinism.py` (adapter contract: canonical edges only), shared request validation (`src/er/request-validation.ts`, records 2..5000), blocking pytest gate (`.gitea/workflows/python-engine.yml`) |
 | **P066** Iceberg | out of scope — ST0RE is owned by another workflow | **NOT TOUCHED** | Contract read only (`src/services/iceberg-write-service.ts`, `src/clients/st0re-lakehouse-client.ts`) |
 | **P067** Lineage: OpenLineage/AUD0 events; source → output lineage is queryable | **PARTIAL.** `src/lineage/lineage-emitter.ts` emitted OpenLineage to AUD0, but nothing in this repo could ANSWER a lineage question — no store, no query, no route | **INTEGRATION_TESTED** | `src/lineage/lineage-edge.ts`, `src/lineage/lineage-store.ts` (local queryable projection; AUD0 stays the system of record), `GET /api/0pt/lineage`; `tests/p0/ingestion-runner.test.ts` queries both directions and proves tenant scoping |
 | **P068** Durable bounded backfill/replay; repeated backfill is idempotent | **MISSING**, documented as such: `services/api/tests/test_known_gaps.py:44-52` skipped it because "no delivery-store persistence layer exists to dedupe a backfill against" | **INTEGRATION_TESTED** | `opt_ingestion_ledger` + `IngestionRunner.backfill`; `tests/p0/ingestion-runner.test.ts` (repeat emits 0, live checkpoint untouched, a real content change still re-emits); same against real Postgres |
@@ -167,22 +183,24 @@ production, and P060's durability guarantee cannot be exercised there,
 because production has no database configured and does not serve the routes.
 Verified with `curl`; not inferred.
 
-### P065 — why it was not closed
+### P065 — closure record (2026-09-07)
 
-Closing P065 means turning `services/api/workers/resolve.py` into an adapter
-over the canonical `src/er/entity-resolution-service.ts`. That change 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. A blind rewrite would ship untested
-into a path nobody can exercise — the exact failure mode this programme
-exists to stop.
+The removal plan below was executed in full and is verified in CI:
 
-The removal plan, for whoever picks it up:
-1. Expose the canonical ER capability as `POST /api/0pt/er/candidates`
-   (service already exists; only a route + Zod schema are missing).
-2. Add a Python CI job that actually runs `services/api/tests`.
-3. Replace `resolve.py`'s body with an HTTP adapter over that route,
-   **fail-closed** when it is unreachable — never falling back to the local
-   scorer, which would preserve the duplicate under a different name.
-4. Port `test_resolve_determinism.py` to assert the adapter's contract, and
-   delete the fuzzy scorer.
+1. ✅ Canonical ER exposed as `POST /api/0pt/er/candidates` (public,
+   tenant-safed) AND `POST /api/0pt/internal/er/candidates` (service-token,
+   fails closed without `X-Internal-Token`).
+2. ✅ Python CI job: `.gitea/workflows/python-engine.yml` runs
+   `services/api/tests` as a BLOCKING pytest gate.
+3. ✅ `workers/resolve.py` rewritten as the HTTP adapter — fail-closed to
+   UNMERGED records when the bridge is unreachable; no fallback scorer
+   remains. Cluster-wide hard-negative constraints prevent transitive
+   matches from overriding canonical identity decisions.
+4. ✅ `services/api/tests/test_resolve_determinism.py` asserts the adapter
+   contract: canonical match edges merge; review/non-match edges do not;
+   hard negatives are respected transitively.
+
+Residual honesty: the PYTHON-side tests run in CI, but the full
+adapter→Next.js route chain has not been exercised against a live deployed
+edge in this branch (the route's own fail-closed behavior is unit-proven).
+That keeps P065 at INTEGRATION_TESTED, not LIVE_VERIFIED.

src/repositories/dataset-version-repository.ts

@@ -143,7 +143,7 @@ export class PostgresDatasetVersionRepository implements DatasetVersionRepositor
            correlation_id, idempotency_key)
         VALUES
           (${input.dataset_id}, ${version}, ${input.tenant_id}, ${input.project_id},
-           ${tx.array(input.artifact_refs)}, ${input.schema_version}, ${input.record_count ?? null},
+           ${tx.array([...input.artifact_refs])}, ${input.schema_version}, ${input.record_count ?? null},
            ${input.description ?? null}, ${input.created_by}, ${input.correlation_id ?? null},
            ${input.idempotency_key ?? null})
         RETURNING *`;

src/types/dataset-version.ts

@@ -17,7 +17,8 @@ export interface CreateDatasetVersionInput {
   dataset_id: string;
   tenant_id: string;
   project_id: string;
-  artifact_refs: string[];
+  /** Immutable input — callers may hand us readonly arrays (as const); we never mutate. */
+  readonly artifact_refs: readonly string[];
   schema_version: string;
   record_count?: number;
   description?: string;

tests/p0/dataset-postgres.realinfra.test.ts

@@ -34,6 +34,10 @@ beforeAll(async () => {
     for (const migration of ["003_datasets.sql", "011_aud0_evidence_outbox.sql", "012_dataset_versions.sql"]) {
       await admin.unsafe(fs.readFileSync(path.join(process.cwd(), "sql", migration), "utf8"));
     }
+    // Aktivera den durabla AUD0-outboxen: createAud0OutboxRepositoryFromEnv()
+    // läser DATABASE_URL (se src/repositories/db.ts) — utan den blir klienten
+    // Noop och inga producer-rader skrivs (testet skulle ljugit om durabilitet).
+    process.env.DATABASE_URL = TEST_DATABASE_URL;
     reachable = true;
   } catch (error) {
     unreachableReason = error instanceof Error ? error.message : String(error);
@@ -41,6 +45,7 @@ beforeAll(async () => {
 }, 30_000);
 
 afterAll(async () => {
+  delete process.env.DATABASE_URL;
   await admin?.end({ timeout: 5 });
   await closeRepositorySqlForTests();
 });

tests/p0/entity-resolution-http.test.ts

@@ -40,13 +40,15 @@ afterEach(() => {
 
 describe("POST /api/0pt/internal/er/candidates", () => {
   it("fails closed without the internal service token", async () => {
+    // OBS: request(body, undefined) triggar default-parametern (skickar GILTIGT
+    // token) — skicka tom string för "utan token" (fail-closed → 401).
     const res = await candidatesPOST(
       request(
         baseBody([
           { record_id: "a", fields: { name: "Acme AB", city: "Stockholm" } },
           { record_id: "b", fields: { name: "Acme AB", city: "Stockholm" } },
         ]),
-        undefined,
+        "",
       ),
     );
     expect(res.status).toBe(401);

tests/p0/entity-resolution.test.ts

@@ -97,7 +97,7 @@ describe("EntityResolutionService — candidate API", () => {
     expect(() =>
       er.generateCandidates(request({ auto_match_threshold: 0.5, review_threshold: 0.6 }))
     ).toThrowError(/review_threshold/);
-    expect(() => er.generateCandidates(request({ records: [RECORDS[0]] }))).toThrowError(/two records/);
+    expect(() => er.generateCandidates(request({ records: [RECORDS[0]] }))).toThrowError(/between 2 and 5000/);
     expect(() => er.generateCandidates(request({ tenant_id: "" }))).toThrowError(/tenant_id/);
   });
 

tests/p0/search-service.test.ts

@@ -160,12 +160,26 @@ describe("SearchService — tenant + purpose/rights enforcement (§10/§31)", ()
     const { search } = await build([
       doc("1", "t_a", "Tagged", { purpose_tags: ["company_enrichment"] }),
       doc("2", "t_a", "OtherPurpose", { purpose_tags: ["compliance"] }),
-      doc("3", "t_a", "Untagged", { purpose_tags: [] }), // empty ⇒ never returnable
     ]);
     const res = await search.query("companies", { tenant_id: "t_a", purpose: "company_enrichment" });
     expect(res.hits.map((h) => h.doc_id)).toEqual(["1"]);
   });
 
+  it("fails closed: rebuild rejects documents without purpose_tags (§10)", async () => {
+    const fake = new FakeOpenSearch();
+    const search = new SearchService(fake);
+    search.registerIndexDefinition({ ...DEF, fields: { ...DEF.fields } });
+    await expect(
+      search.rebuild(
+        "companies",
+        docsSource([
+          doc("1", "t_a", "Ok"),
+          doc("3", "t_a", "Untagged", { purpose_tags: [] }), // empty ⇒ INVALID_REQUEST, generation discarded
+        ]),
+      ),
+    ).rejects.toThrowError(/missing purpose_tags/);
+  });
+
   it("rights filtering: default rights tag is 'internal'; deliverable docs need the tag", async () => {
     const { search } = await build([
       doc("1", "t_a", "Internal", { rights_tags: ["internal"] }),