feat(persistence): live SQL transport and DATABASE_URL gate

CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to TEPP are documented here. The format follows Keep a Chang
 - `relation_graph` forward-only state-transition DAG with past-pointing provenance edges and cycle rejection.
 - `tepp_simulation` deterministic truth-corpus generator with delayed reporting, multilevel memberships, method-effect variants, relation noise, and digest-bound truth manifests.
 - `corpus_split` leakage-safe knowledge-cutoff snapshots, relation-connected co-partition groups, rolling-origin windows, and group-normalized ESS weight contracts.
+- `persistence_postgres` live SQL port: `SqlSession` transport, migration batch applicator, document/audit SQL contracts, `LiveDocumentRepository`, and fail-closed `DATABASE_URL`/`LiveSqlxConfig` gate for SQLx pool wiring (live pool/query driver remains accepted-target).
 - `persistence_postgres` bitemporal foundation: multi-word migration contracts, knowledge-cutoff eligibility, and in-memory as-known-at / as-valid-at document replay (live SQLx/PostgreSQL execution remains accepted-target).
 - `event_core` mention/instance separation with explicit promotion, typed roles, event-time validity, and fail-closed mention-as-instance refusal.
 - `membership_core` time-varying weighted multiple-membership network with contextual roles, event-time validity, and atomistic-fallacy prevention contracts.

crates/persistence_postgres/src/document_sql.rs

@@ -0,0 +1,215 @@
+//! Parameterized SQL contracts for bitemporal document rows.
+
+use crate::PersistenceError;
+use crate::document_store::{AuditEvent, DocumentRecord};
+
+/// Validate digest and render an insert for an open document version.
+///
+/// The rendered statement binds values as quoted `RFC 3339` / UUID literals so a
+/// live `SQLx` transport can execute the same text contract after server-side
+/// parameterization is layered on.
+///
+/// # Errors
+///
+/// Returns [`PersistenceError::InvalidContentDigest`] when the digest is not a
+/// 64-character hexadecimal `SHA-256` string.
+pub fn insert_document_sql(record: &DocumentRecord) -> Result<String, PersistenceError> {
+    validate_digest(&record.content_digest)?;
+    Ok(format!(
+        "INSERT INTO document_record (\
+            document_record_id, tenant_record_id, source_artifact_id, content_sha256, \
+            language_profile_code, assertion_time, document_time, valid_from, valid_to, \
+            system_from, system_to, available_time, revision_number\
+        ) VALUES (\
+            '{document_id}'::uuid, '{tenant_id}'::uuid, '{document_id}'::uuid, '{digest}', \
+            'und', NULL, NULL, '{valid_from}'::timestamptz, {valid_to}, \
+            '{system_from}'::timestamptz, NULL, '{available}'::timestamptz, {revision}\
+        )",
+        document_id = record.document_record_id,
+        tenant_id = record.tenant_record_id,
+        digest = record.content_digest,
+        valid_from = record.valid_from.to_rfc3339(),
+        valid_to = optional_timestamptz(record.valid_to.map(temporal_core::EventTime::to_rfc3339)),
+        system_from = record.system_from.to_rfc3339(),
+        available = record.available_time.to_rfc3339(),
+        revision = record.revision_number,
+    ))
+}
+
+/// Render the close + insert pair used when revising a document identity.
+///
+/// # Errors
+///
+/// Returns digest validation failures for the revised row.
+pub fn revise_document_sqls(record: &DocumentRecord) -> Result<[String; 2], PersistenceError> {
+    validate_digest(&record.content_digest)?;
+    let close = format!(
+        "UPDATE document_record SET system_to = '{system_from}'::timestamptz \
+         WHERE document_record_id = '{document_id}'::uuid AND system_to IS NULL",
+        system_from = record.system_from.to_rfc3339(),
+        document_id = record.document_record_id,
+    );
+    let insert = insert_document_sql(record)?;
+    Ok([close, insert])
+}
+
+/// Render as-known-at selection for one document identity.
+#[must_use]
+pub fn as_known_at_sql(document_record_id: uuid::Uuid, known_at_rfc3339: &str) -> String {
+    format!(
+        "SELECT document_record_id, tenant_record_id, content_sha256, available_time, \
+                valid_from, valid_to, system_from, system_to, revision_number \
+         FROM document_record \
+         WHERE document_record_id = '{document_record_id}'::uuid \
+           AND system_from <= '{known_at_rfc3339}'::timestamptz \
+           AND (system_to IS NULL OR '{known_at_rfc3339}'::timestamptz < system_to) \
+         ORDER BY revision_number DESC \
+         LIMIT 1"
+    )
+}
+
+/// Render as-valid-at selection under a system-time as-of.
+#[must_use]
+pub fn as_valid_at_sql(
+    document_record_id: uuid::Uuid,
+    valid_at_rfc3339: &str,
+    known_at_rfc3339: &str,
+) -> String {
+    format!(
+        "SELECT document_record_id, tenant_record_id, content_sha256, available_time, \
+                valid_from, valid_to, system_from, system_to, revision_number \
+         FROM document_record \
+         WHERE document_record_id = '{document_record_id}'::uuid \
+           AND system_from <= '{known_at_rfc3339}'::timestamptz \
+           AND (system_to IS NULL OR '{known_at_rfc3339}'::timestamptz < system_to) \
+           AND valid_from <= '{valid_at_rfc3339}'::timestamptz \
+           AND (valid_to IS NULL OR '{valid_at_rfc3339}'::timestamptz < valid_to) \
+         ORDER BY revision_number DESC \
+         LIMIT 1"
+    )
+}
+
+/// Render append-only audit insert.
+#[must_use]
+pub fn append_audit_sql(event: &AuditEvent) -> String {
+    format!(
+        "INSERT INTO audit_event (\
+            audit_event_id, tenant_record_id, action_code, subject_record_id, recorded_system_time\
+        ) VALUES (\
+            '{audit_id}'::uuid, '{tenant_id}'::uuid, '{action}', '{subject}'::uuid, \
+            '{recorded}'::timestamptz\
+        )",
+        audit_id = event.audit_event_id,
+        tenant_id = event.tenant_record_id,
+        action = escape_literal(&event.action_code),
+        subject = event.subject_record_id,
+        recorded = event.recorded_system_time.to_rfc3339(),
+    )
+}
+
+fn optional_timestamptz(value: Option<String>) -> String {
+    match value {
+        Some(stamp) => format!("'{stamp}'::timestamptz"),
+        None => "NULL".to_owned(),
+    }
+}
+
+fn escape_literal(value: &str) -> String {
+    value.replace('\'', "''")
+}
+
+fn validate_digest(digest: &str) -> Result<(), PersistenceError> {
+    let length_ok = digest.len() == 64;
+    let hex_ok = digest.chars().all(|ch| ch.is_ascii_hexdigit());
+    if length_ok & hex_ok {
+        Ok(())
+    } else {
+        Err(PersistenceError::InvalidContentDigest)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        append_audit_sql, as_known_at_sql, as_valid_at_sql, escape_literal, insert_document_sql,
+        optional_timestamptz, revise_document_sqls, validate_digest,
+    };
+    use crate::PersistenceError;
+    use crate::document_store::{AuditEvent, DocumentRecord};
+    use temporal_core::{AvailableTime, EventTime, SystemTime};
+
+    fn sample_record() -> DocumentRecord {
+        DocumentRecord {
+            document_record_id: uuid::Uuid::nil(),
+            tenant_record_id: uuid::Uuid::nil(),
+            content_digest: "ab".repeat(32),
+            available_time: AvailableTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("a"),
+            valid_from: EventTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("v"),
+            valid_to: None,
+            system_from: SystemTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("s"),
+            system_to: None,
+            revision_number: 1,
+        }
+    }
+
+    #[test]
+    fn document_sql_covers_insert_revise_and_queries() {
+        let record = sample_record();
+        let insert = insert_document_sql(&record).expect("insert");
+        assert!(insert.contains("INSERT INTO document_record"));
+        assert!(insert.contains("NULL"));
+
+        let mut bounded = record.clone();
+        bounded.valid_to = Some(EventTime::parse_rfc3339("2026-02-01T00:00:00Z").expect("vt"));
+        let insert_bounded = insert_document_sql(&bounded).expect("bounded");
+        assert!(insert_bounded.contains("2026-02-01T00:00:00Z"));
+
+        let [close, reopen] = revise_document_sqls(&record).expect("revise");
+        assert!(close.contains("UPDATE document_record"));
+        assert!(close.contains("system_to IS NULL"));
+        assert!(reopen.contains("INSERT INTO document_record"));
+
+        let known = as_known_at_sql(uuid::Uuid::nil(), "2026-03-01T00:00:00Z");
+        assert!(known.contains("system_from <="));
+        let valid = as_valid_at_sql(
+            uuid::Uuid::nil(),
+            "2026-01-15T00:00:00Z",
+            "2026-03-01T00:00:00Z",
+        );
+        assert!(valid.contains("valid_from <="));
+
+        let audit = AuditEvent {
+            audit_event_id: uuid::Uuid::nil(),
+            tenant_record_id: uuid::Uuid::nil(),
+            action_code: "revise'attempt".into(),
+            subject_record_id: uuid::Uuid::nil(),
+            recorded_system_time: SystemTime::parse_rfc3339("2026-01-01T00:00:00Z").expect("s"),
+        };
+        let audit_sql = append_audit_sql(&audit);
+        assert!(audit_sql.contains("INSERT INTO audit_event"));
+        assert!(audit_sql.contains("revise''attempt"));
+
+        assert_eq!(
+            insert_document_sql(&DocumentRecord {
+                content_digest: "short".into(),
+                ..record
+            }),
+            Err(PersistenceError::InvalidContentDigest)
+        );
+        assert_eq!(
+            revise_document_sqls(&DocumentRecord {
+                content_digest: "nope".into(),
+                ..sample_record()
+            }),
+            Err(PersistenceError::InvalidContentDigest)
+        );
+        assert_eq!(optional_timestamptz(None), "NULL");
+        assert_eq!(
+            optional_timestamptz(Some("2026-01-01T00:00:00Z".into())),
+            "'2026-01-01T00:00:00Z'::timestamptz"
+        );
+        assert_eq!(escape_literal("a'b"), "a''b");
+        assert!(validate_digest(&"ff".repeat(32)).is_ok());
+        assert!(validate_digest("x").is_err());
+    }
+}

crates/persistence_postgres/src/error.rs

@@ -14,6 +14,14 @@ pub enum PersistenceError {
     HistoricalVersionNotFound,
     /// A document digest failed closed validation.
     InvalidContentDigest,
+    /// A live SQL batch was empty after comment/whitespace stripping.
+    EmptySqlBatch,
+    /// A live SQL statement or transport operation failed closed.
+    SqlExecutionFailed,
+    /// A live database URL failed scheme/host validation.
+    DatabaseUrlInvalid,
+    /// Live `SQLx` wiring was requested without a configured transport URL.
+    LiveAdapterNotConfigured,
 }
 
 impl fmt::Display for PersistenceError {
@@ -23,6 +31,10 @@ impl fmt::Display for PersistenceError {
             Self::ImmutableAuditViolation => "immutable audit violation",
             Self::HistoricalVersionNotFound => "historical version not found",
             Self::InvalidContentDigest => "invalid content digest",
+            Self::EmptySqlBatch => "empty sql batch",
+            Self::SqlExecutionFailed => "sql execution failed",
+            Self::DatabaseUrlInvalid => "database url invalid",
+            Self::LiveAdapterNotConfigured => "live adapter not configured",
         };
         formatter.write_str(message)
     }
@@ -80,6 +92,22 @@ mod tests {
             PersistenceError::InvalidContentDigest.to_string(),
             "invalid content digest"
         );
+        assert_eq!(
+            PersistenceError::EmptySqlBatch.to_string(),
+            "empty sql batch"
+        );
+        assert_eq!(
+            PersistenceError::SqlExecutionFailed.to_string(),
+            "sql execution failed"
+        );
+        assert_eq!(
+            PersistenceError::DatabaseUrlInvalid.to_string(),
+            "database url invalid"
+        );
+        assert_eq!(
+            PersistenceError::LiveAdapterNotConfigured.to_string(),
+            "live adapter not configured"
+        );
         assert_eq!(
             MigrationContractError::SingleWordObjectName.to_string(),
             "single-word database object name"

crates/persistence_postgres/src/lib.rs

@@ -1,21 +1,35 @@
 #![forbid(unsafe_code)]
 #![deny(missing_docs)]
-//! PostgreSQL-oriented bitemporal persistence contracts for TEPP.
+//! `PostgreSQL`-oriented bitemporal persistence contracts for TEPP.
 //!
-//! This crate owns migration SQL contracts, knowledge-cutoff eligibility, and
-//! in-memory bitemporal adapters that encode `as_known_at` / `as_valid_at`
-//! replay without requiring a live database in CI. Live `SQLx` execution against
-//! `PostgreSQL` remains an accepted-target follow-on behind the same contracts
-//! (ADR 0013).
+//! This crate owns migration SQL contracts, knowledge-cutoff eligibility,
+//! in-memory bitemporal adapters, live SQL session/migration ports, document
+//! SQL contracts, and a fail-closed `DATABASE_URL` gate for `SQLx` pool wiring
+//! (ADR 0013). In-process transports keep CI deterministic; a validated live
+//! URL is required before any production pool is opened.
 
 mod cutoff;
+mod document_sql;
 mod document_store;
 mod error;
+mod live_repository;
 mod migration;
 mod naming;
+mod sql_session;
+mod sqlx_gate;
 
 /// Knowledge-cutoff eligibility for historical analytical reads.
 pub use cutoff::is_cutoff_eligible;
+/// Render append-only audit insert SQL.
+pub use document_sql::append_audit_sql;
+/// Render as-known-at selection SQL.
+pub use document_sql::as_known_at_sql;
+/// Render as-valid-at selection SQL.
+pub use document_sql::as_valid_at_sql;
+/// Render open-document insert SQL.
+pub use document_sql::insert_document_sql;
+/// Render revise close+insert SQL pair.
+pub use document_sql::revise_document_sqls;
 /// Append-only audit event.
 pub use document_store::AuditEvent;
 /// Bitemporal document version.
@@ -26,9 +40,29 @@ pub use document_store::DocumentStore;
 pub use error::MigrationContractError;
 /// Fail-closed persistence domain errors.
 pub use error::PersistenceError;
+/// Live document repository over a SQL transport.
+pub use live_repository::LiveDocumentRepository;
+/// Migration application failures on the live path.
+pub use live_repository::LiveMigrationError;
 /// Embedded and ad-hoc migration catalogs.
 pub use migration::MigrationCatalog;
 /// Validate migration SQL against TEPP contracts.
 pub use migration::validate_migration_catalog;
 /// Multi-word `snake_case` database object naming.
 pub use naming::is_multi_word_snake_case;
+/// Recording SQL transport for offline contract tests.
+pub use sql_session::RecordingSqlSession;
+/// Live SQL transport contract.
+pub use sql_session::SqlSession;
+/// Apply a SQL batch through a live session.
+pub use sql_session::apply_sql_batch;
+/// Split migration SQL into executable statements.
+pub use sql_session::split_sql_statements;
+/// Environment variable name for live `SQLx` configuration.
+pub use sqlx_gate::DATABASE_URL_ENV;
+/// Validated live `SQLx` connection configuration.
+pub use sqlx_gate::LiveSqlxConfig;
+/// Require a validated live `SQLx` configuration from the environment.
+pub use sqlx_gate::require_live_sqlx_config;
+/// Require live `SQLx` configuration from an explicit optional value.
+pub use sqlx_gate::require_live_sqlx_config_from;