feat: in_transaction() runtime predicate on Backend protocol (Gap 4)

src/mountainash_data/backends/ibis/_transaction.py

@@ -132,3 +132,16 @@ def _exec(sql: str) -> None:
     finally:
         with _LOCK:
             state.depth -= 1
+
+
+def is_active(raw_handle: t.Any) -> bool:
+    """True if a unit of work is registered on this raw handle (any depth).
+
+    Read-only companion to run_transaction(): reads the ambient registry under
+    the same lock the register/unregister critical sections take, so the read
+    observes a fully-committed registry mutation. Never mutates _ACTIVE. Keyed
+    on id(raw_handle), matching run_transaction's reentrancy key, so distinct
+    IbisBackend wrappers of one raw connection agree.
+    """
+    with _LOCK:
+        return id(raw_handle) in _ACTIVE

src/mountainash_data/backends/ibis/backend.py

@@ -11,7 +11,7 @@
 import typing as t
 
 from mountainash_data.backends.ibis.dialects._registry import DIALECTS, DialectSpec, TransactionSupport
-from mountainash_data.backends.ibis._transaction import run_transaction
+from mountainash_data.backends.ibis._transaction import run_transaction, is_active
 from mountainash_data.backends.ibis._adoption import (
     apply_options, snapshot_options, restore_options,
 )
@@ -567,6 +567,30 @@ def transaction(self, *, required: bool = True) -> t.ContextManager[None]:
             raw_execute_hook=self._spec.raw_execute_hook,
         )
 
+    def in_transaction(self) -> bool:
+        """True if a unit of work opened via transaction() is currently active
+        on this backend's raw connection (any nesting depth).
+
+        Runtime companion to the static supports_transactions flag. Total:
+        returns False — never raises — for NONE dialects, a backend that was
+        never connected or has been closed, and a connection whose native
+        handle has gone away. A point-in-time snapshot, not a lock.
+        """
+        if self._spec.transaction_support is TransactionSupport.NONE:
+            return False
+        if self._conn is None:
+            return False
+        try:
+            raw = self.raw_driver_connection()
+        except Exception:
+            # Unresolvable/absent native handle == no live unit of work.
+            # raw_driver_connection() raises RuntimeError on an absent handle
+            # attr; a property-backed handle could raise a driver-specific
+            # error on a dropped connection. Either way the answer is "no
+            # active tx"; a genuine fault surfaces on the caller's next op.
+            return False
+        return is_active(raw)
+
     # --- Inspection (terminal — delegates to IbisConnection) ---
 
     def list_tables(self, namespace: NamespaceLike = None) -> list[str]:

src/mountainash_data/backends/iceberg/backend.py

@@ -103,3 +103,9 @@ def transaction(self, *, required: bool = True):
             )
         warn_once("iceberg", "iceberg has no transaction support; transaction() is a no-op.")
         yield
+
+    def in_transaction(self) -> bool:
+        """Iceberg has no connection-level unit of work (supports_transactions
+        is False); nothing can be active. pyiceberg table-scoped transactions
+        are a separate future capability."""
+        return False

src/mountainash_data/core/protocol.py

@@ -70,3 +70,19 @@ def transaction(self, *, required: bool = True) -> t.ContextManager[None]:
         (an adopted autocommit-off connection is refused with
         TransactionIntegrityError). See spec §5.1–5.3."""
         ...
+
+    def in_transaction(self) -> bool:
+        """True if a unit of work opened via transaction() is currently active
+        on this backend's connection (any nesting depth). Runtime companion to
+        the static supports_transactions flag.
+
+        Total: returns False — never raises — for backends with no transaction
+        concept, a backend that was never connected or has been closed, and a
+        connection whose native handle has gone away. This is a point-in-time
+        snapshot, NOT a lock: an answer may be stale the instant it returns, so
+        it must not be used as a correctness-critical mutual-exclusion gate
+        without external coordination. Intended use is an ownership guard
+        ("refuse to run nested inside an ambient unit of work"). Note: True
+        means a unit of work is open, not that a fresh transaction() join would
+        succeed — a poisoned unit is still reported active until it unwinds."""
+        ...

tests/test_integration/test_transaction_live.py

@@ -1,29 +1,21 @@
-import os
 import pytest
-from mountainash_data import IbisBackend
 
 pytestmark = pytest.mark.integration
 
-REQUIRE = os.environ.get("MOUNTAINASH_REQUIRE_LIVE_DB") == "1"
-PG_URL = os.environ.get("MOUNTAINASH_TEST_POSTGRES_URL")
 
-
-def _skip_or_fail(reason):
-    if REQUIRE:
-        pytest.fail(reason)
-    pytest.skip(reason)
-
-
-def test_postgres_transaction_rollback():
-    if not PG_URL:
-        _skip_or_fail("MOUNTAINASH_TEST_POSTGRES_URL not set")
-    with IbisBackend(PG_URL) as be:
-        raw = be.raw_driver_connection()
-        cur = raw.cursor()
-        cur.execute("CREATE TEMP TABLE t_tx (x INT)")
-        with pytest.raises(ValueError):
-            with be.transaction():
-                cur.execute("INSERT INTO t_tx VALUES (1)")
-                raise ValueError("boom")
-        cur.execute("SELECT count(*) FROM t_tx")
-        assert cur.fetchone()[0] == 0
+def test_postgres_transaction_rollback(postgres_backend):
+    """A transaction() that raises must roll the whole unit of work back on a
+    real postgres connection. Uses the shared postgres_backend fixture (reads
+    IBIS_TEST_POSTGRES_*, connects to the live service, and honours
+    MOUNTAINASH_REQUIRE_LIVE_DB=1 by fail-closing only when the service is
+    genuinely unreachable) — the same convention as every other live test."""
+    be = postgres_backend
+    raw = be.raw_driver_connection()
+    cur = raw.cursor()
+    cur.execute("CREATE TEMP TABLE t_tx (x INT)")
+    with pytest.raises(ValueError):
+        with be.transaction():
+            cur.execute("INSERT INTO t_tx VALUES (1)")
+            raise ValueError("boom")
+    cur.execute("SELECT count(*) FROM t_tx")
+    assert cur.fetchone()[0] == 0

tests/test_unit/backends/ibis/test_backend.py

@@ -572,3 +572,84 @@
             with be.transaction():
                 raw.execute("INSERT INTO t VALUES (1)")
         assert raw.execute("SELECT count(*) FROM t").fetchone()[0] == 1
+
+
+def test_in_transaction_false_outside_and_true_inside():
+    with IbisBackend(dialect="duckdb", database=":memory:") as be:
+        assert be.in_transaction() is False
+        with be.transaction():
+            assert be.in_transaction() is True
+        assert be.in_transaction() is False
+
+
+def test_in_transaction_true_at_nested_depth():
+    with IbisBackend(dialect="duckdb", database=":memory:") as be:
+        with be.transaction():
+            with be.transaction():
+                assert be.in_transaction() is True
+            assert be.in_transaction() is True
+        assert be.in_transaction() is False
+
+
+def test_in_transaction_false_after_rollback():
+    with IbisBackend(dialect="duckdb", database=":memory:") as be:
+        with pytest.raises(ValueError):
+            with be.transaction():
+                raise ValueError("boom")
+        assert be.in_transaction() is False
+
+
+def test_in_transaction_true_while_poisoned_before_unwind():
+    # Public-surface pin of spec §5.4: poisoned-but-open reads True.
+    from mountainash_data.core.errors import TransactionPoisonedError
+    with IbisBackend(dialect="duckdb", database=":memory:") as be:
+        with pytest.raises(TransactionPoisonedError):
+            with be.transaction():
+                try:
+                    with be.transaction():
+                        raise ValueError("inner")
+                except ValueError:
+                    pass
+                assert be.in_transaction() is True  # poisoned, still open
+        assert be.in_transaction() is False
+
+
+def test_in_transaction_none_dialect_returns_false():
+    be = IbisBackend(dialect="clickhouse")  # TransactionSupport.NONE, not connected
+    assert be.in_transaction() is False
+
+
+def test_in_transaction_not_connected_returns_false():
+    be = IbisBackend(dialect="duckdb", database=":memory:")  # never connect()ed
+    assert be.in_transaction() is False
+
+
+def test_in_transaction_after_close_returns_false():
+    be = IbisBackend(dialect="duckdb", database=":memory:")
+    be.connect()
+    be.close()
+    assert be.in_transaction() is False
+
+
+def test_in_transaction_returns_false_when_handle_resolution_raises(monkeypatch):
+    # A property-backed raw_handle_attr could raise a driver-specific (non-
+    # RuntimeError) error on a dropped connection. The predicate must swallow
+    # it and answer False, never propagate. monkeypatch replaces the bound
+    # method with a zero-arg callable; in_transaction() calls it with no args.
+    be = IbisBackend(dialect="duckdb", database=":memory:")
+    be.connect()
+
+    def _boom():
+        raise OSError("driver connection dropped")
+
+    monkeypatch.setattr(be, "raw_driver_connection", _boom)
+    assert be.in_transaction() is False
+
+
+@pytest.mark.parametrize("dialect", ["duckdb", "sqlite"])
+def test_raw_driver_connection_identity_is_stable(dialect):
+    # Load-bearing for cross-wrapper correctness (spec §3.4/§5.6): the handle
+    # must be the SAME object on every call so two wrappers over one raw conn
+    # compute one id. Pinned for every locally testable dialect.
+    with IbisBackend(dialect=dialect, database=":memory:") as be:
+        assert be.raw_driver_connection() is be.raw_driver_connection()

tests/test_unit/backends/ibis/test_backend_adopt.py

@@ -166,3 +166,29 @@ def test_from_ibis_connection_default_unchanged():
     # no apply -> ibis default stands
     assert raw.execute("SELECT current_setting('python_enable_replacements')").fetchone()[0] is False
     be.close()
+
+
+def test_in_transaction_visible_across_wrappers_from_ibis_connection(raw_db):
+    # Two IbisBackends adopting ONE raw connection must agree on tx state —
+    # the pointbreak F-09 scenario (guard runs on a different wrapper).
+    ibis_conn = ibis.duckdb.from_connection(raw_db)
+    a = IbisBackend.from_ibis_connection(ibis_conn, dialect="duckdb")
+    b = IbisBackend.from_ibis_connection(ibis_conn, dialect="duckdb")
+    assert a.in_transaction() is False
+    assert b.in_transaction() is False
+    with a.transaction():
+        assert b.in_transaction() is True  # B sees A's open unit of work
+    assert a.in_transaction() is False
+    assert b.in_transaction() is False
+
+
+def test_in_transaction_visible_across_wrappers_from_raw_connection(raw_db):
+    # from_raw_connection resolves the handle differently; must still agree.
+    # (duckdb has raw_adoption_verified=True.) ibis stores the passed raw conn
+    # directly as .con, so both wrappers key on id(raw_db).
+    a = IbisBackend.from_raw_connection(raw_db, dialect="duckdb")
+    b = IbisBackend.from_raw_connection(raw_db, dialect="duckdb")
+    with a.transaction():
+        assert a.in_transaction() is True
+        assert b.in_transaction() is True
+    assert b.in_transaction() is False

tests/test_unit/backends/ibis/test_transaction.py

@@ -142,3 +142,58 @@
         pass
     assert ("cur", "BEGIN") in h.log and ("cur", "COMMIT") in h.log
     assert ("close", None) in h.log
+
+
+from mountainash_data.backends.ibis._transaction import is_active
+
+
+def test_is_active_false_when_not_registered():
+    h = FakeHandle()
+    assert is_active(h) is False
+
+
+def test_is_active_true_inside_outer_transaction():
+    h = FakeHandle()
+    with _tx(h):
+        assert is_active(h) is True
+    assert is_active(h) is False
+
+
+def test_is_active_true_at_nested_depth():
+    h = FakeHandle()
+    with _tx(h):
+        with _tx(h):
+            assert is_active(h) is True
+        assert is_active(h) is True  # still open at depth 1
+    assert is_active(h) is False
+
+
+def test_is_active_false_after_rollback():
+    h = FakeHandle()
+    with pytest.raises(ValueError):
+        with _tx(h):
+            assert is_active(h) is True
+            raise ValueError("boom")
+    assert is_active(h) is False
+
+
+def test_is_active_true_while_poisoned_before_unwind():
+    # A caught nested failure poisons the unit; while the outer block is still
+    # open the handle stays registered, so is_active must report True.
+    h = FakeHandle()
+    with pytest.raises(TransactionPoisonedError):
+        with _tx(h):
+            try:
+                with _tx(h):
+                    raise ValueError("inner")
+            except ValueError:
+                pass
+            assert is_active(h) is True  # poisoned but still an open unit of work
+    assert is_active(h) is False  # cleared after outer unwind
+
+
+def test_is_active_does_not_mutate_registry():
+    h = FakeHandle()
+    before = dict(_ACTIVE)
+    assert is_active(h) is False
+    assert _ACTIVE == before  # pure read

tests/test_unit/backends/iceberg/test_backend.py

@@ -67,3 +67,17 @@ def test_transaction_not_required_noops():
         with be.transaction(required=False):
             pass
     assert any("iceberg" in str(x.message).lower() for x in w)
+
+
+def test_in_transaction_always_false():
+    be = IcebergBackend(catalog="rest", uri="http://localhost:8181")
+    # No connection-level unit of work — consistent with supports_transactions.
+    assert be.in_transaction() is False
+    assert be.supports_transactions is False
+
+
+def test_iceberg_satisfies_protocol_including_in_transaction():
+    from mountainash_data.core.protocol import Backend
+    be = IcebergBackend(catalog="rest", uri="http://localhost:8181")
+    assert isinstance(be, Backend)
+    assert callable(be.in_transaction)

tests/test_unit/core/test_protocol.py

@@ -78,3 +78,14 @@ def test_iceberg_backend_satisfies_widened_protocol():
 def test_protocol_declares_transaction():
     from mountainash_data.core.protocol import Backend
     assert hasattr(Backend, "transaction")
+
+
+def test_backend_protocol_declares_in_transaction():
+    assert hasattr(Backend, "in_transaction")
+
+
+def test_ibis_backend_satisfies_protocol_including_in_transaction():
+    from mountainash_data.backends.ibis.backend import IbisBackend
+    be = IbisBackend(dialect="duckdb", database=":memory:")
+    assert isinstance(be, Backend)  # runtime_checkable: presence of protocol methods
+    assert callable(be.in_transaction)