Recover retained phase-goal continuations

apps/decodex/src/orchestrator/execution.rs

@@ -1330,6 +1330,9 @@ fn latest_active_phase_goal_recovery_candidate(
 			| "phase_goal_transition"
 			| "review_completion_intent"
 			| "terminal_finalize" => return Ok(None),
+			AUTHORITY_DECISION_REQUEST_EVENT_TYPE => return Ok(None),
+			"progress_checkpoint" if progress_checkpoint_has_blockers(event.payload()) =>
+				return Ok(None),
 			"phase_goal_set" | "phase_goal_status" => {
 				let Some(phase) = phase_goal_event_phase(event.payload()) else {
 					return Ok(None);
@@ -1347,6 +1350,14 @@ fn latest_active_phase_goal_recovery_candidate(
 	Ok(None)
 }
 
+fn progress_checkpoint_has_blockers(payload: &Value) -> bool {
+	payload.get("blockers").is_some_and(|blockers| match blockers {
+		Value::Array(items) => !items.is_empty(),
+		Value::Null => false,
+		_ => true,
+	})
+}
+
 fn phase_goal_event_phase(payload: &Value) -> Option<PhaseGoalKind> {
 	payload
 		.get("phase")

apps/decodex/src/orchestrator/git_ops.rs

@@ -615,14 +615,81 @@ fn ensure_repo_gate_left_no_tracked_changes(cwd: &Path, phase: &str) -> Result<(
 	Ok(())
 }
 
+fn read_repo_gate_tracked_diff(cwd: &Path, phase: &str) -> Result<String> {
+	let output = Command::new("git")
+		.arg("-C")
+		.arg(cwd)
+		.args(["diff", "--no-ext-diff", "--binary", "HEAD", "--"])
+		.output()
+		.map_err(|error| {
+			Report::new(RepoGateFailure::new(
+				RepoGateFailureKind::CommandSpawnFailed,
+				format!(
+					"Failed to spawn tracked-file diff check after repo gate {phase} in `{}`: {error}",
+					cwd.display()
+				),
+			))
+		})?;
+
+	if !output.status.success() {
+		let output_text = repo_gate_output_text(&output);
+
+		return Err(Report::new(RepoGateFailure::new(
+			repo_gate_failure_kind_for_output(
+				RepoGateFailureKind::CleanlinessCheckFailed,
+				&output_text,
+			),
+			format!(
+				"Failed to inspect tracked-file diff after repo gate {phase} in `{}`: {}",
+				cwd.display(),
+				output_text
+			),
+		)));
+	}
+
+	Ok(String::from_utf8_lossy(&output.stdout).into_owned())
+}
+
+fn ensure_repo_gate_preserved_tracked_diff(
+	cwd: &Path,
+	phase: &str,
+	before_diff: &str,
+) -> Result<()> {
+	let after_diff = read_repo_gate_tracked_diff(cwd, phase)?;
+
+	if after_diff == before_diff {
+		return Ok(());
+	}
+
+	let output = run_repo_gate_cleanliness_check_with_git(std::ffi::OsStr::new("git"), cwd)?;
+	let output_text = repo_gate_output_text(&output);
+	let dirty_entries = output_text.trim();
+
+	Err(Report::new(RepoGateFailure::new(
+		RepoGateFailureKind::TrackedRewritesLeft,
+		format!(
+			"Repo gate {phase} rewrote tracked files in `{}`; commit or revert these changes before continuing:\n{}",
+			cwd.display(),
+			dirty_entries
+		),
+	)))
+}
+
 fn run_repo_gate_commands(
 	canonicalize_commands: &[String],
 	verify_commands: &[String],
 	cwd: &Path,
 ) -> Result<()> {
+	let baseline_tracked_diff = read_repo_gate_tracked_diff(cwd, "baseline")?;
+
 	run_canonicalize_commands(canonicalize_commands, cwd)?;
 	run_verify_commands(verify_commands, cwd)?;
-	ensure_repo_gate_left_no_tracked_changes(cwd, "verification")?;
+
+	if baseline_tracked_diff.is_empty() {
+		ensure_repo_gate_left_no_tracked_changes(cwd, "verification")?;
+	} else {
+		ensure_repo_gate_preserved_tracked_diff(cwd, "verification", &baseline_tracked_diff)?;
+	}
 
 	Ok(())
 }

apps/decodex/src/orchestrator/reconciliation.rs

@@ -423,10 +423,40 @@ where
 		"Reconciling stalled run with retained partial progress."
 	);
 
+	let issue_run = stalled_reconciliation_issue_run(state_store, worktree_manager, action)?;
+	let recovered = match try_recover_stalled_retained_phase_goal(
+		project,
+		&action.workflow,
+		state_store,
+		&action.issue,
+		&issue_run,
+	) {
+		Ok(recovered) => recovered,
+		Err(error) if run_failure_requires_terminal_attention(&error) => {
+			state_store.update_run_status(action.run_attempt.run_id(), "stalled")?;
+			state_store.clear_lease(&action.issue.id)?;
+
+			handle_failure(
+				tracker,
+				project,
+				&action.workflow,
+				state_store,
+				&issue_run,
+				&error,
+			)?;
+
+			return Ok(());
+		},
+		Err(error) => return Err(error),
+	};
+
+	if recovered {
+		return Ok(());
+	}
+
 	state_store.update_run_status(action.run_attempt.run_id(), "stalled")?;
 	state_store.clear_lease(&action.issue.id)?;
 
-	let issue_run = stalled_reconciliation_issue_run(state_store, worktree_manager, action)?;
 	let worktree_path = relative_worktree_path(project, &issue_run.worktree);
 
 	write_reconciliation_operation_marker_best_effort(
@@ -452,6 +482,71 @@ where
 	Ok(())
 }
 
+fn try_recover_stalled_retained_phase_goal(
+	project: &ServiceConfig,
+	workflow: &WorkflowDocument,
+	state_store: &StateStore,
+	issue: &TrackerIssue,
+	issue_run: &IssueRunPlan,
+) -> Result<bool> {
+	write_reconciliation_operation_marker_best_effort(
+		&issue_run.worktree.path,
+		&issue_run.run_id,
+		issue_run.attempt_number,
+		RUN_OPERATION_RECONCILIATION,
+	);
+
+	let recovery = recover_active_phase_goal_continuation(
+		project,
+		workflow,
+		state_store,
+		issue_run,
+		"stalled_run_detected",
+	)?;
+	let Some(recovery) = recovery else {
+		return Ok(false);
+	};
+
+	state_store.update_run_status(&issue_run.run_id, CONTINUATION_PENDING_RUN_STATUS)?;
+	state_store.clear_lease(&issue.id)?;
+
+	write_stalled_phase_goal_continuation_retry_marker(state_store, workflow, issue_run)?;
+
+	tracing::warn!(
+		project_id = project.service_id(),
+		issue_id = issue.id,
+		issue = issue.identifier,
+		run_id = issue_run.run_id,
+		attempt = issue_run.attempt_number,
+		source_phase = recovery.source_phase.as_str(),
+		next_phase = recovery.next_phase.as_str(),
+		"Recovered stalled retained phase goal; scheduling continuation instead of manual attention."
+	);
+
+	Ok(true)
+}
+
+fn write_stalled_phase_goal_continuation_retry_marker(
+	state_store: &StateStore,
+	workflow: &WorkflowDocument,
+	issue_run: &IssueRunPlan,
+) -> Result<()> {
+	let attempt = u32::try_from(issue_run.attempt_number).unwrap_or(u32::MAX).max(1);
+	let delay = retry_delay(RetryKind::Continuation, attempt, workflow);
+	let retry_ready_at_unix_epoch = OffsetDateTime::now_utc().unix_timestamp().saturating_add(
+		i64::try_from((delay.as_millis().saturating_add(999)) / 1_000).unwrap_or(i64::MAX),
+	);
+
+	write_retry_schedule_for_run(
+		state_store,
+		&issue_run.issue.id,
+		&issue_run.run_id,
+		issue_run.attempt_number,
+		RetryKind::Continuation,
+		retry_ready_at_unix_epoch,
+	)
+}
+
 fn stalled_reconciliation_issue_run(
 	state_store: &StateStore,
 	worktree_manager: &WorktreeManager,
@@ -590,6 +685,9 @@ fn stalled_idle_duration(
 	if !matches!(run_attempt.status(), "starting" | "running") {
 		return Ok(None);
 	}
+	if stalled_reconciliation_deferred_by_marker(run_attempt, worktree_mapping)? {
+		return Ok(None);
+	}
 
 	let Some(last_activity) =
 		last_observed_run_activity_unix_epoch(state_store, run_attempt, worktree_mapping)?
@@ -660,6 +758,10 @@ fn stalled_protocol_idle_duration(
 	worktree_mapping: Option<&WorktreeMapping>,
 	now_unix_epoch: i64,
 ) -> Result<Option<Duration>> {
+	if stalled_reconciliation_deferred_by_marker(run_attempt, worktree_mapping)? {
+		return Ok(None);
+	}
+
 	let Some(last_activity) =
 		last_observed_protocol_activity_unix_epoch(state_store, run_attempt, worktree_mapping)?
 	else {
@@ -705,3 +807,40 @@ fn observed_idle_duration(last_activity_unix_epoch: i64, now_unix_epoch: i64) ->
 		.and_then(|idle_seconds| u64::try_from(idle_seconds).ok())
 		.map(Duration::from_secs)
 }
+
+fn stalled_reconciliation_deferred_by_marker(
+	run_attempt: &RunAttempt,
+	worktree_mapping: Option<&WorktreeMapping>,
+) -> Result<bool> {
+	let Some(marker) = current_run_activity_marker(run_attempt, worktree_mapping)? else {
+		return Ok(false);
+	};
+
+	if marker.retry_kind().is_some() {
+		return Ok(true);
+	}
+
+	Ok(marker.current_operation() == Some(RUN_OPERATION_REPO_GATE)
+		&& marker_process_is_alive(&marker))
+}
+
+fn current_run_activity_marker(
+	run_attempt: &RunAttempt,
+	worktree_mapping: Option<&WorktreeMapping>,
+) -> Result<Option<RunActivityMarker>> {
+	let Some(worktree_mapping) = worktree_mapping else {
+		return Ok(None);
+	};
+	let Some(marker) = state::read_run_activity_marker_snapshot(worktree_mapping.worktree_path())?
+	else {
+		return Ok(None);
+	};
+
+	if marker.run_id() == run_attempt.run_id()
+		&& marker.attempt_number() == run_attempt.attempt_number()
+	{
+		return Ok(Some(marker));
+	}
+
+	Ok(None)
+}