Statement log: every operation statement, with secrets redacted #765

Parent

#700

What to build

Every statement an operation sends is in the statement log, with its parameters, and a secret never is. After this ticket:

class Account(Model):
    id: int = Field(primary_key=True)
    email: str = Field(unique=True)
    token: SecretStr

logging.getLogger("ferro.sql").setLevel(logging.DEBUG)
async with engines.session():
    await Account(id=1, email="ada@example.com", token=SecretStr("sk_live_1")).save()
    # ferro.sql DEBUG INSERT INTO "account" ("id", "email", "token") VALUES ($1, $2, $3) params=(1, 'ada@example.com', ***) 0.6ms
    await Account.where(lambda account: account.token == "sk_live_1").first()
    # ferro.sql DEBUG SELECT ... WHERE "account"."token" = $1 LIMIT $2 params=(***, 1) 0.3ms
    await raw.execute("UPDATE account SET token = $1", SecretStr("sk_live_2"))
    # ferro.sql DEBUG UPDATE account SET token = $1 params=(***,) 0.2ms

Every FFI operation entry opens the statement context the logger ticket built, naming its operation (fetch, count, exists, save, update, delete, bulk_create, m2m_add, m2m_remove, m2m_clear, raw, transaction, settings, catalog) and its connection (the route's connection name). Transaction control (BEGIN, COMMIT, ROLLBACK, SAVEPOINT, RELEASE, ROLLBACK TO), the session-settings SET LOCAL batch and the Postgres catalog probe are logged under their own operation names; a chunked bulk_create or a many-to-many add leaves one record per statement under the same name. rows carries the driver's affected-row count for execute statements and None for fetches.

Redaction is one rule applied at emit time: a bound value is masked when it is in the operation's secret-value set, which is decided once per operation from two facts Python owns. Python collects the values of every secret-typed literal and field value it unwraps (compile_query for predicate literals, the bind-payload helpers for save/update/bulk_create values, raw for its arguments) and passes them as secret_values; Rust, at entry, adds the values bound to columns whose field the model declares SecretStr/SecretBytes (read from the hydration-converter registry #760 added, for the root model and every hop class). The engine layer masks a bind that equals any value in the set. This is where the ADR's "per-bind flag" lives: a set of values rather than a positional list, because the binds are made by the query builder in Rust and never pass through Python one by one. Session settings are statement text (SET LOCAL takes no parameters) and are not masked.

.sql() does not exist yet; this ticket is complete when the log tells the whole truth for every operation and no secret reaches it.

Acceptance criteria

• Each of these leaves records with the expected operation, on SQLite and Postgres: a where/order_by/include fetch (fetch, one statement), count and exists, save for a new and a fetched instance (save), bulk_create across a chunk boundary (bulk_create, one record per chunk), Query.update() and Query.delete(), Model.delete(), a many-to-many add/remove/clear, raw.execute/fetch_all/fetch_one (raw).
• A transaction() block with a nested transaction() leaves BEGIN, SAVEPOINT …, RELEASE …/ROLLBACK TO … and COMMIT records with operation == "transaction", in order, between the statements of the block.
• A session with settings on Postgres leaves the SET LOCAL records (operation == "settings") before the operation's statement inside the same transaction; the catalog probe leaves a catalog record the first time a table is touched and none the second time.
• Every record's connection is the route's connection name and dialect the backend's; rows is the affected-row count for execute statements and None for fetches.
• A SecretStr/SecretBytes field value in save, bulk_create and update, a plain literal compared against a secret column, a secret-typed literal compared against any column, and a secret-typed raw argument each appear as REDACTED (by identity) in params, and the real value appears in no record's message or params.
• A non-secret value is never masked unless it equals a secret value of the same operation (documented false-positive direction).
• With ferro.sql not enabled for DEBUG, no operation calls the emit callable and no secret-value set is built (asserted with a spy and a counter on the collecting helpers).
• A statement that fails (a unique violation on save) leaves a record with error set; the IntegrityError still propagates.
• compile_query's artifact carries secret_values and the wire JSON is unchanged (golden vectors untouched); the compiled-query pin covers the new field.

Files owned

• Modify: src/operations.rs, src/session_settings.rs, src/sql_log.rs, src/hydration.rs, src/ferro/query/wire.py, src/ferro/query/builder.py, src/ferro/_bind_payload.py, src/ferro/models.py, src/ferro/raw.py, src/ferro/_core.pyi
• Test: tests/test_statement_log.py (the operation, transaction, settings, catalog and redaction cases), tests/test_compiled_query.py (the secret_values field)

Interfaces

• Consumes: sql_log::{scope, StatementContext, observe} and ferro._sql_log.{emit, REDACTED} from Statement log: the ferro.sql logger and the Rust emit path, proven on migration DDL #764 (shipped in PR feat: statement log records on ferro.sql for migration DDL (#764) #768; scope is async fn scope<F, T>(ctx: Option<StatementContext>, fut: F) -> PyResult<T> where F: Future<Output = PyResult<T>>, so an emitter error comes back as the operation's result: wrap each FFI entry's future, which already returns PyResult); the statement_records fixture.
• Consumes: unwrap_secret and the hydration-converter registry (cls._hydration_converters, kind "secret") from SecretStr and SecretBytes fields save as '**********' and hydrate as plain str #760; field_converters_for in src/hydration.rs.
• Produces: CompiledQuery.secret_values: tuple[object, ...] (the unwrapped values of secret-typed literals, collected by compile_query; empty when none), used by the rendering ticket.
• Produces: secret_values(instance) -> tuple[object, ...] and secret_values_of(fields: Mapping[str, object]) -> tuple[object, ...] in src/ferro/_bind_payload.py, used by save, bulk_create, update.
• Produces: a keyword secret_values: list[object] on the FFI entries fetch_filtered, count_filtered, update_filtered, delete_filtered, save_record, update_record, save_bulk_records, raw_execute, raw_fetch_all, raw_fetch_one, declared in _core.pyi; the rendering ticket's render_* twins take the same keyword.
• Produces (Rust): sql_log::secret_columns_for(py, cls) -> PyResult<HashSet<String>> (reads the converter registry), and a StatementContext opened at every operation entry with operation and connection set and secret_values filled from the keyword plus the secret-column values of the plan/payload.

Test scenarios

• Happy path: each operation family above → records with the expected operation, connection, dialect, rows, and one record per statement sent, on both backends.
• Happy path: nested transaction() → BEGIN, SAVEPOINT, inner statement, RELEASE, outer statement, COMMIT, in that order.
• Happy path (Postgres): session settings → SET LOCAL records before the statement, inside the same transaction; first touch of a table → one catalog record, second touch → none.
• Happy path: the four redaction routes (secret field value on save/update/bulk; plain literal against a secret column; secret literal against a plain column; secret raw argument) → REDACTED by identity, real value absent from every message.
• Edge case: bulk_create of two rows where one non-secret value equals another row's secret value → both masked (false positive is the safe direction; asserted so the behaviour is pinned).
• Edge case: ferro.sql at WARNING → emit never called, secret-value collection never invoked.
• Error path: save() violating a unique index → record with error, IntegrityError propagates.
• Integration: tests/test_compiled_query.py → CompiledQuery.secret_values is () for a query with no secret literal and ("hunter2",) for one with SecretStr("hunter2"); wire JSON identical in both cases.

Verify

uv run maturin develop
uv run --no-sync pytest tests/test_statement_log.py tests/test_compiled_query.py tests/test_ir_vectors_contract.py tests/test_secret_fields.py tests/test_session_set_config.py -q --no-cov -p no:cacheprovider
cargo test --no-default-features --features testing
uv run --no-sync ruff check src/ferro tests
uv run --no-sync ty check src/ferro/query tests/test_query_typing.py tests/test_static_contracts.py tests/test_alembic_typing.py

Set FERRO_POSTGRES_URL and pass --db-backends=sqlite,postgres so the Postgres cases run rather than skip.

Blocked by

• Statement log: the ferro.sql logger and the Rust emit path, proven on migration DDL #764 (the ferro.sql logger and the Rust emit path).