iii-hq · andersonleal · Jul 28, 2026 · Jul 28, 2026
diff --git a/.github/workflows/database-e2e.yml b/.github/workflows/database-e2e.yml
@@ -49,9 +49,8 @@ jobs:
           cache: 'npm'
           cache-dependency-path: database/tests/e2e/workers/harness/package-lock.json
 
-      # GHA `services:` blocks can't pass `-c wal_level=logical` to postgres,
-      # which the row-change tests require. Reuse the same docker-compose
-      # stack the harness uses locally for dev/CI parity.
+      # The harness brings up its own postgres + mysql via docker-compose
+      # (not GHA `services:`) so local and CI runs share one stack definition.
       - name: Install iii engine (next)
         run: |
           curl -fsSL --retry 3 --retry-connrefused --retry-delay 5 \

diff --git a/database/Cargo.lock b/database/Cargo.lock
diff --git a/database/Cargo.toml b/database/Cargo.toml
@@ -33,13 +33,11 @@ url = "2"
 chrono = { version = "0.4", features = ["serde"] }
 base64 = "0.22"
 bytes = "1"
-futures-util = "0.3"
 
 # Postgres
 tokio-postgres = { version = "0.7", features = ["with-chrono-0_4", "with-serde_json-1", "with-uuid-1"] }
 deadpool-postgres = "0.14"
 postgres-types = { version = "0.2", features = ["with-chrono-0_4", "with-serde_json-1"] }
-postgres-protocol = "0.6"
 tokio-postgres-rustls = "0.13"
 # `db-tokio-postgres` registers FromSql/ToSql for NUMERIC. postgres-types
 # itself ships no NUMERIC FromSql impl — String accepts only TEXT-family OIDs,

diff --git a/database/README.md b/database/README.md
@@ -195,21 +195,47 @@ const { rows } = await iii.trigger({
 
 ## Triggers
 
-### `database::row-change`
-Postgres only. Streams row-level changes via logical replication (`pgoutput`).
+### `database::row-changed`
 
-> **NOTE (v1.0.0):** Event dispatch is not yet functional. The publication and replication slot are created at startup, but the streaming decode loop is stubbed pending an upstream `tokio-postgres` replication API release. Operators can pre-provision slots and publications now; events will start flowing in a later release.
+Fires after this worker commits a row change. Driver-agnostic — no logical
+replication, no per-database setup, identical on SQLite, Postgres and MySQL.
 
 ```yaml
 triggers:
-  - type: database::row-change
+  - type: database::row-changed
     config:
-      db: primary
-      schema: public
-      tables: [orders, payments]
+      db: primary        # required
+      table: orders      # optional; case- and schema-insensitive
+      ops: [insert]      # optional; insert / update / delete / other
 ```
 
-The worker derives slot/publication names from `trigger_id`: `iii_slot_<sanitized>_<8hex>` and `iii_pub_<sanitized>_<8hex>`, where the 8-hex-char suffix is an FNV-1a-32 hash of the original `trigger_id`. The hash guarantees that two distinct trigger_ids (e.g. `orders-v1` vs `orders.v1`) produce distinct names even though both sanitize to `orders_v1`. The sanitized prefix is truncated at 40 chars so the final name fits in Postgres' 63-byte slot-name limit. Operators can override slot/publication names explicitly with `slot_name`/`publication_name`. Drop them with `pg_drop_replication_slot('<slot>')` and `DROP PUBLICATION <name>` if the worker is decommissioned without graceful shutdown.
+Event: `{ db, table, op, affected_rows, returning?, at }`, where `op` is
+`insert` / `update` / `delete` / `other`.
+
+**This is not change data capture.** It reports mutations made *through this
+worker* — `execute`, `executeBatch`, `transaction`, and the interactive
+transaction surface. A write applied by psql, another worker, or a
+database-side trigger is invisible to it. That covers the case it exists for
+(the worker is the only writer, and something needs to know when rows land)
+and nothing more.
+
+Four things worth knowing:
+
+- **Announced on commit, never before.** Statements inside an interactive
+  transaction are buffered until `commitTransaction`; a rollback — including
+  the timeout watcher's — drops the buffer. Atomic batches announce their
+  statements in order only after the whole batch commits.
+- **Delivery is best-effort.** Dispatch happens after commit and is not durable
+  or atomic with the database write. There is no replay, retry, or exactly-once
+  guarantee; a crash between commit and dispatch can lose an event. Subscriber
+  failures are logged and never fail the write.
+- **`table` can be null.** The table is read off the SQL. A CTE-wrapped write
+  (`WITH … INSERT`) still fires, with `table: null`, rather than being dropped;
+  a binding that named a table simply does not match it. Omit `table` to match
+  every write, including these.
+- **`runStatement` does not fire.** The prepared-run path returns rows, not an
+  affected-row count, and an event that invented one would be lying. Use
+  `execute` when you need the change announced.
 
 ## Errors
 
@@ -224,8 +250,6 @@ Returned `IIIError::Handler` bodies carry a stable `code` field:
 | `UNKNOWN_DB` | `db` parameter doesn't match any configured database. |
 | `INVALID_PARAM` | JSON value couldn't be coerced for the target driver, transaction-control SQL was sent to `transactionExecute` (use `commitTransaction` / `rollbackTransaction`), or a `transaction`/`executeBatch` batch contained transaction-control SQL or an empty statement. |
 | `DRIVER_ERROR` | Wraps underlying driver error with `driver` and `inner_code` (nullable). `inner_code` format is per-driver: Postgres = SQLSTATE 5-char string (e.g. `42P01`), MySQL = server error number as string, SQLite = `rusqlite::ErrorCode` debug name. Pool-acquire failures use the message form `pool connection failed (<class>)` where `<class>` is one of `tls`, `auth`, `network`, `server-policy`, or `unknown` — a redacted hint so untrusted callers can self-triage without seeing host/userinfo/db fragments. The full driver error is in the worker's stderr via `tracing::warn!`. |
-| `REPLICATION_SLOT_EXISTS` | Startup-only: another instance owns the slot. |
-| `UNSUPPORTED` | Operation not supported on the chosen driver. |
 | `CONFIG_ERROR` | Config parse or pool init failure. |
 
 ## Driver compatibility
@@ -237,7 +261,6 @@ A few operations are no-ops on certain drivers. They emit a `tracing::warn!` rat
 | `execute` with `returning: [...]` | ✓ | ✓ | warn-once + ignore |
 | `transaction` `isolation: read_committed` / `repeatable_read` | warn + use serializable | ✓ | ✓ |
 | `transaction` `isolation: serializable` | ✓ (`BEGIN IMMEDIATE`) | ✓ | ✓ |
-| `database::row-change` trigger | — | setup-only in v1.0.0 (see above) | — |
 
 
 ## Troubleshooting
@@ -249,7 +272,6 @@ A few operations are no-ops on certain drivers. They emit a `tracing::warn!` rat
     - `(auth)` — credential or pg_hba/SCRAM rejection. Includes Neon's `?channel_binding=require` failing through the pooler endpoint (drop the URL param, use `tls.mode` in YAML).
     - `(network)` — TCP refuse, DNS, route, or peer reset. Check host/port reachability and any firewalls.
     - `(server-policy)` — server reachable and TLS+auth OK, but the server actively refused (e.g. `max_connections` exceeded, admin shutdown). Look at the worker stderr for the underlying driver message.
-- **Replication slot already exists**: another instance is consuming the slot. Either reuse the slot name or run `SELECT pg_drop_replication_slot('<slot>')`.
 
 ## License
 

diff --git a/database/skills/SKILL.md b/database/skills/SKILL.md
@@ -29,14 +29,13 @@ point. Placeholder syntax: `?` for SQLite and MySQL, `$1`/`$2`/… for Postgres.
   `database::runStatement`).
 - You need read-your-writes across round-trips with logic between steps
   (`database::beginTransaction` … `commitTransaction` / `rollbackTransaction`).
-- You want to react to Postgres row-level changes once logical replication
-  streaming ships (`database::row-change` trigger — see below).
 
 ## Boundaries
 
 - Not a migration tool, ORM, or schema designer — pass raw SQL only.
-- Not a general pub/sub bus — use `database::row-change` only for Postgres
-  table change feeds, not for application events.
+- Not a general pub/sub bus. `database::row-changed` reports only what THIS
+  worker wrote, on commit — not change data capture; a write from psql or
+  another worker is invisible to it.
 - `database::query` is read-oriented; use `database::execute` for writes.
   Running a SELECT through `execute` discards rows.
 - Prepared handles pin a pool connection until TTL expiry — not transactions.
@@ -82,45 +81,19 @@ Interactive transactions auto-roll back when `timeout_ms` elapses (default
 30 s, max 5 min). Prepared handles default to a 1 h TTL (max 24 h) with no
 explicit release call — let them expire or stop using them when done.
 
-## Reactive triggers
+## Reacting to writes
 
-Register a `database::row-change` trigger when a function should run
-automatically on Postgres INSERT/UPDATE/DELETE for specific tables — without
-polling with `database::query`.
+Register a `database::row-changed` trigger to be told when this worker commits
+a change, instead of polling:
 
-Reach for it when:
-
-- A downstream worker or workflow must react to row mutations in near real
-  time on Postgres.
-- You need decoded row payloads (old/new values) from logical replication
-  rather than polling an outbox table.
-
-Do not bind when:
-
-- The writer already has the new row in its `execute` or `transactionExecute`
-  return payload.
-- You are on SQLite or MySQL — this trigger type is Postgres-only.
-- You need events today — v1.0.0 returns `UNSUPPORTED` on `registerTrigger`
-  pending an upstream `tokio-postgres` replication API release.
-
-### How to bind
-
-1. Register a handler: `registerFunction('stream::on-row-change', handler)`.
-2. Register the trigger:
-
-```typescript
-iii.registerTrigger({
-  type: 'database::row-change',
-  function_id: 'stream::on-row-change',
-  config: {
-    db: 'primary',
-    schema: 'public',
-    tables: ['orders', 'payments'],
-    // optional: slot_name, publication_name — see get function info
-  },
-})
+```json
+{ "trigger_type": "database::row-changed", "config": { "db": "primary", "table": "orders", "ops": ["insert"] } }
 ```
 
-Config: `db`, `schema` (default `public`), `tables`. Slot/publication names
-derive from `trigger_id` unless overridden. For event payload shape, call
-`get function info` on the trigger type or handler function id.
+The event is `{ db, table, op, affected_rows, returning?, at }`. It fires on
+commit — an interactive transaction's writes are announced by
+`commitTransaction`, and a rollback announces nothing. `table` is null when the
+statement's table cannot be read off the SQL (a CTE-wrapped write), and
+`runStatement` does not fire because it has no affected-row count to report.
+Delivery is best-effort: it is not durable with the commit and has no replay or
+exactly-once guarantee.
diff --git a/database/src/config.rs b/database/src/config.rs
@@ -298,8 +298,8 @@ pub fn redact_url(input: &str) -> String {
 /// Max 63 chars (Postgres NAMEDATALEN - 1).
 ///
 /// This is the chokepoint for any operator-supplied identifier that gets
-/// interpolated into a SQL string via `format!()` (replication slots,
-/// publication names, schema/table names, cursor table). Validation is
+/// interpolated into a SQL string via `format!()` (schema/table names,
+/// cursor table). Validation is
 /// strict ASCII because the alternative — quoting and escaping per-driver —
 /// is fragile and the v1.0 surface does not need unicode identifiers.
 pub fn validate_sql_identifier(s: &str) -> Result<(), String> {

diff --git a/database/src/driver/mod.rs b/database/src/driver/mod.rs
@@ -51,4 +51,5 @@ pub struct TxStatement {
 pub struct TxStepResult {
     pub affected_rows: u64,
     pub rows: Vec<Row>,
+    pub columns: Vec<ColumnMeta>,
 }
diff --git a/database/src/driver/mysql.rs b/database/src/driver/mysql.rs
@@ -205,6 +205,14 @@ pub async fn transaction(
         let step_result: Result<TxStepResult, DbError> = if returns_rows {
             match conn.exec_iter(stmt.sql.as_str(), bound).await {
                 Ok(mut iter) => {
+                    let columns = iter
+                        .columns_ref()
+                        .iter()
+                        .map(|col| ColumnMeta {
+                            name: col.name_str().to_string(),
+                            ty: format!("{:?}", col.column_type()),
+                        })
+                        .collect();
                     let raw: Result<Vec<mysql_async::Row>, _> = iter.collect().await;
                     match raw {
                         Ok(raw_rows) => {
@@ -213,6 +221,7 @@ pub async fn transaction(
                             Ok(TxStepResult {
                                 affected_rows: cells_rows.len() as u64,
                                 rows: cells_rows,
+                                columns,
                             })
                         }
                         Err(e) => Err(step_err(idx, e)),
@@ -225,6 +234,7 @@ pub async fn transaction(
                 Ok(_) => Ok(TxStepResult {
                     affected_rows: conn.affected_rows(),
                     rows: vec![],
+                    columns: vec![],
                 }),
                 Err(e) => Err(step_err(idx, e)),
             }

diff --git a/database/src/driver/postgres.rs b/database/src/driver/postgres.rs
@@ -384,6 +384,18 @@ pub async fn transaction(
         let step = if returns_rows {
             match tx_client.query(&stmt.sql, bound_refs.as_slice()).await {
                 Ok(rows) => {
+                    let columns = rows
+                        .first()
+                        .map(|row| {
+                            row.columns()
+                                .iter()
+                                .map(|col| ColumnMeta {
+                                    name: col.name().to_string(),
+                                    ty: col.type_().name().to_string(),
+                                })
+                                .collect()
+                        })
+                        .unwrap_or_default();
                     let mut cells_rows: Vec<Row> = Vec::with_capacity(rows.len());
                     for row in &rows {
                         let mut cells = Vec::with_capacity(row.columns().len());
@@ -395,6 +407,7 @@ pub async fn transaction(
                     TxStepResult {
                         affected_rows: cells_rows.len() as u64,
                         rows: cells_rows,
+                        columns,
                     }
                 }
                 Err(e) => {
@@ -407,6 +420,7 @@ pub async fn transaction(
                 Ok(n) => TxStepResult {
                     affected_rows: n,
                     rows: vec![],
+                    columns: vec![],
                 },
                 Err(e) => {
                     let _ = tx_client.batch_execute("ROLLBACK").await;
@@ -456,6 +470,10 @@ pub async fn tx_begin(
 
 /// `COMMIT` the in-progress transaction on a pinned client.
 pub async fn tx_commit(client: &mut crate::pool::postgres::PgClient) -> Result<(), DbError> {
+    // PostgreSQL accepts COMMIT in an aborted transaction but reports the
+    // command tag ROLLBACK; batch_execute discards that tag. Probe while the
+    // transaction is still open so 25P02 takes the ordinary commit-error path.
+    client.simple_query("SELECT 1").await.map_err(map_err)?;
     client.batch_execute("COMMIT").await.map_err(map_err)
 }
 
@@ -1093,6 +1111,48 @@ mod tests {
         assert!(matches!(&r.rows[0].0[0], RowValue::BigInt(0)));
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn pg_interactive_commit_rejects_an_aborted_transaction() {
+        let Some(p) = fresh_pool().await else { return };
+        let _ = execute(&p, "DROP TABLE IF EXISTS db_w_aborted_tx", &[], &[]).await;
+        execute(
+            &p,
+            "CREATE TABLE db_w_aborted_tx (n INT NOT NULL)",
+            &[],
+            &[],
+        )
+        .await
+        .unwrap();
+
+        let mut client = p.acquire().await.unwrap();
+        tx_begin(&mut client, None).await.unwrap();
+        tx_execute(
+            &mut client,
+            "INSERT INTO db_w_aborted_tx VALUES (1)",
+            &[],
+            &[],
+        )
+        .await
+        .unwrap();
+        tx_execute(
+            &mut client,
+            "INSERT INTO db_w_aborted_tx VALUES (NULL)",
+            &[],
+            &[],
+        )
+        .await
+        .unwrap_err();
+
+        assert!(tx_commit(&mut client).await.is_err());
+        tx_rollback(&mut client).await.unwrap();
+        drop(client);
+
+        let rows = query(&p, "SELECT COUNT(*) FROM db_w_aborted_tx", &[], 30_000)
+            .await
+            .unwrap();
+        assert!(matches!(&rows.rows[0].0[0], RowValue::BigInt(0)));
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn pg_run_prepared_executes_with_params() {
         let Some(p) = fresh_pool().await else { return };