feat: SQL inspection and logging — the ferro.sql statement log and .sql() on every Query terminal (#700)

AGENTS.md

@@ -246,6 +246,15 @@ Phantom diffs are the canonical symptom that this invariant has been broken.
   pin the names (`test_index_name_matches_rust_runtime_*`).
 - `docs/solutions/patterns/cross-emitter-ddl-parity.md` documents the rule and
   the recipes below.
+- On the query side, the rendered-statement parity pin in
+  `tests/test_rendered_statement.py` is the counterpart of the cross-emitter
+  pins: for every `Query` terminal (`all`, `first`, `count`, `exists`,
+  `update` in both forms, `delete`, through `Relation` and `ProjectedQuery`
+  too) on both dialects, with a secret column, `await query.<terminal>.sql()`
+  equals the `ferro.sql` record the executed call writes in `sql`, `params`
+  (the `REDACTED` sentinel included) and `dialect`. Both doors build the
+  statement through one `build_<terminal>` function in `src/operations.rs`,
+  which the executing entry sends and its `render_<terminal>` twin returns.
 
 ### Adding a new emitter
 

docs/adr/0057-sql-is-seen-through-the-statement-log-and-a-terminals-sql.md

@@ -35,4 +35,10 @@ On Postgres the SQL text depends on the live table: ferro casts bound values (`$
 
 A bound value shows as `***`, a sentinel whose `repr` is `***` so a test can tell it from the string, when it is a `SecretStr`/`SecretBytes` or when it is bound to a column whose field is declared one. Either rule alone leaks: a plain `"hunter2"` compared against a secret column, or a `SecretStr` passed to `raw.execute` where no column is known. The mask is decided in Python, from the model's declared secret fields and the value's type, and travels to Rust beside the binds; Rust never re-derives it, the one-decision pattern I-1 uses. `.sql()`'s `params` are redacted the same way: one rule, applied in one place, and `print(rendered)` in a bug report is the common use. `SET LOCAL` takes no bind parameters on Postgres, so a session setting's value appears inline, unredacted; settings are tenancy scope, not secrets by declaration. Depends on #760, which makes a secret field store its raw value and hydrate as a secret type.
 
+Amended 2026-10-21 (panel F11 on #771): the mask is a set of values, not a positional flag. The binds are made by the query builder in Rust and never pass through Python one by one, so Python cannot mark positions. Instead each operation has one *secret-value set*, built from two facts. Python collects what every `SecretStr` / `SecretBytes` it unwraps hides (a query's literals as it compiles, a write's field values, a raw statement's arguments) through one collector. Rust adds every value the operation binds to a column whose field is declared secret: predicate literals, literal `SET`s, keyset positions (`after` / `before`) on a secret order key, root or traversed, and a write's secret-column values. A bind is masked when it equals a member of the set, one hash lookup per bind, decided before the GIL is taken to write the record. So `Account.where(lambda account: account.token == "sk_live_1")` logs `params=(***,)` though `"sk_live_1"` is a plain `str`. A plain parameter that happens to equal a secret of the same operation is masked too (`bulk_create` of one row whose email equals another row's token shows both as `***`): over-redaction, in the safe direction, never a leak. The switch is still read once per operation, by the engine at the operation's entry. A write's or raw statement's Python-collected values reach the engine as a lazy iterable it reads only when the log is on; a query's are collected as it compiles, because its `.sql()` masks with the same values.
+
 A pin holds the two doors together: for every terminal on both dialects, `.sql()` equals the record the executed call wrote.
+
+## Amendment (2026-10-21): pool hooks are not in the statement log
+
+A Postgres pool sends two statements of its own, outside any operation: the `after_connect` hook's `SET search_path TO …` on each new connection, and, with `settings_delivery="connection"`, the `after_release` guard's `SELECT coalesce(current_setting($1, true), '')` on each returned connection while a session holds a pinned one. They are not in the statement log. A record needs an operation's scope, and these belong to no operation: sqlx runs the release hook on a task it spawns, and opens connections for `min_connections` in its background maintenance, so a scope could reach at most the connects an operation happens to wait on. Logging them would mean asking the logger once per connection opened and once per connection returned, a GIL take on each, which is the per-pool-event cost the one switch exists to avoid. The run lock's statements are not pool hooks: the run sends them on its own session inside its `migrate` scope, so they are records like the DDL around them.

docs/pages/api/queries.md

@@ -37,6 +37,12 @@ An aggregate-only projection collapses to one record (read with `first()`). Mixi
 
 ::: ferro.query.builder.ProjectedQuery
 
+::: ferro.RenderedSql
+
+::: ferro.REDACTED
+
+::: ferro.query.rendered.Terminal
+
 ::: ferro.query.Row
 
 ::: ferro.query.Rows

docs/pages/guide/connections.md

@@ -216,6 +216,56 @@ See [Testing](../howto/testing.md) for a ready-made pytest fixture.
 !!! note "No `disconnect()` yet"
     An explicit `disconnect()` is **not yet implemented** — pools are cleaned up on process exit. See the [Roadmap](../roadmap.md).
 
+## Logging
+
+The `ferro.sql` logger is the statement log: one `DEBUG` record for each statement Ferro sends, written as the statement finishes. Turn it on with your usual logging setup:
+
+```python
+import logging
+
+logging.basicConfig(format="%(name)s %(levelname)s %(message)s")
+logging.getLogger("ferro.sql").setLevel(logging.DEBUG)
+```
+
+A schema change now prints each statement it runs, with its parameters and how long it took:
+
+```text
+ferro.sql DEBUG CREATE TABLE IF NOT EXISTS "post" (...) params=() 1.2ms
+ferro.sql DEBUG ALTER TABLE "author" ADD COLUMN "bio" varchar params=() 0.8ms
+```
+
+A statement the database rejects ends its line with `failed: <error>`, and the error still raises. Each record also carries the statement as attributes, so a structured handler reads fields instead of parsing the line:
+
+| Attribute | Value |
+| --- | --- |
+| `sql` | The statement, placeholders included |
+| `params` | The bound parameters, as a tuple |
+| `dialect` | `"sqlite"` or `"postgres"` |
+| `duration_ms` | Time the database took, in milliseconds |
+| `operation` | What sent it: one of the values below |
+| `subject` | The table or enum type a pass statement changes, or the step file (`0003_add_slug/01_add_slug.up.sql`) a migration statement comes from; `None` otherwise |
+| `connection` | The connection's name |
+| `rows` | Rows affected, or `None` for a read |
+| `error` | The database's error text, or `None` |
+
+| `operation` | Sent by |
+| --- | --- |
+| `"fetch"` | `all()`, `first()`, `get()` and relation loads |
+| `"count"` / `"exists"` | `count()` / `exists()` |
+| `"save"` | `save()`, insert or update |
+| `"bulk_create"` | `bulk_create()`, one record per chunk |
+| `"update"` / `"delete"` | A query's `update()` / `delete()`; `delete()` on an instance |
+| `"m2m_add"` / `"m2m_remove"` / `"m2m_clear"` | A many-to-many relation's `add()` / `remove()` / `clear()` |
+| `"raw"` | `ferro.raw.execute()`, `fetch_all()` and `fetch_one()` |
+| `"transaction"` | A `transaction()` block's `BEGIN`, `SAVEPOINT`, `RELEASE`, `ROLLBACK TO`, `COMMIT` and `ROLLBACK` |
+| `"settings"` | A session's settings batch (`SET LOCAL`, `set_config`) and its reset |
+| `"catalog"` | The Postgres catalog read the first time an operation touches a table |
+| `"migrate"` | The reconciliation pass and `ferro migrate`, the run lock's statements included |
+
+Statements a Postgres pool sends on its own, outside any operation, are not in the log: the `SET search_path` it runs on each new connection, and, with `settings_delivery="connection"`, the check it runs on each returned connection while a session has one pinned.
+
+The logger is the only switch: while `ferro.sql` is not enabled for `DEBUG`, Ferro collects nothing. Ferro installs only a `NullHandler` on its `ferro` logger, which propagates, so your logging configuration (and pytest's `caplog`) decides where the records go. The `ferro` command line installs its own handler, so a data step's `ctx.log.info(...)` lines still print there.
+
 ## See Also
 
 - [Schema Management](schema/overview.md) — auto-migrate, migrations and Alembic

docs/pages/guide/schema/auto-migrate.md

@@ -254,7 +254,7 @@ except ferro.OperationalError as error:
     changed = {s.subject for s in error.report.statements if s.role == "schema"}
 ```
 
-`connect(..., auto_migrate=True)` runs the same pass and returns nothing; its failure carries the same `.report`. The `ferro` logger's debug lines name each statement as it runs, but they are free text: read the report, not the log.
+`connect(..., auto_migrate=True)` runs the same pass and returns nothing; its failure carries the same `.report`. Each statement is also a record on the `ferro.sql` logger as it finishes, with its `sql`, `subject` and duration as attributes (see [Logging](../connections.md#logging)); the log is for watching, and the report is still the structured answer.
 
 ## Two processes at once
 

src/backend.rs

@@ -14,6 +14,7 @@ use std::collections::{HashMap, HashSet};
 use std::fmt;
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, RwLock};
+use std::time::Instant;
 
 /// Infer the SQL dialect / backend from a connection-URL scheme.
 pub fn dialect_from_url(url: &str) -> Result<Dialect, UnsupportedDatabaseUrl> {
@@ -791,12 +792,24 @@ impl EngineHandle {
                 } else {
                     "BEGIN"
                 };
-                sqlx::query(begin).execute(&mut *conn).await?;
+                let started = Instant::now();
+                let result = sqlx::query(begin)
+                    .execute(&mut *conn)
+                    .await
+                    .map(|done| done.rows_affected());
+                observe_rows(begin, Dialect::Sqlite, started, &result);
+                result?;
                 Ok(EngineConnection::Sqlite(conn))
             }
             BackendPool::Postgres(pool) => {
                 let mut conn = pool.acquire().await?;
-                sqlx::query("BEGIN").execute(&mut *conn).await?;
+                let started = Instant::now();
+                let result = sqlx::query("BEGIN")
+                    .execute(&mut *conn)
+                    .await
+                    .map(|done| done.rows_affected());
+                observe_rows("BEGIN", Dialect::Postgres, started, &result);
+                result?;
                 Ok(EngineConnection::Postgres(conn))
             }
         }
@@ -825,22 +838,22 @@ impl EngineConnection {
     /// Migration DDL must use this: caching a statement against a schema the
     /// same migration is about to change would poison the connection.
     pub async fn execute_sql_unprepared(&mut self, sql: &str) -> Result<u64, sqlx::Error> {
-        match self {
-            EngineConnection::Sqlite(conn) => {
-                let result = sqlx::query(sql)
-                    .persistent(false)
-                    .execute(&mut **conn)
-                    .await?;
-                Ok(result.rows_affected())
-            }
-            EngineConnection::Postgres(conn) => {
-                let result = sqlx::query(sql)
-                    .persistent(false)
-                    .execute(&mut **conn)
-                    .await?;
-                Ok(result.rows_affected())
-            }
-        }
+        let dialect = self.dialect();
+        let started = Instant::now();
+        let result = match self {
+            EngineConnection::Sqlite(conn) => sqlx::query(sql)
+                .persistent(false)
+                .execute(&mut **conn)
+                .await
+                .map(|done| done.rows_affected()),
+            EngineConnection::Postgres(conn) => sqlx::query(sql)
+                .persistent(false)
+                .execute(&mut **conn)
+                .await
+                .map(|done| done.rows_affected()),
+        };
+        observe_rows(sql, dialect, started, &result);
+        result
     }
 
     /// Fetch without entering the connection's prepared-statement cache — the
@@ -851,24 +864,23 @@ impl EngineConnection {
         sql: &str,
         values: &[EngineBindValue],
     ) -> Result<Vec<EngineRow>, sqlx::Error> {
-        match self {
+        let dialect = self.dialect();
+        let started = Instant::now();
+        let result = match self {
             EngineConnection::Sqlite(conn) => {
                 let mut query = sqlx::query(sql).persistent(false);
                 for value in values {
                     query = bind_engine_value(query, value);
                 }
-                let rows = query.fetch_all(&mut **conn).await?;
-                Ok(rows.iter().map(materialize_engine_row).collect())
-            }
-            EngineConnection::Postgres(conn) => {
-                let mut query = sqlx::query(sql).persistent(false);
-                for value in values {
-                    query = bind_engine_value(query, value);
-                }
-                let rows = query.fetch_all(&mut **conn).await?;
-                Ok(rows.iter().map(materialize_pg_row).collect())
+                query
+                    .fetch_all(&mut **conn)
+                    .await
+                    .map(|rows| rows.iter().map(materialize_engine_row).collect())
             }
-        }
+            EngineConnection::Postgres(conn) => fetch_all_pg_unprepared(conn, sql, values).await,
+        };
+        observe_fetch(sql, values, dialect, started, &result);
+        result
     }
 
     pub async fn execute_sql_with_binds(
@@ -887,55 +899,77 @@ impl EngineConnection {
         sql: &str,
         values: &[EngineBindValue],
     ) -> Result<EngineExecuteResult, sqlx::Error> {
-        match self {
+        let dialect = self.dialect();
+        let started = Instant::now();
+        let result = match self {
             EngineConnection::Sqlite(conn) => {
                 let mut query = sqlx::query(sql);
                 for value in values {
                     query = bind_engine_value(query, value);
                 }
-                let result = query.execute(&mut **conn).await?;
-                Ok(EngineExecuteResult {
-                    rows_affected: result.rows_affected(),
-                    last_insert_id: Some(result.last_insert_rowid()),
-                })
+                query
+                    .execute(&mut **conn)
+                    .await
+                    .map(|done| EngineExecuteResult {
+                        rows_affected: done.rows_affected(),
+                        last_insert_id: Some(done.last_insert_rowid()),
+                    })
             }
             EngineConnection::Postgres(conn) => {
                 let mut query = sqlx::query(sql);
                 for value in values {
                     query = bind_engine_value(query, value);
                 }
-                let result = query.execute(&mut **conn).await?;
-                Ok(EngineExecuteResult {
-                    rows_affected: result.rows_affected(),
-                    last_insert_id: None,
-                })
+                query
+                    .execute(&mut **conn)
+                    .await
+                    .map(|done| EngineExecuteResult {
+                        rows_affected: done.rows_affected(),
+                        last_insert_id: None,
+                    })
             }
-        }
+        };
+        crate::sql_log::observe(
+            sql,
+            values,
+            dialect,
+            started,
+            result.as_ref().map(|done| Some(done.rows_affected)),
+        );
+        result
     }
 
     pub async fn fetch_all_sql_with_binds(
         &mut self,
         sql: &str,
         values: &[EngineBindValue],
     ) -> Result<Vec<EngineRow>, sqlx::Error> {
-        match self {
+        let dialect = self.dialect();
+        let started = Instant::now();
+        let result = match self {
             EngineConnection::Sqlite(conn) => {
                 let mut query = sqlx::query(sql);
                 for value in values {
                     query = bind_engine_value(query, value);
                 }
-                let rows = query.fetch_all(&mut **conn).await?;
-                Ok(rows.iter().map(materialize_engine_row).collect())
+                query
+                    .fetch_all(&mut **conn)
+                    .await
+                    .map(|rows| rows.iter().map(materialize_engine_row).collect())
             }
             EngineConnection::Postgres(conn) => {
                 let mut query = sqlx::query(sql);
                 for value in values {
                     query = bind_engine_value(query, value);
                 }
-                let rows = query.fetch_all(&mut **conn).await?;
-                Ok(rows.iter().map(materialize_pg_row).collect())
+                query
+                    .fetch_all(&mut **conn)
+                    .await
+                    .map(|rows| rows.iter().map(materialize_pg_row).collect())
             }
-        }
+        };
+        observe_fetch(sql, values, dialect, started, &result);
+        result
     }
 
     pub async fn commit(&mut self) -> Result<(), sqlx::Error> {
@@ -979,6 +1013,65 @@ impl EngineConnection {
     }
 }
 
+/// Fetch on a Postgres connection the pool no longer owns, without entering
+/// its prepared-statement cache, and write the statement to the statement
+/// log: the run lock's own session (`crate::run::RunLock`) sends its
+/// `pg_try_advisory_lock`, its held-by-this-session check and its
+/// `pg_advisory_unlock` through here, so they are records like any other.
+///
+/// # Errors
+/// The `sqlx::Error` the database returned; it is in the record too.
+pub async fn fetch_all_detached_pg_unprepared(
+    conn: &mut sqlx::PgConnection,
+    sql: &str,
+    values: &[EngineBindValue],
+) -> Result<Vec<EngineRow>, sqlx::Error> {
+    let started = Instant::now();
+    let result = fetch_all_pg_unprepared(conn, sql, values).await;
+    observe_fetch(sql, values, Dialect::Postgres, started, &result);
+    result
+}
+
+/// The Postgres fetch both an [`EngineConnection`] and a detached connection
+/// send; the caller writes the record.
+async fn fetch_all_pg_unprepared(
+    conn: &mut sqlx::PgConnection,
+    sql: &str,
+    values: &[EngineBindValue],
+) -> Result<Vec<EngineRow>, sqlx::Error> {
+    let mut query = sqlx::query(sql).persistent(false);
+    for value in values {
+        query = bind_engine_value(query, value);
+    }
+    query
+        .fetch_all(conn)
+        .await
+        .map(|rows| rows.iter().map(materialize_pg_row).collect())
+}
+
+/// The statement log's record of an unbound execute that returned `result`'s
+/// rows affected.
+fn observe_rows(sql: &str, dialect: Dialect, started: Instant, result: &Result<u64, sqlx::Error>) {
+    crate::sql_log::observe(
+        sql,
+        &[],
+        dialect,
+        started,
+        result.as_ref().map(|n| Some(*n)),
+    );
+}
+
+/// The statement log's record of a fetch: it reports no rows affected.
+fn observe_fetch(
+    sql: &str,
+    binds: &[EngineBindValue],
+    dialect: Dialect,
+    started: Instant,
+    result: &Result<Vec<EngineRow>, sqlx::Error>,
+) {
+    crate::sql_log::observe(sql, binds, dialect, started, result.as_ref().map(|_| None));
+}
+
 fn materialize_engine_row<R>(row: &R) -> EngineRow
 where
     R: Row,