diff --git a/src/clients/aud0-client.ts b/src/clients/aud0-client.ts index 60e510f..025eab7 100644 --- a/src/clients/aud0-client.ts +++ b/src/clients/aud0-client.ts @@ -1,41 +1,13 @@ /** - * src/clients/aud0-client.ts - * =========================== - * 0pt's ONLY path for recording "what happened" to AUD0, the SIAX Cloud - * append-only system of record. + * 0pt's sole AUD0 producer boundary. * - * REWRITTEN 2026-08-31 against AUD0's REAL wire contract (sax3l/aud0 branch - * feat/siax-platforms-p0-evidence-os, `src/api/routes.ts` + - * `src/schemas/event.ts` + `src/api/routes/openlineage.ts`, verified this - * session): - * - * ingestEvent -> POST {baseUrl}/v1/aud0/events (snake_case) - * ingestOpenLineageEvent-> POST {baseUrl}/v1/aud0/openlineage?tenant_id=&app_id= - * ingestLineageEvent -> LEGACY SHIM: converts the historical 0pt lineage - * shape to an OpenLineage RunEvent and posts it to - * the endpoint above. - * - * The path this client used to post lineage to — `/v1/aud0/lineage/events` — - * DOES NOT EXIST on AUD0 and never has. AUD0's only lineage ingest is the - * OpenLineage endpoint; `/v1/aud0/lineage/*` routes are read-only projections - * of the generic event stream. Master-DoD §16 requires OpenLineage run/job/ - * dataset ingest — the emitter in src/lineage/ produces spec-shaped - * RunEvents; this client is just the wire. - * - * The audit-event contract additionally carries the P0 evidence-context - * fields (§16: purpose, policy decision, correlation, causation, trace, - * result, references), all optional on the wire; they are surfaced on - * `Aud0Event` so services can populate them. - * - * Production fallback policy (Master-DoD §3/§10): no base URL in a - * production runtime throws `SiaxError("DEPENDENCY_UNAVAILABLE")` at - * construction. The no-op client survives only for non-production runtimes. - * NOTE the deliberate asymmetry: audit emission FAILURE at runtime (AUD0 - * down) still does not roll back the already-performed action — an audit - * emission records what already happened; the durability requirement is - * §70's "durable local/outbox path", tracked honestly in STATUS.md as a gap. + * Default clients are durable: when DATABASE_URL is configured, evidence is + * first written to opt_aud0_evidence_outbox and later delivered by the + * dispatcher. `durable:false` exists only for that dispatcher, which must + * perform the real HTTP delivery rather than enqueue recursively. */ import { SiaxHttp } from "./http"; +import { createAud0OutboxRepositoryFromEnv, type Aud0OutboxRepository } from "../evidence/aud0-outbox"; import { resolveDependencyBaseUrl } from "../platform/dependency-mode"; import type { SiaxRequestContext } from "../platform/siax-context"; import type { OpenLineageRunEvent } from "../lineage/openlineage"; @@ -50,30 +22,21 @@ export interface Aud0Event { capability_id: string; resource_type: string; resource_id: string; - /** Risk level L0–L5 per the SIAX Cloud risk taxonomy. */ risk_level: string; payload?: unknown; created_at: string; correlation_id?: string; - // P0 evidence-context fields (Master-DoD §16), optional on AUD0's wire. - /** Business time, when it differs from ingestion time. */ occurred_at?: string; - /** GDPR/purpose-limitation tag. */ purpose?: string; - /** CL0UD policy decision id/reason for the action being recorded. */ policy_decision?: string; causation_id?: string; trace_id?: string; - /** Outcome, e.g. "success" | "failure". */ result?: string; - /** Canonical resource refs consumed by the action (e.g. st0re:artifact:...). */ input_refs?: string[]; - /** Canonical resource refs produced by the action. */ output_refs?: string[]; idempotency_key?: string; } -/** Historical 0pt lineage shape — kept as the input to the legacy shim. */ export interface Aud0LineageEvent { lineage_event_id: string; tenant_id: string; @@ -84,19 +47,15 @@ export interface Aud0LineageEvent { source_id: string; target_type: string; target_id: string; - /** e.g. "produced" | "derived_from" | "stored_as". */ lineage_type: string; metadata?: unknown; created_at: string; correlation_id?: string; } -/** Alias so both historical import names resolve to the same shape. */ export type Aud0EventInput = Aud0Event; -/** Alias so both historical import names resolve to the same shape. */ export type Aud0LineageEventInput = Aud0LineageEvent; -/** AUD0's OpenLineage ingest acknowledgement. */ export interface Aud0OpenLineageAck { tenant_id: string; run_id: string; @@ -112,24 +71,28 @@ export interface Aud0Client { tenantId: string, appId: string, event: OpenLineageRunEvent, - context?: SiaxRequestContext + context?: SiaxRequestContext, ): Promise; - /** @deprecated Legacy shim — converts to OpenLineage and posts to /v1/aud0/openlineage. */ ingestLineageEvent(event: Aud0LineageEvent, context?: SiaxRequestContext): Promise; } -export function createAud0Client(baseUrl?: string, opts?: { token?: string }): Aud0Client { +export interface CreateAud0ClientOptions { + token?: string; + /** False only for the outbox dispatcher/raw diagnostics. Default true. */ + durable?: boolean; +} + +export function createAud0Client(baseUrl?: string, opts?: CreateAud0ClientOptions): Aud0Client { const resolved = resolveDependencyBaseUrl("aud0", baseUrl ?? process.env.AUD0_URL); - if (!resolved) return new NoopAud0Client(); - return new HttpAud0Client(resolved, opts?.token ?? process.env.AUD0_SERVICE_TOKEN); + const raw: Aud0Client = resolved + ? new HttpAud0Client(resolved, opts?.token ?? process.env.AUD0_SERVICE_TOKEN) + : new NoopAud0Client(); + + if (opts?.durable === false) return raw; + const outbox = createAud0OutboxRepositoryFromEnv(); + return outbox ? new DurableAud0Client(outbox) : raw; } -/** - * Convert the historical 0pt lineage shape into a spec-shaped OpenLineage - * RunEvent (AUD0's only lineage ingest contract). Deterministic: the runId is - * the legacy lineage_event_id, so idempotent re-emission dedups on AUD0's - * `ol:::` key. - */ export function legacyLineageToOpenLineage(event: Aud0LineageEvent): OpenLineageRunEvent { return { eventType: "COMPLETE", @@ -138,9 +101,7 @@ export function legacyLineageToOpenLineage(event: Aud0LineageEvent): OpenLineage run: { runId: event.lineage_event_id, facets: { - ...(event.correlation_id - ? { siaxCorrelation: { correlation_id: event.correlation_id } } - : {}), + ...(event.correlation_id ? { siaxCorrelation: { correlation_id: event.correlation_id } } : {}), siaxLegacyLineage: { lineage_type: event.lineage_type, capability_id: event.capability_id, @@ -155,15 +116,44 @@ export function legacyLineageToOpenLineage(event: Aud0LineageEvent): OpenLineage }; } -/** Non-production runtimes only: audit emission is a no-op echo. */ +class DurableAud0Client implements Aud0Client { + constructor(private readonly outbox: Aud0OutboxRepository) {} + + async ingestEvent(event: Aud0Event): Promise { + await this.outbox.enqueueEvent(event); + return event; + } + + async ingestOpenLineageEvent( + tenantId: string, + appId: string, + event: OpenLineageRunEvent, + ): Promise { + const row = await this.outbox.enqueueOpenLineage(tenantId, appId, event); + return { + tenant_id: tenantId, + run_id: event.run.runId, + job: `${event.job.namespace}/${event.job.name}`, + event_id: `outbox:${row.outbox_id}`, + }; + } + + async ingestLineageEvent(event: Aud0LineageEvent): Promise { + await this.ingestOpenLineageEvent(event.tenant_id, event.app_id, legacyLineageToOpenLineage(event)); + return event; + } +} + +/** Non-production/env-less runtime fallback only. */ class NoopAud0Client implements Aud0Client { async ingestEvent(event: Aud0Event): Promise { return event; } + async ingestOpenLineageEvent( tenantId: string, _appId: string, - event: OpenLineageRunEvent + event: OpenLineageRunEvent, ): Promise { return { tenant_id: tenantId, @@ -172,6 +162,7 @@ class NoopAud0Client implements Aud0Client { event_id: `noop_${event.run.runId}`, }; } + async ingestLineageEvent(event: Aud0LineageEvent): Promise { return event; } @@ -191,10 +182,8 @@ class HttpAud0Client implements Aud0Client { context, headers: { "X-Tenant-Id": event.tenant_id, "X-Actor-Id": event.actor_id }, body: event, - // 200 = idempotent replay (deduplicated: true) — a success. acceptStatuses: [200], }); - // Prefer AUD0's canonical stored record; fall back to what we sent. return (res.json ?? event) as Aud0Event; } @@ -202,13 +191,11 @@ class HttpAud0Client implements Aud0Client { tenantId: string, appId: string, event: OpenLineageRunEvent, - context?: SiaxRequestContext + context?: SiaxRequestContext, ): Promise { const res = await this.http.request({ method: "POST", path: "/v1/aud0/openlineage", - // AUD0's OpenLineage ingest takes the tenant as a QUERY PARAM (verified - // against src/api/routes/openlineage.ts), not a body field. query: { tenant_id: tenantId, app_id: appId }, context, headers: { "X-Tenant-Id": tenantId, "X-Actor-Id": appId },