fix(OMN-19513): work_ledger_rows refuses ledger writes under a test runner

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "omnimarket"
-version = "0.4.277"
+version = "0.4.278"
 description = "OmniMarket - Portable ONEX workflow package registry"
 authors = [
     {name = "OmniNode.ai", email = "contact@omninode.ai"}

src/omnimarket/nodes/node_projection_read_effect/ports/sqlite_row_source.py

@@ -53,13 +53,15 @@ def build_sqlite_window_query(
     tenant_id: str | None,
     since: str | None = None,
     correlation_id: str | None = None,
+    selection: str = "newest",
 ) -> WindowQuery:
     """The SQLite SQL for one exposure's served window.
 
     The window is the one :func:`table_reader.build_window_query` reads: the
-    newest ``limit * 4`` rows by the recency column, or the next ``limit * 4``
-    above a ``since`` cursor. Every identifier comes from the contract; every
-    caller value is a bound parameter.
+    ``limit * 4`` rows ``selection`` names (``newest`` by the recency column,
+    ``walk`` the oldest by it, ``ranked`` the top of the declared order), or
+    the next ``limit * 4`` above a ``since`` cursor. Every identifier comes
+    from the contract; every caller value is a bound parameter.
     """
     relation = quote_identifier(cfg.table)
     retain = cfg.limit * RETAINED_WINDOW_FACTOR
@@ -91,10 +93,13 @@ def build_sqlite_window_query(
         params.append(since)
         where.append(f"{quote_identifier(cfg.cursor_column)} > ?")
         window_order = f"{quote_identifier(cfg.cursor_column)} ASC"
+    elif selection == "ranked":
+        window_order = order_clause(order_spec)
     else:
         recency = recency_column(cfg)
+        direction = "ASC" if selection == "walk" else "DESC"
         window_order = (
-            f"{quote_identifier(recency)} DESC"
+            f"{quote_identifier(recency)} {direction}"
             if recency is not None
             else order_clause(order_spec)
         )
@@ -232,6 +237,7 @@ async def rows(
         tenant_id: str | None,
         since: str | None = None,
         correlation_id: str | None = None,
+        selection: str = "newest",
     ) -> list[dict[str, Any]]:
         """The exposure's served window, ordered by ``order_spec``."""
         query = build_sqlite_window_query(
@@ -240,9 +246,50 @@ async def rows(
             tenant_id=tenant_id,
             since=since,
             correlation_id=correlation_id,
+            selection=selection,
         )
         return await asyncio.to_thread(self._fetch, cfg, order_spec, query)
 
+    def _smallest_cursor(
+        self, cfg: ProjectionTableConfig, cursor_column: str, tenant_id: str | None
+    ) -> object:
+        where = ""
+        params: tuple[Any, ...] = ()
+        if cfg.tenant_column is not None and tenant_id is not None:
+            where = f" WHERE CAST({quote_identifier(cfg.tenant_column)} AS TEXT) = ?"
+            params = (tenant_id,)
+        conn = self._connect()
+        try:
+            _require_relation(conn, cfg, ())
+            row = conn.execute(
+                f"SELECT min({quote_identifier(cursor_column)}) "
+                f"FROM {quote_identifier(cfg.table)}{where}",
+                params,
+            ).fetchone()
+        except sqlite3.Error as exc:
+            raise _driver_refusal(cfg, exc) from exc
+        finally:
+            conn.close()
+        return None if row is None else row[0]
+
+    async def walk_origin(
+        self, cfg: ProjectionTableConfig, *, tenant_id: str | None
+    ) -> str | None:
+        """The ``since`` value that starts an ascending walk at the first row.
+
+        ``since`` is a strict ``>``, so the origin is one below the smallest
+        cursor. Only an integer cursor has one; any other cursor type has no
+        value a caller could be handed, and the answer is ``None``.
+        """
+        if cfg.cursor_column is None:
+            return None
+        smallest = await asyncio.to_thread(
+            self._smallest_cursor, cfg, cfg.cursor_column, tenant_id
+        )
+        if isinstance(smallest, int) and not isinstance(smallest, bool):
+            return str(smallest - 1)
+        return None
+
     async def latest_event_at(
         self,
         cfg: ProjectionTableConfig,

src/omnimarket/nodes/node_projection_work_ledger/handlers/handler_work_ledger_projection.py

@@ -47,6 +47,9 @@
     SUBSCRIBE_TOPICS,
     TYPED_SUBSCRIBE_TOPICS,
 )
+from omnimarket.nodes.node_projection_work_ledger.handlers.handler_work_ledger_write_guard import (
+    HandlerWorkLedgerWriteGuard,
+)
 from omnimarket.nodes.node_projection_work_ledger.handlers.work_ledger_fold import (
     WorkLedgerFoldError,
     fold_records,
@@ -237,6 +240,7 @@ def handle(self, input_data: dict[str, Any]) -> dict[str, Any]:
     async def _project_one_message(
         self, topic: str, data: dict[str, Any], meta: MessageMeta
     ) -> int:
+        self._refuse_real_dsn_under_test()
         try:
             await self.db.connect()
             return await self._project_and_report(topic, data, meta)
@@ -253,6 +257,10 @@ def _run(coro: Coroutine[Any, Any, _T]) -> _T:
         with ThreadPoolExecutor(max_workers=1) as pool:
             return pool.submit(asyncio.run, coro).result()
 
+    def _refuse_real_dsn_under_test(self) -> None:
+        """Judged before any connect or statement: a test process never writes the real DSN."""
+        HandlerWorkLedgerWriteGuard.check_dsn(str(getattr(self._db, "dsn", "") or ""))
+
     async def project_event(
         self, topic: str, data: dict[str, Any], meta: MessageMeta
     ) -> bool:
@@ -263,6 +271,7 @@ async def _project_and_report(
         self, topic: str, data: dict[str, Any], meta: MessageMeta
     ) -> int:
         """Fold, then write the log record and each state op. Returns rows written."""
+        self._refuse_real_dsn_under_test()
         if topic in TYPED_SUBSCRIBE_TOPICS:
             return await self._project_typed(topic, data)
         if topic not in SUBSCRIBE_TOPICS:

src/omnimarket/nodes/node_projection_work_ledger/handlers/handler_work_ledger_write_guard.py

@@ -0,0 +1,109 @@
+# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc.
+# SPDX-License-Identifier: MIT
+"""The work-ledger projection write guard: a test process never writes the real DSN (OMN-19513).
+
+Operator ruling 2026-10-01: "it should be impossible to like fake test like that". A test
+that drives ``WorkLedgerProjectionWriter`` with a real database URL in its environment
+would insert fixture rows (``lane=alpha ticket=OMN-1``) into ``omninode_internal.work_ledger_rows``.
+Isolation by test setup is a convention a test can forget; this guard sits in the write path.
+
+THE RULE. A write is refused when BOTH hold:
+
+1. The process runs under a test runner: ``PYTEST_CURRENT_TEST`` is set, ``pytest`` or
+   ``unittest`` is imported, or ``ONEX_TEST_CONTEXT`` is set to any non-empty value. The
+   environment signals are inherited by a subprocess. No value of ``ONEX_TEST_CONTEXT``
+   removes a signal.
+2. The DSN the writer would connect with is real: its host is not loopback and not a local
+   socket.
+
+The writer calls :meth:`HandlerWorkLedgerWriteGuard.check_dsn` before it connects and before
+it executes a statement. A test writes through a loopback DSN or an injected fake database.
+There is no bypass flag and no allowlist. The command form,
+``python -m omnimarket.nodes.node_projection_work_ledger.handlers.handler_work_ledger_write_guard --dsn <dsn>``,
+exits 0 when the write may proceed and 79 with the refusal on stderr when it may not.
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import sys
+from urllib.parse import urlsplit
+
+GUARD_NAME = "ledger-test-write-guard"
+TEST_CONTEXT_ENV = "ONEX_TEST_CONTEXT"
+EXIT_TEST_WRITE_REFUSED = 79
+_LOOPBACK_HOSTS = frozenset({"", "localhost", "127.0.0.1", "::1"})
+_RUNNER_MODULES = ("pytest", "unittest")
+
+
+class HandlerWorkLedgerWriteGuard:
+    """Judges a projection DSN against the test-runner signals."""
+
+    @staticmethod
+    def test_context() -> str | None:
+        """The first test-runner signal present, named, or None outside a test."""
+        if os.environ.get("PYTEST_CURRENT_TEST"):
+            return "PYTEST_CURRENT_TEST is set"
+        if os.environ.get(TEST_CONTEXT_ENV, "").strip():
+            return f"{TEST_CONTEXT_ENV} is set"
+        for name in _RUNNER_MODULES:
+            if name in sys.modules:
+                return f"{name} is imported"
+        return None
+
+    @staticmethod
+    def dsn_host(dsn: str) -> str:
+        text = dsn.strip()
+        if "://" in text:
+            return urlsplit(text).hostname or ""
+        match = re.search(r"(?:^|\s)host=(\S+)", text)
+        return match.group(1) if match else ""
+
+    @classmethod
+    def dsn_is_real(cls, dsn: str) -> bool:
+        """A DSN reaches a shared database unless its host is loopback or a local socket."""
+        host = cls.dsn_host(dsn).lower()
+        return not (host in _LOOPBACK_HOSTS or host.startswith("/"))
+
+    @classmethod
+    def refusal(cls, dsn: str) -> str | None:
+        """The refusal message for a test process writing a real DSN, else None."""
+        signal = cls.test_context()
+        if signal is None or not dsn.strip() or not cls.dsn_is_real(dsn):
+            return None
+        host = cls.dsn_host(dsn) or "<non-loopback host>"
+        return (
+            f"{GUARD_NAME} REFUSED -- a test process ({signal}) tried to write the real "
+            f"work_ledger_rows DSN (host {host}). Nothing was written. A test writes a "
+            "loopback DSN or an injected fake database "
+            "(node_projection_work_ledger, OMN-19513)."
+        )
+
+    @classmethod
+    def check_dsn(cls, dsn: str) -> None:
+        """Raise SystemExit(79) naming the guard when a test process holds a real DSN."""
+        message = cls.refusal(dsn)
+        if message is not None:
+            sys.stderr.write(message + "\n")
+            raise SystemExit(EXIT_TEST_WRITE_REFUSED)
+
+
+def main(argv: list[str] | None = None) -> int:
+    """The command form: exit 0 when the DSN may be written, 79 when a test may not."""
+    parser = argparse.ArgumentParser(
+        description="Refuse a test process's real-DSN write."
+    )
+    parser.add_argument("--dsn", action="append", default=[])
+    args = parser.parse_args(argv)
+    for dsn in args.dsn:
+        message = HandlerWorkLedgerWriteGuard.refusal(dsn)
+        if message is not None:
+            sys.stderr.write(message + "\n")
+            return EXIT_TEST_WRITE_REFUSED
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/conftest.py

@@ -363,6 +363,20 @@ def _isolate_unit_env(
     monkeypatch.delenv("ONEX_STATE_ROOT", raising=False)
 
 
+@pytest.fixture(autouse=True)
+def _strip_projection_database_bindings(monkeypatch: pytest.MonkeyPatch) -> None:
+    """OMN-19513: no test inherits a real projection DSN from the caller's shell.
+
+    The work-ledger projection writer refuses a real DSN under a test runner
+    (``handler_work_ledger_write_guard``); this fixture also keeps the ambient
+    database URL and runtime-binding overlay out of every test, so a writer built
+    without an explicit DSN has none to dial. A test that needs a database sets its own
+    loopback DSN after this fixture's setup.
+    """
+    monkeypatch.delenv("OMNIDASH_ANALYTICS_DB_URL", raising=False)
+    monkeypatch.delenv("OMNIMARKET_PROJECTION_RUNTIME_BINDING_OVERLAY", raising=False)
+
+
 @pytest.fixture(autouse=True)
 def _default_paid_escalation_for_tests(monkeypatch: pytest.MonkeyPatch) -> None:
     """OMN-14225: run the suite at the production DEFAULT — paid escalation ON.

tests/unit/nodes/node_projection_read_effect/test_projection_read_effect_sqlite_omn20329.py

@@ -248,3 +248,60 @@ def test_other_schemes_are_still_refused(
 )
 def test_sqlite_path_follows_the_writer_convention(dsn: str, expected: Path) -> None:
     assert sqlite_path_from_dsn(dsn) == expected
+
+
+async def test_sqlite_source_serves_the_walk_and_ranked_selections(
+    tmp_path: Path,
+) -> None:
+    source = SqliteTableRowSource(_store(tmp_path))
+    cfg = _cfg(_UNSCOPED, tenant_column=None, limit=1)
+    order_spec = cfg.order_by_spec
+    newest = await source.rows(cfg, order_spec=order_spec, tenant_id=None)
+    walk = await source.rows(
+        cfg, order_spec=order_spec, tenant_id=None, selection="walk"
+    )
+    ranked = await source.rows(
+        cfg, order_spec=order_spec, tenant_id=None, selection="ranked"
+    )
+    assert {row["written_at"] for row in newest} == {
+        "2026-10-01T12:01:00+00:00",
+        "2026-10-01T12:02:00+00:00",
+    }
+    assert {row["written_at"] for row in walk} == {
+        "2026-10-01T12:00:00+00:00",
+        "2026-10-01T12:01:00+00:00",
+    }
+    assert {row["written_at"] for row in ranked} == {
+        row["written_at"] for row in newest
+    }
+
+
+async def test_sqlite_walk_origin_is_one_below_an_integer_cursor(
+    tmp_path: Path,
+) -> None:
+    db_path = _store(tmp_path)
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("CREATE TABLE ledger (seq INTEGER, tenant_id TEXT)")
+        conn.executemany(
+            "INSERT INTO ledger VALUES (?, ?)",
+            [(7, _TENANT), (9, _TENANT), (3, _OTHER_TENANT)],
+        )
+        conn.commit()
+    finally:
+        conn.close()
+    source = SqliteTableRowSource(db_path)
+    ledger = _cfg(
+        _DECISIONS,
+        table="ledger",
+        columns=("seq", "tenant_id"),
+        order_by="seq DESC",
+        order_by_spec=parse_order_by_clauses("seq DESC", ("seq", "tenant_id")),
+        freshness_column=None,
+        cursor_column="seq",
+        key_columns=("seq",),
+    )
+    assert await source.walk_origin(ledger, tenant_id=_TENANT) == "6"
+    assert await source.walk_origin(ledger, tenant_id=_OTHER_TENANT) == "2"
+    text_cursor = _cfg(_UNSCOPED, tenant_column=None)
+    assert await source.walk_origin(text_cursor, tenant_id=None) is None