diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index 4c2894bfce..bbd70e7073 100644 --- a/orchestrator/concurrent_executor.py +++ b/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]: diff --git a/orchestrator/kubernetes_monitor.py b/orchestrator/kubernetes_monitor.py index f6acc8d13e..010f2bb4b2 100644 --- a/orchestrator/kubernetes_monitor.py +++ b/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 diff --git a/orchestrator/models.py b/orchestrator/models.py index 639f9bc2a3..7d1031e340 100644 --- a/orchestrator/models.py +++ b/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-'") + 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") diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 7f9c065b75..2ca2f49506 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -2555,17 +2555,27 @@ 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( @@ -2573,6 +2583,7 @@ def restart_agent(pipeline_id: str, agent_role: str) -> tuple[Response, int]: 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) diff --git a/orchestrator/routes/signals.py b/orchestrator/routes/signals.py index 4ca6a85edb..044748b045 100644 --- a/orchestrator/routes/signals.py +++ b/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)", diff --git a/orchestrator/startup_reconciliation.py b/orchestrator/startup_reconciliation.py index 97a3219dd9..d67ccac41d 100644 --- a/orchestrator/startup_reconciliation.py +++ b/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) diff --git a/orchestrator/tests/test_models.py b/orchestrator/tests/test_models.py index b8f00637c5..af463e3845 100644 --- a/orchestrator/tests/test_models.py +++ b/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-`` 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.""" diff --git a/orchestrator/tests/test_restart_agent.py b/orchestrator/tests/test_restart_agent.py index d4915fbba3..f2111e75d4 100644 --- a/orchestrator/tests/test_restart_agent.py +++ b/orchestrator/tests/test_restart_agent.py @@ -925,6 +925,217 @@ def test_no_slice_id_forwards_none( assert get_count_call.kwargs.get("slice_id") is None +# --------------------------------------------------------------------------- +# Issue #2422: in-place agent-state mutation matches on (role, slice_id) +# --------------------------------------------------------------------------- + + +def _make_pipeline_with_two_slice_coders(): + """Pipeline with concurrent slice-2 + slice-3 coder records. + + The two ``AgentExecution`` records share ``role=CODER`` and only + differ on ``slice_id``. Pre-#2422 the restart route walked + ``phase_exec.agents`` looking for ``agent.role == role`` and mutated + the first match — so a slice-3 restart would clobber slice-2's + record. This fixture lets the test assert the new ``(role, + slice_id)`` predicate keeps the unrelated slice's row untouched. + """ + pipeline = Pipeline( + id="issue-100", + issue_number=100, + repo="owner/repo", + branch="egg/issue-100", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + pipeline.phases = { + "implement": PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.RUNNING, + containers=[ + ContainerInfo( + container_id="container-slice-2", + container_name="egg-issue-100-slice-2-coder", + agent_role=AgentRole.CODER, + status=ContainerStatus.RUNNING, + ), + ContainerInfo( + container_id="container-slice-3", + container_name="egg-issue-100-slice-3-coder", + agent_role=AgentRole.CODER, + status=ContainerStatus.RUNNING, + ), + ], + agents=[ + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.RUNNING, + container_id="container-slice-2", + slice_id="slice-2", + ), + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.RUNNING, + container_id="container-slice-3", + slice_id="slice-3", + ), + ], + ), + } + return pipeline + + +@pytest.mark.skipif(not _HAS_FLASK, reason="Flask not available") +class TestRestartAgentSliceMatching: + """``restart_agent`` mutation predicate matches on ``(role, slice_id)`` (#2422).""" + + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_slice_3_restart_does_not_mutate_slice_2_record( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, client + ): + """Restarting slice-3 coder leaves slice-2 coder's AgentExecution intact.""" + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + pipeline = _make_pipeline_with_two_slice_coders() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + new_container = SpawnedContainer( + container_info=ContainerInfo( + container_id="container-slice-3-RESTARTED", + container_name="egg-issue-100-slice-3-coder", + status=ContainerStatus.RUNNING, + ), + session_info=None, + agent_role=AgentRole.CODER, + pipeline_id="issue-100", + environment={}, + ) + mock_spawner.restart_agent_container.return_value = new_container + mock_spawner.get_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + slice2_before = next( + a for a in pipeline.phases["implement"].agents if a.slice_id == "slice-2" + ) + slice2_container_before = slice2_before.container_id + + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart?slice_id=slice-3", + json={"reason": "slice-3 stall"}, + ) + + assert response.status_code == 200, response.get_json() + + # Assert on the *persisted* state, not just the in-memory pipeline. + # The route serialises the mutated pipeline via + # ``store.update_pipeline(pipeline_id, pipeline.model_dump(mode="json"))``; + # inspecting the call_args guards against a future refactor where + # ``_resolve_pipeline`` returns a copy and the route forgets to + # save the mutation back. + mock_store.update_pipeline.assert_called() + persisted = mock_store.update_pipeline.call_args[0][1] + persisted_agents = persisted["phases"]["implement"]["agents"] + slice2_persisted = next(a for a in persisted_agents if a["slice_id"] == "slice-2") + slice3_persisted = next(a for a in persisted_agents if a["slice_id"] == "slice-3") + + assert slice2_persisted["container_id"] == slice2_container_before, ( + "slice-2 container_id must not change when slice-3 is restarted" + ) + assert slice3_persisted["container_id"] == "container-slice-3-RESTARTED" + assert slice3_persisted["status"] == AgentExecutionStatus.RUNNING.value + + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_slice_3_restart_with_no_existing_record_appends_with_slice_id( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_lock_fn, client + ): + """Fall-through ``AgentExecution`` append carries the route's slice_id.""" + mock_repo.return_value = "/repo" + mock_lock_fn.return_value = MagicMock() + + # Pipeline has only slice-2 coder; slice-3 restart should append + # a new slice-3 row rather than mutating slice-2's record. + pipeline = Pipeline( + id="issue-100", + issue_number=100, + repo="owner/repo", + branch="egg/issue-100", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + pipeline.phases = { + "implement": PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.RUNNING, + containers=[ + ContainerInfo( + container_id="container-slice-2", + container_name="egg-issue-100-slice-2-coder", + agent_role=AgentRole.CODER, + status=ContainerStatus.RUNNING, + ), + ], + agents=[ + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.RUNNING, + container_id="container-slice-2", + slice_id="slice-2", + ), + ], + ), + } + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + new_container = SpawnedContainer( + container_info=ContainerInfo( + container_id="container-slice-3-NEW", + container_name="egg-issue-100-slice-3-coder", + status=ContainerStatus.RUNNING, + ), + session_info=None, + agent_role=AgentRole.CODER, + pipeline_id="issue-100", + environment={}, + ) + mock_spawner.restart_agent_container.return_value = new_container + mock_spawner.get_restart_count.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + response = client.post( + "/api/v1/pipelines/issue-100/agents/coder/restart?slice_id=slice-3", + json={"reason": "slice-3 first restart"}, + ) + + assert response.status_code == 200 + # Assert on the *persisted* state — same robustness rationale as + # the slice-3-does-not-mutate-slice-2 test above. + mock_store.update_pipeline.assert_called() + persisted = mock_store.update_pipeline.call_args[0][1] + persisted_agents = persisted["phases"]["implement"]["agents"] + # slice-2 untouched, slice-3 appended + assert any( + a["slice_id"] == "slice-2" and a["container_id"] == "container-slice-2" + for a in persisted_agents + ) + slice3_rows = [a for a in persisted_agents if a["slice_id"] == "slice-3"] + assert len(slice3_rows) == 1 + assert slice3_rows[0]["container_id"] == "container-slice-3-NEW" + + # --------------------------------------------------------------------------- # Issue #1695: mode=None raises ValueError (issue 7) # --------------------------------------------------------------------------- diff --git a/orchestrator/tests/test_signals.py b/orchestrator/tests/test_signals.py index 3c39287b04..e995b24783 100644 --- a/orchestrator/tests/test_signals.py +++ b/orchestrator/tests/test_signals.py @@ -504,6 +504,169 @@ def test_error_not_suppressed_when_agent_still_running( # Contract should have been updated with the error mock_orch.fail_agent.assert_called_once() + @patch("routes.signals.save_contract") + @patch("routes.signals.resolve_worktree_path") + @patch("routes.signals.get_state_store") + @patch("routes.signals.load_contract") + def test_slice_3_error_not_suppressed_by_slice_2_complete( + self, + mock_load_contract, + mock_get_store, + mock_resolve_wt, + mock_save_contract, + app, + ): + """slice-3 coder error must not be silently swallowed by a slice-2 coder + already-COMPLETE record (#2422). Pre-fix the role-only predicate matched + the slice-2 row first and returned ``already_complete``.""" + from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + PhaseExecution, + Pipeline, + PipelinePhase, + PipelineStatus, + ) + + pipeline = Pipeline( + id="issue-42", + issue_number=42, + repo="owner/repo", + branch="egg/issue-42", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + phase_exec = PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + agents=[ + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.COMPLETE, + slice_id="slice-2", + ), + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.RUNNING, + slice_id="slice-3", + ), + ], + ) + pipeline.phases["implement"] = phase_exec + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_get_store.return_value = mock_store + mock_resolve_wt.return_value = Path("/tmp/worktree") + mock_load_contract.return_value = MagicMock() + + mock_orch = MagicMock() + mock_orch.apply_to_contract.return_value = MagicMock() + + with patch("routes.signals.create_orchestrator", return_value=mock_orch): + with app.app_context(): + from routes.signals import handle_error_signal + + response, status_code = handle_error_signal( + "issue-42", + { + "agent_role": "coder", + "error": "Build failed in slice-3", + "recoverable": False, + "slice_id": "slice-3", + }, + Path("/tmp/repo"), + ) + + assert status_code == 200 + data = json.loads(response.data) + # slice-3 is RUNNING, so it must NOT be suppressed + assert "already_complete" not in data.get("data", {}), ( + "slice-3 error was suppressed by slice-2's COMPLETE record" + ) + mock_orch.fail_agent.assert_called_once() + + @patch("routes.signals.get_state_store") + def test_slice_3_error_suppressed_when_slice_3_complete( + self, + mock_get_store, + app, + ): + """slice-3 coder COMPLETE → slice-3 coder error is suppressed (positive case).""" + from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + PhaseExecution, + Pipeline, + PipelinePhase, + PipelineStatus, + ) + + pipeline = Pipeline( + id="issue-42", + issue_number=42, + repo="owner/repo", + branch="egg/issue-42", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + phase_exec = PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + agents=[ + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.RUNNING, + slice_id="slice-2", + ), + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.COMPLETE, + slice_id="slice-3", + ), + ], + ) + pipeline.phases["implement"] = phase_exec + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_get_store.return_value = mock_store + + with app.app_context(): + from routes.signals import handle_error_signal + + response, status_code = handle_error_signal( + "issue-42", + { + "agent_role": "coder", + "error": "post-consensus SIGTERM", + "recoverable": False, + "slice_id": "slice-3", + }, + Path("/tmp/repo"), + ) + + assert status_code == 200 + data = json.loads(response.data) + assert data["data"]["already_complete"] is True + + def test_invalid_slice_id_returns_400(self, app): + """Malformed slice_id is rejected before touching pipeline state.""" + with app.app_context(): + from routes.signals import handle_error_signal + + response, status_code = handle_error_signal( + "issue-42", + { + "agent_role": "coder", + "error": "x", + "slice_id": "../etc", + }, + Path("/tmp/repo"), + ) + + assert status_code == 400 + # --------------------------------------------------------------------------- # Completion signal branch verification tests (TASK-5-3) diff --git a/sandbox/egg_agent_tools/handlers/progress.py b/sandbox/egg_agent_tools/handlers/progress.py index 6e14cd7b1b..a94a549d23 100644 --- a/sandbox/egg_agent_tools/handlers/progress.py +++ b/sandbox/egg_agent_tools/handlers/progress.py @@ -8,11 +8,30 @@ from egg_agent_tools.handlers._gateway import ( get_agent_role, get_pipeline_id, + get_slice_id, orchestrator_request, ) from egg_agent_tools.handlers.errors import GatewayError, HandlerError _PIPELINE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$") +_SLICE_ID_PATTERN = re.compile(r"^slice-[0-9]+$") + + +def _maybe_attach_slice_id(req: dict[str, Any], data: dict[str, Any]) -> None: + """Forward ``slice_id`` from the request or env onto the signal body. + + Mirrors ``brc._maybe_attach_slice_id`` so the orchestrator-side + ``handle_error_signal`` can scope its "agent already COMPLETE, + suppress error" check by ``(role, slice_id)`` rather than role + alone — without this, slice-2 coder finishing would silently + swallow slice-3 coder's error (#2422). + """ + slice_id = req.get("slice_id") or get_slice_id() + if not slice_id: + return + if not isinstance(slice_id, str) or not _SLICE_ID_PATTERN.fullmatch(slice_id): + raise HandlerError(f"Invalid slice_id {slice_id!r}: must match 'slice-'") + data["slice_id"] = slice_id def _require_pipeline_id(req: dict[str, Any]) -> str: @@ -97,12 +116,13 @@ def progress_signal_error(req: dict[str, Any]) -> dict[str, Any]: raise HandlerError("'error' is required") recoverable = bool(req.get("recoverable", False)) - data = { + data: dict[str, Any] = { "signal_type": "error", "agent_role": role, "error": error, "recoverable": recoverable, } + _maybe_attach_slice_id(req, data) result = orchestrator_request(f"/api/v1/pipelines/{pid}/signal", method="POST", data=data) if not result.get("success"): raise GatewayError(result.get("message", "error signal failed")) diff --git a/sandbox/tests/test_progress_slice_routing.py b/sandbox/tests/test_progress_slice_routing.py new file mode 100644 index 0000000000..bf3244101c --- /dev/null +++ b/sandbox/tests/test_progress_slice_routing.py @@ -0,0 +1,90 @@ +"""Progress error-signal handler threads ``slice_id`` from ``EGG_SLICE_ID`` (#2422). + +Per-slice agents must tag the error signal with their ``slice_id`` so the +orchestrator's "agent already COMPLETE" suppression check can scope by +``(role, slice_id)`` instead of role alone — without it, a slice-2 coder +finishing first would silently swallow a slice-3 coder's error because +both ``AgentExecution`` records share ``phase_exec.agents``. +""" + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest + +_sandbox_path = str(Path(__file__).parent.parent) +if _sandbox_path not in sys.path: + sys.path.insert(0, _sandbox_path) + + +_ERROR_REQ = { + "pipeline_id": "issue-2422", + "role": "coder", + "error": "Build failed", + "recoverable": False, +} + + +def _captured_data(mock_request: Any) -> dict[str, Any]: + assert mock_request.called, "orchestrator_request was not invoked" + return dict(mock_request.call_args.kwargs["data"]) + + +class TestErrorSignalSliceId: + """``EGG_SLICE_ID`` flows onto the error signal body (#2422).""" + + def test_error_signal_attaches_slice_id_from_env(self, monkeypatch): + from egg_agent_tools.handlers import progress as handlers + + monkeypatch.setenv("EGG_SLICE_ID", "slice-3") + + with patch( + "egg_agent_tools.handlers.progress.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.progress_signal_error(dict(_ERROR_REQ)) + + assert _captured_data(mock_request)["slice_id"] == "slice-3" + + def test_request_slice_id_overrides_env(self, monkeypatch): + from egg_agent_tools.handlers import progress as handlers + + monkeypatch.setenv("EGG_SLICE_ID", "slice-9") + + with patch( + "egg_agent_tools.handlers.progress.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.progress_signal_error({**_ERROR_REQ, "slice_id": "slice-3"}) + + assert _captured_data(mock_request)["slice_id"] == "slice-3" + + def test_no_slice_id_omits_field(self, monkeypatch): + from egg_agent_tools.handlers import progress as handlers + + monkeypatch.delenv("EGG_SLICE_ID", raising=False) + + with patch( + "egg_agent_tools.handlers.progress.orchestrator_request", + return_value={"success": True, "data": {}}, + ) as mock_request: + handlers.progress_signal_error(dict(_ERROR_REQ)) + + assert "slice_id" not in _captured_data(mock_request) + + def test_invalid_slice_id_rejected(self, monkeypatch): + from egg_agent_tools.handlers import progress as handlers + from egg_agent_tools.handlers.errors import HandlerError + + # Path-separator values must not reach the wire — defense in depth + # against a malformed env var. + monkeypatch.setenv("EGG_SLICE_ID", "slice-2/../etc") + + with patch( + "egg_agent_tools.handlers.progress.orchestrator_request", + return_value={"success": True, "data": {}}, + ): + with pytest.raises(HandlerError, match="slice_id"): + handlers.progress_signal_error(dict(_ERROR_REQ))