Fix #2535: stop slice-N from inheriting slice-(N-1) consensus, drop gateway import

orchestrator/concurrent_executor.py

@@ -636,27 +636,31 @@ def check_consensus(self) -> dict[str, Any]:
                 pipeline_id=self.pipeline.id,
                 slice_id=self._slice_id,
             )
-            # Attempt lazy reconstruction from message store
-            # Reconstruction does NOT yet support slice scoping — for
-            # slice-scoped trackers we fall back to fetching the bare
-            # pipeline-id tracker (legacy behaviour). This is acceptable
-            # because reconstruction is a backstop for orchestrator
-            # restarts, not the steady-state path; per-slice trackers
-            # are stateless event consumers and are recreated by the
-            # slice scheduler on the next iteration.
-            try:
-                from peer_consensus import reconstruct_tracker_from_messages
-
-                graph = self._get_review_graph()
-                tracker = reconstruct_tracker_from_messages(self.pipeline.id, graph)
-            except ImportError:
-                pass
-            except Exception as e:
-                logger.warning(
-                    "Tracker reconstruction failed",
-                    error=str(e),
-                    pipeline_id=self.pipeline.id,
-                )
+            # Attempt lazy reconstruction from message store — pipeline-
+            # scoped only. Slice-scoped reconstruction is unsafe today:
+            # CONSENSUS_* messages are persisted under the bare
+            # pipeline_id, so a per-slice replay would mingle siblings'
+            # messages and reach false consensus the moment a fresh
+            # slice spawns roles whose names match an already-confirmed
+            # prior slice. Per-slice trackers are stateless event
+            # consumers and are recreated by the slice scheduler on the
+            # next iteration; the pipeline run loop's empty-tracker
+            # iteration will simply observe is_complete=False, which is
+            # the correct answer for a brand-new slice (#2535).
+            if self._slice_id is None:
+                try:
+                    from peer_consensus import reconstruct_tracker_from_messages
+
+                    graph = self._get_review_graph()
+                    tracker = reconstruct_tracker_from_messages(self.pipeline.id, graph)
+                except ImportError:
+                    pass
+                except Exception as e:
+                    logger.warning(
+                        "Tracker reconstruction failed",
+                        error=str(e),
+                        pipeline_id=self.pipeline.id,
+                    )
         if tracker:
             result = tracker.evaluate()
             # Message-bus fallback: if reconstruction produced a tracker but
@@ -678,7 +682,9 @@ def check_consensus(self) -> dict[str, Any]:
                 # Safety net (#1671): if the tracker has all roles in
                 # confirmed_roles but evaluate() returned False due to stale
                 # NACK edges in the approval matrix (common after NACK →
-                # re-propose cycles), trust the confirmed set.
+                # re-propose cycles), trust the confirmed set. This path is
+                # safe for slice-scoped trackers because ``confirmed_roles``
+                # is the per-slice tracker's own state.
                 tracker_confirmed = tracker.confirmed_roles
                 if all_roles and all_roles.issubset(tracker_confirmed):
                     logger.warning(
@@ -691,6 +697,23 @@ def check_consensus(self) -> dict[str, Any]:
                     )
                     result["is_complete"] = True
                     result["fallback"] = "tracker_confirmed"
+                elif self._slice_id is not None:
+                    # Slice-scoped: the message-bus fallback below scans
+                    # ``store.get_messages(pipeline_id)`` pipeline-wide and
+                    # cannot distinguish slice-1's CONFIRMs from slice-2's,
+                    # so it would falsely declare consensus the moment a
+                    # fresh slice spawns roles whose names match an already-
+                    # confirmed prior slice (#2535). The in-memory per-slice
+                    # tracker is the authoritative source for slice work;
+                    # an empty fresh tracker correctly returns
+                    # is_complete=False here so the pipeline run loop keeps
+                    # polling.
+                    logger.info(
+                        "Skipping pipeline-wide message-bus fallback for slice-scoped tracker",
+                        pipeline_id=self.pipeline.id,
+                        slice_id=self._slice_id,
+                        blocking_agents=result.get("blocking_agents", []),
+                    )
                 else:
                     # Message-bus fallback: scan message store for
                     # CONSENSUS_CONFIRMED messages (#1471/#1615).

orchestrator/gateway_client.py

@@ -10,6 +10,7 @@
 
 import json
 import os
+import re
 import socket
 import subprocess
 import sys
@@ -48,6 +49,58 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 logger = get_logger("orchestrator.gateway_client")
 
 
+_REBASE_REF_RE = re.compile(r"^[A-Za-z0-9._/+-][A-Za-z0-9._/+-]*$")
+
+
+def _build_rebase_onto_args(
+    branch: str, new_base: str, old_base: str
+) -> tuple[list[str], bool, str]:
+    """Construct the canonical ``rebase --onto`` argv for the reconciler.
+
+    Performs the same ref-shape sanity checks as
+    :func:`gateway.git_client.build_rebase_onto_args` (reject empty,
+    flag-shaped, whitespace-bearing, or non-git-ref-shaped inputs) but
+    lives in the orchestrator package so the deployed orchestrator image
+    (which does not ship ``gateway/``) can build the argv without an
+    import-time dependency on the gateway code. Two intentional
+    differences from the gateway helper:
+
+    - The argv is emitted with each ref *stripped*, so leading/trailing
+      whitespace that the regex would otherwise reject as the input is
+      normalised before the gateway round-trip.
+    - This helper does NOT call ``gateway.git_client.validate_git_args``
+      — pulling that import in would defeat the point of inlining. The
+      gateway server's ``/git`` endpoint runs the same allowlist
+      validator on every submission, so the security floor is unchanged
+      (audit boundary is the server, not the client-side helper).
+    """
+    if not isinstance(branch, str) or not branch.strip():
+        return [], False, "branch must be a non-empty string"
+    if not isinstance(new_base, str) or not new_base.strip():
+        return [], False, "new_base must be a non-empty string"
+    if not isinstance(old_base, str) or not old_base.strip():
+        return [], False, "old_base must be a non-empty string"
+
+    for label, value in (("branch", branch), ("new_base", new_base), ("old_base", old_base)):
+        v = value.strip()
+        if v.startswith("-"):
+            return [], False, f"{label} must not start with '-' (rejected flag-shaped ref: {v!r})"
+        if any(ch.isspace() or ch == "\x00" for ch in v):
+            return (
+                [],
+                False,
+                f"{label} must not contain whitespace or NUL (rejected: {v!r})",
+            )
+        if not _REBASE_REF_RE.fullmatch(v):
+            return (
+                [],
+                False,
+                f"{label} must look like a git ref (alnum + . _ / + -); got {v!r}",
+            )
+
+    return ["--onto", new_base.strip(), old_base.strip(), branch.strip()], True, ""
+
+
 @dataclass
 class SessionInfo:
     """Information about a gateway session."""
@@ -1390,8 +1443,11 @@ def rebase_onto(
 
         1. ``git rebase --onto <new_base> <old_base> <branch>``
            (via the existing per-agent ``/api/v1/git/execute``
-           endpoint and the canonical argv from
-           :func:`gateway.git_client.build_rebase_onto_args`).
+           endpoint and the canonical argv from the local
+           :func:`_build_rebase_onto_args` helper — which mirrors
+           ``gateway.git_client.build_rebase_onto_args`` so the
+           orchestrator image does not need ``gateway/`` on its
+           Python path).
         2. ``git push --force-with-lease origin <branch>``
            (via the existing per-agent ``/api/v1/git/push``
            endpoint) — propagates the rewritten history to origin
@@ -1420,16 +1476,7 @@ def rebase_onto(
         reconciler counts both ``False`` and exceptions as
         ``rebases_failed``.
         """
-        try:
-            from gateway.git_client import build_rebase_onto_args
-        except ImportError:
-            logger.error(
-                "rebase_onto: gateway/git_client module unavailable",
-                pipeline_id=pipeline_id,
-            )
-            return False
-
-        args, ok, err = build_rebase_onto_args(branch, new_base, old_base)
+        args, ok, err = _build_rebase_onto_args(branch, new_base, old_base)
         if not ok:
             logger.warning(
                 "rebase_onto: argv rejected by allowlist validator",

orchestrator/routes/signals.py

@@ -1465,6 +1465,7 @@ def _existing_confirmed_for_role(
     pipeline_id: str,
     agent_role: str,
     phase: str | None,
+    slice_id: str | None = None,
 ) -> tuple[bool, bool]:
     """Return (has_final, has_pending_acks) for prior CONFIRMED messages.
 
@@ -1475,6 +1476,17 @@ def _existing_confirmed_for_role(
 
     - ``has_final``: a non-pending_acks CONFIRMED message already exists.
     - ``has_pending_acks``: a pending_acks CONFIRMED message exists.
+
+    ``slice_id`` scopes the check to a single slice. The message store
+    keys messages by bare ``pipeline_id``, so without scoping a fresh
+    slice-N coder would falsely appear "already confirmed" because
+    slice-(N-1)'s coder wrote a CONFIRMED message under the same
+    pipeline_id (#2535). Filtering on ``metadata["slice_id"]`` (written
+    by the per-slice tracker path below) confines the lookup to the
+    same slice. Pipeline-scoped (``slice_id is None``) callers continue
+    to see only messages with no ``slice_id`` in metadata, preserving
+    the legacy non-slice behaviour. Tracked under #2409 as part of
+    end-to-end slice-scoped message routing.
     """
     try:
         from message_store import get_message_store
@@ -1510,6 +1522,12 @@ def _existing_confirmed_for_role(
         if phase is not None and msg_phase is not None and msg_phase != phase:
             continue
         metadata = getattr(m, "metadata", None) or {}
+        # Scope idempotency check to the same slice. A None slice_id on
+        # either side (caller or message) only matches the same; this
+        # cleanly separates slice-N from slice-M and from pipeline-level
+        # confirms.
+        if metadata.get("slice_id") != slice_id:
+            continue
         if metadata.get("pending_acks"):
             has_pending = True
         else:
@@ -1581,6 +1599,16 @@ def handle_consensus_confirmed_signal(
                 error=str(recon_err),
             )
 
+        if not tracker and slice_id is not None:
+            # Slice-scoped: pipeline-wide message-bus fallback would mingle
+            # other slices' CONFIRMs and reach false consensus the moment a
+            # fresh slice spawns roles matching an already-confirmed prior
+            # slice (#2535). Per-slice trackers are recreated by the slice
+            # scheduler on the next iteration; surface the missing tracker
+            # rather than guessing from sibling-slice state.
+            scope = f"{pipeline_id}/{slice_id}"
+            return make_error_response(f"No consensus tracker for pipeline {scope}", 404)
+
         if not tracker:
             # Message-bus authoritative fallback: if all expected roles have
             # CONSENSUS_CONFIRMED messages, accept the confirmation directly.
@@ -1659,10 +1687,17 @@ def handle_consensus_confirmed_signal(
         # check_consensus() can detect when all agents have *attempted*
         # confirmation even if the tracker rejected some (#1615).
         current_phase = _resolve_pipeline_phase(pipeline_id, repo_path)
+        # Pass slice_id so the idempotency probe doesn't see sibling-slice
+        # CONFIRMs as "already confirmed for this role" (#2535).
         has_final, has_pending = _existing_confirmed_for_role(
-            pipeline_id, agent_role, current_phase
+            pipeline_id, agent_role, current_phase, slice_id=slice_id
         )
 
+        # Common metadata tag so future _existing_confirmed_for_role probes
+        # can scope by slice (None for pipeline-level callers, matching the
+        # legacy behaviour exactly).
+        _slice_meta = {"slice_id": slice_id} if slice_id is not None else {}
+
         if result.get("status") == "pending_acks":
             # Dedupe pending_acks writes once an agent has already emitted one
             # (or a final) in this phase — the fallback check only needs one
@@ -1680,7 +1715,7 @@ def handle_consensus_confirmed_signal(
                         subject=f"Confirmed by {agent_role} (pending_acks)",
                         body=result.get("message", ""),
                         phase=current_phase,
-                        metadata={"pending_acks": True},
+                        metadata={"pending_acks": True, **_slice_meta},
                     )
                 )
             return make_success_response(result["message"], data=result, status_code=202)
@@ -1701,7 +1736,10 @@ def handle_consensus_confirmed_signal(
                     subject=f"Confirmed by {agent_role}",
                     body="",
                     phase=current_phase,
-                    metadata={"consensus_reached": result.get("consensus_reached", False)},
+                    metadata={
+                        "consensus_reached": result.get("consensus_reached", False),
+                        **_slice_meta,
+                    },
                 )
             )
 

orchestrator/stacked_pr_reconciler.py

@@ -23,12 +23,18 @@
 3. Calls ``GatewayClient.rebase_onto`` (orchestrator-side bridge
    in :mod:`orchestrator.gateway_client`) which forwards the
    request through the gateway's existing per-agent allowlist
-   plumbing — internally constructed via
-   :func:`gateway.git_client.build_rebase_onto_args` and submitted
-   through the same ``/api/v1/git/execute`` endpoint that
-   authorised agents use today. No new privileged
-   orchestrator-role endpoint is introduced (refine-phase
-   decision-15).
+   plumbing — argv is built client-side by
+   :func:`orchestrator.gateway_client._build_rebase_onto_args`
+   (an inlined copy of
+   :func:`gateway.git_client.build_rebase_onto_args` so the
+   orchestrator image, which does not ship ``gateway/``, has no
+   import-time dependency on the gateway package — see #2535)
+   and submitted through the same ``/api/v1/git/execute``
+   endpoint that authorised agents use today. The gateway server
+   re-validates the argv via the same allowlist, so dropping the
+   client-side ``validate_git_args`` call doesn't reduce the
+   security floor. No new privileged orchestrator-role endpoint
+   is introduced (refine-phase decision-15).
 
 This module is pure-Python and side-effect-free at import time —
 the orchestrator's pipeline run loop wires up an async timer that