From 2d34cb162996d8a185a8be023893b4e3463b5d5f Mon Sep 17 00:00:00 2001 From: Simon Axelsson Date: Tue, 8 Sep 2026 22:18:45 +0200 Subject: [PATCH] =?UTF-8?q?feat(P-001,=20P-002):=200pt=E2=86=92N0D=20quali?= =?UTF-8?q?ty=20delegation=20+=20internal=20OpenAPI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- contracts/openapi.yaml | 99 +++++++++++++++++++++ src/clients/n0d-client.ts | 72 ++++++++++++++- src/quality/quality-service.ts | 20 ++--- tests/contracts/n0d-client.contract.test.ts | 64 ++++++++++++- tests/p0/quality-service.test.ts | 18 ++++ 5 files changed, 252 insertions(+), 21 deletions(-) diff --git a/contracts/openapi.yaml b/contracts/openapi.yaml index 08ced4f..6248eef 100644 --- a/contracts/openapi.yaml +++ b/contracts/openapi.yaml @@ -367,6 +367,105 @@ paths: $ref: "#/components/schemas/ProvenanceRecord" "401": $ref: "#/components/responses/Unauthorized" + + # --------------------------------------------------------------------------- + # Quality — P-002 internal OpenAPI (0pt quality delegation to N0D) + # --------------------------------------------------------------------------- + /api/0pt/quality/runs: + post: + operationId: 0pt.quality.run + summary: Execute quality checks for a dataset; delegates to N0D via n0dRunDataQuality when any rule needs the GE backend. + x-siax-status: target + tags: [Quality] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [tenant_id, dataset_id] + properties: + tenant_id: { type: string } + dataset_id: { type: string } + dataset_version: { type: integer } + responses: + "201": + description: Quality run accepted (status=delegated when offloaded to N0D). + "404": + description: No active rules for the dataset. + get: + operationId: 0pt.quality.list_runs + summary: List quality runs for a tenant. + x-siax-status: target + tags: [Quality] + parameters: + - { name: tenant_id, in: query, required: true, schema: { type: string } } + responses: + "200": + description: List of quality runs. + + /api/0pt/quality/bridge-run: + post: + operationId: 0pt.quality.bridge_run + summary: Advisory quality bridge — auto-provisions baseline rules, runs QualityService.runQuality; Python engine calls through here. Fail-open. + x-siax-status: target + tags: [Quality] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [tenant_id, dataset_id] + properties: + tenant_id: { type: string } + dataset_id: { type: string } + responses: + "200": + description: Run + any open issues. + + /api/0pt/quality/checks: + get: + operationId: 0pt.quality.checks + summary: List active quality checks/rules for a tenant+dataset. + x-siax-status: target + tags: [Quality] + parameters: + - { name: tenant_id, in: query, required: true, schema: { type: string } } + - { name: dataset_id, in: query, required: true, schema: { type: string } } + responses: + "200": + description: Active quality rules. + + /api/0pt/quality/reports: + get: + operationId: 0pt.quality.reports + summary: Retrieve quality reports / Great Expectations results for a run. + x-siax-status: target + tags: [Quality] + parameters: + - { name: run_id, in: query, required: true, schema: { type: string } } + responses: + "200": + description: Quality report. + + # --------------------------------------------------------------------------- + # N0D interop — 0pt consumption of N0D owner operations (P-001 owner-seam) + # --------------------------------------------------------------------------- + /api/0pt/n0d/results/{jobId}: + get: + operationId: 0pt.n0dGetJobResult + summary: Retrieve the result of an N0D-delegated quality job (n0dGetJobResult). Polls N0D after the quality run is delegated. + x-siax-status: target + tags: [Quality, N0D] + parameters: + - { name: jobId, in: path, required: true, schema: { type: string } } + responses: + "200": + description: N0D job result (outcome, metrics, ST0RE artifact refs). + "404": + description: Job not found or foreign-tenant probe. + components: securitySchemes: bearerAuth: diff --git a/src/clients/n0d-client.ts b/src/clients/n0d-client.ts index f650a40..b95e8a7 100644 --- a/src/clients/n0d-client.ts +++ b/src/clients/n0d-client.ts @@ -10,9 +10,10 @@ * `src/api/mappers.ts` jobRecordToWire + `src/api/routes/job-lifecycle.ts`, * verified this session): * - * createJob -> POST {baseUrl}/v1/n0d/jobs (202 Accepted) - * getJob -> GET {baseUrl}/v1/n0d/jobs/:jobId - * getJobResult -> GET {baseUrl}/v1/n0d/jobs/:jobId/result + * createJob -> POST {baseUrl}/v1/n0d/jobs (202 Accepted) + * runDataQuality -> POST {baseUrl}/v1/n0d/quality/run (202 Accepted) + * getJob -> GET {baseUrl}/v1/n0d/jobs/:jobId + * getJobResult -> GET {baseUrl}/v1/n0d/jobs/:jobId/result * * Real-contract facts encoded here: * - required create fields: job_type, capability_id, tenant_id, @@ -159,8 +160,29 @@ export interface N0dJobResult { output_artifacts?: { st0re_artifact_id: string; artifact_type?: string }[]; } +/** Quality rule-set body for the N0D data-quality delegation (n0dRunDataQuality). */ +export interface RunDataQualityInput { + tenant_id: string; + dataset_id: string; + quality_rule_set_id: string; + project_id?: string; + app_id?: string; + requested_by?: string; + run_inline?: { + dataset_csv: string; + rules: Record[]; + }; + priority?: "low" | "normal" | "high"; + risk_level?: string; + idempotency_key?: string; + timeout_seconds?: number; + max_retries?: number; + callback_url?: string; +} + export interface N0dClient { createJob(input: N0dCreateJobInput, context?: SiaxRequestContext): Promise; + runDataQuality(input: RunDataQualityInput, context?: SiaxRequestContext): Promise; getJob(jobId: string, tenantId: string, context?: SiaxRequestContext): Promise; getJobResult(jobId: string, tenantId: string, context?: SiaxRequestContext): Promise; } @@ -228,6 +250,23 @@ class MockN0dClient implements N0dClient { await this.getJob(jobId, tenantId); throw new Error(`mock N0D: job ${jobId} has no result (mock jobs never run)`); } + + async runDataQuality(input: RunDataQualityInput): Promise { + const job: N0dJob = { + job_id: `mock_job_${crypto.randomUUID()}`, + job_type: "0pt.data_quality_job", + tenant_id: input.tenant_id, + project_id: input.project_id, + capability_id: "0pt.quality.run.v1", + app_id: input.app_id ?? input.requested_by ?? "0pt", + requested_by: input.requested_by ?? "svc_0pt", + resource_requirements: { cpu: 2, memory_gb: 4, gpu: 0, disk_gb: 10 }, + status: "queued", + created_at: new Date().toISOString(), + }; + this.jobs.set(job.job_id, job); + return job; + } } class HttpN0dClient implements N0dClient { @@ -267,6 +306,33 @@ class HttpN0dClient implements N0dClient { return (res.json ?? {}) as N0dJob; } + async runDataQuality(input: RunDataQualityInput, context?: SiaxRequestContext): Promise { + const res = await this.http.request({ + method: "POST", + path: "/v1/n0d/quality/run", + context, + headers: { "X-Tenant-Id": input.tenant_id }, + body: { + tenant_id: input.tenant_id, + project_id: input.project_id, + dataset_id: input.dataset_id, + quality_rule_set_id: input.quality_rule_set_id, + run_inline: input.run_inline, + app_id: input.app_id, + requested_by: input.requested_by, + priority: input.priority ?? "normal", + risk_level: input.risk_level ?? "L2", + requires_shared_backbone: false, + ...(input.idempotency_key ? { idempotency_key: input.idempotency_key } : {}), + ...(input.timeout_seconds ? { timeout_seconds: input.timeout_seconds } : {}), + ...(input.max_retries !== undefined ? { max_retries: input.max_retries } : {}), + ...(input.callback_url ? { callback_url: input.callback_url } : {}), + }, + acceptStatuses: [202], + }); + return (res.json ?? {}) as N0dJob; + } + async getJob(jobId: string, tenantId: string, context?: SiaxRequestContext): Promise { const res = await this.http.request({ method: "GET", diff --git a/src/quality/quality-service.ts b/src/quality/quality-service.ts index 26c0ac1..a73c018 100644 --- a/src/quality/quality-service.ts +++ b/src/quality/quality-service.ts @@ -164,22 +164,14 @@ export class QualityService { const allLocal = rules.every((r) => this.engine.supports(r.expectation.expectation_type)); if (!allLocal || input.rows === undefined) { - // DELEGATE to the GE backend (0pt-motor) via N0D — contract only here; - // the run stays "delegated" until results are posted back. NEVER faked. - const job = await this.n0d.createJob({ + // DELEGATE to the GE backend (0pt-motor) via N0D — owner-seam: + // n0dRunDataQuality (POST /v1/n0d/quality/run), NOT the generic + // createJob. N0D OWNS quality execution; 0pt only requests it. + const job = await this.n0d.runDataQuality({ tenant_id: input.tenant_id, project_id: input.project_id, - job_type: "0pt.quality.ge_run", - capability_id: "0pt.quality.run.v1", - requested_by_app_id: "0pt", - requested_by_actor_id: "svc_0pt", - output_artifact_type: "quality_result", - requires_backbone: false, - requires_gpu: false, - required_node_roles: ["compute_general"], - resource_profile: "standard", - risk_level: "L2", - correlation_id: input.correlation_id, + dataset_id: input.dataset_id, + quality_rule_set_id: input.dataset_id, idempotency_key: `dq:${input.dataset_id}:${input.dataset_version ?? "latest"}:${runId}`, }); const run: DataQualityRun = { diff --git a/tests/contracts/n0d-client.contract.test.ts b/tests/contracts/n0d-client.contract.test.ts index 6683074..834dd10 100644 --- a/tests/contracts/n0d-client.contract.test.ts +++ b/tests/contracts/n0d-client.contract.test.ts @@ -3,14 +3,18 @@ * contract (sax3l/n0d main `src/api/schemas.ts` createJobSchema + * `src/api/mappers.ts` + job-lifecycle routes, verified 2026-08-31): * - * POST /v1/n0d/jobs (202) — required: job_type, capability_id, tenant_id, - * required_role (enum), resource_requirements {cpu, memory_gb, disk_gb}; - * idempotency is BODY-ONLY (idempotency_key); callback_url = CloudEvents. - * GET /v1/n0d/jobs/:id, GET /v1/n0d/jobs/:id/result + * POST /v1/n0d/jobs (202) — generic job creation (n0dCreateJob) + * POST /v1/n0d/quality/run (202) — data-quality delegation (n0dRunDataQuality) + * GET /v1/n0d/jobs/:id + * GET /v1/n0d/jobs/:id/result * * The pre-2026-08-31 client sent required_node_roles[]/resource_profile/ * output_artifact_type/risk_level — fields N0D's createJobSchema does not * accept. This test pins the mapped wire shape. + * + * P-001 (owner-seam): 0pt delegates quality to N0D via n0dRunDataQuality + * (POST /v1/n0d/quality/run), NOT the generic createJob. N0D OWNS quality + * execution; 0pt only requests it. */ import { describe, it, expect, afterEach } from "vitest"; import { @@ -18,6 +22,7 @@ import { resolveRequiredRole, resolveResourceRequirements, type N0dCreateJobInput, + type RunDataQualityInput, } from "@/src/clients/n0d-client"; import { SiaxError } from "@/src/platform/errors"; import { startContractServer, type ContractServer } from "./contract-test-server"; @@ -150,4 +155,55 @@ describe("N0dClient wire contract", () => { expect((err as SiaxError).code).toBe("NOT_FOUND"); } }); + + it("P-001: runDataQuality POSTs to /v1/n0d/quality/run with dataset_id + quality_rule_set_id", async () => { + server = await startContractServer([ + { method: "POST", path: "/v1/n0d/quality/run", status: 202, respond: () => WIRE_JOB }, + ]); + const client = createN0dClient(server.baseUrl); + const runInput: RunDataQualityInput = { + tenant_id: "t_test", + dataset_id: "ds_1", + quality_rule_set_id: "rs_1", + project_id: "p_test", + run_inline: { dataset_csv: "a,b\n1,2", rules: [{ expectation: "expect_column_to_exist", column: "a" }] }, + idempotency_key: "dq:ds_1:latest:dqrun_", + }; + const job = await client.runDataQuality(runInput); + + const req = server.requests[0]; + expect(req.method).toBe("POST"); + expect(req.path).toBe("/v1/n0d/quality/run"); + expect(req.headers["x-tenant-id"]).toBe("t_test"); + const body = req.body as Record; + expect(body.tenant_id).toBe("t_test"); + expect(body.dataset_id).toBe("ds_1"); + expect(body.quality_rule_set_id).toBe("rs_1"); + expect(body.run_inline).toBeDefined(); + expect(body.job_type).toBeUndefined(); + expect(body.capability_id).toBeUndefined(); + expect(body.required_role).toBeUndefined(); + expect(body.resource_requirements).toBeUndefined(); + expect(body.idempotency_key).toBe("dq:ds_1:latest:dqrun_"); + expect(job.job_id).toBe("job_r3al"); + }); + + it("P-001: runDataQuality without run_inline is enqueue-only (no fabricated result)", async () => { + server = await startContractServer([ + { method: "POST", path: "/v1/n0d/quality/run", status: 202, respond: () => WIRE_JOB }, + ]); + const client = createN0dClient(server.baseUrl); + const job = await client.runDataQuality({ + tenant_id: "t_test", + dataset_id: "ds_2", + quality_rule_set_id: "rs_2", + }); + expect(job.status).toBe("queued"); + expect(job.job_id).toBe("job_r3al"); + // The wire body must NOT carry createJob-only fields and must NOT fabricate results. + const body = server.requests[0].body as Record; + expect(body.run_inline).toBeUndefined(); + expect(body.job_type).toBeUndefined(); + expect(body.capability_id).toBeUndefined(); + }); }); diff --git a/tests/p0/quality-service.test.ts b/tests/p0/quality-service.test.ts index c8e2fb9..49b3d2c 100644 --- a/tests/p0/quality-service.test.ts +++ b/tests/p0/quality-service.test.ts @@ -4,6 +4,9 @@ * GE delegation (never faked), cross-tenant isolation. */ import { describe, it, expect } from "vitest"; +import { readFileSync } from "node:fs"; +import { resolve, dirname } from "node:path"; +import { fileURLToPath } from "node:url"; import { QualityService } from "@/src/quality/quality-service"; import { ExpectationEngine } from "@/src/quality/expectation-engine"; import { SiaxError } from "@/src/platform/errors"; @@ -226,4 +229,19 @@ describe("QualityService — GE delegation (§11 DQ, never faked)", () => { expect(finished.results[0].executed_by).toBe("great_expectations"); await expect(q.recordDelegatedResults(run.run_id, "t_a", [])).rejects.toMatchObject({ code: "CONFLICT" }); }); + + it("P-002: n0dGetJobResult is registered in the 0pt OpenAPI catalog (0pt.n0dGetJobResult)", () => { + // Verify that the n0dGetJobResult operationId is present in the 0pt internal + // OpenAPI contract (contracts/openapi.yaml) — it is the P-001 owner-seam + // retrieval endpoint for delegated quality jobs. + const catalog = readFileSync( + resolve(dirname(fileURLToPath(import.meta.url)), "../../contracts/openapi.yaml"), + "utf8" + ); + expect(catalog).toContain("operationId: 0pt.n0dGetJobResult"); + expect(catalog).toContain("/api/0pt/n0d/results/{jobId}"); + expect(catalog).toContain("operationId: 0pt.quality.run"); + expect(catalog).toContain("operationId: 0pt.quality.checks"); + expect(catalog).toContain("operationId: 0pt.quality.reports"); + }); }); -- 2.54.0