Fix retry phase-goal continuation

apps/decodex/src/orchestrator/execution.rs

@@ -193,6 +193,53 @@ impl RepoGatePhaseGoalController<'_> {
 			.and_then(phase_goal_kind_from_str))
 	}
 
+	fn latest_cross_attempt_phase_goal(&self) -> Result<Option<PhaseGoalKind>> {
+		if self.issue_run.dispatch_mode != IssueDispatchMode::Retry {
+			return Ok(None);
+		}
+
+		// Retry attempts inherit only the immediately previous attempt's open phase
+		// boundary, so older validation passes cannot override newer work.
+		let events = self.state_store.list_private_execution_events_for_issue(
+			self.project.service_id(),
+			&self.issue_run.issue.id,
+		)?;
+		let previous_attempt = self.issue_run.attempt_number - 1;
+
+		if previous_attempt < 1 {
+			return Ok(None);
+		}
+
+		for event in events.iter().rev().filter(|event| {
+			event.attempt_number() == previous_attempt
+				&& event.run_id() != self.issue_run.run_id
+		}) {
+			match event.event_type() {
+				"terminal_finalize"
+				| "review_completion_intent"
+				| AUTHORITY_DECISION_REQUEST_EVENT_TYPE
+				| PHASE_GOAL_RECOVERY_BLOCKED_EVENT_TYPE => return Ok(None),
+				"progress_checkpoint" if progress_checkpoint_has_blockers(event.payload()) =>
+					return Ok(None),
+				PHASE_GOAL_RECOVERY_EVENT_TYPE | "phase_goal_next" | "phase_goal_transition" => {
+					if let Some(phase) =
+						phase_goal_continuation_next_phase(event.event_type(), event.payload())
+					{
+						return Ok(Some(phase));
+					}
+				},
+				"phase_goal_set" | "phase_goal_status" => {
+					if let Some(phase) = phase_goal_active_phase(event.payload()) {
+						return Ok(Some(phase));
+					}
+				},
+				_ => {},
+			}
+		}
+
+		Ok(None)
+	}
+
 	fn validate_phase_goal_output(&self, phase: PhaseGoalKind) -> Result<PhaseGoalTransition> {
 		let selected_repo_gate = select_repo_gate_for_worktree(
 			self.workflow.frontmatter().execution(),
@@ -366,6 +413,9 @@ impl PhaseGoalController for RepoGatePhaseGoalController<'_> {
 		if let Some(phase) = self.latest_persisted_phase_goal()? {
 			return Ok(Some(self.phase_goal_spec(phase, None)));
 		}
+		if let Some(phase) = self.latest_cross_attempt_phase_goal()? {
+			return Ok(Some(self.phase_goal_spec(phase, None)));
+		}
 
 		Ok(Some(self.phase_goal_spec(self.initial_phase_goal_kind(), None)))
 	}
@@ -1469,6 +1519,13 @@ fn phase_goal_recovery_candidate_from_status(
 	}
 }
 
+fn phase_goal_active_phase(payload: &Value) -> Option<PhaseGoalKind> {
+	let phase = phase_goal_event_phase(payload)?;
+	let status = phase_goal_event_status(payload)?;
+
+	(status == "active").then_some(phase)
+}
+
 fn matching_phase_goal_recovery_count(
 	project: &ServiceConfig,
 	state_store: &StateStore,
@@ -1502,6 +1559,19 @@ fn phase_goal_recovery_event_source_error_class(payload: &Value) -> Option<&str>
 	payload.get("payload")?.get("sourceErrorClass")?.as_str()
 }
 
+fn phase_goal_continuation_next_phase(
+	event_type: &str,
+	payload: &Value,
+) -> Option<PhaseGoalKind> {
+	let phase = if event_type == "phase_goal_next" {
+		payload.get("phase")?.as_str()?
+	} else {
+		payload.get("payload")?.get("nextPhase")?.as_str()?
+	};
+
+	phase_goal_kind_from_str(phase)
+}
+
 fn record_phase_goal_recovery_continuation(record: PhaseGoalRecoveryRecord<'_>) -> Result<()> {
 	record.state_store.append_private_execution_event(
 		record.project.service_id(),

apps/decodex/src/orchestrator/status.rs

@@ -480,12 +480,22 @@ fn operator_current_lane_statuses(
 		.collect::<Vec<_>>();
 	let mut current_lane_run_ids =
 		current_lanes.iter().map(|run| run.run_id.clone()).collect::<HashSet<_>>();
+	let current_lane_shadow_keys = current_lanes
+		.iter()
+		.filter(|run| run.run_lease)
+		.map(operator_run_group_key)
+		.collect::<HashSet<_>>();
 
 	for run in recent_runs {
-		if !current_lane_run_ids.contains(&run.run_id) && operator_run_has_live_execution(run) {
-			current_lane_run_ids.insert(run.run_id.clone());
-			current_lanes.push(run.clone());
+		if current_lane_run_ids.contains(&run.run_id)
+			|| current_lane_shadow_keys.contains(&operator_run_group_key(run))
+			|| !operator_run_has_live_execution(run)
+		{
+			continue;
 		}
+
+		current_lane_run_ids.insert(run.run_id.clone());
+		current_lanes.push(run.clone());
 	}
 
 	hydrate_current_lane_lifecycle_metrics(&mut current_lanes, recent_runs);

apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs

@@ -2127,6 +2127,60 @@ fn operator_status_snapshot_counts_stale_starting_run_as_attention_not_running()
 	assert_eq!(project.attention_count, 1);
 }
 
+#[test]
+fn operator_status_snapshot_shadows_stale_attempt_when_newer_leased_attempt_exists() {
+	let (_temp_dir, config, _workflow) = temp_project_layout();
+	let state_store = StateStore::open_in_memory().expect("state store should open");
+	let issue = sample_issue("Todo", &[]);
+	let worktree_path = config.worktree_root().join("PUB-101");
+	let stale_activity =
+		OffsetDateTime::now_utc().unix_timestamp() - RUN_LEASE_IDLE_TIMEOUT.as_secs() as i64 - 30;
+	let stale_run_id = "pub-101-attempt-2-1781621836";
+	let current_run_id = "pub-101-attempt-3-1781623863";
+
+	state_store
+		.record_run_attempt(stale_run_id, &issue.id, 2, "running")
+		.expect("stale run attempt should record");
+	state_store
+		.record_run_attempt(current_run_id, &issue.id, 3, "running")
+		.expect("current run attempt should record");
+	state_store
+		.upsert_lease("pubfi", &issue.id, current_run_id, "In Progress")
+		.expect("current run lease should record");
+	state_store
+		.upsert_worktree(
+			"pubfi",
+			&issue.id,
+			"x/pubfi-pub-101",
+			&worktree_path.display().to_string(),
+		)
+		.expect("worktree should record");
+
+	fs::create_dir_all(&worktree_path).expect("worktree path should exist");
+	fs::write(
+		worktree_path.join(RUN_ACTIVITY_MARKER_FILE),
+		format!(
+			"run_id={stale_run_id}\nattempt_number=2\nlast_activity_unix_epoch={stale_activity}\nlast_protocol_activity_unix_epoch={stale_activity}\nevent_count=1\nlast_event_type=skills/changed\n"
+		),
+	)
+	.expect("stale protocol marker should write");
+
+	let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 10)
+		.expect("snapshot should build");
+	let project = snapshot.projects.first().expect("project summary should exist");
+	let rendered = orchestrator::render_operator_status(&snapshot);
+
+	assert_eq!(snapshot.current_lanes.len(), 1);
+	assert_eq!(snapshot.current_lanes[0].run_id, current_run_id);
+	assert!(snapshot.recent_runs.iter().any(|run| run.run_id == stale_run_id));
+	assert_eq!(project.current_lane_count, 1);
+	assert_eq!(project.running_lane_count, 1);
+	assert_eq!(project.attention_count, 0);
+	assert!(rendered.contains("Current lanes: 1"));
+	assert!(rendered.contains("Running lanes: 1"));
+	assert!(!rendered.contains(stale_run_id));
+}
+
 #[test]
 fn operator_status_snapshot_excludes_completed_lingering_lease_from_current_lanes() {
 	let (_temp_dir, config, _workflow) = temp_project_layout();