fix(OMN-19833): projection writers report rows_upserted and route no pure fold, and a gate keeps it so

.pre-commit-config.yaml

@@ -486,6 +486,20 @@ repos:
         pass_filenames: false
         stages: [pre-commit]
 
+      # OMN-19833: the runtime counts a projection write only under `rows_upserted`
+      # and reads any other key as zero rows, so a writer reporting `rows_written`
+      # wrote its rows and drove projection_apply_divergence DEGRADED. A pure fold
+      # routed beside an in-process writer is handed the raw event it cannot
+      # validate. Syntax-tree and YAML reads only; also run in CI by
+      # tests/validators/test_projection_writer_contract_omn19833.py.
+      - id: projection-writer-contract
+        name: Projection writers report rows_upserted and route no pure fold (OMN-19833)
+        language: system
+        entry: uv run python -m omnimarket.validators.projection_writer_contract
+        files: ^(src/omnimarket/nodes/.*/(contract\.yaml|handlers/.*\.py)|src/omnimarket/validators/projection_writer_contract\.py)$
+        pass_filenames: false
+        stages: [pre-commit]
+
       # OMN-18013: a burned-down baseline may not come back, and a baseline still held
       # open for another lane may not grow past that lane's rows. Pure filesystem
       # assertion — no scan, no derivation, nothing to soften.

src/omnimarket/nodes/node_projection_board_probe_results/contract.yaml

@@ -23,11 +23,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: board_probe_results_projection_runner
-      handler:
-        name: HandlerProjectionBoardProbeResults
-        module: omnimarket.nodes.node_projection_board_probe_results.handlers.board_probe_results_fold
-      handler_module: omnimarket.nodes.node_projection_board_probe_results.handlers.board_probe_results_fold
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     - operation: board_probe_results_projection_writer
       handler:
         name: BoardProbeResultsProjectionWriter

src/omnimarket/nodes/node_projection_ci_attempt_outcome/contract.yaml

@@ -40,11 +40,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: projection_ci_attempt_outcome
-      handler:
-        name: HandlerProjectionCiAttemptOutcome
-        module: omnimarket.nodes.node_projection_ci_attempt_outcome.handlers.handler_projection_ci_attempt_outcome
-      handler_module: omnimarket.nodes.node_projection_ci_attempt_outcome.handlers.handler_projection_ci_attempt_outcome
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     - operation: ci_attempt_outcome_projection_writer
       handler:
         name: CiAttemptOutcomeProjectionWriter

src/omnimarket/nodes/node_projection_ci_attempt_outcome/handlers/handler_ci_attempt_outcome_writer.py

@@ -147,7 +147,7 @@ async def _project_one_message(
     ) -> dict[str, Any]:
         """Project one runtime-dispatched message and report what changed.
 
-        The returned mapping carries a row COUNT, which the runtime's
+        The returned mapping carries a row COUNT under ``rows_upserted``, which the runtime's
         write-path guard reads. A bare truthy acknowledgement over a message
         that wrote nothing is indistinguishable from one that wrote a hundred
         rows, and telling those apart is the whole reason this projection
@@ -159,7 +159,7 @@ async def _project_one_message(
         finally:
             await self.db.close()
         return {
-            "rows_written": len(written),
+            "rows_upserted": len(written),
             "attempt_rows": written,
             "unattributed_row_count": unattributed,
             "skipped_check_count": skipped,

src/omnimarket/nodes/node_projection_consumer_flow/contract.yaml

@@ -27,11 +27,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: projection_consumer_flow
-      handler:
-        name: HandlerProjectionConsumerFlow
-        module: omnimarket.nodes.node_projection_consumer_flow.handlers.handler_projection_consumer_flow
-      handler_module: omnimarket.nodes.node_projection_consumer_flow.handlers.handler_projection_consumer_flow
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     - operation: consumer_flow_projection_runner
       handler:
         name: ConsumerFlowProjectionWriter

src/omnimarket/nodes/node_projection_lab_container_memory/contract.yaml

@@ -39,11 +39,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: projection_lab_container_memory
-      handler:
-        name: HandlerContainerMemoryFold
-        module: omnimarket.nodes.node_projection_lab_container_memory.handlers.handler_container_memory_fold
-      handler_module: omnimarket.nodes.node_projection_lab_container_memory.handlers.handler_container_memory_fold
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     # The DURABLE half, and the entry the projection wiring path calls: the
     # runtime injects the bare event dict and expects the entry it invokes to
     # write, which the pure fold above cannot do.

src/omnimarket/nodes/node_projection_lab_lane_health/contract.yaml

@@ -43,11 +43,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: lab_lane_health_projection_runner
-      handler:
-        name: HandlerProjectionLabLaneHealth
-        module: omnimarket.nodes.node_projection_lab_lane_health.handlers.handler_lab_lane_health_runner
-      handler_module: omnimarket.nodes.node_projection_lab_lane_health.handlers.handler_lab_lane_health_runner
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     # The DURABLE half, and the one the projection wiring path actually calls.
     # Declared exactly as node_projection_runner_fleet declares its own writer
     # beside its pure derivation: the runtime injects _db and _topic and

src/omnimarket/nodes/node_projection_pr_landing/contract.yaml

@@ -35,11 +35,11 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: projection_pr_landing
-      handler:
-        name: HandlerProjectionPrLanding
-        module: omnimarket.nodes.node_projection_pr_landing.handlers.handler_projection_pr_landing
-      handler_module: omnimarket.nodes.node_projection_pr_landing.handlers.handler_projection_pr_landing
+    # Only the writer is routed (OMN-19721, OMN-19833). A routed pure fold is
+    # handed the raw event beside the writer, fails to validate it as its
+    # wrapped request, and dead-letters it, and its zero upserts count against
+    # the projection_apply_divergence health dimension. The writer calls the
+    # fold in process (rule 7a).
     - operation: pr_landing_projection_writer
       handler:
         name: PrLandingProjectionWriter

src/omnimarket/nodes/node_projection_pr_landing/handlers/handler_pr_landing_writer.py

@@ -13,14 +13,29 @@
 projection arm validates and returns without writing. A runner-shaped class
 that does not declare in-process dispatch is skipped by the shared runtime.
 Both are pinned by tests.
+
+Three more ways it went wrong on the runtime, all fixed here and pinned
+(OMN-19833, measured on the .201 dev lane 2026-09-30T03:27Z to 03:39Z):
+
+* The runtime counts a write only under ``rows_upserted``. A count under any
+  other key reads as zero, so a writer that did write tripped the
+  ``projection_apply_divergence`` health dimension and turned the runtime
+  DEGRADED.
+* One writer instance is routed on all four topics, and the runtime runs
+  handlers in worker threads, so two topics' messages can be in ``handle()`` at
+  once. A pool held on the shared adapter was then opened on one thread's loop
+  and used or closed from the other's. Each message opens its own adapter.
+* A redelivery the ordering guard refuses is reported under the runtime's
+  refusal key, so it is logged as a deliberate refusal and not as a writer that
+  silently wrote nothing.
 """
 
 from __future__ import annotations
 
 import asyncio
 from collections.abc import Mapping
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import yaml
 
@@ -41,9 +56,22 @@
 )
 from omnimarket.projection.runner import BaseProjectionRunner, MessageMeta
 
+if TYPE_CHECKING:
+    from omnimarket.adapters.asyncpg_adapter import AsyncpgAdapter
+
 TABLE_PR_LANDING_STATE = "omninode_internal.pr_landing_state"
 TABLE_PR_LANDING_TRANSITIONS = "omninode_internal.pr_landing_transitions"
 
+#: The key the runtime's write-path guard and apply counters read the row count
+#: from (omnibase_infra ``_extract_rows_upserted``). Any other key reads as 0.
+ROWS_UPSERTED_KEY = "rows_upserted"
+
+#: The key the runtime reads for statements an ordering guard declined
+#: (omnibase_infra ``ROWS_REFUSED_KEY``, OMN-18992). Restated rather than
+#: imported: this module is loaded by processes that do not import the
+#: runtime's wiring package.
+ROWS_REFUSED_KEY = "rows_refused_by_ordering_guard"
+
 #: Which request field each subscribed topic fills.
 TOPIC_EVENT_KIND: Mapping[str, EnumPrLandingProjectionEventKind] = {
     PR_LANDING_TRANSITIONED_TOPIC_V1: EnumPrLandingProjectionEventKind.TRANSITIONED,
@@ -214,26 +242,40 @@ def handle(self, input_data: dict[str, Any]) -> dict[str, Any]:
         )
         return asyncio.run(self._project_one_message(topic, data, meta))
 
+    def _adapter_for_one_message(self) -> AsyncpgAdapter:
+        """A database adapter owned by one message and the loop that serves it.
+
+        Never the shared ``self.db``: the runtime routes this one instance on
+        all four topics and calls ``handle()`` from worker threads, so a pool
+        on the shared adapter would be created on one thread's loop and used or
+        closed from another's (``PoolConnectionHolder.wait_until_released`` on
+        the dev lane). The DSN is the one the runtime bound on ``self.db``.
+        """
+        from omnimarket.adapters.asyncpg_adapter import AsyncpgAdapter
+
+        return AsyncpgAdapter(dsn=self.db.dsn, min_size=1, max_size=2)
+
     async def _project_one_message(
         self, topic: str, data: dict[str, Any], meta: MessageMeta
     ) -> dict[str, Any]:
         """Project one message and report what it wrote, as a row COUNT.
 
-        The runtime's write-path guard reads ``rows_written``. A truthy
+        The runtime's write-path guard reads ``rows_upserted``. A truthy
         acknowledgement over a message that wrote nothing would be
         indistinguishable from one that wrote rows.
         """
-        await self.db.connect()
+        db = self._adapter_for_one_message()
+        await db.connect()
         try:
-            return await self._project_event(topic, data)
+            return await self._project_event(topic, data, db)
         finally:
-            await self.db.close()
+            await db.close()
 
     async def project_event(
         self, topic: str, data: dict[str, Any], meta: MessageMeta
     ) -> bool:
         """Standalone-runner entrypoint: project one message, report success."""
-        await self._project_event(topic, data)
+        await self._project_event(topic, data, self.db)
         return True
 
     @staticmethod
@@ -253,37 +295,42 @@ def build_request(
         payload = {key: value for key, value in data.items() if not key.startswith("_")}
         return ModelPrLandingProjectionRequest.model_validate({kind.value: payload})
 
-    async def _project_event(self, topic: str, data: dict[str, Any]) -> dict[str, Any]:
+    async def _project_event(
+        self, topic: str, data: dict[str, Any], db: AsyncpgAdapter
+    ) -> dict[str, Any]:
         request = self.build_request(topic, data)
         result = self._derive.handle(request)
 
         transition_rows: list[dict[str, Any]] = []
         if result.transition_row is not None:
-            appended = await self.db.execute(
+            appended = await db.execute(
                 _APPEND_TRANSITION, *_transition_params(result.transition_row)
             )
             transition_rows = [dict(row) for row in appended]
 
-        upserted = await self.db.execute(
-            _UPSERT_STATE, *_state_params(result.state_row)
-        )
+        upserted = await db.execute(_UPSERT_STATE, *_state_params(result.state_row))
         state_rows = [dict(row) for row in upserted]
 
+        attempted = 1 + (1 if result.transition_row is not None else 0)
+        written = len(transition_rows) + len(state_rows)
         return {
             "event_kind": request.event_kind.value,
             "repository": result.state_row.repository,
             "pr_number": result.state_row.pr_number,
             "seq": result.state_row.seq,
-            "rows_written": len(transition_rows) + len(state_rows),
+            ROWS_UPSERTED_KEY: written,
+            # Refused by the stale-write guard: an older seq, or a redelivered
+            # transition. Not an error, and deliberately not counted as written.
+            ROWS_REFUSED_KEY: attempted - written,
             "state_rows": state_rows,
             "transition_rows": transition_rows,
-            # Refused by the stale-write guard: an older seq, or a redelivered
-            # transition. Not an error, and deliberately not counted.
             "state_write_refused": not state_rows,
         }
 
 
 __all__ = [
+    "ROWS_REFUSED_KEY",
+    "ROWS_UPSERTED_KEY",
     "TABLE_PR_LANDING_STATE",
     "TABLE_PR_LANDING_TRANSITIONS",
     "TOPIC_EVENT_KIND",

src/omnimarket/nodes/node_projection_prod_promotion_gate/contract.yaml

@@ -39,11 +39,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: projection_prod_promotion_gate
-      handler:
-        name: HandlerProjectionProdPromotionGate
-        module: omnimarket.nodes.node_projection_prod_promotion_gate.handlers.handler_projection_prod_promotion_gate
-      handler_module: omnimarket.nodes.node_projection_prod_promotion_gate.handlers.handler_projection_prod_promotion_gate
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     # The DURABLE half, and the one the projection wiring path actually calls.
     # The runtime injects _db and _topic into the bare event dict and expects
     # the entry it invokes to WRITE, which the pure fold above cannot do.

src/omnimarket/nodes/node_projection_runner_fleet/contract.yaml

@@ -37,11 +37,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: projection_runner_fleet
-      handler:
-        name: HandlerProjectionRunnerFleet
-        module: omnimarket.nodes.node_projection_runner_fleet.handlers.handler_projection_runner_fleet
-      handler_module: omnimarket.nodes.node_projection_runner_fleet.handlers.handler_projection_runner_fleet
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     - operation: fleet_liveness_projection_writer
       handler:
         name: FleetLivenessProjectionWriter

src/omnimarket/nodes/node_projection_session_content/contract.yaml

@@ -33,11 +33,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-    - operation: session_content_projection_fold
-      handler:
-        name: HandlerProjectionSessionContent
-        module: omnimarket.nodes.node_projection_session_content.handlers.handler_session_content
-      handler_module: omnimarket.nodes.node_projection_session_content.handlers.handler_session_content
+    # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+    # event beside the writer, cannot report a write, and fails or runs on defaults
+    # when its request wraps the event. The writer calls the fold in process (rule 7a).
     # The DURABLE half, and the one the projection wiring path calls with _db
     # and _topic injected. The pure fold above cannot write, and a node that
     # declared only the fold would consume, commit and store nothing (rule 7a,

src/omnimarket/nodes/node_projection_worktree_reconcile/contract.yaml

@@ -20,11 +20,9 @@ handler:
 handler_routing:
   routing_strategy: operation_match
   handlers:
-  - operation: projection_worktree_reconcile
-    handler:
-      name: HandlerProjectionWorktreeReconcile
-      module: omnimarket.nodes.node_projection_worktree_reconcile.handlers.handler_projection_worktree_reconcile
-    handler_module: omnimarket.nodes.node_projection_worktree_reconcile.handlers.handler_projection_worktree_reconcile
+  # Only the writer is routed (OMN-19833). A routed pure fold is handed the raw
+  # event beside the writer, cannot report a write, and fails or runs on defaults
+  # when its request wraps the event. The writer calls the fold in process (rule 7a).
   - operation: worktree_reconcile_projection_writer
     handler:
       name: WorktreeReconcileProjectionWriter

src/omnimarket/nodes/node_projection_worktree_reconcile/handlers/handler_worktree_reconcile_writer.py

@@ -105,7 +105,7 @@ async def _project_event(self, topic: str, data: dict[str, Any]) -> dict[str, An
         )
         return {
             "host": row.host,
-            "rows_written": len(written),
+            "rows_upserted": len(written),
             "state_rows": [dict(item) for item in written],
             "state_write_refused": not written,
         }