From be4043791f11167c01211cac9fb7c84f96069cc3 Mon Sep 17 00:00:00 2001 From: Yvette Carlisle Date: Sun, 7 Jun 2026 22:29:20 +0800 Subject: [PATCH] {"schema":"decodex/commit/1","summary":"Recover lane controls after active lease loss","authority":"XY-777"} --- apps/decodex/src/orchestrator/lane_control.rs | 99 ++++++-- apps/decodex/src/orchestrator/status.rs | 50 +++- .../tests/operator/status/http.rs | 240 ++++++++++++++++++ .../tests/operator/status/running_lanes.rs | 43 ++++ apps/decodex/src/state/models.rs | 8 + apps/decodex/src/state/store.rs | 81 +++++- apps/decodex/src/state/tests.rs | 39 +++ docs/reference/operator-control-plane.md | 12 +- docs/runbook/lane-control-recovery.md | 13 + docs/spec/lane-control.md | 22 +- docs/spec/runtime.md | 5 + 11 files changed, 579 insertions(+), 33 deletions(-) diff --git a/apps/decodex/src/orchestrator/lane_control.rs b/apps/decodex/src/orchestrator/lane_control.rs index 386117a5d..8bb0487d0 100644 --- a/apps/decodex/src/orchestrator/lane_control.rs +++ b/apps/decodex/src/orchestrator/lane_control.rs @@ -8,6 +8,7 @@ use std::{ use libc::{ESRCH, SIGKILL, SIGTERM, c_int, pid_t}; use serde::Serialize; +use serde_json::Value; use crate::{ config::ServiceConfig, @@ -393,8 +394,12 @@ pub(super) fn steer_lane_with_state( } fn soft_interrupt_allows_hard_fallback(soft: &LaneSoftInterruptReport) -> bool { - matches!(soft.status.as_str(), "pending" | "failed" | "unavailable") - && soft.error_class.as_deref() != Some("lane_not_active") + match soft.status.as_str() { + "pending" | "failed" | "unavailable" => + soft.error_class.as_deref() != Some("lane_not_active"), + "rejected" => soft.error_class.as_deref() == Some("active_lease_missing"), + _ => false, + } } fn load_lane_control_project( @@ -510,6 +515,45 @@ fn soft_interrupt_available_for_run(run: &OperatorRunStatus) -> bool { && run.control_capability.as_ref().is_some_and(|capability| capability.status == "active") } +fn lane_control_operator_context(run: &OperatorRunStatus) -> Value { + let control_capability = run.control_capability.as_ref().map(|capability| { + serde_json::json!({ + "project_id": capability.project_id.as_str(), + "issue_id": capability.issue_id.as_str(), + "run_id": capability.run_id.as_str(), + "attempt_number": capability.attempt_number, + "thread_id": capability.thread_id.as_deref(), + "turn_id": capability.turn_id.as_deref(), + "transport": capability.transport.as_str(), + "channel_path": capability.channel_path.as_str(), + "status": capability.status.as_str(), + "published_at": capability.published_at.as_str(), + "updated_at": capability.updated_at.as_str(), + }) + }); + + serde_json::json!({ + "status": run.status.as_str(), + "attempt_status": run.attempt_status.as_str(), + "phase": run.phase.as_str(), + "wait_reason": run.wait_reason.as_deref(), + "current_operation": run.current_operation.as_str(), + "active_lease": run.active_lease, + "queue_lease_state": run.queue_lease_state.as_str(), + "execution_liveness": run.execution_liveness.as_str(), + "thread_status": run.thread_status.as_deref(), + "process_id": run.process_id, + "process_alive": run.process_alive, + "process_liveness_reason": run.process_liveness_reason.as_deref(), + "branch": run.branch_name.as_deref(), + "worktree_path": run.worktree_path.as_deref(), + "last_event_type": run.last_event_type.as_deref(), + "last_event_at": run.last_event_at.as_deref(), + "event_count": run.event_count, + "control_capability": control_capability, + }) +} + fn attempt_lane_steer( state_store: &StateStore, project: &ServiceConfig, @@ -518,6 +562,7 @@ fn attempt_lane_steer( ) -> Result { let message_byte_count = request.message.len(); let message_line_count = lane_steer_message_line_count(request.message); + let context = lane_control_operator_context(run); let metadata = serde_json::json!({ "expectedTurnId": request.expected_turn_id, "messageByteCount": message_byte_count, @@ -534,6 +579,7 @@ fn attempt_lane_steer( action: "steer", timeout_ms: Some(i64::try_from(request.wait_timeout.as_millis()).unwrap_or(i64::MAX)), metadata: Some(&metadata), + context: Some(&context), })?; if receipt.outcome() != RUN_CONTROL_ACTION_ACCEPTED { @@ -793,20 +839,14 @@ fn attempt_soft_lane_interrupt( )); } - let receipt = state_store.resolve_run_control_action(RunControlActionRequest { - project_id: project.service_id(), - issue_id: &run.issue_id, - run_id: &run.run_id, - attempt_number: run.attempt_number, - thread_id: Some(thread_id), - turn_id: Some(turn_id), + let receipt = resolve_soft_interrupt_control_action( + state_store, + project, + run, + thread_id, + turn_id, source, - action: "interrupt", - timeout_ms: Some( - i64::try_from(LANE_INTERRUPT_RESPONSE_WAIT.as_millis()).unwrap_or(i64::MAX), - ), - metadata: None, - })?; + )?; if receipt.outcome() != "accepted" { return Ok(LaneSoftInterruptReport::from_control_rejection(&receipt)); @@ -886,6 +926,33 @@ fn attempt_soft_lane_interrupt( } } +fn resolve_soft_interrupt_control_action( + state_store: &StateStore, + project: &ServiceConfig, + run: &OperatorRunStatus, + thread_id: &str, + turn_id: &str, + source: &str, +) -> Result { + let context = lane_control_operator_context(run); + + state_store.resolve_run_control_action(RunControlActionRequest { + project_id: project.service_id(), + issue_id: &run.issue_id, + run_id: &run.run_id, + attempt_number: run.attempt_number, + thread_id: Some(thread_id), + turn_id: Some(turn_id), + source, + action: "interrupt", + timeout_ms: Some( + i64::try_from(LANE_INTERRUPT_RESPONSE_WAIT.as_millis()).unwrap_or(i64::MAX), + ), + metadata: None, + context: Some(&context), + }) +} + fn attempt_hard_lane_interrupt( state_store: &StateStore, run: &OperatorRunStatus, @@ -978,6 +1045,7 @@ fn record_hard_interrupt_control_fallback( run: &OperatorRunStatus, _reason: Option<&str>, ) -> Result<()> { + let context = lane_control_operator_context(run); let receipt = state_store.resolve_run_control_action(RunControlActionRequest { project_id: &run.project_id, issue_id: &run.issue_id, @@ -989,6 +1057,7 @@ fn record_hard_interrupt_control_fallback( action: "interrupt", timeout_ms: None, metadata: None, + context: Some(&context), })?; state_store.record_run_control_action_outcome( diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index 61e0472e0..0bb80b601 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -297,7 +297,7 @@ fn build_operator_status_snapshot_with_account_mode( active_runs.iter().map(|run| run.run_id.clone()).collect::>(); for run in &recent_runs { - if !active_run_ids.contains(&run.run_id) && operator_run_has_live_process(run) { + if !active_run_ids.contains(&run.run_id) && operator_run_has_live_execution(run) { active_run_ids.insert(run.run_id.clone()); active_runs.push(run.clone()); } @@ -1050,12 +1050,16 @@ fn worktree_cleanup_key(worktree: &OperatorWorktreeStatus) -> String { } fn operator_run_counts_as_active(run: &OperatorRunStatus) -> bool { - (run.active_lease || operator_run_has_live_process(run)) + (run.active_lease || operator_run_has_live_execution(run)) && !matches!(run.phase.as_str(), "completed" | "failed" | "terminated") } -fn operator_run_has_live_process(run: &OperatorRunStatus) -> bool { - matches!(run.status.as_str(), "starting" | "running") && run.process_alive == Some(true) +fn operator_run_has_live_execution(run: &OperatorRunStatus) -> bool { + matches!(run.status.as_str(), "starting" | "running") + && matches!( + run.execution_liveness.as_str(), + "process_alive" | "thread_active" | "protocol_observed" + ) } fn operator_run_counts_as_running(run: &OperatorRunStatus) -> bool { @@ -4348,8 +4352,14 @@ fn operator_run_status( let timing = operator_run_timing(&run, marker.as_ref(), now_unix_epoch); let app_server_state = operator_run_app_server_state(&run, marker.as_ref()); let protocol_summary = operator_run_protocol_summary(&run, marker.as_ref()); - let status = - operator_run_visible_status(run.status(), &app_server_state, &protocol_summary, &timing); + let marker_current_operation = marker.as_ref().and_then(RunActivityMarker::current_operation); + let status = operator_run_visible_status( + run.status(), + &app_server_state, + &protocol_summary, + &timing, + marker_current_operation, + ); let (retry_kind, retry_ready_at_unix_epoch) = visible_operator_run_retry_schedule( &status, marker.as_ref().and_then(RunActivityMarker::retry_kind), @@ -4364,7 +4374,7 @@ fn operator_run_status( ); let current_operation = classify_operator_run_operation( &phase, - marker.as_ref().and_then(RunActivityMarker::current_operation), + marker_current_operation, ); let suspected_stall = operator_run_is_suspected_stall( &phase, @@ -4706,6 +4716,7 @@ fn operator_run_visible_status( app_server_state: &OperatorRunAppServerState, protocol_summary: &OperatorRunProtocolSummary, timing: &OperatorRunTiming, + marker_current_operation: Option<&str>, ) -> String { if attempt_status == "starting" && operator_run_has_app_server_execution_evidence( @@ -4716,10 +4727,35 @@ fn operator_run_visible_status( { 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, timing) + { + return String::from("running"); + } attempt_status.to_owned() } +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, + 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() + || timing.protocol_idle_for_seconds.is_some_and(|idle_for| { + u64::try_from(idle_for) + .is_ok_and(|idle_for| idle_for < ACTIVE_RUN_IDLE_TIMEOUT.as_secs()) + }) +} + fn operator_run_has_app_server_execution_evidence( app_server_state: &OperatorRunAppServerState, protocol_summary: &OperatorRunProtocolSummary, diff --git a/apps/decodex/src/orchestrator/tests/operator/status/http.rs b/apps/decodex/src/orchestrator/tests/operator/status/http.rs index 3c9b0770c..bbc0a2d54 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/http.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/http.rs @@ -4,9 +4,27 @@ use std::net::SocketAddr; use orchestrator::OperatorControlRequests; use state::RUN_CONTROL_CHANNEL_DIR; use state::RUN_CONTROL_CHANNEL_TRANSPORT_LOCAL_FILE; +use process::Child; use crate::runtime; +#[cfg(unix)] +struct ActiveLeaseMissingControlFixture { + issue: TrackerIssue, + channel_path: PathBuf, + child: Child, + child_process_id: u32, +} +#[cfg(unix)] +impl Drop for ActiveLeaseMissingControlFixture { + fn drop(&mut self) { + if matches!(self.child.try_wait(), Ok(None)) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + } +} + #[test] fn operator_state_endpoint_serves_dashboard_html_from_root_and_dashboard_route() { for path in [ @@ -1785,6 +1803,228 @@ fn operator_lane_interrupt_api_force_does_not_hard_fallback_after_control_reject })); } +#[cfg(unix)] +fn active_lease_missing_control_fixture( + config: &ServiceConfig, + state_store: &StateStore, +) -> ActiveLeaseMissingControlFixture { + let registration = ProjectRegistration::from_config( + config.service_id(), + &service_config_path(config.repo_root()), + config, + true, + "test-fingerprint", + ); + let issue = sample_issue("In Progress", &[]); + let worktree_path = config.worktree_root().join(&issue.identifier); + let channel_path = + worktree_path.join(RUN_CONTROL_CHANNEL_DIR).join("pub-101-attempt-1.channel"); + let child = Command::new("/bin/sh") + .args(["-c", "exec sleep 60"]) + .spawn() + .expect("lane child process should start"); + let child_process_id = child.id(); + + assert!( + orchestrator::process_is_alive(child_process_id), + "lane child process should be live before control request" + ); + + fs::create_dir_all(channel_path.parent().expect("channel path should have parent")) + .expect("run-control channel dir should exist"); + fs::write(&channel_path, "ready\n").expect("control channel should write"); + state::write_run_activity_marker_for_process( + &worktree_path, + "pub-101-attempt-1", + 1, + child_process_id, + ) + .expect("activity marker should write"); + + state_store.upsert_project(®istration).expect("project should register"); + state_store + .record_run_attempt("pub-101-attempt-1", &issue.id, 1, "running") + .expect("run attempt should record"); + state_store + .update_run_thread("pub-101-attempt-1", "thread-1") + .expect("thread should record"); + state_store + .update_run_turn("pub-101-attempt-1", "turn-1") + .expect("turn should record"); + state_store + .upsert_lease(config.service_id(), &issue.id, "pub-101-attempt-1", "In Progress") + .expect("lease should record"); + state_store + .upsert_worktree( + config.service_id(), + &issue.id, + "x/pubfi-pub-101", + &worktree_path.display().to_string(), + ) + .expect("worktree should record"); + state_store + .publish_run_control_channel_for_active_attempt( + "pub-101-attempt-1", + 1, + &channel_path, + RUN_CONTROL_CHANNEL_TRANSPORT_LOCAL_FILE, + ) + .expect("control channel should publish"); + state_store.clear_lease(&issue.id).expect("lease should clear"); + + ActiveLeaseMissingControlFixture { issue, channel_path, child, child_process_id } +} + +#[cfg(unix)] +fn operator_json_response_body(response: &str, context: &str) -> Value { + let body = response + .split_once("\r\n\r\n") + .map(|(_, body)| body) + .unwrap_or_else(|| panic!("{context} response should include body")); + + serde_json::from_str(body).unwrap_or_else(|_| panic!("{context} response should be json")) +} + +#[cfg(unix)] +fn reap_active_lease_missing_child(fixture: &mut ActiveLeaseMissingControlFixture) { + if orchestrator::process_is_alive(fixture.child_process_id) { + fixture + .child + .kill() + .expect("lane child process should be killable after failed fallback"); + } + + fixture.child.wait().expect("lane child process should reap"); +} + +#[cfg(unix)] +fn assert_active_lease_missing_control_audit( + config: &ServiceConfig, + state_store: &StateStore, + fixture: &ActiveLeaseMissingControlFixture, + run_id: &str, +) { + let events = state_store + .list_private_execution_events(config.service_id(), &fixture.issue.id, run_id, 1) + .expect("private control audit should read"); + let missing_lease_steer_event = events + .iter() + .find(|event| { + event.event_type() == "control_action" + && event.payload()["action"] == "steer" + && event.payload()["reason"] == "active_lease_missing" + }) + .expect("active lease steer rejection should be audited"); + let missing_lease_interrupt_event = events + .iter() + .find(|event| { + event.event_type() == "control_action" + && event.payload()["action"] == "interrupt" + && event.payload()["reason"] == "active_lease_missing" + }) + .expect("active lease interrupt rejection should be audited"); + let expected_channel_path = fixture.channel_path.display().to_string(); + + assert_eq!( + missing_lease_steer_event.payload()["context"]["process_alive"].as_bool(), + Some(true) + ); + assert_eq!( + missing_lease_steer_event.payload()["channel"]["channel_path"].as_str(), + Some(expected_channel_path.as_str()) + ); + assert_eq!( + missing_lease_interrupt_event.payload()["lane"]["active_lease"].as_bool(), + Some(false) + ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["process_alive"].as_bool(), + Some(true) + ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["execution_liveness"].as_str(), + Some("process_alive") + ); + assert_eq!( + missing_lease_interrupt_event.payload()["context"]["control_capability"]["channel_path"] + .as_str(), + Some(expected_channel_path.as_str()) + ); + assert_eq!( + missing_lease_interrupt_event.payload()["channel"]["channel_path"].as_str(), + Some(expected_channel_path.as_str()) + ); + assert!(events.iter().any(|event| { + event.event_type() == "control_action" + && event.payload()["outcome"] == "fallback" + && event.payload()["reason"] == "hard_interrupt_fallback" + })); +} + +#[cfg(unix)] +#[test] +fn operator_lane_interrupt_api_force_hard_fallbacks_after_active_lease_missing() { + let (_temp_dir, config, _workflow) = temp_project_layout(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let mut fixture = active_lease_missing_control_fixture(&config, &state_store); + let project_id = config.service_id().to_owned(); + let issue_identifier = fixture.issue.identifier.clone(); + let run_id = "pub-101-attempt-1"; + let body = format!( + r#"{{"projectId":"{project_id}","issue":"{issue_identifier}","runId":"{run_id}","force":true}}"# + ); + let request = format!( + "POST {} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + orchestrator::OPERATOR_LANE_INTERRUPT_ENDPOINT_PATH, + body.len(), + body + ); + let steer_body = format!( + r#"{{"projectId":"{project_id}","issue":"{issue_identifier}","runId":"{run_id}","expectedTurnId":"turn-1","message":"preserve partial work and report current state"}}"# + ); + let steer_request = format!( + "POST {} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + orchestrator::OPERATOR_LANE_STEER_ENDPOINT_PATH, + steer_body.len(), + steer_body + ); + let steer_response = String::from_utf8(orchestrator::build_operator_lane_steer_http_response( + &state_store, + steer_request.as_bytes(), + )) + .expect("lane steer response should be utf-8"); + let steer_data = operator_json_response_body(&steer_response, "lane steer"); + let response = String::from_utf8(orchestrator::build_operator_lane_interrupt_http_response( + &state_store, + request.as_bytes(), + )) + .expect("lane interrupt response should be utf-8"); + let data = operator_json_response_body(&response, "lane interrupt"); + + reap_active_lease_missing_child(&mut fixture); + + assert!( + steer_response.starts_with("HTTP/1.1 409 Conflict\r\n"), + "{steer_response}" + ); + assert_eq!(steer_data["outcome"], "rejected"); + assert_eq!(steer_data["reason"], "active_lease_missing"); + assert_eq!(steer_data["failureClass"], "run_control_action_failed"); + assert!(response.starts_with("HTTP/1.1 200 OK\r\n")); + assert_eq!(data["classification"], "hard_interrupt_fallback"); + assert_eq!(data["softInterrupt"]["status"], "rejected"); + assert_eq!(data["softInterrupt"]["errorClass"], "active_lease_missing"); + assert_eq!(data["hardInterrupt"]["classification"], "hard_interrupt_fallback"); + assert_eq!(data["hardInterrupt"]["status"], "sent"); + assert_eq!( + data["hardInterrupt"]["processId"].as_u64(), + Some(u64::from(fixture.child_process_id)) + ); + assert_eq!(data["hardInterrupt"]["processAliveAfter"], false); + + assert_active_lease_missing_control_audit(&config, &state_store, &fixture, run_id); +} + #[test] fn operator_lane_steer_api_rejects_stale_expected_turn_id() { let (_temp_dir, config, _workflow) = temp_project_layout(); 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 1f2bc83d7..a7c99e530 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs @@ -1150,6 +1150,49 @@ fn operator_status_snapshot_keeps_unleased_live_process_in_running_lanes() { assert_eq!(project.retained_worktree_count, 0); } +#[test] +fn operator_status_snapshot_keeps_terminal_status_live_process_in_running_lanes() { + 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"); + + state_store + .record_run_attempt("run-1", &issue.id, 1, "failed") + .expect("run attempt 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"); + state::write_run_activity_marker_for_process(&worktree_path, "run-1", 1, process::id()) + .expect("live process marker should write"); + + 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 project = snapshot.projects.first().expect("project summary should exist"); + + assert_eq!(snapshot.active_runs.len(), 1); + assert_eq!(run.run_id, "run-1"); + assert_eq!(run.status, "running"); + assert_eq!(run.attempt_status, "failed"); + assert_eq!(run.phase, "executing"); + assert!(!run.active_lease); + assert_eq!(run.queue_lease_state, "not_held"); + 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!(project.active_run_count, 1); + assert_eq!(project.running_lane_count, 1); + assert_eq!(project.retained_worktree_count, 0); +} + #[test] fn operator_status_snapshot_promotes_starting_after_app_server_activity() { let (_temp_dir, config, _workflow) = temp_project_layout(); diff --git a/apps/decodex/src/state/models.rs b/apps/decodex/src/state/models.rs index 15ab70511..ed7ecee9a 100644 --- a/apps/decodex/src/state/models.rs +++ b/apps/decodex/src/state/models.rs @@ -149,6 +149,7 @@ pub struct RunControlActionReceipt { reason: String, audit_record_id: i64, metadata: Option, + context: Option, channel: Option, } impl RunControlActionReceipt { @@ -222,6 +223,11 @@ impl RunControlActionReceipt { self.metadata.as_ref() } + /// Optional compact lane context captured with the audit event. + pub fn context(&self) -> Option<&Value> { + self.context.as_ref() + } + /// Control channel selected for an accepted request. pub fn channel(&self) -> Option<&RunControlChannel> { self.channel.as_ref() @@ -573,6 +579,8 @@ pub(crate) struct RunControlActionRequest<'a> { pub(crate) timeout_ms: Option, /// Optional compact, non-secret action metadata to include in audit evidence. pub(crate) metadata: Option<&'a Value>, + /// Optional compact lane context to include in audit evidence. + pub(crate) context: Option<&'a Value>, } /// Follow-up outcome for a run-control action handled after initial resolution. diff --git a/apps/decodex/src/state/store.rs b/apps/decodex/src/state/store.rs index 14eed24cf..c231a555c 100644 --- a/apps/decodex/src/state/store.rs +++ b/apps/decodex/src/state/store.rs @@ -857,6 +857,10 @@ impl StateStore { &resolution.reason, None, )?; + let receipt_channel = resolution + .channel + .clone() + .or_else(|| resolution.audit_target.channel.clone()); Ok(RunControlActionReceipt { project_id: resolution.audit_target.project_id, @@ -873,7 +877,8 @@ impl StateStore { reason: resolution.reason, audit_record_id: event.record_id(), metadata: resolution.audit_target.metadata, - channel: resolution.channel, + context: resolution.audit_target.context, + channel: receipt_channel, }) } @@ -901,6 +906,14 @@ impl StateStore { action: receipt.action.clone(), timeout_ms: None, metadata: receipt.metadata.clone(), + context: receipt.context.clone(), + attempt_status: None, + branch_name: None, + worktree_path: None, + active_lease: None, + event_count: None, + last_event_type: None, + last_event_at: None, channel: receipt.channel.clone(), }; @@ -934,6 +947,14 @@ impl StateStore { action: request.action.to_owned(), timeout_ms: request.timeout_ms, metadata: request.metadata.cloned(), + context: None, + attempt_status: None, + branch_name: None, + worktree_path: None, + active_lease: None, + event_count: None, + last_event_type: None, + last_event_at: None, channel: request.channel.cloned(), }; @@ -975,10 +996,20 @@ impl StateStore { "timeout_ms": target.timeout_ms, }, "observed": { - "thread_id": target.current_thread_id, - "turn_id": target.current_turn_id, + "thread_id": target.current_thread_id.as_deref(), + "turn_id": target.current_turn_id.as_deref(), }, - "metadata": target.metadata, + "lane": { + "attempt_status": target.attempt_status.as_deref(), + "active_lease": target.active_lease, + "branch": target.branch_name.as_deref(), + "worktree_path": target.worktree_path.as_ref().map(|path| path.display().to_string()), + "event_count": target.event_count, + "last_event_type": target.last_event_type.as_deref(), + "last_event_at": target.last_event_at.as_deref(), + }, + "metadata": target.metadata.as_ref(), + "context": target.context.as_ref(), "channel": channel.map(|channel| serde_json::json!({ "transport": channel.transport(), "channel_path": channel.channel_path().display().to_string(), @@ -2115,6 +2146,7 @@ struct RunControlAuditTarget { issue_id: String, run_id: String, attempt_number: i64, + attempt_status: Option, thread_id: Option, turn_id: Option, source: String, @@ -2123,6 +2155,13 @@ struct RunControlAuditTarget { current_thread_id: Option, current_turn_id: Option, metadata: Option, + context: Option, + branch_name: Option, + worktree_path: Option, + active_lease: Option, + event_count: Option, + last_event_type: Option, + last_event_at: Option, channel: Option, } @@ -2246,11 +2285,17 @@ fn resolve_run_control_action_locked( .map(|channel| channel.project_id.clone()) .or_else(|| state.project_id_for_run(&attempt.issue_id, &attempt.run_id)) .unwrap_or_else(|| request.project_id.to_owned()); + let project_run_status = state.project_run_status(&audit_project_id, attempt); + let control_channel = project_run_status + .as_ref() + .and_then(|status| status.control_channel().cloned()) + .or_else(|| state.control_channels.get(request.run_id).map(RunControlChannelRecord::as_public)); let audit_target = RunControlAuditTarget { project_id: audit_project_id, issue_id: attempt.issue_id.clone(), run_id: attempt.run_id.clone(), attempt_number: attempt.attempt_number, + attempt_status: Some(attempt.status.clone()), thread_id: request.thread_id.map(str::to_owned), turn_id: request.turn_id.map(str::to_owned), current_thread_id: attempt.thread_id.clone(), @@ -2259,7 +2304,22 @@ fn resolve_run_control_action_locked( action: request.action.to_owned(), timeout_ms: request.timeout_ms, metadata: request.metadata.cloned(), - channel: None, + context: request.context.cloned(), + branch_name: project_run_status + .as_ref() + .and_then(|status| status.branch_name().map(str::to_owned)), + worktree_path: project_run_status + .as_ref() + .and_then(|status| status.worktree_path().map(Path::to_path_buf)), + active_lease: project_run_status.as_ref().map(ProjectRunStatus::active_lease), + event_count: project_run_status.as_ref().map(ProjectRunStatus::event_count), + last_event_type: project_run_status + .as_ref() + .and_then(|status| status.last_event_type().map(str::to_owned)), + last_event_at: project_run_status + .as_ref() + .and_then(|status| status.last_event_at().map(str::to_owned)), + channel: control_channel.clone(), }; if attempt.issue_id != request.issue_id { @@ -2291,10 +2351,9 @@ fn resolve_run_control_action_locked( return rejected_run_control_resolution(request, Some(audit_target), "run_not_active"); } - let Some(channel) = state.control_channels.get(request.run_id).cloned() else { + let Some(channel) = control_channel else { return rejected_run_control_resolution(request, Some(audit_target), "control_channel_missing"); }; - let channel = channel.as_public(); let audit_target = RunControlAuditTarget { channel: Some(channel.clone()), ..audit_target }; if channel.project_id() != request.project_id @@ -2342,6 +2401,7 @@ fn rejected_run_control_resolution( issue_id: request.issue_id.to_owned(), run_id: request.run_id.to_owned(), attempt_number: request.attempt_number, + attempt_status: None, thread_id: request.thread_id.map(str::to_owned), turn_id: request.turn_id.map(str::to_owned), current_thread_id: None, @@ -2350,6 +2410,13 @@ fn rejected_run_control_resolution( action: request.action.to_owned(), timeout_ms: request.timeout_ms, metadata: request.metadata.cloned(), + context: request.context.cloned(), + branch_name: None, + worktree_path: None, + active_lease: None, + event_count: None, + last_event_type: None, + last_event_at: None, channel: None, }), outcome: RUN_CONTROL_ACTION_REJECTED.to_owned(), diff --git a/apps/decodex/src/state/tests.rs b/apps/decodex/src/state/tests.rs index a2e28c8b6..9545e1020 100644 --- a/apps/decodex/src/state/tests.rs +++ b/apps/decodex/src/state/tests.rs @@ -2684,6 +2684,7 @@ fn run_control_accepts_active_attempt_and_persists_audit() { action: "noop", timeout_ms: Some(500), metadata: None, + context: None, }) .expect("control request should resolve"); @@ -2755,6 +2756,7 @@ fn run_control_rejects_stale_turn_and_run_mismatch() { action: "steer", timeout_ms: None, metadata: None, + context: None, }) .expect("stale turn should be audited"); let stale_run = store @@ -2769,6 +2771,7 @@ fn run_control_rejects_stale_turn_and_run_mismatch() { action: "noop", timeout_ms: None, metadata: None, + context: None, }) .expect("stale run should be audited"); @@ -2824,6 +2827,7 @@ fn run_control_rejects_missing_channel_file() { action: "noop", timeout_ms: None, metadata: None, + context: None, }) .expect("missing channel should be audited"); @@ -2835,6 +2839,7 @@ fn run_control_rejects_missing_channel_file() { fn run_control_requires_active_run_ownership() { let temp_dir = TempDir::new().expect("tempdir should create"); let channel_path = temp_dir.path().join("control.channel"); + let worktree_path = temp_dir.path().join("PUB-101"); fs::write(&channel_path, "ready\n").expect("control channel should write"); @@ -2844,6 +2849,14 @@ fn run_control_requires_active_run_ownership() { .upsert_lease("pubfi", "issue-1", "run-1", IN_PROGRESS_STATE) .expect("lease should record"); store.record_run_attempt("run-1", "issue-1", 1, "running").expect("attempt should record"); + store + .upsert_worktree( + "pubfi", + "issue-1", + "x/pubfi-issue-1", + &worktree_path.display().to_string(), + ) + .expect("worktree mapping should record"); store .publish_run_control_channel_for_active_attempt("run-1", 1, &channel_path, "local_file") .expect("control channel should publish"); @@ -2861,6 +2874,7 @@ fn run_control_requires_active_run_ownership() { action: "noop", timeout_ms: None, metadata: None, + context: None, }) .expect("missing lease should be audited"); @@ -2880,6 +2894,7 @@ fn run_control_requires_active_run_ownership() { action: "noop", timeout_ms: None, metadata: None, + context: None, }) .expect("wrong active run should be audited"); @@ -2887,4 +2902,28 @@ fn run_control_requires_active_run_ownership() { assert_eq!(no_lease.reason(), "active_lease_missing"); assert_eq!(wrong_run.outcome(), "rejected"); assert_eq!(wrong_run.reason(), "active_run_mismatch"); + + let events = store + .list_private_execution_events("pubfi", "issue-1", "run-1", 1) + .expect("private control audit should read"); + let no_lease_event = events + .iter() + .find(|event| event.record_id() == no_lease.audit_record_id()) + .expect("missing lease audit event should exist"); + let expected_worktree_path = worktree_path.display().to_string(); + let expected_channel_path = channel_path.display().to_string(); + + assert_eq!(no_lease_event.payload()["lane"]["active_lease"].as_bool(), Some(false)); + assert_eq!(no_lease_event.payload()["lane"]["attempt_status"].as_str(), Some("running")); + assert_eq!(no_lease_event.payload()["lane"]["branch"].as_str(), Some("x/pubfi-issue-1")); + assert_eq!( + no_lease_event.payload()["lane"]["worktree_path"].as_str(), + Some(expected_worktree_path.as_str()) + ); + assert_eq!(no_lease_event.payload()["channel"]["status"].as_str(), Some("active")); + assert_eq!(no_lease_event.payload()["channel"]["path_exists"].as_bool(), Some(true)); + assert_eq!( + no_lease_event.payload()["channel"]["channel_path"].as_str(), + Some(expected_channel_path.as_str()) + ); } diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index 670b1f3f3..1da21fb7f 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -77,8 +77,11 @@ curl -sS -X POST http://127.0.0.1:8192/api/lane/interrupt \ `POST /api/lane/interrupt` first writes a soft interrupt request for the active app-server child to deliver with `turn/interrupt`. Add `"force": true` only when the operator explicitly wants hard process-kill fallback after soft interrupt is -unavailable or does not return in the local wait window. Hard fallback is reported as -`hard_interrupt_fallback`, not as a graceful stop. +unavailable, does not return in the local wait window, or is rejected with +`active_lease_missing` while the same lane still has recorded live process evidence. +Hard fallback is reported as `hard_interrupt_fallback`, not as a graceful stop. If no +signalable child process is recorded, the force response remains a recovery/inspection +result and must not be read as a successful soft interrupt. Use `--dev` only for isolated local development: @@ -277,6 +280,11 @@ Worktree visibility follows the owning dashboard section: when `process_alive` is false. `active_lease` is queue lease ownership only; `execution_liveness` explains why the lane is still visible when the queue lease is not held. +- Lane steer and interrupt rejections such as `active_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 + process/protocol liveness so operators can decide between wait, force interrupt, + retained resume, or manual attention without reconstructing state from local paths. - In the JSON snapshot, `active_run_count` follows the same visibility boundary as the top-level `active_runs` list. The `Projects` table's `running` work number uses `running_lane_count`, so stopped, stale, or attention lanes can stay visible as diff --git a/docs/runbook/lane-control-recovery.md b/docs/runbook/lane-control-recovery.md index b4a42539e..40d4d8a93 100644 --- a/docs/runbook/lane-control-recovery.md +++ b/docs/runbook/lane-control-recovery.md @@ -65,6 +65,8 @@ Before mutating anything, confirm: - run id, attempt, thread id, current turn id, and process/protocol liveness - control outcome such as accepted, rejected, timed out, failed, or `hard_interrupt_fallback` +- `active_lease_missing` rejections together with process, protocol, channel, branch, + and retained worktree evidence when the lane still appears live - private evidence and public lifecycle signal - PR URL, head branch, and head SHA when the lane has crossed review handoff @@ -77,6 +79,8 @@ or clean labels. | --- | --- | --- | | Active lane still matches the issue, branch, run id, attempt, and turn. | Let the runtime continue or wait for the control result. | No label change. Use the next CLI/API control only when the operator explicitly asks. | | Soft interrupt was accepted and the runtime is still resolving the attempt. | Wait for status, protocol activity, or evidence to settle. | Re-inspect; do not requeue or force-kill. | +| Soft control was rejected with `active_lease_missing`, but inspect/status still shows the same run id, attempt, branch, active channel, and live child process or protocol activity. | Treat the lane as degraded active execution, not cleanup-only state. | Re-inspect with `decodex lane inspect` or use `decodex lane interrupt --run-id --force` only when the operator explicitly wants hard process fallback. | +| Forced interrupt after `active_lease_missing` reports no signalable process. | Treat force as non-mutating for the child process. | Inspect retained worktree and private evidence; do not claim the interrupt succeeded or clear attention labels. | | Hard fallback reports `hard_interrupt_fallback`. | Treat it as an interrupted runtime event, not a graceful completion. | Inspect retained worktree and evidence; resume only if lineage is exact. | | Retained worktree has useful local changes and lineage matches issue, branch, runtime evidence, and PR when present. | Resume or repair the same lane. | Use `decodex run ` when the registered workflow makes it eligible, or use the specific retained recovery runbook. | | Review handoff marker is missing or stale but the retained PR lane appears recoverable. | Diagnose before rebind. | Run `decodex recover review-handoff diagnose ` and follow [`recover-review-handoff.md`](./recover-review-handoff.md). | @@ -111,6 +115,15 @@ request. Re-run `decodex lane inspect` or `decodex status --json`. If protocol activity shows the same turn is still stopping, wait or inspect private evidence; do not kill the child process from the side. +Example: `decodex lane steer XY-123 --run-id run-abc --expected-turn-id turn-1 ...` +or `decodex lane interrupt XY-123 --run-id run-abc` reports +`active_lease_missing` while `decodex lane inspect` still shows the same branch, +attempt, active channel, and live process/protocol state. Do not treat the lane as +cleanup-only and do not clear `decodex:needs-attention`. If the operator needs to stop +the child immediately, retry interrupt with `--force`; otherwise preserve the retained +worktree and use the private evidence readback to decide whether the lane can be +resumed or needs manual attention. + Example: `decodex lane interrupt XY-123 --run-id run-abc --force` reports `hard_interrupt_fallback`. Inspect the retained worktree before retry. If the worktree contains a partial patch that still belongs to `XY-123`, resume through diff --git a/docs/spec/lane-control.md b/docs/spec/lane-control.md index 4d6d59856..0b951f26e 100644 --- a/docs/spec/lane-control.md +++ b/docs/spec/lane-control.md @@ -41,8 +41,8 @@ agent-facing skills must guide responsible use. | Project dispatch resume | Supported for future dispatch | `decodex project enable ` and the runtime project enabled flag | Resume re-enables future dispatch after the operator has inspected blockers, capacity, and queue state. | | Linear scan request | Supported | `POST /api/linear-scan` with optional `projectId` | Queue a scan for the next control-plane tick while respecting tracker backoff. This is an intake/status refresh request, not an execution command. | | Run-control channel foundation | Supported foundation | Active attempts publish a local `.decodex-run-control/*` channel record, runtime SQLite `run_control_channels`, operator status `control_capability`, and private `control_action` audit events | Route lane-control mutations through the active attempt's project, issue, run id, attempt, thread id, current turn id, active lease, and local channel metadata. Invalid or stale requests fail closed and remain local audit evidence. | -| Soft interrupt | Supported CLI/API control | `decodex lane interrupt --run-id ` and `POST /api/lane/interrupt` write a run-control request that the active app-server child delivers with `turn/interrupt` | Prefer soft interrupt before hard interruption when the active turn id is known and the app-server capability is present. Soft interrupt requests a graceful turn stop and records the protocol outcome when app-server returns one. | -| Hard interrupt fallback | Explicit fallback only | `decodex lane interrupt --run-id --force` and `POST /api/lane/interrupt` with `"force": true` classify process signaling as `hard_interrupt_fallback` | Use only when soft interrupt is unavailable, timed out, or impossible because the process or app-server boundary cannot be reached. Preserve retained worktree evidence and runtime classification. | +| Soft interrupt | Supported CLI/API control | `decodex lane interrupt --run-id ` and `POST /api/lane/interrupt` write a run-control request that the active app-server child delivers with `turn/interrupt` | Prefer soft interrupt before hard interruption when the active turn id is known and the app-server capability is present. Soft interrupt requests a graceful turn stop and records the protocol outcome when app-server returns one. If the run-control resolver rejects soft delivery with `active_lease_missing`, preserve the observed process, channel, branch, and retained worktree evidence instead of hiding the lane as inactive. | +| Hard interrupt fallback | Explicit fallback only | `decodex lane interrupt --run-id --force` and `POST /api/lane/interrupt` with `"force": true` classify process signaling as `hard_interrupt_fallback` | Use only when soft interrupt is unavailable, timed out, or impossible because the process, app-server boundary, or active lease cannot be reached. A forced interrupt may signal the recorded child process after `active_lease_missing` only when inspection still identifies the same issue, run id, attempt, channel, and live process. Preserve retained worktree evidence and runtime classification. | | Steer active lane | Supported CLI/API control; bottom-layer method stays broad | `decodex lane steer --run-id --expected-turn-id --message `, canonical `POST /api/lane/steer`, legacy alias `POST /api/lane-steer`, local run-control steer request/response files, app-server `turn/steer`, private `control_action` audit events, and protocol activity `turn/steer` summaries | Pass operator-supplied steer text through CLI/API to the current active turn. Require `expectedTurnId`; stale turn ids fail closed. Do not narrow the protocol to a fixed set of task-content categories. Apply policy, audit, privacy, and lifecycle guardrails above the protocol. | | Retained resume/retry | Supported through runtime lifecycle | `decodex run `, retry scheduling, retained worktree recovery, and `thread/resume` for same-thread app-server continuation | Resume only when retained worktree, issue, branch, PR, and runtime evidence still prove the same lane. Treat ambiguous lineage as manual attention. | | Manual attention | Supported terminal control path | `decodex:needs-attention`, `issue_comment(kind = "manual_attention")`, and `issue_terminal_finalize(path = "manual_attention")` | Stop automation when policy requires a human decision. Explain the blocker through structured public fields and keep private evidence local. | @@ -94,6 +94,14 @@ A control request is valid only when all of the following hold: Any mismatch fails closed. Rejections are not converted into process signals, tracker state changes, or worktree mutations. +`active_lease_missing` is a soft-control rejection, not proof that execution is gone. +The audit payload must retain the requested project, issue, run id, attempt, current +thread and turn ids, active channel metadata when present, branch, retained worktree +mapping, and operator-local process/protocol liveness context. If the operator used +`--force` or `"force": true`, the interrupt command may then take the explicit hard +fallback path against the recorded child process without rebinding the queue lease or +pretending that soft app-server control succeeded. + ## Soft And Hard Interrupts Soft interrupt is the preferred active-turn stop path. A compliant soft interrupt: @@ -130,6 +138,13 @@ evidence, marks an active attempt as `interrupted` when a recorded child is sign clears or retains ownership according to the runtime contract, and avoids pretending the agent completed a terminal path. +When forced interrupt follows an `active_lease_missing` soft-control rejection, the +CLI/API response must show both facts: soft control was rejected, and hard fallback was +attempted only because force was explicit and process evidence was still present. If no +signalable child process is recorded, the hard-fallback report must say it was +unavailable or found no process and direct the operator to inspect retained evidence; +it must not imply a graceful interrupt or successful recovery. + ## Steer Steer is an active-lane instruction from the operator to the running Codex turn. The @@ -208,6 +223,9 @@ the configured lifecycle. Every supported control mutation should create local runtime evidence. At minimum, a control audit record should identify the project, issue, run id, attempt, branch, operator command source, requested capability, normalized result, and next action. +Rejected control evidence must also preserve queue lease state, execution liveness, +process id/aliveness when observed, active channel path/status when present, current +thread/turn ids, retained worktree path, and latest protocol event summary. The run-control foundation records local `control_action` private execution events for accepted, rejected, completed, failed, timed out, and fallback outcomes. These records diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index 84181be0f..b26ef5253 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -393,6 +393,11 @@ The runtime database stores at least: - typed connector health and external API backoff For child supervision, the active lane may also carry a short-lived worktree heartbeat marker at `.worktrees//.decodex-run-activity`. That marker is advisory, keyed to the current `run_id` plus `attempt_number`, and exists so the control plane can observe child activity across process boundaries, surface active thread and protocol progress in operator status, and keep high-frequency telemetry out of Linear. When the marker records process liveness, it must pair `process_id` with both host boot identity (`host_boot_id`) and process start identity (`process_start_identity`). A marker from a previous boot, a marker missing either identity, a marker whose process start identity no longer matches the live PID, a marker whose PID has exited into an unreaped zombie state, or a marker observed while Decodex cannot read the current host or process identity must not be treated as a live process even if that PID currently exists. Operator snapshots expose `process_liveness_reason` so operators can distinguish stopped processes, previous-boot markers, and same-boot PID reuse from genuine live execution. The marker may also carry additive `child_agent_activity`, protocol, account, and legacy review-policy JSON or scalar fields for operator diagnostics. Legacy review-policy marker fields are breadcrumbs only: review-policy gating belongs to the runtime store and must not be overwritten from marker values. Operator snapshots must keep queue ownership separate from execution liveness: `active_lease` and `queue_lease_state` describe the local queue lease, while `execution_liveness` describes the observed process, app-server thread, or protocol marker that keeps an active lane visible. If a raw attempt is still `starting` after app-server thread, model, or protocol activity is observed, operator-facing `status` must report `running` and preserve the raw value in `attempt_status`. High-frequency heartbeat, child-agent buckets, token counts, idle ages, and other transient liveness details stay local/operator-only under the boundary defined by [`linear-execution-ledger.md`](./linear-execution-ledger.md). +If a persisted attempt has a terminal-looking status such as `failed`, `interrupted`, +or `stalled` while current marker or protocol evidence still identifies the same +`run_id` and `attempt_number` as live, operator status must keep the lane visible with +the process/protocol liveness details instead of hiding it as only terminal history or +cleanup work. Post-review ownership is stored in the runtime database. Retained handoff rows record the authoritative PR URL, branch lineage, validated PR head OID, run id, and attempt number that completed the `In Review` handoff. Retained orchestration rows record the current post-review phase for that exact handoff identity. If the matching database row is missing, post-review ownership must block as unresolved instead of rebinding from branch-name, current-head, Linear comments, or other heuristics. If a retained review marker exists but a stored handoff or orchestration head no longer matches a clean retained worktree and matching PR head, operator status must keep the marker PR URL visible when known and recovery diagnosis must report the concrete mismatched field before any explicit rebind refresh. When retained PR readback degrades but the handoff identity is still safe to preserve, operator-local status may expose a typed `readback_root_cause` diagnostic such as missing GitHub CLI, missing GitHub token, GitHub auth failure, API/read failure, parse/shape failure, or lineage validation failure while keeping public-safe warning reasons such as `pull_request_state_read_failed` stable. The only source-tree runtime artifacts that clean-source checks may ignore are the untracked top-level `.decodex-run-activity` heartbeat marker and `.decodex-run-control/` local control-channel directory. Durable review handoff, orchestration, review-policy checkpoints, retry, phase timing, and retained PR state belong in the Decodex runtime database, not in root-level or worktree-local review marker files. If the heartbeat marker carries similarly named fields for compatibility or operator diagnostics, those breadcrumb values cannot override runtime-store rows.