Implement lane orchestration kernel

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,
 	},
 };

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 =

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,26 +1085,50 @@ 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<Option<ReviewPolicyStopRequested>> {
+		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<ReviewPolicyStopRequested> {
 		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 =>
 				ReviewPolicyStopReason::ArchitectureReviewRequired,
 			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 {

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]

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<DynamicToolCallResponse> {
+		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::<ProgressCheckpointArgs>(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)
 	}
 

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<String>,
 	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<String> {
+	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(),

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<String>,
 	thread_id: Option<String>,
 	turn_id: Option<String>,
 	thread_status: Option<String>,
@@ -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"),