From 367cd059a7873301466a13e40fb786c213d1a2ef Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Tue, 16 Jun 2026 15:27:56 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Implement lane orchestration kernel","authority":"XY-962"} --- apps/decodex/src/agent.rs | 7 +- apps/decodex/src/agent/tracker_tool_bridge.rs | 2 +- .../src/agent/tracker_tool_bridge/review.rs | 37 +- .../tests/review/policy.rs | 44 +- .../src/agent/tracker_tool_bridge/tools.rs | 46 +- .../src/orchestrator/agent_evidence.rs | 43 +- apps/decodex/src/orchestrator/lane_control.rs | 33 +- .../src/orchestrator/operator_dashboard.html | 33 ++ apps/decodex/src/orchestrator/status.rs | 451 +++++++++++++++--- .../tests/operator/status/dashboard.rs | 14 + .../tests/operator/status/http.rs | 30 +- .../tests/operator/status/running_lanes.rs | 148 +++--- .../tests/operator/status_support.rs | 20 +- apps/decodex/src/orchestrator/types.rs | 7 + docs/spec/index.md | 3 + docs/spec/lane-control-state.md | 149 ++++++ 16 files changed, 912 insertions(+), 155 deletions(-) create mode 100644 docs/spec/lane-control-state.md diff --git a/apps/decodex/src/agent.rs b/apps/decodex/src/agent.rs index 656c8280d..9a560c155 100644 --- a/apps/decodex/src/agent.rs +++ b/apps/decodex/src/agent.rs @@ -24,8 +24,9 @@ pub(crate) use self::{ ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME, ISSUE_LABEL_ADD_TOOL_NAME, ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_CHECKPOINT_TOOL_NAME, ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, - ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, ReviewExecutionMode, - ReviewHandoffContext, ReviewHandoffWritebackFailed, ReviewPolicyStopReason, - ReviewPolicyStopRequested, RunCompletionDisposition, TrackerToolBridge, + ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, + REVIEW_POLICY_CONVERGENCE_BUDGET, ReviewExecutionMode, ReviewHandoffContext, + ReviewHandoffWritebackFailed, ReviewPolicyStopReason, ReviewPolicyStopRequested, + RunCompletionDisposition, TrackerToolBridge, }, }; diff --git a/apps/decodex/src/agent/tracker_tool_bridge.rs b/apps/decodex/src/agent/tracker_tool_bridge.rs index 7c31e477b..a60011377 100644 --- a/apps/decodex/src/agent/tracker_tool_bridge.rs +++ b/apps/decodex/src/agent/tracker_tool_bridge.rs @@ -39,8 +39,8 @@ pub(crate) const ISSUE_REVIEW_HANDOFF_TOOL_NAME: &str = "issue_review_handoff"; pub(crate) const ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME: &str = "issue_review_repair_complete"; pub(crate) const ISSUE_DELIVERY_CLOSEOUT_COMPLETE_TOOL_NAME: &str = "issue_closeout_complete"; pub(crate) const ISSUE_TERMINAL_FINALIZE_TOOL_NAME: &str = "issue_terminal_finalize"; +pub(crate) const REVIEW_POLICY_CONVERGENCE_BUDGET: i64 = 3; -const REVIEW_POLICY_CONVERGENCE_BUDGET: i64 = 3; const REVIEW_HANDOFF_PUBLIC_SUMMARY_FALLBACK: &str = "Implementation completed and the PR is ready for review."; const REVIEW_REPAIR_PUBLIC_SUMMARY_FALLBACK: &str = diff --git a/apps/decodex/src/agent/tracker_tool_bridge/review.rs b/apps/decodex/src/agent/tracker_tool_bridge/review.rs index 3e47fc68a..ed994389f 100644 --- a/apps/decodex/src/agent/tracker_tool_bridge/review.rs +++ b/apps/decodex/src/agent/tracker_tool_bridge/review.rs @@ -653,9 +653,8 @@ impl<'a> TrackerToolBridge<'a> { }) .transpose()? .flatten(); - let external_round_count = persisted_orchestration.map_or(0, |marker| { - if marker.external_round_count() >= 4 { 0 } else { marker.external_round_count() } - }); + let external_round_count = + persisted_orchestration.map_or(0, |marker| marker.external_round_count()); tracker::create_prepared_linear_execution_event_comment( self.tracker, @@ -1086,12 +1085,36 @@ impl<'a> TrackerToolBridge<'a> { let Some(checkpoint) = self.review_policy_state_for_current_head(review_context)? else { return Ok(None); }; + + Ok(self.review_policy_stop_from_checkpoint(review_context, checkpoint)) + } + + pub(super) fn review_policy_stop_requested_for_current_phase( + &self, + review_context: &ReviewHandoffContext, + ) -> crate::prelude::Result> { + if !review_context.decodex_review_checkpoint_enabled() { + return Ok(None); + } + + let Some(checkpoint) = self.review_policy_state_for_current_phase(review_context)? else { + return Ok(None); + }; + + Ok(self.review_policy_stop_from_checkpoint(review_context, checkpoint)) + } + + fn review_policy_stop_from_checkpoint( + &self, + review_context: &ReviewHandoffContext, + checkpoint: ReviewPolicyState, + ) -> Option { let stop_reason = match checkpoint.status { - ReviewPolicyStatus::Clean => return Ok(None), + ReviewPolicyStatus::Clean => return None, ReviewPolicyStatus::Findings if checkpoint.nonclean_rounds < REVIEW_POLICY_CONVERGENCE_BUDGET => { - return Ok(None); + return None; }, ReviewPolicyStatus::Findings => ReviewPolicyStopReason::Exhausted, ReviewPolicyStatus::NeedsArchitectureReview => @@ -1099,13 +1122,13 @@ impl<'a> TrackerToolBridge<'a> { ReviewPolicyStatus::Blocked => ReviewPolicyStopReason::Blocked, }; - Ok(Some(ReviewPolicyStopRequested { + Some(ReviewPolicyStopRequested { head_sha: checkpoint.head_sha, issue_identifier: self.issue.identifier.clone(), nonclean_rounds: Some(checkpoint.nonclean_rounds), reason: stop_reason, run_id: review_context.run_id.clone(), - })) + }) } pub(super) fn required_pr_completion_tool_name(&self) -> &'static str { diff --git a/apps/decodex/src/agent/tracker_tool_bridge/tests/review/policy.rs b/apps/decodex/src/agent/tracker_tool_bridge/tests/review/policy.rs index 4301c8146..8d2a67a0d 100644 --- a/apps/decodex/src/agent/tracker_tool_bridge/tests/review/policy.rs +++ b/apps/decodex/src/agent/tracker_tool_bridge/tests/review/policy.rs @@ -828,7 +828,15 @@ fn review_checkpoint_findings_continue_until_budget_then_stop() { }), ); - assert!(response.success); + assert!(!response.success); + assert!( + matches!( + response.content_items.first(), + Some(DynamicToolContentItem::InputText { text }) + if text.contains("Review churn threshold exceeded") + ), + "third consecutive findings checkpoint should fail immediately: {response:?}" + ); let error = DynamicToolHandler::classify_turn_completion(&bridge, "stop") .expect_err("third consecutive findings checkpoint should stop the lane"); @@ -838,6 +846,36 @@ fn review_checkpoint_findings_continue_until_budget_then_stop() { assert_eq!(stop.reason, ReviewPolicyStopReason::Exhausted); assert_eq!(stop.nonclean_rounds, Some(3)); + + let fenced_response = DynamicToolHandler::handle_call( + &bridge, + ISSUE_PROGRESS_CHECKPOINT_TOOL_NAME, + serde_json::json!({ + "phase": "implementing", + "focus": "Continue repairing after review findings.", + "next_action": "Keep editing the same repair strategy.", + "blockers": [], + "evidence": ["The review checkpoint already exceeded the convergence budget."], + "verification": [], + "head_sha": sample_local_repo().head_oid, + "branch": "x/decodex-1" + }), + ); + + assert!(!fenced_response.success); + assert!( + matches!( + fenced_response.content_items.first(), + Some(DynamicToolContentItem::InputText { text }) + if text.contains("Review policy stop `review_policy_exhausted` is active") + && text.contains("issue_progress_checkpoint") + ), + "review policy stop should fence mutable progress writes: {fenced_response:?}" + ); + assert!( + tracker.comments.borrow().is_empty(), + "fenced progress checkpoint must not write a tracker comment" + ); } #[test] @@ -1564,7 +1602,7 @@ fn review_repair_apply_persists_updated_handoff_marker_without_tracker_transitio } #[test] -fn review_repair_apply_resets_external_round_budget_after_fourth_round() { +fn review_repair_apply_does_not_reset_external_round_budget_after_fourth_round() { let temp_dir = TempDir::new().expect("tempdir should create"); let tracker = FakeTracker::new(); let issue = sample_review_issue(); @@ -1616,7 +1654,7 @@ fn review_repair_apply_resets_external_round_budget_after_fourth_round() { persisted_review_orchestration_marker(&bridge, &issue, &review_context, &marker); assert_eq!(orchestration_marker.phase(), "request_pending"); - assert_eq!(orchestration_marker.external_round_count(), 0); + assert_eq!(orchestration_marker.external_round_count(), 4); } #[test] diff --git a/apps/decodex/src/agent/tracker_tool_bridge/tools.rs b/apps/decodex/src/agent/tracker_tool_bridge/tools.rs index 10d6224c2..b9e788135 100644 --- a/apps/decodex/src/agent/tracker_tool_bridge/tools.rs +++ b/apps/decodex/src/agent/tracker_tool_bridge/tools.rs @@ -9,11 +9,11 @@ use crate::{ ISSUE_REVIEW_HANDOFF_TOOL_NAME, ISSUE_REVIEW_REPAIR_COMPLETE_TOOL_NAME, ISSUE_TERMINAL_FINALIZE_TOOL_NAME, ISSUE_TRANSITION_TOOL_NAME, LabelArgs, NormalizedProgressCheckpoint, NormalizedReviewCheckpointPayload, PendingReviewAction, - PendingReviewCompletion, ProgressCheckpointArgs, PullRequestDetails, ReviewCheckpointArgs, - ReviewCheckpointChecksArgs, ReviewCheckpointFindingArgs, - ReviewCheckpointRejectedFindingArgs, ReviewExecutionMode, ReviewHandoffArgs, - ReviewHandoffContext, ReviewPolicyPhase, ReviewPolicyStatus, RunCompletionDisposition, - TerminalFinalizeArgs, TrackerToolBridge, TransitionArgs, + PendingReviewCompletion, ProgressCheckpointArgs, PullRequestDetails, + REVIEW_POLICY_CONVERGENCE_BUDGET, ReviewCheckpointArgs, ReviewCheckpointChecksArgs, + ReviewCheckpointFindingArgs, ReviewCheckpointRejectedFindingArgs, ReviewExecutionMode, + ReviewHandoffArgs, ReviewHandoffContext, ReviewPolicyPhase, ReviewPolicyStatus, + RunCompletionDisposition, TerminalFinalizeArgs, TrackerToolBridge, TransitionArgs, }, orchestrator::{ self, AUTHORITY_BOUNDARY_CHECK_EVENT_TYPE, AuthorityDecisionOption, @@ -521,6 +521,10 @@ impl<'a> TrackerToolBridge<'a> { tool_name: &str, arguments: Value, ) -> DynamicToolCallResponse { + if let Some(response) = self.review_policy_mutation_fence(tool_name) { + return response; + } + match tool_name { ISSUE_TRANSITION_TOOL_NAME => self.handle_transition(arguments), ISSUE_COMMENT_TOOL_NAME => self.handle_comment(arguments), @@ -536,6 +540,30 @@ impl<'a> TrackerToolBridge<'a> { } } + fn review_policy_mutation_fence(&self, tool_name: &str) -> Option { + if matches!( + tool_name, + ISSUE_REVIEW_CHECKPOINT_TOOL_NAME | ISSUE_TERMINAL_FINALIZE_TOOL_NAME + ) { + return None; + } + + let review_context = self.review_context.as_ref()?; + + match self.review_policy_stop_requested_for_current_phase(review_context) { + Ok(Some(stop)) => Some(DynamicToolCallResponse::failure(format!( + "Review policy stop `{}` is active for issue `{}` after `{}` non-clean rounds; `{tool_name}` is fenced until architecture recovery or human attention resolves the lane.", + stop.reason.error_class(), + stop.issue_identifier, + stop.nonclean_rounds.unwrap_or_default() + ))), + Ok(None) => None, + Err(error) => Some(DynamicToolCallResponse::failure(format!( + "Failed to evaluate review policy mutation fence for `{tool_name}`: {error}" + ))), + } + } + pub(super) fn handle_progress_checkpoint(&self, arguments: Value) -> DynamicToolCallResponse { let parsed = match serde_json::from_value::(arguments) { Ok(parsed) => parsed, @@ -1348,6 +1376,14 @@ impl<'a> TrackerToolBridge<'a> { }, ); + if review_policy_status == ReviewPolicyStatus::Findings + && nonclean_rounds >= REVIEW_POLICY_CONVERGENCE_BUDGET + { + return DynamicToolCallResponse::failure(format!( + "{message} Review churn threshold exceeded; stop the current repair strategy now and route through architecture recovery or human attention before making further repair mutations." + )); + } + DynamicToolCallResponse::success(message) } diff --git a/apps/decodex/src/orchestrator/agent_evidence.rs b/apps/decodex/src/orchestrator/agent_evidence.rs index c32e46d0a..47c3eb589 100644 --- a/apps/decodex/src/orchestrator/agent_evidence.rs +++ b/apps/decodex/src/orchestrator/agent_evidence.rs @@ -150,6 +150,12 @@ struct AgentRunCapsule { current_operation: String, queue_lease_state: String, execution_liveness: String, + ownership_state: String, + liveness_state: String, + policy_state: String, + terminalization_state: String, + lane_control_next_action: String, + lane_control_conditions: Vec, active_lease: bool, continuation_pending: bool, suspected_stall: bool, @@ -1567,6 +1573,12 @@ fn agent_run_capsule( current_operation: run.current_operation.clone(), queue_lease_state: run.queue_lease_state.clone(), execution_liveness: run.execution_liveness.clone(), + ownership_state: run.ownership_state.clone(), + liveness_state: run.liveness_state.clone(), + policy_state: run.policy_state.clone(), + terminalization_state: run.terminalization_state.clone(), + lane_control_next_action: run.lane_control_next_action.clone(), + lane_control_conditions: run.lane_control_conditions.clone(), active_lease: run.active_lease, continuation_pending: run.continuation_pending, suspected_stall: run.suspected_stall, @@ -1623,11 +1635,23 @@ fn agent_run_diagnosis(run: &OperatorRunStatus) -> AgentRunDiagnosis { AgentRunDiagnosis { attention_required: reason.is_some(), reason_code: reason.map(str::to_owned), - next_action: agent_run_next_action(run).map(str::to_owned), + next_action: agent_run_next_action(run), } } fn agent_run_blocker_reason(run: &OperatorRunStatus) -> Option<&'static str> { + if run.policy_state == "review_churn_exceeded" { + return Some("review_churn_exceeded"); + } + if run.ownership_state == "retained_attention" { + return Some("retained_attention"); + } + if run.ownership_state == "orphaned_live_thread" { + return Some("orphaned_live_thread"); + } + if run.ownership_state == "terminalizing" { + return Some("terminalizing"); + } if run.suspected_stall { return Some("suspected_stall"); } @@ -1650,15 +1674,19 @@ fn agent_run_blocker_reason(run: &OperatorRunStatus) -> Option<&'static str> { None } -fn agent_run_next_action(run: &OperatorRunStatus) -> Option<&'static str> { +fn agent_run_next_action(run: &OperatorRunStatus) -> Option { + if !run.lane_control_next_action.trim().is_empty() { + return Some(run.lane_control_next_action.clone()); + } + match agent_run_blocker_reason(run) { Some("suspected_stall" | "run_stalled" | "stale_execution_without_known_process") => - Some("Inspect the run capsule, retained worktree, protocol activity, and process state before retrying."), + Some(String::from("Inspect the run capsule, retained worktree, protocol activity, and process state before retrying.")), Some("process_exited_without_terminal_status") => - Some("Inspect the retained worktree and runtime markers; reconcile or retry only after preserving useful local changes."), + Some(String::from("Inspect the retained worktree and runtime markers; reconcile or retry only after preserving useful local changes.")), Some("run_waiting") => - Some("Inspect wait_reason, thread status, and protocol activity before deciding whether the agent can continue."), - Some("retry_backoff") => Some("Wait until next_retry_at or run an explicit operator retry after reviewing the retained state."), + Some(String::from("Inspect wait_reason, thread status, and protocol activity before deciding whether the agent can continue.")), + Some("retry_backoff") => Some(String::from("Wait until next_retry_at or run an explicit operator retry after reviewing the retained state.")), _ => None, } } @@ -1705,7 +1733,8 @@ fn push_run_blockers( .wait_reason .clone() .unwrap_or_else(|| reason_code.to_owned()), - next_action: agent_run_next_action(run).unwrap_or("Inspect the run capsule.").to_owned(), + next_action: agent_run_next_action(run) + .unwrap_or_else(|| String::from("Inspect the run capsule.")), blocker_snapshot_path: blocker_snapshot_path(blockers_dir, &issue_key) .display() .to_string(), diff --git a/apps/decodex/src/orchestrator/lane_control.rs b/apps/decodex/src/orchestrator/lane_control.rs index bc98360a7..743413126 100644 --- a/apps/decodex/src/orchestrator/lane_control.rs +++ b/apps/decodex/src/orchestrator/lane_control.rs @@ -107,6 +107,12 @@ struct LaneRunInspect { current_operation: String, active_lease: bool, execution_liveness: String, + ownership_state: String, + liveness_state: String, + policy_state: String, + terminalization_state: String, + lane_control_next_action: String, + lane_control_conditions: Vec, thread_id: Option, turn_id: Option, thread_status: Option, @@ -136,6 +142,12 @@ impl LaneRunInspect { current_operation: run.current_operation.clone(), active_lease: run.active_lease, execution_liveness: run.execution_liveness.clone(), + ownership_state: run.ownership_state.clone(), + liveness_state: run.liveness_state.clone(), + policy_state: run.policy_state.clone(), + terminalization_state: run.terminalization_state.clone(), + lane_control_next_action: run.lane_control_next_action.clone(), + lane_control_conditions: run.lane_control_conditions.clone(), thread_id: run.thread_id.clone(), turn_id: run.turn_id.clone(), thread_status: run.thread_status.clone(), @@ -554,6 +566,12 @@ fn lane_control_operator_context(run: &OperatorRunStatus) -> Value { "active_lease": run.active_lease, "queue_lease_state": run.queue_lease_state.as_str(), "execution_liveness": run.execution_liveness.as_str(), + "ownership_state": run.ownership_state.as_str(), + "liveness_state": run.liveness_state.as_str(), + "policy_state": run.policy_state.as_str(), + "terminalization_state": run.terminalization_state.as_str(), + "lane_control_next_action": run.lane_control_next_action.as_str(), + "lane_control_conditions": &run.lane_control_conditions, "thread_status": run.thread_status.as_deref(), "process_id": run.process_id, "process_alive": run.process_alive, @@ -1185,14 +1203,27 @@ fn render_lane_inspect_report(report: &LaneInspectReport) -> String { for run in &report.runs { output.push_str(&format!( - "- {} attempt {}: status={}, phase={}, activeLease={}, liveness={}\n", + "- {} attempt {}: status={}, phase={}, activeLease={}, owner={}, liveness={}\n", run.run_id, run.attempt_number, run.status, run.phase, run.active_lease, + run.ownership_state, run.execution_liveness )); + output.push_str(&format!( + " laneControl: livenessState={}, policyState={}, terminalization={}, nextAction={}, conditions={}\n", + run.liveness_state, + run.policy_state, + run.terminalization_state, + run.lane_control_next_action, + if run.lane_control_conditions.is_empty() { + String::from("none") + } else { + run.lane_control_conditions.join(",") + } + )); output.push_str(&format!( " appServer: thread={}, turn={}, softInterruptAvailable={}\n", run.thread_id.as_deref().unwrap_or("none"), diff --git a/apps/decodex/src/orchestrator/operator_dashboard.html b/apps/decodex/src/orchestrator/operator_dashboard.html index 7139fabb8..d72f7987e 100644 --- a/apps/decodex/src/orchestrator/operator_dashboard.html +++ b/apps/decodex/src/orchestrator/operator_dashboard.html @@ -5987,6 +5987,27 @@

Run History

return displayToken(run.execution_liveness || "liveness_unknown"); } + function runOwnershipSummary(run) { + return displayToken(run.ownership_state || (runCountsAsRunning(run) ? "owned_active" : "unknown")); + } + + function runLivenessStateSummary(run) { + return displayToken(run.liveness_state || "unknown"); + } + + function runPolicyStateSummary(run) { + return displayToken(run.policy_state || "allowed"); + } + + function runTerminalizationSummary(run) { + return displayToken(run.terminalization_state || "none"); + } + + function runLaneControlConditionsSummary(run) { + const conditions = run.lane_control_conditions ?? []; + return conditions.length ? conditions.map(displayToken).join(", ") : "none"; + } + function runQueueLeaseSummary(run) { const leaseState = run.queue_lease_state || (run.active_lease ? "held" : "not_held"); @@ -10602,6 +10623,12 @@

${escapeHtml(item.title)}

const issueKey = issueDisplayKey(run); const issueTitle = runIssueTitle(run, derived); const summary = activeRunSummary(run); + if (run.ownership_state && run.ownership_state !== "owned_active") { + statusBits.push(inlineStatusFact("Owner", displayToken(run.ownership_state))); + } + if (run.policy_state && !["allowed", "review_findings", "review_pending"].includes(run.policy_state)) { + statusBits.push(inlineStatusFact("Policy", displayToken(run.policy_state))); + } if (run.wait_reason && !runWaitReasonShowsExecutionProgress(run)) { const waitReason = displayToken(run.wait_reason); @@ -10671,6 +10698,12 @@

${escapeHtml(issueTitle)}

${field("Worktree", run.worktree_path || "none")} ${field("Queue lease", runQueueLeaseSummary(run))} ${field("Execution liveness", runExecutionLivenessSummary(run))} + ${field("Ownership", runOwnershipSummary(run))} + ${field("Liveness state", runLivenessStateSummary(run))} + ${field("Policy state", runPolicyStateSummary(run))} + ${field("Terminalization", runTerminalizationSummary(run))} + ${field("Lane next action", capturedValue(run.lane_control_next_action))} + ${field("Lane conditions", runLaneControlConditionsSummary(run))} ${field("Loop", run.loop_status?.summary || "none")} ${field("Review loop", loopStatusFacts(run.loop_status).map(([label, value]) => `${label}: ${value}`).join("; ") || "none")} ${field("Model", runModelSummary(run))} diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 769c68ae5..e310a742a 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -19,6 +19,7 @@ use state::WORKTREE_PROVENANCE_GIT_HYGIENE_SCAN; use state::WORKTREE_PROVENANCE_LEGACY_UNKNOWN; use state::ProjectLoopEvidenceSnapshot; +use crate::agent::REVIEW_POLICY_CONVERGENCE_BUDGET; use crate::pull_request::{self, PullRequestLandingGateView}; use crate::worktree; use crate::worktree::MergedWorktreeCleanupDebt; @@ -280,6 +281,15 @@ struct OperatorLifecycleMetricPhase { rank: u8, } +struct OperatorLaneControlProjection { + ownership_state: String, + liveness_state: String, + policy_state: String, + terminalization_state: String, + next_action: String, + conditions: Vec, +} + pub(crate) fn ensure_project_has_no_merged_worktree_cleanup_debt( project: &ServiceConfig, ) -> crate::prelude::Result<()> { @@ -1202,11 +1212,37 @@ fn worktree_ownership( completed_state: Option<&str>, ) -> WorktreeOwnership { if let Some(run) = worktree_active_run_owner(worktree, snapshot) { - return WorktreeOwnership { - kind: "active_lane", - reason: format!("Active lane `{}` owns this worktree.", run.run_id), - next_action: None, - audit_required: false, + return match run.ownership_state.as_str() { + "owned_active" => WorktreeOwnership { + kind: "active_lane", + reason: format!("Active lane `{}` owns this worktree.", run.run_id), + next_action: None, + audit_required: false, + }, + "retained_attention" => WorktreeOwnership { + kind: "retained_attention", + reason: format!("Lane `{}` requires operator attention before it can own this worktree.", run.run_id), + next_action: Some(run.lane_control_next_action.clone()), + audit_required: true, + }, + "orphaned_live_thread" => WorktreeOwnership { + kind: "orphaned_live_thread", + reason: format!("Lane `{}` has live evidence but no active Decodex lease.", run.run_id), + next_action: Some(run.lane_control_next_action.clone()), + audit_required: true, + }, + "terminalizing" => WorktreeOwnership { + kind: "terminalizing_lane", + reason: format!("Lane `{}` is inside terminalization and no longer counts as running.", run.run_id), + next_action: Some(run.lane_control_next_action.clone()), + audit_required: true, + }, + _ => WorktreeOwnership { + kind: "orphaned_local_worktree", + reason: format!("Lane `{}` is not an active owner for this worktree.", run.run_id), + next_action: Some(run.lane_control_next_action.clone()), + audit_required: true, + }, }; } if let Some(lane) = worktree_post_review_owner(worktree, snapshot) { @@ -1278,11 +1314,18 @@ fn worktree_active_run_owner<'a>( worktree: &OperatorWorktreeStatus, snapshot: &'a OperatorStatusSnapshot, ) -> Option<&'a OperatorRunStatus> { - snapshot.active_runs.iter().find(|run| { - run.worktree_path.as_deref() == Some(worktree.worktree_path.as_str()) - || run.branch_name.as_deref() == Some(worktree.branch_name.as_str()) - || run.issue_id == worktree.issue_id - }) + snapshot + .active_runs + .iter() + .chain(snapshot.recent_runs.iter()) + .find(|run| { + matches!( + run.ownership_state.as_str(), + "owned_active" | "retained_attention" | "orphaned_live_thread" | "terminalizing" + ) && (run.worktree_path.as_deref() == Some(worktree.worktree_path.as_str()) + || run.branch_name.as_deref() == Some(worktree.branch_name.as_str()) + || run.issue_id == worktree.issue_id) + }) } fn worktree_post_review_owner<'a>( @@ -1453,7 +1496,7 @@ fn project_attention_count( for run in snapshot .active_runs .iter() - .filter(|run| operator_run_needs_attention(run)) + .filter(|run| operator_run_counts_as_attention(run)) { attention_keys.insert(operator_run_group_key(run)); } @@ -1674,7 +1717,7 @@ fn hydrate_post_review_lane_active_run_shadowing(snapshot: &mut OperatorStatusSn let active_issue_keys = snapshot .active_runs .iter() - .filter(|run| run.counts_as_running || run.has_fresh_execution) + .filter(|run| run.counts_as_running) .map(|run| operator_run_group_key(run).to_ascii_uppercase()) .collect::>(); @@ -1707,12 +1750,22 @@ fn operator_run_has_live_execution(run: &OperatorRunStatus) -> bool { } fn operator_run_counts_as_running(run: &OperatorRunStatus) -> bool { - matches!(run.status.as_str(), "starting" | "running") + run.ownership_state == "owned_active" + && matches!(run.status.as_str(), "starting" | "running") && run.phase == "executing" && (run.process_alive != Some(false) || run.has_fresh_execution) && !operator_run_needs_attention(run) } +fn operator_run_counts_as_attention(run: &OperatorRunStatus) -> bool { + run.needs_attention + || run.ownership_state == "retained_attention" + || matches!( + run.policy_state.as_str(), + "review_churn_exceeded" | "authority_boundary_required" | "human_attention_required" + ) +} + fn operator_run_needs_attention(run: &OperatorRunStatus) -> bool { matches!(run.status.as_str(), "needs_attention" | "terminal_failure") || run.phase == "needs_attention" @@ -5528,7 +5581,50 @@ fn operator_run_status( &lifecycle.current_operation, )?; - Ok(hydrate_operator_run_derived_status(OperatorRunStatus { + Ok(hydrate_operator_run_derived_status(operator_run_status_from_parts( + project, + project_display_name, + &run, + lifecycle, + wait_reason, + app_server_state, + timing, + protocol_summary, + child_agent_activity, + protocol_activity, + progress_diagnostic, + account, + accounts, + branch_name, + worktree_path, + issue_identifier, + private_evidence, + loop_status, + ))) +} + +#[allow(clippy::too_many_arguments)] +fn operator_run_status_from_parts( + project: &ServiceConfig, + project_display_name: &str, + run: &ProjectRunStatus, + lifecycle: OperatorRunLifecycleProjection, + wait_reason: Option, + app_server_state: OperatorRunAppServerState, + timing: OperatorRunTiming, + protocol_summary: OperatorRunProtocolSummary, + child_agent_activity: Option, + protocol_activity: Option, + progress_diagnostic: Option, + account: Option, + accounts: Vec, + branch_name: Option, + worktree_path: Option, + issue_identifier: Option, + private_evidence: AgentPrivateEvidenceRef, + loop_status: OperatorLoopStatus, +) -> OperatorRunStatus { + OperatorRunStatus { project_id: project.service_id().to_owned(), project_display_name: project_display_name.to_owned(), run_id: run.run_id().to_owned(), @@ -5540,10 +5636,16 @@ fn operator_run_status( status: lifecycle.status, attempt_status: run.status().to_owned(), status_projection_reason: lifecycle.status_projection_reason, + ownership_state: String::new(), + liveness_state: String::new(), + policy_state: String::new(), + terminalization_state: String::new(), + lane_control_next_action: String::new(), + lane_control_conditions: Vec::new(), phase: lifecycle.phase, wait_reason, current_operation: lifecycle.current_operation, - control_capability: operator_run_control_capability(&run, &app_server_state), + control_capability: operator_run_control_capability(run, &app_server_state), thread_id: app_server_state.thread_id, turn_id: app_server_state.turn_id, thread_status: app_server_state.thread_status, @@ -5589,17 +5691,264 @@ fn operator_run_status( accounts, branch_name, worktree_path, - })) + } } fn hydrate_operator_run_derived_status(mut status: OperatorRunStatus) -> OperatorRunStatus { status.has_fresh_execution = operator_run_has_fresh_execution(&status); status.needs_attention = operator_run_needs_attention(&status); + + let lane_control_state = operator_lane_control_state(&status); + + status.ownership_state = lane_control_state.ownership_state; + status.liveness_state = lane_control_state.liveness_state; + status.policy_state = lane_control_state.policy_state; + status.terminalization_state = lane_control_state.terminalization_state; + status.lane_control_next_action = lane_control_state.next_action; + status.lane_control_conditions = lane_control_state.conditions; + status.needs_attention = operator_run_counts_as_attention(&status); status.counts_as_running = operator_run_counts_as_running(&status); status } +fn operator_lane_control_state(run: &OperatorRunStatus) -> OperatorLaneControlProjection { + let liveness_state = operator_run_liveness_state(run); + let policy_state = operator_run_policy_state(run); + let terminalization_state = operator_run_terminalization_state(run, &liveness_state); + let ownership_state = + operator_run_ownership_state(run, &liveness_state, &policy_state, &terminalization_state); + let next_action = operator_run_lane_control_next_action( + run, + &ownership_state, + &liveness_state, + &policy_state, + &terminalization_state, + ); + let mut conditions = operator_run_lane_control_conditions(run, &liveness_state, &policy_state); + + if ownership_state == "owned_active" && !run.active_lease { + conditions.push(String::from("invalid_owned_active_without_lease")); + } + + OperatorLaneControlProjection { + ownership_state, + liveness_state, + policy_state, + terminalization_state, + next_action, + conditions, + } +} + +fn operator_run_ownership_state( + run: &OperatorRunStatus, + liveness_state: &str, + policy_state: &str, + terminalization_state: &str, +) -> String { + if run.active_lease + && matches!(run.attempt_status.as_str(), "starting" | "running" | "continuation_pending") + && !matches!( + policy_state, + "review_churn_exceeded" | "authority_boundary_required" | "human_attention_required" + ) + { + return String::from("owned_active"); + } + if matches!( + policy_state, + "review_churn_exceeded" | "authority_boundary_required" | "human_attention_required" + ) || run.needs_attention + || (!run.active_lease && liveness_state == "host_boot_mismatch") + { + return String::from("retained_attention"); + } + if !run.active_lease + && matches!(liveness_state, "process_alive" | "thread_active" | "protocol_recent") + { + return String::from("orphaned_live_thread"); + } + if terminalization_state != "none" && terminalization_state != "cleanup_complete" { + return String::from("terminalizing"); + } + if matches!(run.attempt_status.as_str(), "starting" | "running" | "continuation_pending") { + return String::from("pending"); + } + + String::from("closed") +} + +fn operator_run_liveness_state(run: &OperatorRunStatus) -> String { + if matches!(run.process_liveness_reason.as_deref(), Some("host_boot_id_mismatch")) { + return String::from("host_boot_mismatch"); + } + if run.process_alive == Some(true) { + return String::from("process_alive"); + } + if matches!(run.thread_status.as_deref(), Some("active")) || !run.thread_active_flags.is_empty() { + return String::from("thread_active"); + } + if operator_run_has_recent_app_server_execution(run) { + return String::from("protocol_recent"); + } + if run.process_alive == Some(false) + || matches!( + run.execution_liveness.as_str(), + "not_running" | "process_identity_mismatch" + ) + { + return String::from("not_running"); + } + + String::from("unknown") +} + +fn operator_run_policy_state(run: &OperatorRunStatus) -> String { + let Some(loop_status) = run.loop_status.as_ref() else { + return String::from("allowed"); + }; + + if loop_status.decision_request.is_some() { + return String::from("authority_boundary_required"); + } + if loop_status.autonomy == "human_required" { + return String::from("human_attention_required"); + } + + if let Some(recovery) = loop_status.architecture_recovery.as_ref() { + return if recovery.status == "active" { + String::from("architecture_recovery_pending") + } else { + String::from("human_attention_required") + }; + } + if let Some(review) = loop_status.review.as_ref() { + return match review.status.as_str() { + "pending" => String::from("review_pending"), + "findings" => { + if review + .checkpoint + .as_ref() + .is_some_and(|checkpoint| checkpoint.nonclean_rounds >= REVIEW_POLICY_CONVERGENCE_BUDGET) + { + String::from("review_churn_exceeded") + } else { + String::from("review_findings") + } + }, + "blocked" | "needs_architecture_review" => String::from("human_attention_required"), + _ => String::from("allowed"), + }; + } + + String::from("allowed") +} + +fn operator_run_terminalization_state(run: &OperatorRunStatus, liveness_state: &str) -> String { + if matches!( + run.status.as_str(), + "cleanup_complete" | "merged_closeout_reconciled" + ) || matches!(run.current_operation.as_str(), "ledger_outcome") + && matches!(run.phase.as_str(), "completed") + { + return String::from("cleanup_complete"); + } + if matches!(run.phase.as_str(), "completed" | "failed" | "terminated") + && !run.active_lease + && matches!(liveness_state, "not_running" | "unknown") + { + return String::from("cleanup_complete"); + } + if matches!(run.phase.as_str(), "completed" | "failed" | "terminated") { + return String::from("barrier_started"); + } + + String::from("none") +} + +fn operator_run_lane_control_conditions( + run: &OperatorRunStatus, + liveness_state: &str, + policy_state: &str, +) -> Vec { + let mut conditions = Vec::new(); + + if !run.active_lease + && matches!(run.attempt_status.as_str(), "starting" | "running" | "continuation_pending") + { + conditions.push(String::from("active_lease_missing")); + } + if matches!( + run.attempt_status.as_str(), + "failed" | "interrupted" | "stalled" | "succeeded" + ) && matches!(liveness_state, "process_alive" | "thread_active" | "protocol_recent") + { + conditions.push(String::from("terminal_attempt_has_live_evidence")); + } + if liveness_state == "host_boot_mismatch" { + conditions.push(String::from("host_boot_id_mismatch")); + } + if policy_state == "review_churn_exceeded" { + conditions.push(String::from("review_churn_threshold_exceeded")); + } + if matches!( + policy_state, + "authority_boundary_required" | "human_attention_required" + ) { + conditions.push(String::from("policy_requires_human_attention")); + } + + conditions +} + +fn operator_run_lane_control_next_action( + run: &OperatorRunStatus, + ownership_state: &str, + liveness_state: &str, + policy_state: &str, + terminalization_state: &str, +) -> String { + if policy_state == "review_churn_exceeded" { + return String::from("start_architecture_recovery_or_stop_for_human_attention"); + } + if matches!( + policy_state, + "authority_boundary_required" | "human_attention_required" + ) { + return String::from("resolve_policy_stop_before_mutating_lane"); + } + if ownership_state == "orphaned_live_thread" { + return String::from("inspect_or_interrupt_orphaned_live_thread"); + } + if liveness_state == "host_boot_mismatch" { + return String::from("inspect_recovery_evidence"); + } + if terminalization_state != "none" && terminalization_state != "cleanup_complete" { + return String::from("finish_terminalization"); + } + if ownership_state == "owned_active" { + if let Some(next_action) = + run.loop_status.as_ref().and_then(|loop_status| loop_status.next_action.clone()) + { + return next_action; + } + + return String::from("continue_owned_attempt"); + } + if ownership_state == "closed" { + return String::from("no_action"); + } + + if let Some(next_action) = + run.loop_status.as_ref().and_then(|loop_status| loop_status.next_action.clone()) + { + return next_action; + } + + String::from("inspect_lane_state") +} + fn operator_run_lifecycle_projection( run: &ProjectRunStatus, marker: Option<&RunActivityMarker>, @@ -6399,7 +6748,7 @@ fn operator_run_visible_status( app_server_state: &OperatorRunAppServerState, protocol_summary: &OperatorRunProtocolSummary, timing: &OperatorRunTiming, - marker_current_operation: Option<&str>, + _marker_current_operation: Option<&str>, ) -> String { if attempt_status == "starting" && operator_run_has_app_server_execution_evidence( @@ -6410,18 +6759,6 @@ fn operator_run_visible_status( { return String::from("running"); } - if attempt_status == "succeeded" - && operator_marker_operation_allows_terminal_status_promotion(marker_current_operation) - && operator_run_has_live_process_or_thread_evidence(app_server_state, timing) - { - return String::from("running"); - } - if matches!(attempt_status, "failed" | "interrupted" | "stalled") - && operator_marker_operation_allows_terminal_status_promotion(marker_current_operation) - && operator_run_has_live_execution_evidence(app_server_state, protocol_summary, timing) - { - return String::from("running"); - } attempt_status.to_owned() } @@ -6432,7 +6769,7 @@ fn operator_run_status_projection_reason( app_server_state: &OperatorRunAppServerState, protocol_summary: &OperatorRunProtocolSummary, timing: &OperatorRunTiming, - marker_current_operation: Option<&str>, + _marker_current_operation: Option<&str>, ) -> Option { if attempt_status == visible_status || visible_status != "running" { return None; @@ -6440,10 +6777,6 @@ fn operator_run_status_projection_reason( let projection_kind = if attempt_status == "starting" { "starting_attempt" - } else if matches!(attempt_status, "failed" | "interrupted" | "stalled" | "succeeded") - && operator_marker_operation_allows_terminal_status_promotion(marker_current_operation) - { - "terminal_attempt" } else { return None; }; @@ -6483,30 +6816,6 @@ fn operator_run_live_evidence_source( None } -fn operator_run_has_live_process_or_thread_evidence( - app_server_state: &OperatorRunAppServerState, - timing: &OperatorRunTiming, -) -> bool { - timing.process_alive == Some(true) - || matches!(app_server_state.thread_status.as_deref(), Some("active")) - || !app_server_state.thread_active_flags.is_empty() -} - -fn operator_marker_operation_allows_terminal_status_promotion( - marker_current_operation: Option<&str>, -) -> bool { - matches!(marker_current_operation, None | Some(RUN_OPERATION_AGENT_RUN)) -} - -fn operator_run_has_live_execution_evidence( - app_server_state: &OperatorRunAppServerState, - protocol_summary: &OperatorRunProtocolSummary, - timing: &OperatorRunTiming, -) -> bool { - operator_run_has_live_process_or_thread_evidence(app_server_state, timing) - || operator_run_has_recent_protocol_execution_evidence(protocol_summary, timing) -} - fn operator_run_has_recent_protocol_execution_evidence( protocol_summary: &OperatorRunProtocolSummary, timing: &OperatorRunTiming, @@ -7744,7 +8053,7 @@ fn active_run_id_for_queue_candidate<'a>( snapshot .active_runs .iter() - .find(|run| run.issue_id == queued_issue.issue_id) + .find(|run| run.issue_id == queued_issue.issue_id && run.counts_as_running) .map(|run| run.run_id.as_str()) } @@ -7828,9 +8137,10 @@ fn rendered_worktree_role<'a>( return worktree.ownership.as_str(); } if snapshot.active_runs.iter().any(|run| { - run.worktree_path.as_deref() == Some(worktree.worktree_path.as_str()) - || run.branch_name.as_deref() == Some(worktree.branch_name.as_str()) - || run.issue_id == worktree.issue_id + run.ownership_state == "owned_active" + && (run.worktree_path.as_deref() == Some(worktree.worktree_path.as_str()) + || run.branch_name.as_deref() == Some(worktree.branch_name.as_str()) + || run.issue_id == worktree.issue_id) }) { return "active_lane"; } @@ -8321,9 +8631,10 @@ fn append_rendered_run(output: &mut String, run: &OperatorRunStatus) { render_loop_architecture_recovery_summary(run.loop_status.as_ref()); let loop_boundary = render_loop_boundary_summary(run.loop_status.as_ref()); let control_capability = render_control_capability_summary(run.control_capability.as_ref()); + let lane_control_conditions = render_lane_control_conditions(run); output.push_str(&format!( - "- run_id: {}\n project_id: {}\n issue_id: {}\n issue_identifier: {}\n title: {}\n attempt: {}\n status: {}\n attempt_status: {}\n status_projection_reason: {}\n phase: {}\n wait_reason: {}\n current_operation: {}\n active_lease: {}\n queue_lease_state: {}\n queue_lease: {}\n execution_liveness: {}\n has_fresh_execution: {}\n counts_as_running: {}\n needs_attention: {}\n freshness_at: {}\n freshness_source: {}\n timing: run_idle={} protocol_idle={} last_progress={} protocol_event={} events={}\n account: {}\n accounts: {}\n child_agent_activity: {}\n protocol_activity: {}\n context_pressure: {}\n private_evidence: {}\n loop_status: {}\n loop_review: {}\n loop_architecture_recovery: {}\n loop_boundary: {}\n control_capability: {}\n thread_id: {}\n turn_id: {}\n thread_status: {}\n thread_active_flags: {}\n interactive_requested: {}\n continuation_pending: {}\n branch: {}\n worktree_path: {}\n updated_at: {}\n last_run_activity_at: {}\n last_protocol_activity_at: {}\n last_progress_at: {}\n idle_for_seconds: {}\n protocol_idle_for_seconds: {}\n suspected_stall: {}\n progress_diagnostic: {}\n process_id: {}\n process_alive: {}\n process_liveness_reason: {}\n retry_kind: {}\n next_retry_at: {}\n effective_model: {}\n effective_model_provider: {}\n effective_cwd: {}\n effective_approval_policy: {}\n effective_approvals_reviewer: {}\n effective_sandbox_mode: {}\n protocol_event: {}\n event_count: {}\n", + "- run_id: {}\n project_id: {}\n issue_id: {}\n issue_identifier: {}\n title: {}\n attempt: {}\n status: {}\n attempt_status: {}\n status_projection_reason: {}\n ownership_state: {}\n liveness_state: {}\n policy_state: {}\n terminalization_state: {}\n lane_control_next_action: {}\n lane_control_conditions: {}\n phase: {}\n wait_reason: {}\n current_operation: {}\n active_lease: {}\n queue_lease_state: {}\n queue_lease: {}\n execution_liveness: {}\n has_fresh_execution: {}\n counts_as_running: {}\n needs_attention: {}\n freshness_at: {}\n freshness_source: {}\n timing: run_idle={} protocol_idle={} last_progress={} protocol_event={} events={}\n account: {}\n accounts: {}\n child_agent_activity: {}\n protocol_activity: {}\n context_pressure: {}\n private_evidence: {}\n loop_status: {}\n loop_review: {}\n loop_architecture_recovery: {}\n loop_boundary: {}\n control_capability: {}\n thread_id: {}\n turn_id: {}\n thread_status: {}\n thread_active_flags: {}\n interactive_requested: {}\n continuation_pending: {}\n branch: {}\n worktree_path: {}\n updated_at: {}\n last_run_activity_at: {}\n last_protocol_activity_at: {}\n last_progress_at: {}\n idle_for_seconds: {}\n protocol_idle_for_seconds: {}\n suspected_stall: {}\n progress_diagnostic: {}\n process_id: {}\n process_alive: {}\n process_liveness_reason: {}\n retry_kind: {}\n next_retry_at: {}\n effective_model: {}\n effective_model_provider: {}\n effective_cwd: {}\n effective_approval_policy: {}\n effective_approvals_reviewer: {}\n effective_sandbox_mode: {}\n protocol_event: {}\n event_count: {}\n", run.run_id, run.project_id, run.issue_id, @@ -8333,6 +8644,12 @@ fn append_rendered_run(output: &mut String, run: &OperatorRunStatus) { run.status, run.attempt_status, run.status_projection_reason.as_deref().unwrap_or("none"), + run.ownership_state, + run.liveness_state, + run.policy_state, + run.terminalization_state, + run.lane_control_next_action, + lane_control_conditions, run.phase, run.wait_reason.as_deref().unwrap_or("none"), run.current_operation, @@ -8396,6 +8713,14 @@ fn append_rendered_run(output: &mut String, run: &OperatorRunStatus) { )); } +fn render_lane_control_conditions(run: &OperatorRunStatus) -> String { + if run.lane_control_conditions.is_empty() { + String::from("none") + } else { + run.lane_control_conditions.join(",") + } +} + fn operator_run_queue_lease_summary(run: &OperatorRunStatus) -> String { if run.active_lease { return String::from("held"); diff --git a/apps/decodex/src/orchestrator/tests/operator/status/dashboard.rs b/apps/decodex/src/orchestrator/tests/operator/status/dashboard.rs index 28788599d..c5e7af63f 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/dashboard.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/dashboard.rs @@ -507,11 +507,25 @@ fn assert_liveness_and_cleanup_contract(response: &str) { "typeof run?.counts_as_running === \"boolean\"", "runStaleWithoutKnownProcessNeedsAttention", "runExecutionLivenessSummary", + "runOwnershipSummary", + "runLivenessStateSummary", + "runPolicyStateSummary", + "runTerminalizationSummary", + "runLaneControlConditionsSummary", "runQueueLeaseSummary", "return displayToken(run.execution_liveness || \"liveness_unknown\");", + "return displayToken(run.ownership_state || (runCountsAsRunning(run) ? \"owned_active\" : \"unknown\"));", "field(\"Attempt status\", run.attempt_status || run.status)", "field(\"Queue lease\", runQueueLeaseSummary(run))", "field(\"Execution liveness\", runExecutionLivenessSummary(run))", + "field(\"Ownership\", runOwnershipSummary(run))", + "field(\"Liveness state\", runLivenessStateSummary(run))", + "field(\"Policy state\", runPolicyStateSummary(run))", + "field(\"Terminalization\", runTerminalizationSummary(run))", + "field(\"Lane next action\", capturedValue(run.lane_control_next_action))", + "field(\"Lane conditions\", runLaneControlConditionsSummary(run))", + "inlineStatusFact(\"Owner\", displayToken(run.ownership_state))", + "inlineStatusFact(\"Policy\", displayToken(run.policy_state))", "live_no_queue_lease", "return `${leaseState}; ${displayToken(run.execution_liveness || \"liveness_unknown\")}`;", "attention.worktree_path", diff --git a/apps/decodex/src/orchestrator/tests/operator/status/http.rs b/apps/decodex/src/orchestrator/tests/operator/status/http.rs index 323681788..2e78fb427 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/http.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/http.rs @@ -1786,6 +1786,14 @@ fn operator_lane_inspect_api_returns_lane_identity() { assert_eq!(data["runs"][0]["runId"], "pub-101-attempt-1"); assert_eq!(data["runs"][0]["attemptStatus"], "running"); assert_eq!(data["runs"][0]["activeLease"], true); + assert_eq!(data["runs"][0]["ownershipState"], "owned_active"); + assert_eq!(data["runs"][0]["livenessState"], "unknown"); + assert_eq!(data["runs"][0]["policyState"], "review_pending"); + assert_eq!(data["runs"][0]["terminalizationState"], "none"); + assert_eq!( + data["runs"][0]["laneControlNextAction"], + "Record the independent Decodex Review checkpoint for the current lane head." + ); assert_eq!(data["runs"][0]["threadId"], "thread-1"); assert_eq!(data["runs"][0]["turnId"], "turn-1"); assert_eq!(data["runs"][0]["softInterruptAvailable"], false); @@ -2177,6 +2185,22 @@ fn assert_active_lease_missing_control_audit( missing_lease_interrupt_event.payload()["context"]["execution_liveness"].as_str(), Some("process_alive") ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["ownership_state"].as_str(), + Some("orphaned_live_thread") + ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["liveness_state"].as_str(), + Some("process_alive") + ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["lane_control_next_action"].as_str(), + Some("inspect_or_interrupt_orphaned_live_thread") + ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["lane_control_conditions"][0].as_str(), + Some("active_lease_missing") + ); assert_eq!( missing_lease_interrupt_event.payload()["context"]["control_capability"]["channel_path"] .as_str(), @@ -2259,7 +2283,7 @@ fn operator_lane_interrupt_api_force_hard_fallbacks_after_active_lease_missing() #[cfg(unix)] #[test] -fn operator_lane_interrupt_api_force_hard_fallbacks_after_succeeded_status_with_live_process() { +fn operator_lane_interrupt_api_force_hard_fallbacks_terminal_live_process_without_soft_owner() { let (_temp_dir, config, _workflow) = temp_project_layout(); let state_store = StateStore::open_in_memory().expect("state store should open"); let registration = ProjectRegistration::from_config( @@ -2326,8 +2350,8 @@ fn operator_lane_interrupt_api_force_hard_fallbacks_after_succeeded_status_with_ assert!(response.starts_with("HTTP/1.1 200 OK\r\n"), "{response}"); assert_eq!(data["classification"], "hard_interrupt_fallback"); - assert_eq!(data["softInterrupt"]["status"], "rejected"); - assert_eq!(data["softInterrupt"]["errorClass"], "active_lease_missing"); + assert_eq!(data["softInterrupt"]["status"], "unavailable"); + assert_eq!(data["softInterrupt"]["errorClass"], "lane_not_active"); assert_eq!(data["hardInterrupt"]["classification"], "hard_interrupt_fallback"); assert_eq!(data["hardInterrupt"]["status"], "sent"); assert_eq!( 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 ca834e554..157d5f73d 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs @@ -1332,7 +1332,7 @@ fn operator_status_snapshot_counts_previous_boot_process_as_attention_not_runnin #[cfg(any(target_os = "linux", target_os = "macos"))] #[test] -fn operator_status_snapshot_keeps_unleased_app_server_active_run_with_stale_process_marker() { +fn operator_status_snapshot_projects_unleased_app_server_active_run_as_retained_attention() { 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", &[]); @@ -1378,18 +1378,25 @@ fn operator_status_snapshot_keeps_unleased_app_server_active_run_with_stale_proc assert_eq!(run.process_alive, Some(false)); assert_eq!(run.process_liveness_reason.as_deref(), Some("host_boot_id_mismatch")); assert_eq!(run.thread_status.as_deref(), Some("active")); + assert_eq!(run.ownership_state, "retained_attention"); + assert_eq!(run.liveness_state, "host_boot_mismatch"); + assert_eq!(run.lane_control_next_action, "inspect_recovery_evidence"); + assert!(run + .lane_control_conditions + .iter() + .any(|condition| condition == "host_boot_id_mismatch")); assert!(run.has_fresh_execution); - assert!(run.counts_as_running); - assert!(!run.needs_attention); + assert!(!run.counts_as_running); + assert!(run.needs_attention); assert_eq!(project.active_run_count, 1); - assert_eq!(project.running_lane_count, 1); - assert_eq!(project.attention_count, 0); - assert_eq!(snapshot.worktrees[0].ownership, "active_lane"); + assert_eq!(project.running_lane_count, 0); + assert_eq!(project.attention_count, 1); + assert_eq!(snapshot.worktrees[0].ownership, "retained_attention"); } #[cfg(any(target_os = "linux", target_os = "macos"))] #[test] -fn operator_status_snapshot_shadows_post_review_lane_when_active_run_is_fresh() { +fn operator_status_snapshot_does_not_shadow_post_review_lane_with_retained_attention_run() { 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", &[]); @@ -1453,13 +1460,15 @@ fn operator_status_snapshot_shadows_post_review_lane_when_active_run_is_fresh() let rendered = orchestrator::render_operator_status(&snapshot); assert!(snapshot.active_runs[0].has_fresh_execution); - assert!(snapshot.active_runs[0].counts_as_running); - assert!(lane.shadowed_by_active_run); - assert_eq!(project.running_lane_count, 1); - assert_eq!(project.post_review_lane_count, 0); + assert!(!snapshot.active_runs[0].counts_as_running); + assert!(snapshot.active_runs[0].needs_attention); + assert_eq!(snapshot.active_runs[0].ownership_state, "retained_attention"); + assert!(!lane.shadowed_by_active_run); + assert_eq!(project.running_lane_count, 0); + assert_eq!(project.post_review_lane_count, 1); assert_eq!(project.waiting_lane_count, 0); - assert_eq!(project.attention_count, 0); - assert!(rendered.contains("shadowed_by_active_run: yes")); + assert_eq!(project.attention_count, 1); + assert!(rendered.contains("shadowed_by_active_run: no")); assert!(rendered.contains("readback_root_cause: lineage_validation_failed")); } @@ -1511,7 +1520,7 @@ fn operator_status_snapshot_counts_reused_pid_as_attention_not_running() { } #[test] -fn operator_status_snapshot_keeps_unleased_live_process_in_running_lanes() { +fn operator_status_snapshot_keeps_unleased_live_process_visible_but_not_running() { 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", &[]); @@ -1547,16 +1556,27 @@ fn operator_status_snapshot_keeps_unleased_live_process_in_running_lanes() { assert_eq!(run.execution_liveness, "process_alive"); assert_eq!(run.process_alive, Some(true)); assert_eq!(run.process_liveness_reason.as_deref(), Some("process_alive")); + assert_eq!(run.ownership_state, "orphaned_live_thread"); + assert_eq!(run.liveness_state, "process_alive"); + assert_eq!( + run.lane_control_next_action, + "inspect_or_interrupt_orphaned_live_thread" + ); + assert!(run + .lane_control_conditions + .iter() + .any(|condition| condition == "active_lease_missing")); assert!(run.has_fresh_execution); - assert!(run.counts_as_running); + assert!(!run.counts_as_running); assert!(!run.needs_attention); assert_eq!(project.active_run_count, 1); - assert_eq!(project.running_lane_count, 1); - assert_eq!(project.retained_worktree_count, 0); + assert_eq!(project.running_lane_count, 0); + assert_eq!(project.retained_worktree_count, 1); + assert_eq!(snapshot.worktrees[0].ownership, "orphaned_live_thread"); } #[test] -fn operator_status_snapshot_keeps_terminal_status_live_process_in_running_lanes() { +fn operator_status_snapshot_keeps_terminal_status_live_process_in_recent_orphan_bucket() { 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", &[]); @@ -1580,26 +1600,35 @@ fn operator_status_snapshot_keeps_terminal_status_live_process_in_running_lanes( let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 10) .expect("snapshot should build"); - let run = snapshot.active_runs.first().expect("live terminal run should remain visible"); + let run = snapshot + .recent_runs + .iter() + .find(|run| run.run_id == "run-1") + .expect("live terminal run should remain inspectable"); let project = snapshot.projects.first().expect("project summary should exist"); - assert_eq!(snapshot.active_runs.len(), 1); + assert!(snapshot.active_runs.is_empty()); assert_eq!(run.run_id, "run-1"); - assert_eq!(run.status, "running"); + assert_eq!(run.status, "failed"); assert_eq!(run.attempt_status, "failed"); - assert_eq!( - run.status_projection_reason.as_deref(), - Some("terminal_attempt_promoted_by_process_alive") - ); - assert_eq!(run.phase, "executing"); + assert_eq!(run.status_projection_reason, None); + assert_eq!(run.phase, "failed"); assert!(!run.active_lease); assert_eq!(run.queue_lease_state, "not_held"); - assert_eq!(run.execution_liveness, "process_alive"); + assert_eq!(run.execution_liveness, "not_running"); + assert_eq!(run.ownership_state, "orphaned_live_thread"); + assert_eq!(run.liveness_state, "process_alive"); + assert_eq!(run.terminalization_state, "barrier_started"); + assert!(run + .lane_control_conditions + .iter() + .any(|condition| condition == "terminal_attempt_has_live_evidence")); assert_eq!(run.process_alive, Some(true)); assert_eq!(run.process_liveness_reason.as_deref(), Some("process_alive")); - assert_eq!(project.active_run_count, 1); - assert_eq!(project.running_lane_count, 1); - assert_eq!(project.retained_worktree_count, 0); + assert_eq!(project.active_run_count, 0); + assert_eq!(project.running_lane_count, 0); + assert_eq!(project.retained_worktree_count, 1); + assert_eq!(snapshot.worktrees[0].ownership, "orphaned_live_thread"); } #[test] @@ -1632,7 +1661,7 @@ fn operator_status_snapshot_excludes_terminal_thread_archive_from_running_lanes( } #[test] -fn operator_status_snapshot_explains_terminal_run_promoted_by_active_thread() { +fn operator_status_snapshot_projects_terminal_run_with_active_thread_as_retained_attention() { 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", &[]); @@ -1665,28 +1694,32 @@ fn operator_status_snapshot_explains_terminal_run_promoted_by_active_thread() { let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 10) .expect("snapshot should build"); - let run = snapshot.active_runs.first().expect("active run should remain visible"); - let rendered = orchestrator::render_operator_status(&snapshot); + let run = snapshot + .recent_runs + .iter() + .find(|run| run.run_id == "run-1") + .expect("terminal run should remain inspectable"); - assert_eq!(run.status, "running"); + assert!(snapshot.active_runs.is_empty()); + assert_eq!(run.status, "stalled"); assert_eq!(run.attempt_status, "stalled"); - assert_eq!( - run.status_projection_reason.as_deref(), - Some("terminal_attempt_promoted_by_thread_active") - ); + assert_eq!(run.status_projection_reason, None); assert!(!run.active_lease); assert_eq!(run.queue_lease_state, "not_held"); - assert_eq!(run.execution_liveness, "process_identity_mismatch"); + assert_eq!(run.execution_liveness, "not_running"); + assert_eq!(run.ownership_state, "retained_attention"); + assert_eq!(run.liveness_state, "host_boot_mismatch"); assert_eq!(run.process_alive, Some(false)); assert_eq!(run.process_liveness_reason.as_deref(), Some("host_boot_id_mismatch")); assert_eq!(run.thread_status.as_deref(), Some("active")); - assert!(rendered.contains( - "status_projection_reason: terminal_attempt_promoted_by_thread_active" - )); + assert!(run + .lane_control_conditions + .iter() + .any(|condition| condition == "host_boot_id_mismatch")); } #[test] -fn operator_status_snapshot_keeps_succeeded_status_live_process_in_running_lanes() { +fn operator_status_snapshot_keeps_succeeded_status_live_process_in_recent_orphan_bucket() { 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", &[]); @@ -1710,26 +1743,31 @@ fn operator_status_snapshot_keeps_succeeded_status_live_process_in_running_lanes let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 10) .expect("snapshot should build"); - let run = snapshot.active_runs.first().expect("live succeeded run should remain visible"); + let run = snapshot + .recent_runs + .iter() + .find(|run| run.run_id == "run-1") + .expect("live succeeded run should remain inspectable"); let project = snapshot.projects.first().expect("project summary should exist"); - assert_eq!(snapshot.active_runs.len(), 1); + assert!(snapshot.active_runs.is_empty()); assert_eq!(run.run_id, "run-1"); - assert_eq!(run.status, "running"); + assert_eq!(run.status, "succeeded"); assert_eq!(run.attempt_status, "succeeded"); - assert_eq!( - run.status_projection_reason.as_deref(), - Some("terminal_attempt_promoted_by_process_alive") - ); - assert_eq!(run.phase, "executing"); + assert_eq!(run.status_projection_reason, None); + assert_eq!(run.phase, "completed"); assert!(!run.active_lease); assert_eq!(run.queue_lease_state, "not_held"); - assert_eq!(run.execution_liveness, "process_alive"); + assert_eq!(run.execution_liveness, "not_running"); + assert_eq!(run.ownership_state, "orphaned_live_thread"); + assert_eq!(run.liveness_state, "process_alive"); + assert_eq!(run.terminalization_state, "barrier_started"); assert_eq!(run.process_alive, Some(true)); assert_eq!(run.process_liveness_reason.as_deref(), Some("process_alive")); - assert_eq!(project.active_run_count, 1); - assert_eq!(project.running_lane_count, 1); - assert_eq!(project.retained_worktree_count, 0); + assert_eq!(project.active_run_count, 0); + assert_eq!(project.running_lane_count, 0); + assert_eq!(project.retained_worktree_count, 1); + assert_eq!(snapshot.worktrees[0].ownership, "orphaned_live_thread"); } #[test] diff --git a/apps/decodex/src/orchestrator/tests/operator/status_support.rs b/apps/decodex/src/orchestrator/tests/operator/status_support.rs index d1acdee1a..ea3bc0587 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status_support.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status_support.rs @@ -344,6 +344,12 @@ fn operator_status_text_active_run() -> OperatorRunStatus { status: String::from("running"), attempt_status: String::from("running"), status_projection_reason: None, + ownership_state: String::from("owned_active"), + liveness_state: String::from("process_alive"), + policy_state: String::from("allowed"), + terminalization_state: String::from("none"), + lane_control_next_action: String::from("continue_owned_attempt"), + lane_control_conditions: Vec::new(), phase: String::from("executing"), wait_reason: None, current_operation: String::from(RUN_OPERATION_AGENT_RUN), @@ -353,13 +359,13 @@ fn operator_status_text_active_run() -> OperatorRunStatus { thread_active_flags: vec![String::from("waitingOnApproval")], interactive_requested: true, continuation_pending: false, - active_lease: true, - queue_lease_state: String::from("held"), - execution_liveness: String::from("process_alive"), - has_fresh_execution: true, - counts_as_running: true, - needs_attention: false, - updated_at: String::from("2026-03-14 09:00:00"), + active_lease: true, + queue_lease_state: String::from("held"), + execution_liveness: String::from("process_alive"), + has_fresh_execution: true, + counts_as_running: true, + needs_attention: false, + updated_at: String::from("2026-03-14 09:00:00"), last_run_activity_at: Some(String::from("2026-03-14 10:00:00Z")), last_protocol_activity_at: Some(String::from("2026-03-14 10:00:01Z")), last_progress_at: Some(String::from("2026-03-14 10:00:01Z")), diff --git a/apps/decodex/src/orchestrator/types.rs b/apps/decodex/src/orchestrator/types.rs index 7988ea6d2..e42832230 100644 --- a/apps/decodex/src/orchestrator/types.rs +++ b/apps/decodex/src/orchestrator/types.rs @@ -1522,6 +1522,13 @@ struct OperatorRunStatus { attempt_status: String, #[serde(skip_serializing_if = "Option::is_none")] status_projection_reason: Option, + ownership_state: String, + liveness_state: String, + policy_state: String, + terminalization_state: String, + lane_control_next_action: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + lane_control_conditions: Vec, phase: String, wait_reason: Option, current_operation: String, diff --git a/docs/spec/index.md b/docs/spec/index.md index 8b87a3f07..ca0d9bbf1 100644 --- a/docs/spec/index.md +++ b/docs/spec/index.md @@ -65,6 +65,9 @@ Then keep the body explicit: - [`lane-control.md`](./lane-control.md) defines the CLI/API-first operator lane-control capability matrix and the boundary between bottom-layer steer support and higher-level policy guardrails. +- [`lane-control-state.md`](./lane-control-state.md) defines the authoritative lane + control state axes, invariants, guard semantics, terminal barrier, and projection + rules used by scheduler decisions and operator status. - [`github-change-bundle.md`](./github-change-bundle.md) defines the normalized GitHub input model for PR-first public signal analysis. - [`signal-entry.md`](./signal-entry.md) defines the published signal-entry schema used diff --git a/docs/spec/lane-control-state.md b/docs/spec/lane-control-state.md new file mode 100644 index 000000000..f38d9b77e --- /dev/null +++ b/docs/spec/lane-control-state.md @@ -0,0 +1,149 @@ +# Lane-Control State Specification + +Purpose: Define the authoritative Decodex lane-control state model used by +scheduler decisions, policy guards, terminal cleanup, and operator projections. +Status: normative +Read this when: You are implementing or validating lane scheduling, active-run +projection, review-policy stops, retained recovery, closeout, or dashboard status. +Not this document: The operator command sequence for steering or interrupting a lane. +Use [`lane-control.md`](./lane-control.md) for CLI/API controls and +[`runtime.md`](./runtime.md) for broader runtime reconciliation rules. +Defines: The lane control state axes, invariants, guard semantics, terminal barrier, +and projection rules that prevent liveness evidence from re-creating ownership. + +## Resource Model + +Each Decodex lane is a control-plane resource with four separate state axes: + +- `ownership_state`: who, if anyone, is authorized to mutate the lane. +- `liveness_state`: what local process, app-server, or protocol evidence is visible. +- `policy_state`: whether review, retry, architecture, or authority rules allow the + current strategy to continue. +- `terminalization_state`: whether finalization has started, retired run control, or + completed cleanup. + +Operator snapshots may include derived fields for readability, but scheduler +decisions and running-lane counts must be based on the state axes rather than inferred +from protocol activity. + +## Canonical Values + +`ownership_state` values: + +- `pending`: eligible or waiting before an owned attempt starts. +- `owned_active`: Decodex owns the active lease and the active attempt may mutate the + lane. +- `terminalizing`: Decodex is retiring run control, finishing writeback, archiving the + app-server thread, or cleaning up an owned attempt. +- `retained_attention`: useful retained state exists but autonomous mutation is stopped + until recovery or human attention resolves the blocker. +- `orphaned_live_thread`: liveness evidence remains after Decodex lost active ownership. +- `closed`: no active ownership remains and no retained recovery bucket is required. + +`liveness_state` values: + +- `unknown`: no useful live evidence exists. +- `process_alive`: the recorded child process is still alive. +- `thread_active`: app-server reports an active thread or active flags. +- `protocol_recent`: recent protocol work evidence exists without a live process. +- `not_running`: the owned process/thread is stopped or archived. +- `host_boot_mismatch`: the process marker belongs to a previous host boot. +- `late_protocol_activity`: protocol activity arrived after a terminal barrier. + +`policy_state` values: + +- `allowed`: no policy stop is active. +- `review_pending`: a required review checkpoint has not been recorded. +- `review_findings`: the latest review checkpoint has non-clean findings but remains + within the repair budget. +- `review_churn_exceeded`: repeated non-clean findings exceeded the convergence budget + and the current repair strategy must stop. +- `architecture_recovery_pending`: architecture recovery is the next autonomous path. +- `authority_boundary_required`: a boundary decision or human authority is required. +- `human_attention_required`: the lane is blocked for explicit operator attention. + +`terminalization_state` values: + +- `none`: no terminal barrier is active. +- `barrier_started`: a terminal transition has started. +- `run_control_retired`: the run-control channel is no longer authoritative. +- `thread_archive_requested`: the app-server thread is being archived or has been + asked to stop. +- `cleanup_pending`: closeout or worktree cleanup remains. +- `cleanup_complete`: the lane is closed out and cleanup is complete. + +## Invariants + +- A lane counts as a running lane only when `ownership_state` is `owned_active`. +- Liveness evidence may update `liveness_state`, but it must not create or restore + `owned_active` ownership. +- `active_lease=false` is incompatible with `ownership_state=owned_active`. +- Terminal attempt statuses such as `failed`, `interrupted`, `stalled`, or `succeeded` + must not be promoted to `running` by live process, thread, or protocol evidence. +- `policy_state=review_churn_exceeded` blocks further review-repair mutation for the + same strategy until architecture recovery or human attention changes the policy + state. +- Review checkpoints that would move the current phase to + `review_churn_exceeded` must fail immediately after persisting the checkpoint + evidence; the runtime must not wait for turn-end classification to enforce the + stop. +- Review-repair external round counters are monotonic for a retained review lineage; + there is no fourth-round reset path. +- Protocol events after a terminal barrier are retained as evidence and projected as + `late_protocol_activity`; they do not change `ownership_state`. +- Merged pull request plus completed tracker issue plus no owned active attempt + projects as cleanup or closed state, not as a running lane. + +## Guard Semantics + +Every mutating lane tool must consult lane control state before writing tracker, +worktree, review, closeout, or run-control data. A guard decision has three outcomes: + +- `allow`: the tool may proceed. +- `deny_terminal`: the lane is terminalizing or closed; the tool must not mutate it. +- `deny_policy`: policy state requires architecture recovery or human attention before + another repair or handoff mutation. + +`issue_review_checkpoint` may record the checkpoint that triggers a policy stop, and +`issue_terminal_finalize` may record terminal retention/cleanup after the stop. +Progress, handoff, repair-complete, closeout, transition, label, and comment tools +must be fenced while `policy_state=review_churn_exceeded` remains current for the +run's review phase and repair strategy. Changing the lane HEAD does not clear this +guard; architecture recovery or human attention must explicitly change the policy +state. + +Guard decisions are private runtime evidence. Public tracker projections may summarize +the stop reason, but raw lane evidence remains local unless an allowlisted lifecycle +projection renders it. + +## Projection Rules + +Operator status and dashboard views must show the four state axes and `next_action`. +Legacy fields such as `status`, `phase`, and `execution_liveness` may remain for +compatibility, but they must not be used to infer ownership in new scheduler code. + +Examples: + +```text +ownership_state: closed +liveness_state: late_protocol_activity +policy_state: allowed +terminalization_state: cleanup_complete +next_action: ignore_late_activity +``` + +```text +ownership_state: retained_attention +liveness_state: host_boot_mismatch +policy_state: allowed +terminalization_state: none +next_action: inspect_recovery_evidence +``` + +```text +ownership_state: owned_active +liveness_state: process_alive +policy_state: review_findings +terminalization_state: none +next_action: repair_review_findings +```