feat(lineage): route production evidence through durable AUD0 outbox

src/lineage/lineage-emitter.ts

@@ -1,24 +1,18 @@
 /**
- * src/lineage/lineage-emitter.ts
- * ===============================
- * 0pt's OpenLineage emitter (Master-DoD §10 Lineage). One narrow API for
- * capability services to record "this run read X and produced Y" against
- * AUD0's real OpenLineage ingest (`POST /v1/aud0/openlineage` — see
- * src/clients/aud0-client.ts for the wire).
+ * 0pt OpenLineage emitter.
  *
- * Design points:
- *   - Deterministic runIds (caller-supplied), so an idempotent retry of a
- *     capability re-emits the SAME event and AUD0 dedups it instead of
- *     recording a phantom second run.
- *   - Canonical SIAX resource refs (`st0re:artifact:...`, `0pt:dataset:...`)
- *     go in dataset facets so evidence chains can be joined across products
- *     (§16 "Evidence chains: Link source → transform → model → decision").
- *   - Emission is evidence of what already happened — a lineage POST failure
- *     is surfaced to the caller as a warning value, never an exception that
- *     rolls back the completed work. Callers that need hard evidence
- *     guarantees must use an outbox (tracked honestly in STATUS.md).
+ * AUD0 is canonical evidence. When 0pt has a repository DATABASE_URL the
+ * emitter first persists the exact RunEvent to the local AUD0 outbox and
+ * returns `queued:true`; the dispatcher is then the only delivery path. This
+ * preserves the historic never-throw-after-domain-work contract while removing
+ * AUD0 availability as an evidence-loss window. In env-less/unit runtimes the
+ * emitter keeps the direct client behavior.
  */
 import type { Aud0Client } from "../clients/aud0-client";
+import {
+  createAud0OutboxRepositoryFromEnv,
+  type Aud0OutboxRepository,
+} from "../evidence/aud0-outbox";
 import {
   OPT_OPENLINEAGE_PRODUCER,
   type OpenLineageDatasetRef,
@@ -29,35 +23,30 @@ import {
 export interface LineageEmitInput {
   tenant_id: string;
   project_id?: string;
-  /** The 0pt capability that ran, e.g. "0pt.quality.run.v1". Becomes job.name. */
   capability_id: string;
-  /** Deterministic run id (e.g. the run/version/export id being recorded). */
   run_id: string;
   event_type: OpenLineageEventType;
-  /** ISO timestamp; defaults to now. */
   event_time?: string;
-  /** Canonical SIAX refs and/or plain names consumed by the run. */
   inputs?: LineageDatasetInput[];
-  /** Canonical SIAX refs and/or plain names produced by the run. */
   outputs?: LineageDatasetInput[];
   correlation_id?: string;
-  /** Extra run facets (namespaced siax*). */
   run_facets?: Record<string, unknown>;
 }
 
 export interface LineageDatasetInput {
-  /** Human/system dataset name, e.g. "dataset/ds_x" or "opensearch/opt-companies__g2". */
   name: string;
-  /** Optional canonical SIAX resource ref for cross-product joining. */
   ref?: string;
   namespace?: string;
   facets?: Record<string, unknown>;
 }
 
 export interface LineageEmitResult {
+  /** True only when AUD0 acknowledged the event synchronously. */
   emitted: boolean;
+  /** True when the exact event is durably persisted for at-least-once delivery. */
+  queued: boolean;
   run_id: string;
-  /** Present when emission failed (AUD0 unreachable etc.). */
+  outbox_id?: string;
   warning?: string;
 }
 
@@ -74,8 +63,21 @@ function toDatasetRef(d: LineageDatasetInput): OpenLineageDatasetRef {
   };
 }
 
+function safeWarning(error: unknown): string {
+  return error instanceof Error ? error.name : "UnknownError";
+}
+
 export class LineageEmitter {
-  constructor(private readonly aud0: Aud0Client) {}
+  private readonly outbox?: Aud0OutboxRepository;
+
+  constructor(
+    private readonly aud0: Aud0Client,
+    opts?: { outbox?: Aud0OutboxRepository | null },
+  ) {
+    this.outbox = opts && "outbox" in opts
+      ? opts.outbox ?? undefined
+      : createAud0OutboxRepositoryFromEnv();
+  }
 
   buildRunEvent(input: LineageEmitInput): OpenLineageRunEvent {
     return {
@@ -96,17 +98,56 @@ export class LineageEmitter {
     };
   }
 
-  /** Emit one run event. Never throws — see module doc comment. */
+  /**
+   * Record one run event without throwing after domain work has completed.
+   *
+   * With an outbox: enqueue only; the dispatcher owns delivery and retry.
+   * Without an outbox: direct AUD0 call for local/tests/backwards compatibility.
+   * If durable enqueue itself fails, a direct AUD0 attempt is still made so an
+   * available AUD0 can preserve evidence; the warning makes the durability
+   * degradation explicit rather than silently green.
+   */
   async emit(input: LineageEmitInput): Promise<LineageEmitResult> {
     const event = this.buildRunEvent(input);
+
+    if (this.outbox) {
+      try {
+        const row = await this.outbox.enqueueOpenLineage(input.tenant_id, APP_ID, event);
+        return {
+          emitted: row.status === "delivered",
+          queued: row.status === "pending",
+          run_id: input.run_id,
+          outbox_id: row.outbox_id,
+        };
+      } catch (queueError) {
+        try {
+          await this.aud0.ingestOpenLineageEvent(input.tenant_id, APP_ID, event);
+          return {
+            emitted: true,
+            queued: false,
+            run_id: input.run_id,
+            warning: `durable outbox enqueue failed (${safeWarning(queueError)}); AUD0 accepted direct fallback`,
+          };
+        } catch (aud0Error) {
+          return {
+            emitted: false,
+            queued: false,
+            run_id: input.run_id,
+            warning: `durable outbox enqueue failed (${safeWarning(queueError)}); direct AUD0 delivery failed (${safeWarning(aud0Error)})`,
+          };
+        }
+      }
+    }
+
     try {
       await this.aud0.ingestOpenLineageEvent(input.tenant_id, APP_ID, event);
-      return { emitted: true, run_id: input.run_id };
-    } catch (err) {
+      return { emitted: true, queued: false, run_id: input.run_id };
+    } catch (error) {
       return {
         emitted: false,
+        queued: false,
         run_id: input.run_id,
-        warning: `openlineage emission failed: ${err instanceof Error ? err.message : String(err)}`,
+        warning: `openlineage emission failed: ${safeWarning(error)}`,
       };
     }
   }