From d88a1eba2f52a82f870847c3a95ac24168a5479c Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Wed, 17 Jun 2026 00:33:28 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Fix retry phase-goal continuation","authority":"manual"} --- apps/decodex/src/orchestrator/execution.rs | 70 ++++ apps/decodex/src/orchestrator/status.rs | 16 +- .../tests/operator/status/running_lanes.rs | 54 +++ .../orchestrator/tests/runtime/repo_gate.rs | 324 ++++++++++++++++++ docs/reference/operator-control-plane.md | 6 + docs/spec/lane-control-state.md | 7 + docs/spec/loop-runtime.md | 5 + 7 files changed, 479 insertions(+), 3 deletions(-) diff --git a/apps/decodex/src/orchestrator/execution.rs b/apps/decodex/src/orchestrator/execution.rs index c89249ba3..9778083f0 100644 --- a/apps/decodex/src/orchestrator/execution.rs +++ b/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> { + 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 { 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 { + 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 { + 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(), diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index c951342af..a0f7d9137 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -480,12 +480,22 @@ fn operator_current_lane_statuses( .collect::>(); let mut current_lane_run_ids = current_lanes.iter().map(|run| run.run_id.clone()).collect::>(); + let current_lane_shadow_keys = current_lanes + .iter() + .filter(|run| run.run_lease) + .map(operator_run_group_key) + .collect::>(); 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); diff --git a/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs b/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs index 69d0b7878..4cf063f82 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs +++ b/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(); diff --git a/apps/decodex/src/orchestrator/tests/runtime/repo_gate.rs b/apps/decodex/src/orchestrator/tests/runtime/repo_gate.rs index 6608f7836..c4015a44e 100644 --- a/apps/decodex/src/orchestrator/tests/runtime/repo_gate.rs +++ b/apps/decodex/src/orchestrator/tests/runtime/repo_gate.rs @@ -203,6 +203,330 @@ fn phase_goal_completion_runs_repo_gate_and_persists_handoff_phase() { })); } +#[test] +fn retry_phase_goal_resumes_cross_attempt_handoff_after_recovered_validation_pass() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let issue = sample_issue("In Progress", &[tracker::automation_active_label(TEST_SERVICE_ID).as_str()]); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let first_issue_run = IssueRunPlan { + issue: issue.clone(), + issue_state: String::from("In Progress"), + initial_issue_state: String::from("Todo"), + worktree: WorktreeSpec { + branch_name: String::from("x/pubfi-pub-101"), + issue_identifier: issue.identifier.clone(), + path: config.repo_root().to_path_buf(), + reused_existing: false, + }, + retry_project_slug: String::from("pubfi"), + dispatch_mode: IssueDispatchMode::Normal, + attempt_number: 1, + run_id: String::from("pub-101-attempt-1"), + retry_budget_base: 0, + }; + + RepoGatePhaseGoalController { + project: &config, + workflow: &workflow, + state_store: &state_store, + issue_run: &first_issue_run, + } + .phase_goal_completed(PhaseGoalKind::ImplementToValidationReady) + .expect("completed implementation phase should persist handoff phase"); + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + &first_issue_run.run_id, + first_issue_run.attempt_number, + PHASE_GOAL_RECOVERY_EVENT_TYPE, + serde_json::json!({ + "schema": "decodex.phase_goal_signal/1", + "phase": "implement_to_validation_ready", + "signal": "phase_goal_recovered", + "payload": { + "nextPhase": "handoff_evidence", + "sourceErrorClass": "app_server_run_failed", + }, + }), + ) + .expect("phase goal recovery should record"); + + let retry_issue_run = IssueRunPlan { + issue: issue.clone(), + issue_state: String::from("In Progress"), + initial_issue_state: String::from("Todo"), + worktree: first_issue_run.worktree.clone(), + retry_project_slug: String::from("pubfi"), + dispatch_mode: IssueDispatchMode::Retry, + attempt_number: 2, + run_id: String::from("pub-101-attempt-2"), + retry_budget_base: 1, + }; + let goal = RepoGatePhaseGoalController { + project: &config, + workflow: &workflow, + state_store: &state_store, + issue_run: &retry_issue_run, + } + .initial_phase_goal() + .expect("retry phase goal should build") + .expect("retry should still set a phase goal"); + + assert_eq!(goal.phase, PhaseGoalKind::HandoffEvidence); + assert!( + goal.objective.contains("prepare PR-backed handoff evidence"), + "retry should continue to handoff instead of repeating implementation" + ); +} + +#[test] +fn retry_phase_goal_resumes_cross_attempt_active_handoff_phase() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let issue = sample_issue("In Progress", &[tracker::automation_active_label(TEST_SERVICE_ID).as_str()]); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let worktree = WorktreeSpec { + branch_name: String::from("x/pubfi-pub-101"), + issue_identifier: issue.identifier.clone(), + path: config.repo_root().to_path_buf(), + reused_existing: false, + }; + + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + "pub-101-attempt-2", + 2, + "phase_goal_set", + serde_json::json!({ + "schema": "decodex.phase_goal_signal/1", + "phase": "handoff_evidence", + "payload": { + "phase": "handoff_evidence", + "status": "active", + }, + }), + ) + .expect("active handoff phase should record"); + + let retry_issue_run = IssueRunPlan { + issue: issue.clone(), + issue_state: String::from("In Progress"), + initial_issue_state: String::from("Todo"), + worktree, + retry_project_slug: String::from("pubfi"), + dispatch_mode: IssueDispatchMode::Retry, + attempt_number: 3, + run_id: String::from("pub-101-attempt-3"), + retry_budget_base: 2, + }; + let goal = RepoGatePhaseGoalController { + project: &config, + workflow: &workflow, + state_store: &state_store, + issue_run: &retry_issue_run, + } + .initial_phase_goal() + .expect("retry phase goal should build") + .expect("retry should still set a phase goal"); + + assert_eq!(goal.phase, PhaseGoalKind::HandoffEvidence); + assert!( + goal.objective.contains("prepare PR-backed handoff evidence"), + "retry should resume handoff evidence instead of repeating implementation" + ); +} + +#[test] +fn retry_phase_goal_does_not_resume_cross_attempt_phase_after_terminal_finalize() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let issue = sample_issue("In Progress", &[tracker::automation_active_label(TEST_SERVICE_ID).as_str()]); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let worktree = WorktreeSpec { + branch_name: String::from("x/pubfi-pub-101"), + issue_identifier: issue.identifier.clone(), + path: config.repo_root().to_path_buf(), + reused_existing: false, + }; + + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + "pub-101-attempt-2", + 2, + "phase_goal_set", + serde_json::json!({ + "schema": "decodex.phase_goal_signal/1", + "phase": "handoff_evidence", + "payload": { + "phase": "handoff_evidence", + "status": "active", + }, + }), + ) + .expect("active handoff phase should record"); + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + "pub-101-attempt-2", + 2, + "terminal_finalize", + serde_json::json!({ + "schema": "decodex.terminal_finalize/1", + "path": "review_handoff", + }), + ) + .expect("terminal finalize should record"); + + let retry_issue_run = IssueRunPlan { + issue: issue.clone(), + issue_state: String::from("In Progress"), + initial_issue_state: String::from("Todo"), + worktree, + retry_project_slug: String::from("pubfi"), + dispatch_mode: IssueDispatchMode::Retry, + attempt_number: 3, + run_id: String::from("pub-101-attempt-3"), + retry_budget_base: 2, + }; + let goal = RepoGatePhaseGoalController { + project: &config, + workflow: &workflow, + state_store: &state_store, + issue_run: &retry_issue_run, + } + .initial_phase_goal() + .expect("retry phase goal should build") + .expect("retry should still set a phase goal"); + + assert_eq!(goal.phase, PhaseGoalKind::ImplementToValidationReady); +} + +#[test] +fn retry_phase_goal_uses_only_latest_previous_attempt_for_cross_attempt_resume() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let issue = sample_issue("In Progress", &[tracker::automation_active_label(TEST_SERVICE_ID).as_str()]); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let worktree = WorktreeSpec { + branch_name: String::from("x/pubfi-pub-101"), + issue_identifier: issue.identifier.clone(), + path: config.repo_root().to_path_buf(), + reused_existing: false, + }; + + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + "pub-101-attempt-1", + 1, + "phase_goal_next", + serde_json::json!({ + "schema": "decodex.phase_goal_signal/1", + "phase": "handoff_evidence", + "reason": "validation_pass", + }), + ) + .expect("older handoff phase should record"); + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + "pub-101-attempt-2", + 2, + "phase_goal_set", + serde_json::json!({ + "schema": "decodex.phase_goal_signal/1", + "phase": "implement_to_validation_ready", + "payload": { + "phase": "implement_to_validation_ready", + "status": "active", + }, + }), + ) + .expect("newer implementation phase should record"); + + let retry_issue_run = IssueRunPlan { + issue: issue.clone(), + issue_state: String::from("In Progress"), + initial_issue_state: String::from("Todo"), + worktree, + retry_project_slug: String::from("pubfi"), + dispatch_mode: IssueDispatchMode::Retry, + attempt_number: 3, + run_id: String::from("pub-101-attempt-3"), + retry_budget_base: 2, + }; + let goal = RepoGatePhaseGoalController { + project: &config, + workflow: &workflow, + state_store: &state_store, + issue_run: &retry_issue_run, + } + .initial_phase_goal() + .expect("retry phase goal should build") + .expect("retry should still set a phase goal"); + + assert_eq!(goal.phase, PhaseGoalKind::ImplementToValidationReady); +} + +#[test] +fn retry_phase_goal_does_not_skip_empty_previous_attempt_for_cross_attempt_resume() { + let (_temp_dir, config, workflow) = temp_project_layout(); + let issue = sample_issue("In Progress", &[tracker::automation_active_label(TEST_SERVICE_ID).as_str()]); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let worktree = WorktreeSpec { + branch_name: String::from("x/pubfi-pub-101"), + issue_identifier: issue.identifier.clone(), + path: config.repo_root().to_path_buf(), + reused_existing: false, + }; + + state_store + .append_private_execution_event( + TEST_SERVICE_ID, + &issue.id, + "pub-101-attempt-1", + 1, + "phase_goal_next", + serde_json::json!({ + "schema": "decodex.phase_goal_signal/1", + "phase": "handoff_evidence", + "reason": "validation_pass", + }), + ) + .expect("older handoff phase should record"); + state_store + .record_run_attempt("pub-101-attempt-2", &issue.id, 2, "failed") + .expect("empty previous attempt should record"); + + let retry_issue_run = IssueRunPlan { + issue: issue.clone(), + issue_state: String::from("In Progress"), + initial_issue_state: String::from("Todo"), + worktree, + retry_project_slug: String::from("pubfi"), + dispatch_mode: IssueDispatchMode::Retry, + attempt_number: 3, + run_id: String::from("pub-101-attempt-3"), + retry_budget_base: 2, + }; + let goal = RepoGatePhaseGoalController { + project: &config, + workflow: &workflow, + state_store: &state_store, + issue_run: &retry_issue_run, + } + .initial_phase_goal() + .expect("retry phase goal should build") + .expect("retry should still set a phase goal"); + + assert_eq!(goal.phase, PhaseGoalKind::ImplementToValidationReady); +} + #[test] fn open_phase_goal_tracked_rewrites_stop_instead_of_repair_continuation() { let (_temp_dir, config, workflow) = temp_project_layout_with_workflow_markdown( diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index 8b131a426..812e495ca 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -400,6 +400,12 @@ Worktree visibility follows the owning dashboard section: when `process_alive` is false. `run_lease` is queue lease ownership only; `execution_liveness` explains why the lane is still visible when the queue lease is not held. + If a newer attempt for the same issue holds the run lease, older unleased attempts + with stale protocol/process evidence are shadowed out of `Running Lanes` and + current-attention counts so one issue does not appear as simultaneous current work. + Retry attempts also read the immediately previous attempt's unterminated phase-goal + state, so a lane already in `handoff_evidence` resumes handoff instead of repeating + implementation just because the attempt identity changed. - Lane steer and interrupt rejections such as `run_lease_missing` are private runtime evidence. They should preserve the queue lease state, branch, retained worktree path, current run id and attempt, active channel metadata, and observed diff --git a/docs/spec/lane-control-state.md b/docs/spec/lane-control-state.md index 628ee48f3..af90b3855 100644 --- a/docs/spec/lane-control-state.md +++ b/docs/spec/lane-control-state.md @@ -77,6 +77,13 @@ from protocol activity. - A lane counts as a running lane only when `ownership_state` is `leased_run`. - Liveness evidence may update `liveness_state`, but it must not create or restore `leased_run` ownership. +- When a newer attempt for the same issue has `ownership_state=leased_run`, older + unleased attempts with stale or protocol-only liveness evidence stay out of + current-lane projection and current-attention counts. +- A retry or automatic continuation for the same issue must resume the latest + unterminated phase-goal state from the immediately previous attempt. A validated + or active `handoff_evidence` phase must not be reset to + `implement_to_validation_ready` merely because the runtime created a new attempt. - `run_lease=false` is incompatible with `ownership_state=leased_run`. - Terminal attempt statuses such as `failed`, `interrupted`, `stalled`, or `succeeded` must not be promoted to `running` by live process, thread, or protocol evidence. diff --git a/docs/spec/loop-runtime.md b/docs/spec/loop-runtime.md index bd11fd034..8f7e9f68f 100644 --- a/docs/spec/loop-runtime.md +++ b/docs/spec/loop-runtime.md @@ -514,6 +514,11 @@ the only valid way to exit a satisfied phase and hand control back to Decodex's repo-gate transition. A progress checkpoint or final message that says the lane is validation-ready and waiting for the next phase is only evidence; it must not replace the Codex goal-complete signal. +When Decodex has already recorded a valid phase-goal continuation or active phase in +the immediately previous attempt and must create a retry or automatic continuation +attempt, the new attempt resumes that unterminated phase state instead of restarting +implementation. This preserves the state-machine boundary between validated work and +the later review/handoff contract. ## Validation And Review