fix: enable production PostgreSQL persistence

.github/workflows/rust.yml

@@ -48,3 +48,47 @@ jobs:
 
       - name: Run tests
         run: cargo test
+
+  postgres:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17-alpine@sha256:742f40ea20b9ff2ff31db5458d127452988a2164df9e17441e191f3b72252193
+        env:
+          POSTGRES_DB: agentic_api_test
+          POSTGRES_PASSWORD: postgres
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U postgres -d agentic_api_test"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 10
+    env:
+      TEST_POSTGRES_URL: postgresql://postgres:postgres@localhost:5432/agentic_api_test?sslmode=disable
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@e97e2d8cc328f1b50210efc529dca0028893a2d9 # v1
+        with:
+          toolchain: stable
+
+      - name: Cache cargo registry and build
+        uses: actions/cache@668228422ae6a00e4ad889ee87cd7109ec5666a7 # v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: cargo-postgres-${{ runner.os }}-${{ hashFiles('Cargo.lock') }}
+          restore-keys: |
+            cargo-postgres-${{ runner.os }}-
+            cargo-${{ runner.os }}-
+
+      - name: Verify PostgreSQL upgrade preservation
+        run: cargo test -p agentic-server-core --lib postgres_integer_widening_upgrade_preserves_existing_state -- --ignored
+
+      - name: Verify PostgreSQL migrations and restart persistence
+        run: cargo test -p agentic-server-core --test postgres_storage_integration -- --ignored

crates/agentic-server-core/migrations/0003_index_conversation_sequence.sql

@@ -0,0 +1,2 @@
+DROP INDEX IF EXISTS idx_items_conversation_id;
+CREATE INDEX idx_items_conversation_id ON items (conversation_id, seq);

crates/agentic-server-core/src/config.rs

@@ -1,7 +1,41 @@
+use std::time::Duration;
+
+pub const DEFAULT_POSTGRES_MAX_CONNECTIONS: u32 = 10;
+pub const DEFAULT_POSTGRES_ACQUIRE_TIMEOUT_SECONDS: u64 = 30;
+pub const DEFAULT_POSTGRES_IDLE_TIMEOUT_SECONDS: u64 = 600;
+pub const DEFAULT_POSTGRES_LOCK_TIMEOUT_SECONDS: u64 = 5;
+pub const DEFAULT_POSTGRES_MAX_LIFETIME_SECONDS: u64 = 1_800;
+pub const DEFAULT_POSTGRES_MIGRATION_TIMEOUT_SECONDS: u64 = 300;
+pub const DEFAULT_POSTGRES_STATEMENT_TIMEOUT_SECONDS: u64 = 30;
 pub const DEFAULT_SQLITE_MAX_CONNECTIONS: u32 = 4;
 pub const DEFAULT_SQLITE_JOURNAL_SIZE_LIMIT_BYTES: u64 = 6_144_000;
 pub const DEFAULT_SQLITE_MMAP_SIZE_BYTES: u64 = 268_435_456;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PostgresConfig {
+    pub max_connections: u32,
+    pub acquire_timeout: Duration,
+    pub lock_timeout: Duration,
+    pub migration_timeout: Duration,
+    pub statement_timeout: Duration,
+    pub idle_timeout: Option<Duration>,
+    pub max_lifetime: Option<Duration>,
+}
+
+impl Default for PostgresConfig {
+    fn default() -> Self {
+        Self {
+            max_connections: DEFAULT_POSTGRES_MAX_CONNECTIONS,
+            acquire_timeout: Duration::from_secs(DEFAULT_POSTGRES_ACQUIRE_TIMEOUT_SECONDS),
+            lock_timeout: Duration::from_secs(DEFAULT_POSTGRES_LOCK_TIMEOUT_SECONDS),
+            migration_timeout: Duration::from_secs(DEFAULT_POSTGRES_MIGRATION_TIMEOUT_SECONDS),
+            statement_timeout: Duration::from_secs(DEFAULT_POSTGRES_STATEMENT_TIMEOUT_SECONDS),
+            idle_timeout: Some(Duration::from_secs(DEFAULT_POSTGRES_IDLE_TIMEOUT_SECONDS)),
+            max_lifetime: Some(Duration::from_secs(DEFAULT_POSTGRES_MAX_LIFETIME_SECONDS)),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub enum SqliteTempStore {
     Default,
@@ -50,6 +84,7 @@ pub struct Config {
     /// Database URL for conversation and response storage.
     /// `None` means stateful features are disabled; all requests are proxied.
     pub db_url: Option<String>,
+    pub postgres: PostgresConfig,
     pub sqlite: SqliteConfig,
 }
 

crates/agentic-server-core/src/executor/engine.rs

@@ -10,7 +10,7 @@ use std::sync::Arc;
 use async_stream::stream;
 use either::Either;
 use tokio::sync::mpsc;
-use tracing::{debug, warn};
+use tracing::debug;
 
 use super::gateway::{
     GatewayCallResult, LoopDecision, append_gateway_calls_to_new_input, append_output_items_to_input,
@@ -332,9 +332,7 @@ async fn run_blocking(
 
     let ch = exec_ctx.conv_handler.clone();
     let rh = exec_ctx.resp_handler.clone();
-    if let Err(e) = persist_if_needed(payload.clone(), ctx, ch, rh).await {
-        warn!("persist failed: {e}");
-    }
+    persist_if_needed(payload.clone(), ctx, ch, rh).await?;
 
     Ok(payload)
 }
@@ -389,12 +387,11 @@ fn run_stream(ctx: RequestContext, exec_ctx: Arc<ExecutionContext>, auth: Option
                             let terminal_chunk = stream_accumulator.terminal_response_chunk(&payload);
                             let ch = exec_ctx.conv_handler.clone();
                             let rh = exec_ctx.resp_handler.clone();
-                            if let Err(e) = persist_if_needed(payload, ctx, ch, rh).await {
-                                warn!("persist failed: {e}");
-                            }
-
-                            match terminal_chunk {
-                                Ok(chunk) => yield chunk,
+                            match persist_if_needed(payload, ctx, ch, rh).await {
+                                Ok(()) => match terminal_chunk {
+                                    Ok(chunk) => yield chunk,
+                                    Err(e) => yield stream_accumulator.error_chunk(&e.to_string()),
+                                },
                                 Err(e) => yield stream_accumulator.error_chunk(&e.to_string()),
                             }
                             yield DONE_MARKER.to_string();

crates/agentic-server-core/src/executor/request.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use crate::config::Config;
 use crate::error::Error;
 use crate::executor::modes::{ConversationHandler, ResponseHandler};
-use crate::storage::{ConversationStore, ResponseStore, create_pool_with_schema_and_sqlite_config};
+use crate::storage::{ConversationStore, ResponseStore, create_pool_with_schema_and_configs};
 use crate::tool::{GatewayExecutor, GatewayExecutors};
 use crate::types::io::InputItem;
 use crate::types::messages::GatewayToolMap;
@@ -113,9 +113,10 @@ impl ExecutionContext {
     /// migration fails.
     pub async fn from_config(cfg: &Config) -> Result<Self, Error> {
         let db_url = cfg.db_url.as_deref().unwrap_or("sqlite://./agentic_api.db");
-        let pool = create_pool_with_schema_and_sqlite_config(Some(db_url), cfg.sqlite)
+        let database_backend = database_backend(db_url);
+        let pool = create_pool_with_schema_and_configs(Some(db_url), cfg.sqlite, cfg.postgres)
             .await
-            .map_err(|e| Error::Config(format!("failed to open database '{db_url}': {e}")))?;
+            .map_err(|error| database_open_error(database_backend, &error))?;
 
         let conv_handler = ConversationHandler::new(ConversationStore::new(pool.clone()));
         let resp_handler = ResponseHandler::new(ResponseStore::new(pool));
@@ -134,10 +135,109 @@ impl ExecutionContext {
     }
 }
 
+fn database_open_error(database_backend: &str, error: &sqlx::Error) -> Error {
+    let category = match error {
+        sqlx::Error::Configuration(_) | sqlx::Error::InvalidArgument(_) => "configuration error".to_owned(),
+        sqlx::Error::Database(database_error) => database_error
+            .code()
+            .filter(|code| code.len() <= 5 && code.bytes().all(|byte| byte.is_ascii_alphanumeric()))
+            .map_or_else(
+                || "database error".to_owned(),
+                |code| format!("database error (SQLSTATE {code})"),
+            ),
+        sqlx::Error::Io(_) => "database I/O error".to_owned(),
+        sqlx::Error::Tls(_) => "database TLS error".to_owned(),
+        sqlx::Error::Protocol(_) => "database protocol error".to_owned(),
+        sqlx::Error::PoolTimedOut => "connection pool timeout".to_owned(),
+        sqlx::Error::PoolClosed => "connection pool closed".to_owned(),
+        sqlx::Error::WorkerCrashed => "database worker crashed".to_owned(),
+        sqlx::Error::Migrate(_) => "database migration error".to_owned(),
+        _ => "database error".to_owned(),
+    };
+    let detail = redact_database_urls(&error.to_string());
+    Error::Config(format!(
+        "failed to open {database_backend} database: {category}: {detail}"
+    ))
+}
+
+fn redact_database_urls(message: &str) -> String {
+    const DATABASE_SCHEMES: [&str; 3] = ["postgresql://", "postgres://", "mysql://"];
+
+    let mut redacted = message.to_owned();
+    for scheme in DATABASE_SCHEMES {
+        let mut search_from = 0;
+        while let Some(offset) = redacted[search_from..].find(scheme) {
+            let start = search_from + offset;
+            let end = redacted[start..]
+                .find(char::is_whitespace)
+                .map_or(redacted.len(), |length| start + length);
+            let replacement = format!("{scheme}[redacted]");
+            redacted.replace_range(start..end, &replacement);
+            search_from = start + replacement.len();
+        }
+    }
+    redacted
+}
+
+fn database_backend(database_url: &str) -> &'static str {
+    if database_url.starts_with("postgres://") || database_url.starts_with("postgresql://") {
+        "PostgreSQL"
+    } else if database_url.starts_with("sqlite") {
+        "SQLite"
+    } else {
+        "configured"
+    }
+}
+
 /// Load the `/v1/messages` gateway-tool alias map from the environment.
 fn messages_gateway_tools_from_env() -> GatewayToolMap {
     std::env::var(GATEWAY_TOOL_ALIASES_ENV)
         .ok()
         .map(|raw| GatewayToolMap::from_env_str(&raw))
         .unwrap_or_default()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{database_backend, database_open_error};
+
+    #[test]
+    fn database_errors_are_actionable_without_exposing_credentials() {
+        let database_url = "postgresql://agentic-api:super'secret@postgres.example.com/agentic_api";
+        let backend = database_backend(database_url);
+        let tls_error = sqlx::Error::Tls(Box::new(std::io::Error::other(format!(
+            "{database_url} certificate verify failed"
+        ))));
+        let pool_error = sqlx::Error::PoolTimedOut;
+        let tls_message = database_open_error(backend, &tls_error).to_string();
+        let pool_message = database_open_error(backend, &pool_error).to_string();
+
+        assert!(tls_message.contains("database TLS error"));
+        assert!(tls_message.contains("certificate verify failed"));
+        assert!(tls_message.contains("postgresql://[redacted]"));
+        assert_eq!(
+            pool_message,
+            "failed to open PostgreSQL database: connection pool timeout: \
+             pool timed out while waiting for an open connection"
+        );
+        assert!(!tls_message.contains("super-secret"));
+        assert!(!tls_message.contains("secret"));
+        assert!(!tls_message.contains("agentic-api"));
+        assert_ne!(tls_message, pool_message);
+
+        let mysql_error = sqlx::Error::Io(std::io::Error::other(
+            "mysql://gateway:mysql-secret@mysql.example.com/agentic_api refused",
+        ));
+        let mysql_message = database_open_error("configured", &mysql_error).to_string();
+        assert!(mysql_message.contains("mysql://[redacted]"));
+        assert!(!mysql_message.contains("mysql-secret"));
+
+        let short_postgres_url = "postgres://gateway:postgres-secret@postgres.example.com/agentic_api";
+        let short_postgres_error = sqlx::Error::Io(std::io::Error::other(format!("{short_postgres_url} refused")));
+        let short_postgres_message =
+            database_open_error(database_backend(short_postgres_url), &short_postgres_error).to_string();
+        assert!(short_postgres_message.contains("failed to open PostgreSQL database"));
+        assert!(short_postgres_message.contains("postgres://[redacted]"));
+        assert!(!short_postgres_message.contains("postgres-secret"));
+    }
+}

crates/agentic-server-core/src/proxy.rs

@@ -317,6 +317,7 @@ mod tests {
             llm_ready_interval_s: 0.1,
             skip_llm_ready_check: false,
             db_url: None,
+            postgres: crate::config::PostgresConfig::default(),
             sqlite: crate::config::SqliteConfig::default(),
         }
     }

crates/agentic-server-core/src/storage/conversation.rs

@@ -113,6 +113,7 @@ impl ConversationStore {
 
         let mut tx = pool.begin().await?;
 
+        conversation::lock_in_tx(&mut tx, conversation_id).await?;
         item::create_in_tx(&mut tx, items_, Some(conversation_id)).await?;
 
         response::create_in_tx(

crates/agentic-server-core/src/storage/mod.rs

@@ -24,8 +24,8 @@ pub use models::Conversation as DbConversation;
 pub use models::Item;
 pub use models::Response as DbResponse;
 pub use pool::{
-    DbPool, DbResult, DbTransaction, create_pool, create_pool_with_schema, create_pool_with_schema_and_sqlite_config,
-    create_pool_with_sqlite_config,
+    DbPool, DbResult, DbTransaction, create_pool, create_pool_with_configs, create_pool_with_schema,
+    create_pool_with_schema_and_configs, create_pool_with_schema_and_sqlite_config, create_pool_with_sqlite_config,
 };
 pub use response::ResponseStore;
 pub use schema::{PoolWithSchema, SchemaManager};

crates/agentic-server-core/src/storage/models/conversation.rs

@@ -1,6 +1,6 @@
 //! Conversation context and history.
 
-use super::super::pool::{DbPool, DbResult};
+use super::super::pool::{DbPool, DbResult, DbTransaction};
 use crate::utils::common::utcnow_str;
 
 /// Conversation context and history.
@@ -27,7 +27,7 @@ pub async fn create(pool: &DbPool, id: &str) -> DbResult<Conversation> {
     let now = utcnow_str();
     sqlx::query_as::<_, Conversation>(
         "INSERT INTO conversations (id, created_at) \
-         VALUES (?, ?) RETURNING *",
+         VALUES ($1, $2) RETURNING *",
     )
     .bind(id)
     .bind(now)
@@ -43,7 +43,7 @@ pub async fn get_or_create(pool: &DbPool, id: &str) -> DbResult<Conversation> {
     let now = utcnow_str();
     sqlx::query_as::<_, Conversation>(
         "INSERT INTO conversations (id, created_at) \
-         VALUES (?, ?) \
+         VALUES ($1, $2) \
          ON CONFLICT (id) DO UPDATE SET created_at = created_at \
          RETURNING *",
     )
@@ -58,12 +58,39 @@ pub async fn get_or_create(pool: &DbPool, id: &str) -> DbResult<Conversation> {
 /// # Errors
 /// Returns `DbResult::Err` if the database query fails.
 pub async fn get(pool: &DbPool, id: &str) -> DbResult<Option<Conversation>> {
-    sqlx::query_as::<_, Conversation>("SELECT * FROM conversations WHERE id = ?")
+    sqlx::query_as::<_, Conversation>("SELECT * FROM conversations WHERE id = $1")
         .bind(id)
         .fetch_optional(pool)
         .await
 }
 
+/// Locks an existing conversation for the lifetime of the transaction.
+///
+/// `PostgreSQL` takes a row lock without writing the row. `SQLite` uses a no-op
+/// update to acquire its write lock. Both serialize per-conversation sequence
+/// allocation when multiple gateway replicas persist turns concurrently.
+///
+/// # Errors
+/// Returns `DbResult::Err` if the database query fails or the conversation does not exist.
+pub async fn lock_in_tx(tx: &mut DbTransaction<'_>, id: &str) -> DbResult<()> {
+    if tx.as_mut().backend_name() == "PostgreSQL" {
+        let locked_id = sqlx::query_scalar::<_, String>("SELECT id FROM conversations WHERE id = $1 FOR UPDATE")
+            .bind(id)
+            .fetch_optional(&mut **tx)
+            .await?;
+        return locked_id.map(|_| ()).ok_or(sqlx::Error::RowNotFound);
+    }
+
+    let result = sqlx::query("UPDATE conversations SET created_at = created_at WHERE id = $1")
+        .bind(id)
+        .execute(&mut **tx)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(sqlx::Error::RowNotFound);
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;