feat(evidence): make standard AUD0 client durable by default
This commit is contained in:
+55
-68
@@ -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<Aud0OpenLineageAck>;
|
||||
/** @deprecated Legacy shim — converts to OpenLineage and posts to /v1/aud0/openlineage. */
|
||||
ingestLineageEvent(event: Aud0LineageEvent, context?: SiaxRequestContext): Promise<Aud0LineageEvent>;
|
||||
}
|
||||
|
||||
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:<runId>:<eventType>:<eventTime>` 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<Aud0Event> {
|
||||
await this.outbox.enqueueEvent(event);
|
||||
return event;
|
||||
}
|
||||
|
||||
async ingestOpenLineageEvent(
|
||||
tenantId: string,
|
||||
appId: string,
|
||||
event: OpenLineageRunEvent,
|
||||
): Promise<Aud0OpenLineageAck> {
|
||||
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<Aud0LineageEvent> {
|
||||
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<Aud0Event> {
|
||||
return event;
|
||||
}
|
||||
|
||||
async ingestOpenLineageEvent(
|
||||
tenantId: string,
|
||||
_appId: string,
|
||||
event: OpenLineageRunEvent
|
||||
event: OpenLineageRunEvent,
|
||||
): Promise<Aud0OpenLineageAck> {
|
||||
return {
|
||||
tenant_id: tenantId,
|
||||
@@ -172,6 +162,7 @@ class NoopAud0Client implements Aud0Client {
|
||||
event_id: `noop_${event.run.runId}`,
|
||||
};
|
||||
}
|
||||
|
||||
async ingestLineageEvent(event: Aud0LineageEvent): Promise<Aud0LineageEvent> {
|
||||
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<Aud0OpenLineageAck> {
|
||||
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 },
|
||||
|
||||
Reference in New Issue
Block a user