@@ -20,6 +20,7 @@ fensu-policy.workspace = true
 mimalloc.workspace = true
 pyo3.workspace = true
 rayon.workspace = true
+rusqlite.workspace = true
 rustc-hash.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 
@@ -41,7 +41,6 @@ use sqlbuild_state::execution_history::models::{
 use sqlbuild_state::execution_history::types::UtcTime;
 
 use crate::bindings::_helpers::boundary::python_calls::imported;
-use crate::bindings::_helpers::boundary::python_calls::keywords;
 use crate::bindings::_helpers::execution_history::appends::prepared_batch;
 use crate::bindings::_helpers::execution_history::conversion::{
     checked, event_filter_spec, exact_int, history_error_of_kind, new_namespace, now,
@@ -53,12 +52,13 @@ use crate::bindings::_helpers::execution_history::statements::{
     column, execute, execute_plain, execute_query, fetch_one, tuple_of,
 };
 use crate::bindings::_helpers::native_host::host;
+use crate::bindings::_helpers::native_host::sqlite_connection::{SqliteConnection, sqlite_params};
+use crate::bindings::_helpers::native_host::sqlite_cursor::SqliteCursor;
 use crate::bindings::_helpers::native_threads::reentrant_lock::ReentrantLock;
 use crate::bindings::_helpers::records::fields::field as native_field;
 
 const SQLITE_MODULE: &str = "sqlite3";
 const PATHLIB_MODULE: &str = "pathlib";
-const MILLIS_PER_SECOND: f64 = 1000.0;
 const CLOSED: &str = "SQLite execution history storage is closed";
 const NOT_AWARE: &str = "SQLite history timestamp is not timezone-aware";
 const INTEGRITY_CONFLICT: ExecutorHost = ExecutorHost::IntegrityConflictError;
@@ -233,7 +233,10 @@ impl SqliteExecutionHistory {
             };
             if let Err(error) = execute_plain(&connection, SQLITE_COMMIT) {
                 if sqlite_failure(py, &error)?
-                    && native_field(&connection, "in_transaction")?.is_truthy()?
+                    && match connection.cast::<SqliteConnection>() {
+                        Ok(connection) => connection.get().in_transaction(py)?,
+                        Err(_) => native_field(&connection, "in_transaction")?.is_truthy()?,
+                    }
                 {
                     let _ = execute_plain(&connection, SQLITE_ROLLBACK);
                 }
@@ -310,6 +313,16 @@ impl SqliteExecutionHistory {
     fn replace_projection(connection: &Bound<'_, PyAny>, projected: &[RunRow]) -> PyResult<()> {
         let py: Python<'_> = connection.py();
         execute_plain(connection, SQLITE_DELETE_RUNS)?;
+        if let Ok(connection) = connection.cast::<SqliteConnection>() {
+            for run in projected {
+                connection.get().execute_native(
+                    py,
+                    SQLITE_RUN_INSERT,
+                    &sqlite_params(&run_insert_query(Dialect::Sqlite, run).params)?,
+                )?;
+            }
+            return Ok(());
+        }
         let rows: Vec<Bound<'_, PyAny>> = projected
             .iter()
             .map(|run| {
@@ -347,10 +360,16 @@ impl SqliteExecutionHistory {
                 None => {
                     let query = event_insert_query(Dialect::Sqlite, event, &now(py)?);
                     let cursor: Bound<'py, PyAny> = execute_query(connection, &query)?;
-                    let storage_id: Bound<'py, PyAny> = native_field(&cursor, "lastrowid")?;
-                    connection
-                        .call_method1("execute", (SQLITE_EVENT_BY_STORAGE_ID, (storage_id,)))?
-                        .call_method0("fetchone")?
+                    let storage_id: i64 = match cursor.cast::<SqliteCursor>() {
+                        Ok(cursor) => cursor.get().last_row_id,
+                        Err(_) => native_field(&cursor, "lastrowid")?.extract()?,
+                    };
+                    fetch_one(&execute(
+                        connection,
+                        SQLITE_EVENT_BY_STORAGE_ID,
+                        &[SqlParam::Integer(storage_id)],
+                    )?)?
+                    .unwrap_or_else(|| py.None().into_bound(py))
                 }
             };
             if namespace.is_none() {
@@ -491,25 +510,10 @@ impl SqliteExecutionHistory {
         let mut connection: Option<Bound<'_, PyAny>> = None;
         let opened: PyResult<()> = (|| {
             prepare_history_directory(&database)?;
-            let sqlite = py.import(SQLITE_MODULE)?;
-            let opened: Bound<'_, PyAny> = native_field(&sqlite, "connect")?.call(
-                (database.as_str(),),
-                Some(&keywords(
-                    py,
-                    &[
-                        (
-                            "timeout",
-                            (busy_timeout_ms as f64 / MILLIS_PER_SECOND).into_bound_py_any(py)?,
-                        ),
-                        ("isolation_level", py.None().into_bound(py)),
-                        ("check_same_thread", false.into_bound_py_any(py)?),
-                    ],
-                )?),
-            )?;
+            let opened = Bound::new(py, SqliteConnection::open(py, &database)?)?.into_any();
             connection = Some(opened.clone());
             slf.setattr("_connection", &opened)?;
             this.with_lock(py, || {
-                opened.setattr("row_factory", native_field(&sqlite, "Row")?)?;
                 execute_plain(&opened, &format!("PRAGMA busy_timeout = {busy_timeout_ms}"))?;
                 execute_plain(&opened, SQLITE_FOREIGN_KEYS)?;
                 if database != MEMORY_PATH {
 
@@ -5,6 +5,8 @@ use pyo3::types::PyTuple;
 use sqlbuild_state::execution_history::models::{Query, SqlParam};
 
 use crate::bindings::_helpers::execution_history::conversion::python_params;
+use crate::bindings::_helpers::native_host::sqlite_connection::{SqliteConnection, sqlite_params};
+use crate::bindings::_helpers::native_host::sqlite_cursor::{SqliteCursor, SqliteRow};
 
 /// `target.execute(sql, params)`.
 pub(crate) fn execute<'py>(
@@ -13,6 +15,15 @@ pub(crate) fn execute<'py>(
     params: &[SqlParam],
 ) -> PyResult<Bound<'py, PyAny>> {
     let py: Python<'py> = target.py();
+    if let Ok(connection) = target.cast::<SqliteConnection>() {
+        return Bound::new(
+            py,
+            connection
+                .get()
+                .execute_native(py, sql, &sqlite_params(params)?)?,
+        )
+        .map(Bound::into_any);
+    }
     target.call_method1("execute", (sql, python_params(py, params)?))
 }
 
@@ -21,6 +32,9 @@ pub(crate) fn execute_plain<'py>(
     target: &Bound<'py, PyAny>,
     sql: &str,
 ) -> PyResult<Bound<'py, PyAny>> {
+    if target.is_instance_of::<SqliteConnection>() {
+        return execute(target, sql, &[]);
+    }
     target.call_method1("execute", (sql,))
 }
 
@@ -34,6 +48,13 @@ pub(crate) fn execute_query<'py>(
 
 /// The first row `cursor.fetchone()` returns, or `None`.
 pub(crate) fn fetch_one<'py>(cursor: &Bound<'py, PyAny>) -> PyResult<Option<Bound<'py, PyAny>>> {
+    if let Ok(cursor) = cursor.cast::<SqliteCursor>() {
+        return cursor
+            .get()
+            .next_row()
+            .map(|row| Bound::new(cursor.py(), row).map(Bound::into_any))
+            .transpose();
+    }
     let row: Bound<'py, PyAny> = cursor.call_method0("fetchone")?;
     Ok((!row.is_none()).then_some(row))
 }
@@ -45,6 +66,9 @@ pub(crate) fn fetch_all<'py>(cursor: &Bound<'py, PyAny>) -> PyResult<Vec<Bound<'
 
 /// `row[name]`.
 pub(crate) fn column<'py>(row: &Bound<'py, PyAny>, name: &str) -> PyResult<Bound<'py, PyAny>> {
+    if let Ok(row) = row.cast::<SqliteRow>() {
+        return row.get().column(row.py(), name);
+    }
     row.get_item(name)
 }
 
 
@@ -1,6 +1,8 @@
-//! The one Python module native code imports by path, and the staying-Python entries it names.
+//! Permanent Python host calls and compatibility views of native-owned execution data.
 
 pub(crate) mod connections;
 pub(crate) mod host;
 pub(crate) mod registration;
 pub(crate) mod results;
+pub(crate) mod sqlite_connection;
+pub(crate) mod sqlite_cursor;
@@ -0,0 +1,189 @@
+//! Rust-owned SQLite connection; Python views exist only for connection overrides.
+
+use std::collections::VecDeque;
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex, PoisonError};
+
+use pyo3::exceptions::PyValueError;
+use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyErr, PyResult, Python, pyclass, pymethods};
+use rusqlite::types::Value;
+use rusqlite::{Connection, Error, ErrorCode, OpenFlags, params_from_iter};
+use sqlbuild_state::execution_history::constants::MEMORY_PATH;
+use sqlbuild_state::execution_history::models::SqlParam;
+
+use crate::bindings::_helpers::boundary::python_calls::imported;
+use crate::bindings::_helpers::native_host::sqlite_cursor::{SqliteCursor, SqliteRow};
+
+/// Preserve DB-API exception classes at the public storage-error boundary.
+fn database_error(py: Python<'_>, error: Error) -> PyErr {
+    let class: &str = match &error {
+        Error::SqliteFailure(error, _) if error.code == ErrorCode::ConstraintViolation => {
+            "IntegrityError"
+        }
+        Error::SqliteFailure(error, _)
+            if matches!(
+                error.code,
+                ErrorCode::DatabaseCorrupt | ErrorCode::NotADatabase
+            ) =>
+        {
+            "DatabaseError"
+        }
+        _ => "OperationalError",
+    };
+    match imported(py, "sqlite3", class).and_then(|class| class.call1((error.to_string(),))) {
+        Ok(error) => PyErr::from_value(error),
+        Err(error) => error,
+    }
+}
+
+/// Parameters of history SQL contain no Python-owned values.
+pub(crate) fn sqlite_params(params: &[SqlParam]) -> PyResult<Vec<Value>> {
+    params
+        .iter()
+        .map(|value| {
+            Ok(match value {
+                SqlParam::Null => Value::Null,
+                SqlParam::Text(value) => Value::Text(value.clone()),
+                SqlParam::Integer(value) => Value::Integer(*value),
+                SqlParam::Flag(value) => Value::Integer(i64::from(*value)),
+                SqlParam::Timestamp(_) | SqlParam::TextList(_) => {
+                    return Err(PyValueError::new_err("Postgres-only history parameter"));
+                }
+            })
+        })
+        .collect()
+}
+
+#[pyclass(frozen, module = "sqlbuild._native", name = "_HistorySqliteConnection")]
+pub(crate) struct SqliteConnection {
+    connection: Mutex<Option<Connection>>,
+}
+
+impl SqliteConnection {
+    /// Open literal filenames, using absolute paths because bundled SQLite enables URI handling globally.
+    pub(crate) fn open(py: Python<'_>, path: &str) -> PyResult<Self> {
+        let flags = OpenFlags::default() & !OpenFlags::SQLITE_OPEN_URI;
+        let database: PathBuf = if path == MEMORY_PATH {
+            PathBuf::from(path)
+        } else {
+            std::env::current_dir()?.join(path)
+        };
+        let connection = py
+            .detach(|| Connection::open_with_flags(database, flags))
+            .map_err(|error| database_error(py, error))?;
+        Ok(Self {
+            connection: Mutex::new(Some(connection)),
+        })
+    }
+
+    fn with_connection<T: Send>(
+        &self,
+        py: Python<'_>,
+        body: impl FnOnce(&Connection) -> rusqlite::Result<T> + Send,
+    ) -> PyResult<T> {
+        let result = py.detach(|| {
+            let guard = self
+                .connection
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner);
+            guard.as_ref().map(body)
+        });
+        match result {
+            Some(result) => result.map_err(|error| database_error(py, error)),
+            None => Err(PyErr::from_value(
+                imported(py, "sqlite3", "ProgrammingError")?
+                    .call1(("Cannot operate on a closed database.",))?,
+            )),
+        }
+    }
+
+    pub(crate) fn execute_native(
+        &self,
+        py: Python<'_>,
+        sql: &str,
+        params: &[Value],
+    ) -> PyResult<SqliteCursor> {
+        self.with_connection(py, |connection| {
+            let mut statement = connection.prepare_cached(sql)?;
+            let columns = Arc::new(
+                statement
+                    .column_names()
+                    .into_iter()
+                    .map(str::to_owned)
+                    .collect(),
+            );
+            let count: usize = statement.column_count();
+            let mut cursor = statement.query(params_from_iter(params))?;
+            let mut rows: VecDeque<SqliteRow> = VecDeque::new();
+            while let Some(row) = cursor.next()? {
+                rows.push_back(SqliteRow {
+                    columns: Arc::clone(&columns),
+                    values: (0..count)
+                        .map(|index| row.get(index))
+                        .collect::<rusqlite::Result<_>>()?,
+                });
+            }
+            Ok(SqliteCursor {
+                rows: Mutex::new(rows),
+                last_row_id: connection.last_insert_rowid(),
+            })
+        })
+    }
+}
+
+#[pymethods]
+impl SqliteConnection {
+    #[pyo3(signature = (sql, parameters = None))]
+    fn execute(
+        &self,
+        py: Python<'_>,
+        sql: &str,
+        parameters: Option<&Bound<'_, PyAny>>,
+    ) -> PyResult<SqliteCursor> {
+        let mut params: Vec<Value> = Vec::new();
+        if let Some(parameters) = parameters {
+            for value in parameters.try_iter()? {
+                let value = value?;
+                params.push(if value.is_none() {
+                    Value::Null
+                } else if let Ok(value) = value.extract::<i64>() {
+                    Value::Integer(value)
+                } else {
+                    Value::Text(value.extract()?)
+                });
+            }
+        }
+        self.execute_native(py, sql, &params)
+    }
+
+    fn executemany(
+        &self,
+        py: Python<'_>,
+        sql: &str,
+        parameters: &Bound<'_, PyAny>,
+    ) -> PyResult<SqliteCursor> {
+        let mut result = SqliteCursor {
+            rows: Mutex::default(),
+            last_row_id: 0,
+        };
+        for row in parameters.try_iter()? {
+            result = self.execute(py, sql, Some(&row?))?;
+        }
+        Ok(result)
+    }
+
+    #[getter]
+    pub(crate) fn in_transaction(&self, py: Python<'_>) -> PyResult<bool> {
+        self.with_connection(py, |connection| Ok(!connection.is_autocommit()))
+    }
+
+    pub(crate) fn close(&self, py: Python<'_>) -> PyResult<()> {
+        py.detach(|| {
+            self.connection
+                .lock()
+                .unwrap_or_else(PoisonError::into_inner)
+                .take();
+        });
+        Ok(())
+    }
+}