Fix #2422: scope phase_exec.agents walks by (role, slice_id)

orchestrator/concurrent_executor.py

@@ -460,6 +460,7 @@ def _spawn_roles(
                             role=role,
                             status=AgentExecutionStatus.FAILED,
                             error=str(e),
+                            slice_id=self._slice_id,
                         )
                     )
 
@@ -500,6 +501,7 @@ def _spawn_agent(self, role: AgentRole, prompt_text: str = "") -> AgentExecution
             container_id=container_id,
             container_info=result.container_info,
             started_at=datetime.now(UTC),
+            slice_id=self._slice_id,
         )
 
     def handle_agent_failure(self, role: str, error: str) -> dict[str, Any]:

orchestrator/kubernetes_monitor.py

@@ -746,6 +746,14 @@ def _handle_consensus_stall_recovery(
                 if phase_key is None:
                     return
 
+                # ``slice_id`` is optional in the health-check details dict
+                # (the consensus_stall check is currently pipeline-level
+                # only, but #2422's audit asks every walker of
+                # ``phase_exec.agents`` to scope by ``(role, slice_id)`` so
+                # the moment it becomes slice-aware this path doesn't flip
+                # other slices' agents to COMPLETE).
+                stall_slice_id = details.get("slice_id")
+
                 fresh_pipeline = store.load_pipeline(pipeline_id)
                 original_version = fresh_pipeline.version
 
@@ -764,6 +772,8 @@ def _handle_consensus_stall_recovery(
                 now = datetime.now(UTC)
                 completed_container_ids: set[str] = set()
                 for agent in phase_exec.agents:
+                    if getattr(agent, "slice_id", None) != stall_slice_id:
+                        continue
                     if agent.status in (AgentExecutionStatus.RUNNING, AgentExecutionStatus.FAILED):
                         agent.status = AgentExecutionStatus.COMPLETE
                         agent.completed_at = now
@@ -798,6 +808,14 @@ def _handle_consensus_stall_recovery(
                 if phase_exec.cycle_timings and phase_exec.cycle_timings[-1].completed_at is None:
                     phase_exec.cycle_timings[-1].completed_at = now
 
+                # TODO(#2441): the phase-level mutations below are unconditional
+                # even though the agent walk above is now slice-scoped. Safe
+                # today because ``consensus_stall`` is pipeline-level only and
+                # ``stall_slice_id`` is always ``None`` here, but the moment
+                # the upstream check becomes slice-aware this path will mark
+                # the whole phase COMPLETE while other slices are still
+                # RUNNING. Scope to "no other slice still active" or split
+                # ``phase_exec`` into per-slice status when #2441 lands.
                 phase_exec.status = PipelineStatus.COMPLETE
                 phase_exec.completed_at = now
 

orchestrator/models.py

@@ -13,6 +13,7 @@
 
 from egg_contracts.models import PipelinePhase
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+from slice_id_validation import SLICE_ID_PATTERN
 
 # Phase-aware fallback defaults for consensus timeout. Calibrated against
 # producer/reviewer fan-out and iteration profile per phase — see #2263.
@@ -229,6 +230,37 @@ def _migrate_removed_roles(cls, data: Any) -> Any:
             "Optional for backward compatibility with older state files."
         ),
     )
+    slice_id: str | None = Field(
+        default=None,
+        description=(
+            "Slice scope (e.g. ``slice-2``) when the agent runs as part of a "
+            "per-slice team in a multi-slice phase (#2137). ``None`` for "
+            "pipeline-level (non-sliced) agents. Distinguishes concurrent "
+            "same-role agents in the same ``phase_exec.agents`` list so "
+            "consumers that walk by role match on ``(role, slice_id)`` "
+            "rather than role alone (#2422)."
+        ),
+    )
+
+    @field_validator("slice_id")
+    @classmethod
+    def _validate_slice_id(cls, v: str | None) -> str | None:
+        """Defense-in-depth: reject non-canonical ``slice_id`` values.
+
+        Production write paths populate this field from validated values
+        produced by ``extract_slice_id`` / ``concurrent_executor._slice_id``,
+        which already enforce ``SLICE_ID_PATTERN``. This validator closes
+        the gap for hand-built fixtures, migration tools, or any future
+        caller that constructs ``AgentExecution`` directly — a non-canonical
+        value would silently break the ``(role, slice_id)`` walks that
+        consumers rely on.
+        """
+        if v is None:
+            return None
+        if not SLICE_ID_PATTERN.fullmatch(v):
+            raise ValueError(f"Invalid slice_id {v!r}: must match 'slice-<N>'")
+        return v
+
     started_at: datetime | None = Field(default=None, description="When started")
     completed_at: datetime | None = Field(default=None, description="When completed")
     commit: str | None = Field(default=None, description="Commit SHA if changes made")

orchestrator/routes/pipelines.py

@@ -2555,24 +2555,35 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]:
                 # ``agent-heartbeat-stall`` trigger is structurally dead on
                 # the ``restart_agent`` path (issue #2084).
                 respawn_started_at = datetime.now(UTC)
+                # Match on ``(role, slice_id)`` — without the slice tiebreaker
+                # the first matching role wins, which on a multi-slice phase
+                # mutates the wrong slice's record (#2422). ``slice_id`` is
+                # the route-level scope already plumbed into the spawner and
+                # consensus tracker above.
                 found = False
                 for agent in fresh_phase_exec.agents:
-                    if hasattr(agent, "role") and (
-                        agent.role == role
-                        or (hasattr(agent.role, "value") and agent.role.value == role.value)
-                    ):
-                        agent.container_id = spawned.container_info.container_id
-                        agent.status = AgentExecutionStatus.RUNNING
-                        agent.started_at = respawn_started_at
-                        found = True
-                        break
+                    if not hasattr(agent, "role"):
+                        continue
+                    role_match = agent.role == role or (
+                        hasattr(agent.role, "value") and agent.role.value == role.value
+                    )
+                    if not role_match:
+                        continue
+                    if getattr(agent, "slice_id", None) != slice_id:
+                        continue
+                    agent.container_id = spawned.container_info.container_id
+                    agent.status = AgentExecutionStatus.RUNNING
+                    agent.started_at = respawn_started_at
+                    found = True
+                    break
                 if not found:
                     fresh_phase_exec.agents.append(
                         AgentExecution(
                             role=role,
                             container_id=spawned.container_info.container_id,
                             status=AgentExecutionStatus.RUNNING,
                             started_at=respawn_started_at,
+                            slice_id=slice_id,
                         )
                     )
 
@@ -11903,6 +11914,7 @@ def _run_concurrent_phase(
                         ),
                         container_id=exec_info.container_id,
                         started_at=datetime.now(UTC),
+                        slice_id=slice_id,
                     )
                     phase_execution.agents.append(agent_state)
                 store.save_pipeline(pip)
@@ -12124,7 +12136,14 @@ def _update_agents_complete() -> None:
                     except Exception:
                         pass
 
+                # Filter to this slice's agents — without the filter, slice-2
+                # BRC completing flips slice-3's still-running agents to
+                # COMPLETE because they share ``pe.agents`` (#2422). For
+                # pipeline-level (non-sliced) phases ``slice_id`` is ``None``
+                # and we still match all agents whose ``slice_id`` is ``None``.
                 for agent in pe.agents:
+                    if getattr(agent, "slice_id", None) != slice_id:
+                        continue
                     if agent.status in (StateAgentStatus.RUNNING, StateAgentStatus.FAILED):
                         agent.status = StateAgentStatus.COMPLETE
                         agent.completed_at = datetime.now(UTC)
@@ -13046,11 +13065,20 @@ def _spawn_and_wait(
                 )
                 phase_execution.containers.append(container_info)
 
-                # Track agent execution
+                # Track agent execution.
+                #
+                # ``slice_id`` is explicitly ``None`` because this helper has
+                # no production callers today and is reachable only from
+                # tests that mock-patch it. If a future change resurrects
+                # this path for a sliced spawn, the caller MUST plumb a
+                # ``slice_id`` through here — otherwise the new
+                # ``(role, slice_id)`` walks added in #2422 will not see
+                # the record. See PR #2435 review thread.
                 agent_execution = AgentExecution(
                     role=agent_role,
                     status=AgentExecutionStatus.RUNNING,
                     container_id=spawned.container_info.container_id,
+                    slice_id=None,
                     started_at=datetime.now(UTC),
                 )
                 phase_execution.agents.append(agent_execution)

orchestrator/routes/signals.py

@@ -573,6 +573,19 @@ def handle_error_signal(
     error_message = data.get("error", "Unknown error")
     recoverable = data.get("recoverable", False)
 
+    # Slice-scope the "already COMPLETE" suppression check below — without
+    # this, a slice-2 coder finishing would silently swallow a slice-3
+    # coder's error because both records share ``phase_execution.agents``
+    # (#2422). The sandbox attaches ``slice_id`` on per-slice agents via
+    # ``progress._maybe_attach_slice_id``; pipeline-level agents send no
+    # ``slice_id`` and this resolves to ``None`` (matches non-sliced
+    # records). ``_extract_slice_id`` rejects malformed values the same
+    # way the BRC handlers do.
+    try:
+        signal_slice_id = _extract_slice_id(data)
+    except ValueError as exc:
+        return make_error_response(f"Invalid slice_id: {exc}")
+
     try:
         store = get_state_store(repo_path)
         pipeline = store.load_pipeline(pipeline_id)
@@ -602,11 +615,16 @@ def handle_error_signal(
         phase_execution = pipeline.phases.get(phase_key)
         if phase_execution is not None:
             for agent in phase_execution.agents:
-                if agent.role == agent_role and agent.status == AgentExecutionStatus.COMPLETE:
+                if (
+                    agent.role == agent_role
+                    and getattr(agent, "slice_id", None) == signal_slice_id
+                    and agent.status == AgentExecutionStatus.COMPLETE
+                ):
                     logger.info(
                         "Agent already COMPLETE, suppressing error signal (consensus path)",
                         pipeline_id=pipeline_id,
                         role=agent_role.value,
+                        slice_id=signal_slice_id,
                     )
                     return make_success_response(
                         "Error suppressed (agent already complete)",

orchestrator/startup_reconciliation.py

@@ -316,10 +316,28 @@ def reconcile_stale_containers(store: object, docker_client: object) -> int:
 
                         phase_exec = pipeline.phases.get(pipeline.current_phase.value)
                         if phase_exec is not None:
+                            # The reconstructed tracker is the pipeline-level
+                            # one (``get_peer_consensus_tracker(pipeline_id)``
+                            # — no slice arg), so only mark pipeline-level
+                            # agents COMPLETE. Per-slice tracker
+                            # reconstruction would have to evaluate each
+                            # slice's tracker separately; flipping every
+                            # agent regardless of slice would prematurely
+                            # complete agents whose slice-scoped consensus
+                            # hadn't actually reached terminal state (#2422).
                             for agent in phase_exec.agents:
+                                if getattr(agent, "slice_id", None) is not None:
+                                    continue
                                 if agent.status == AgentExecutionStatus.RUNNING:
                                     agent.status = AgentExecutionStatus.COMPLETE
                                     agent.completed_at = datetime.now(UTC)
+                            # TODO(#2441): phase-level mutations are
+                            # unconditional even though the agent walk above
+                            # is slice-scoped (only pipeline-level agents
+                            # flipped). Marks the whole phase COMPLETE even
+                            # if per-slice trackers are still RUNNING; safe
+                            # today because per-slice tracker reconstruction
+                            # isn't wired in here yet.
                             phase_exec.status = PipelineStatus.COMPLETE
                             phase_exec.completed_at = datetime.now(UTC)
                             store.save_pipeline(pipeline)

orchestrator/tests/test_models.py

@@ -4,6 +4,7 @@
 
 from datetime import UTC, datetime
 
+import pytest
 from models import (
     PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN,
     AgentExecution,
@@ -75,6 +76,32 @@ def test_agent_with_outputs(self):
         assert agent.commit == "abc1234"
         assert agent.outputs["files_changed"] == ["src/main.py"]
 
+    def test_slice_id_none_allowed(self):
+        """``slice_id=None`` (the default) is the pipeline-level scope."""
+        agent = AgentExecution(role=AgentRole.CODER)
+        assert agent.slice_id is None
+
+    def test_slice_id_canonical_accepted(self):
+        """Canonical ``slice-<N>`` ids pass the validator."""
+        agent = AgentExecution(role=AgentRole.CODER, slice_id="slice-2")
+        assert agent.slice_id == "slice-2"
+
+    @pytest.mark.parametrize(
+        "bad_value",
+        ["phase-2", "slice-", "slice-2a", "Slice-2", " slice-2", "slice-2 ", ""],
+    )
+    def test_slice_id_non_canonical_rejected(self, bad_value):
+        """Non-canonical ``slice_id`` values are rejected at construction.
+
+        Defense-in-depth (#2422 review): production write paths use
+        ``extract_slice_id`` / ``concurrent_executor._slice_id`` which
+        already enforce ``SLICE_ID_PATTERN``, but a hand-built fixture
+        or migration tool must not be able to smuggle a non-canonical
+        value through ``AgentExecution(...)``.
+        """
+        with pytest.raises(ValueError, match="Invalid slice_id"):
+            AgentExecution(role=AgentRole.CODER, slice_id=bad_value)
+
 
 class TestHITLDecision:
     """Tests for HITLDecision model."""