refactor(callcenter): unify on AuditSink trait per OQ-7-2 (drop UnifiedAuditSink shim)

Sprint-7 meta-review CC-7-1 fix: UnifiedBridge::audit_sink was typed
Arc<dyn UnifiedAuditSink> (old D-SDR-4 placeholder trait) while
sprint-7 W6 production sinks impl Arc<dyn AuditSink>. W6 shipped
orphaned. Per OQ-7-2 decision (locked 2026-05-13 EPIPHANIES):
**full migrate, no adapter** — CLAUDE.md "no abstractions beyond
what the task requires".

Migration:
* crates/lance-graph-callcenter/src/audit_sink/mod.rs — add
  NoopAuditSink (Default impl AuditSink). Update doc comment so
  the canonical trait surface is here, not unified_audit.
* crates/lance-graph-callcenter/src/unified_audit.rs — drop
  UnifiedAuditSink trait + NoopUnifiedAuditSink struct. Leave a
  one-paragraph migration note. Doc-comment references updated.
* crates/lance-graph-callcenter/src/unified_bridge.rs — audit_sink
  field retyped to Arc<dyn AuditSink>. Constructor uses NoopAuditSink.
  with_audit_chain / with_audit_chain_resume builders take
  Arc<dyn AuditSink>. emit_audit() now `let _ = sink.emit(event);`
  (move + best-effort discard of Result — failures must not block
  the authorize hot path).
* crates/lance-graph-callcenter/src/lib.rs — drop the
  NoopUnifiedAuditSink / UnifiedAuditSink re-exports.
* crates/lance-graph-callcenter/src/cognitive_bridge_gate.rs —
  test RecordingSink retyped to impl AuditSink (3 methods).
* crates/lance-graph-consumer-conformance/src/harness.rs —
  RecordingSink retyped to impl AuditSink. Conformance imports
  audit_sink::{AuditSink, AuditError, MerkleRoot}.

OQ-7-3 also locked: new UnifiedBridge::with_jsonl_audit(super_domain,
salt, base_path) ergonomic constructor for MedCare-rs sprint-2 item
5's "JSONL primary + optional Lance projection" pattern. Default
remains NoopAuditSink (explicit opt-in to durable sinks — no silent
disk writes).

Verification:
* cargo clippy --workspace --tests --no-deps -- -D warnings exits 0
* cargo test -p lance-graph-callcenter -p lance-graph-consumer-conformance
  passes (all 8 conformance assertions + cognitive_bridge_gate +
  unified_audit tests, including noop_sink_swallows_events).
* Full workspace test running in background.

UnifiedAuditEvent::canonical_bytes byte layout unchanged (still 26
bytes) — emit() takes the event by value but doesn't change the wire
representation. W6 sinks' prev_merkle field is recorded outside
canonical_bytes (per W6 design).

Author: claude

crates/lance-graph-callcenter/src/audit_sink/mod.rs

@@ -1,13 +1,14 @@
-//! D-SDR-4b audit sink infrastructure.
+//! D-SDR-4b audit sink infrastructure — the canonical sink trait.
 //!
-//! Defines the `AuditSink` trait and `AuditError` enum used by both
-//! `LanceAuditSink` (columnar) and `JsonlAuditSink` (plain text) as
-//! production persistence sinks for `UnifiedAuditEvent`.
+//! Defines the `AuditSink` trait, `AuditError` enum, and `NoopAuditSink`
+//! used by `UnifiedBridge` and the production sinks (`LanceAuditSink`
+//! columnar, `JsonlAuditSink` plain text).
 //!
-//! The legacy `UnifiedAuditSink` in `unified_audit.rs` takes `&UnifiedAuditEvent`
-//! and returns `()`. This module introduces `AuditSink` as the D-SDR-4b
-//! production interface: `emit()` returns `Result<_, AuditError>`, and
-//! `flush()` + `checkpoint()` provide durability guarantees.
+//! Per OQ-7-2 (locked 2026-05-13): this is the ONLY audit sink trait.
+//! The earlier `UnifiedAuditSink` shim from D-SDR-4 was migrated to this
+//! interface in sprint-7. `emit()` returns `Result<_, AuditError>` and
+//! moves the event (not `&event`); `flush()` + `checkpoint()` provide
+//! durability guarantees that the legacy trait lacked.
 
 use crate::unified_audit::UnifiedAuditEvent;
 
@@ -54,6 +55,26 @@ pub trait AuditSink: Send + Sync {
     fn checkpoint(&self) -> Result<(), AuditError>;
 }
 
+/// No-op sink — discards every event. Default for `UnifiedBridge::new()`
+/// when `super_domain.audit_required = false` (no compliance regime requires
+/// audit), and for tests. Per OQ-7-3 (locked 2026-05-13): silent default;
+/// explicit opt-in to durable sinks via `UnifiedBridge::with_jsonl_audit()` /
+/// `with_audit_chain()`.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct NoopAuditSink;
+
+impl AuditSink for NoopAuditSink {
+    fn emit(&self, _event: UnifiedAuditEvent) -> Result<(), AuditError> {
+        Ok(())
+    }
+    fn flush(&self) -> Result<MerkleRoot, AuditError> {
+        Ok(0)
+    }
+    fn checkpoint(&self) -> Result<(), AuditError> {
+        Ok(())
+    }
+}
+
 pub mod composite;
 
 #[cfg(feature = "jsonl")]

crates/lance-graph-callcenter/src/cognitive_bridge_gate.rs

@@ -195,7 +195,8 @@ mod tests {
     use lance_graph_rbac::policy::smb_policy;
 
     use crate::unified_bridge::{TenantId, UnifiedBridge};
-    use crate::unified_audit::{UnifiedAuditEvent, UnifiedAuditSink};
+    use crate::audit_sink::{AuditError, AuditSink, MerkleRoot};
+    use crate::unified_audit::UnifiedAuditEvent;
     use crate::super_domain::SuperDomain as SD;
     use thinking_engine::bridge_gate::{CognitiveBridgeGate, CognitiveOpKind, CognitiveAuthResult};
 
@@ -230,10 +231,13 @@ mod tests {
     impl RecordingSink {
         fn count(&self) -> usize { self.events.lock().unwrap().len() }
     }
-    impl UnifiedAuditSink for RecordingSink {
-        fn emit(&self, event: &UnifiedAuditEvent) {
-            self.events.lock().unwrap().push(*event);
+    impl AuditSink for RecordingSink {
+        fn emit(&self, event: UnifiedAuditEvent) -> Result<(), AuditError> {
+            self.events.lock().unwrap().push(event);
+            Ok(())
         }
+        fn flush(&self) -> Result<MerkleRoot, AuditError> { Ok(0) }
+        fn checkpoint(&self) -> Result<(), AuditError> { Ok(()) }
     }
 
     // ── Chinese-wall tests ───────────────────────────────────────────────────

crates/lance-graph-callcenter/src/lib.rs

@@ -187,7 +187,7 @@ pub use cognitive_bridge_gate::UnifiedBridgeGate;
 pub mod unified_audit;
 pub use unified_audit::{
     verify_chain, AuditChain, AuditMerkleRoot, AuthDecision, AuthOp, HydrationRefreshAudit,
-    NoopUnifiedAuditSink, UnifiedAuditEvent, UnifiedAuditSink,
+    UnifiedAuditEvent,
 };
 
 // D-SDR-4b — Production audit sinks: LanceAuditSink (columnar) and

crates/lance-graph-callcenter/src/unified_audit.rs

@@ -3,7 +3,7 @@
 //!
 //! Each `UnifiedBridge::authorize()` call that materially gates access
 //! (Deny / Escalate / Audit-required Allow) emits one `UnifiedAuditEvent`
-//! through a `UnifiedAuditSink`. Events form a chain: the merkle root of
+//! through an `AuditSink` (see `crate::audit_sink`). Events form a chain: the merkle root of
 //! event N includes the merkle root of event N-1 plus a per-super-domain
 //! `merkle_salt` (§13.4 hard-lock — cross-domain audit logs are
 //! unlinkable). Tampering with any past event is detectable by chain
@@ -24,9 +24,9 @@
 //! durable record   (JSON Lines / Lance dataset / no-op)
 //! ```
 //!
-//! D-SDR-4 scope: type system + chain mechanics + `NoopUnifiedAuditSink`
+//! D-SDR-4 scope: type system + chain mechanics + sink trait
 //! reference impl + tamper detection helper. Production sinks
-//! (`JsonLinesUnifiedAuditSink`, `LanceUnifiedAuditSink`) are D-SDR-4b.
+//! (`JsonlAuditSink`, `LanceAuditSink` in `crate::audit_sink`) are D-SDR-4b/sprint-7.
 //! Wiring into `UnifiedBridge::authorize()` is D-SDR-5.
 //!
 //! ## Separate from `crate::audit`
@@ -301,34 +301,10 @@ impl HydrationRefreshAudit {
     }
 }
 
-// ═══════════════════════════════════════════════════════════════════════════
-// UnifiedAuditSink — pluggable persistence
-// ═══════════════════════════════════════════════════════════════════════════
-
-/// Sink trait — pluggable persistence backend. Implementations:
-/// - `NoopUnifiedAuditSink` (this module) — discards events; default for
-///   tests and `audit_required = false` policies.
-/// - `JsonLinesUnifiedAuditSink` (D-SDR-4b) — appends to a JSONL file.
-/// - `LanceUnifiedAuditSink` (D-SDR-4b) — appends to a Lance dataset,
-///   indexed by `(tenant, super_domain, ts_unix_ms)`.
-pub trait UnifiedAuditSink: Send + Sync
-{
-    /// Persist one event. **Must not block on I/O** for >1ms — the
-    /// authorize() hot path calls this synchronously. Production sinks
-    /// buffer asynchronously and flush on a separate task.
-    fn emit(&self, event: &UnifiedAuditEvent);
-}
-
-/// No-op sink — discards every event. Use as the default sink when
-/// `super_domain.audit_required = false` (no compliance regime requires
-/// audit), or in tests.
-#[derive(Clone, Copy, Debug, Default)]
-pub struct NoopUnifiedAuditSink;
-
-impl UnifiedAuditSink for NoopUnifiedAuditSink
-{
-    fn emit(&self, _event: &UnifiedAuditEvent) {}
-}
+// The audit sink trait + NoopAuditSink moved to crate::audit_sink in
+// sprint-7 (OQ-7-2 locked 2026-05-13). UnifiedAuditSink and
+// NoopUnifiedAuditSink were the D-SDR-4 placeholders; production
+// consumers use `AuditSink` from `crate::audit_sink` directly.
 
 // ═══════════════════════════════════════════════════════════════════════════
 // Chain verification — tamper detection
@@ -494,12 +470,12 @@ mod tests
     }
 
     #[test]
-    fn noop_sink_swallows_events()
-    {
-        let sink = NoopUnifiedAuditSink;
+    fn noop_sink_swallows_events() {
+        use crate::audit_sink::{AuditSink, NoopAuditSink};
+        let sink = NoopAuditSink;
         let mut chain = AuditChain::new(SuperDomain::Healthcare, 0xC0FFEE);
         let ev = chain.advance(fresh_event());
-        sink.emit(&ev); // doesn't panic, doesn't observe — by design
+        sink.emit(ev).expect("noop never errors"); // doesn't observe — by design
     }
 
     #[test]

crates/lance-graph-callcenter/src/unified_bridge.rs

@@ -47,9 +47,9 @@ use lance_graph_rbac::policy::{Operation, Policy};
 
 use crate::super_domain::SuperDomain;
 use crate::unified_audit::{
-    AuditChain, AuditMerkleRoot, AuthDecision, AuthOp, NoopUnifiedAuditSink, UnifiedAuditEvent,
-    UnifiedAuditSink,
+    AuditChain, AuditMerkleRoot, AuthDecision, AuthOp, UnifiedAuditEvent,
 };
+use crate::audit_sink::{AuditSink, NoopAuditSink};
 
 /// Extract the canonical ontology entity type name from a resolved
 /// [`MappingRow`], for use as the [`Policy::evaluate`] key.
@@ -249,9 +249,9 @@ pub struct UnifiedBridge<B: NamespaceBridge> {
     tenant: TenantId,
     /// Audit sink — every `authorize_*` call that reaches the policy
     /// evaluation step emits one `UnifiedAuditEvent`. Default is
-    /// `NoopUnifiedAuditSink` (zero overhead, no persistence). Swap via
+    /// `NoopAuditSink` (zero overhead, no persistence). Swap via
     /// [`Self::with_audit_chain`].
-    audit_sink: Arc<dyn UnifiedAuditSink>,
+    audit_sink: Arc<dyn AuditSink>,
     /// Merkle-chained audit advancer. Holds the prior event's root +
     /// per-super-domain salt so each new event chains off it. Mutex
     /// guards the `last_root` advance under concurrent `authorize_*`
@@ -262,7 +262,7 @@ pub struct UnifiedBridge<B: NamespaceBridge> {
 impl<B: NamespaceBridge> UnifiedBridge<B> {
     /// Construct a new unified bridge.
     ///
-    /// Defaults audit to `NoopUnifiedAuditSink` + a chain anchored at
+    /// Defaults audit to `NoopAuditSink` + a chain anchored at
     /// `SuperDomain::Unknown` with salt 0. Call
     /// [`Self::with_audit_chain`] to swap in a real sink + the
     /// super-domain-specific salt before authorization traffic starts.
@@ -278,20 +278,20 @@ impl<B: NamespaceBridge> UnifiedBridge<B> {
             actor_role,
             actor_role_hash: fnv1a_str(actor_role),
             tenant,
-            audit_sink: Arc::new(NoopUnifiedAuditSink),
+            audit_sink: Arc::new(NoopAuditSink),
             audit_chain: Mutex::new(AuditChain::new(SuperDomain::Unknown, 0)),
         }
     }
 
-    /// Builder: swap in a real `UnifiedAuditSink` + the super-domain's
+    /// Builder: swap in a real `AuditSink` + the super-domain's
     /// `merkle_salt` (§13.4 — cross-domain audit logs unlinkable). Resets
     /// the chain to GENESIS; pass a resume root via
     /// [`Self::with_audit_chain_resume`] if continuing a persisted chain.
     pub fn with_audit_chain(
         mut self,
         super_domain: SuperDomain,
         salt: u64,
-        sink: Arc<dyn UnifiedAuditSink>,
+        sink: Arc<dyn AuditSink>,
     ) -> Self {
         self.audit_sink = sink;
         self.audit_chain = Mutex::new(AuditChain::new(super_domain, salt));
@@ -306,13 +306,30 @@ impl<B: NamespaceBridge> UnifiedBridge<B> {
         super_domain: SuperDomain,
         salt: u64,
         last_root: AuditMerkleRoot,
-        sink: Arc<dyn UnifiedAuditSink>,
+        sink: Arc<dyn AuditSink>,
     ) -> Self {
         self.audit_sink = sink;
         self.audit_chain = Mutex::new(AuditChain::resume(super_domain, salt, last_root));
         self
     }
 
+    /// Ergonomic constructor: wire a `JsonlAuditSink` at `base_path` as
+    /// the primary audit destination. Per OQ-7-3 (locked 2026-05-13):
+    /// `new()` defaults to `NoopAuditSink`; this constructor is the
+    /// explicit opt-in for the production "JSONL primary + optional Lance
+    /// projection" pattern (MedCare-rs sprint-2 item 5). Only available
+    /// when the `jsonl` feature is enabled.
+    #[cfg(feature = "jsonl")]
+    pub fn with_jsonl_audit(
+        self,
+        super_domain: SuperDomain,
+        salt: u64,
+        base_path: impl Into<std::path::PathBuf>,
+    ) -> std::io::Result<Self> {
+        let sink = Arc::new(crate::audit_sink::JsonlAuditSink::new(base_path.into())?);
+        Ok(self.with_audit_chain(super_domain, salt, sink))
+    }
+
     /// Returns the underlying namespace bridge.
     pub fn bridge(&self) -> &B {
         &self.bridge
@@ -350,7 +367,7 @@ impl<B: NamespaceBridge> UnifiedBridge<B> {
     /// from policy authorship.
     ///
     /// On policy evaluation reaching, one `UnifiedAuditEvent` is emitted
-    /// through the configured `UnifiedAuditSink` carrying tenant +
+    /// through the configured `AuditSink` carrying tenant +
     /// super-domain + owl + decision. **`BridgeError` short-circuits
     /// before audit** — bad input names aren't auth decisions, they're
     /// invalid requests (D-SDR-5 minimum; revisit if probing detection
@@ -428,7 +445,10 @@ impl<B: NamespaceBridge> UnifiedBridge<B> {
         };
         let stamped = chain.advance(event);
         drop(chain);
-        self.audit_sink.emit(&stamped);
+        // Best-effort: audit emission failures must not block the authorize
+        // hot path. Sinks are responsible for their own buffering/backpressure
+        // (see audit_sink::{JsonlAuditSink, LanceAuditSink} BestEffort mode).
+        let _ = self.audit_sink.emit(stamped);
     }
 }
 
@@ -821,9 +841,16 @@ mod tests {
         }
     }
 
-    impl UnifiedAuditSink for RecordingSink {
-        fn emit(&self, event: &UnifiedAuditEvent) {
-            self.events.lock().unwrap().push(*event);
+    impl AuditSink for RecordingSink {
+        fn emit(&self, event: UnifiedAuditEvent) -> Result<(), crate::audit_sink::AuditError> {
+            self.events.lock().unwrap().push(event);
+            Ok(())
+        }
+        fn flush(&self) -> Result<crate::audit_sink::MerkleRoot, crate::audit_sink::AuditError> {
+            Ok(0)
+        }
+        fn checkpoint(&self) -> Result<(), crate::audit_sink::AuditError> {
+            Ok(())
         }
     }
 

crates/lance-graph-consumer-conformance/src/harness.rs

@@ -22,9 +22,8 @@
 use std::sync::Arc;
 
 use lance_graph_callcenter::super_domain::SuperDomain;
-use lance_graph_callcenter::unified_audit::{
-    AuditMerkleRoot, UnifiedAuditEvent, UnifiedAuditSink,
-};
+use lance_graph_callcenter::audit_sink::{AuditError, AuditSink, MerkleRoot};
+use lance_graph_callcenter::unified_audit::{AuditMerkleRoot, UnifiedAuditEvent};
 use lance_graph_callcenter::unified_bridge::{TenantId, UnifiedBridge};
 use lance_graph_contract::hash::fnv1a_str;
 use lance_graph_contract::property::PrefetchDepth;
@@ -59,9 +58,16 @@ impl RecordingSink {
     }
 }
 
-impl UnifiedAuditSink for RecordingSink {
-    fn emit(&self, event: &UnifiedAuditEvent) {
-        self.events.lock().unwrap().push(*event);
+impl AuditSink for RecordingSink {
+    fn emit(&self, event: UnifiedAuditEvent) -> Result<(), AuditError> {
+        self.events.lock().unwrap().push(event);
+        Ok(())
+    }
+    fn flush(&self) -> Result<MerkleRoot, AuditError> {
+        Ok(0)
+    }
+    fn checkpoint(&self) -> Result<(), AuditError> {
+        Ok(())
     }
 }